powershell - ouput as json (#3574)

* powershell - ouput as json

This patch enables the run_cmdlet method in lisa/tools/powershell.py to output the result as JSON when the output_json parameter is set to True.

Getting Powershell output as JSON is beneficial because it allows for easier parsing and processing of the output data.

* Update powershell.py

* mypy fixes

* Update hyperv.py

lisa/sut_orchestrator/hyperv/get_assignable_devices.py

@@ -106,7 +106,7 @@ def __get_pnp_device_property(self, device_id: str, property_name: str) -> str:
             sudo=True,
             force_run=True,
         )
-        return output.strip()
+        return str(output.strip())
 
     def __load_pnp_allocated_resources(self) -> List[Dict[str, str]]:
         # Command output result (just 2 device properties)

lisa/tools/hyperv.py

@@ -41,7 +41,7 @@ def exists_vm(self, name: str) -> bool:
             force_run=True,
         )
 
-        return output.strip() != ""
+        return bool(output.strip() != "")
 
     def delete_vm_async(self, name: str) -> Optional[Process]:
         # check if vm is present
@@ -219,7 +219,7 @@ def exists_switch(self, name: str) -> bool:
             fail_on_error=False,
             force_run=True,
         )
-        return output.strip() != ""
+        return bool(output.strip() != "")
 
     def delete_switch(self, name: str) -> None:
         if self.exists_switch(name):
@@ -251,7 +251,7 @@ def exists_nat(self, name: str) -> bool:
             fail_on_error=False,
             force_run=True,
         )
-        return output.strip() != ""
+        return bool(output.strip() != "")
 
     def delete_nat(self, name: str) -> None:
         if self.exists_nat(name):
@@ -317,7 +317,7 @@ def get_ip_address(self, name: str) -> str:
         if not ip_address:
             raise LisaException(f"Could not find IP address for VM {name}")
 
-        return ip_address
+        return str(ip_address)
 
     def exist_port_forwarding(
         self,
@@ -328,7 +328,7 @@ def exist_port_forwarding(
             fail_on_error=False,
             force_run=True,
         )
-        return output.strip() != ""
+        return bool(output.strip() != "")
 
     def delete_port_forwarding(self, nat_name: str) -> None:
         if self.exist_port_forwarding(nat_name):
@@ -359,7 +359,7 @@ def exists_virtual_disk(self, name: str) -> bool:
             fail_on_error=False,
             force_run=True,
         )
-        return output.strip() != ""
+        return bool(output.strip() != "")
 
     def delete_virtual_disk(self, name: str) -> None:
         if self.exists_virtual_disk(name):
@@ -417,6 +417,8 @@ def _run_hyperv_cmdlet(
         pwsh = self.node.tools[PowerShell]
         if not extra_args:
             extra_args = {}
-        return pwsh.run_cmdlet(
-            f"{cmd} {args} {extra_args.get(cmd.lower(), '')}", force_run=force_run
+        return str(
+            pwsh.run_cmdlet(
+                f"{cmd} {args} {extra_args.get(cmd.lower(), '')}", force_run=force_run
+            )
         )

lisa/tools/ls.py

@@ -88,15 +88,15 @@ def is_file(self, path: PurePath, sudo: bool = False) -> bool:
             sudo=sudo,
         )
 
-        return output.strip() == "True"
+        return bool(output.strip() == "True")
 
     def path_exists(self, path: str, sudo: bool = False) -> bool:
         output = self.node.tools[PowerShell].run_cmdlet(
             f"Test-Path {path}",
             force_run=True,
             sudo=sudo,
         )
-        return output.strip() == "True"
+        return bool(output.strip() == "True")
 
     def list(self, path: str, sudo: bool = False) -> List[str]:
         command = f'Get-ChildItem -Path "{path}" | Select-Object -ExpandProperty Name'
@@ -106,7 +106,7 @@ def list(self, path: str, sudo: bool = False) -> List[str]:
             sudo=sudo,
         )
         if output:
-            return output.split()
+            return List(output.split())
         else:
             return []
 

lisa/tools/mdadm.py

@@ -171,7 +171,7 @@ def _exists_pool(self, pool_name: str) -> bool:
             fail_on_error=False,
             force_run=True,
         )
-        return output.strip() != ""
+        return bool(output.strip() != "")
 
     def _delete_pool(self, pool_name: str) -> None:
         if self._exists_pool(pool_name):

lisa/tools/powershell.py

@@ -1,7 +1,7 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT license.
-
 import base64
+import json
 from typing import Any
 from xml.etree import ElementTree
 
@@ -48,14 +48,17 @@ def run_cmdlet_async(
     def run_cmdlet(
         self,
         cmdlet: str,
+        output_json: bool = False,
         force_run: bool = False,
         sudo: bool = False,
         fail_on_error: bool = True,
         timeout: int = 600,
         # Powershell error log is the xml format, it needs extra decoding. But
         # for long running script, it needs to look real time results.
         no_debug_log: bool = True,
-    ) -> str:
+    ) -> Any:
+        if output_json:
+            cmdlet = f"{cmdlet} | ConvertTo-Json"
         process = self.run_cmdlet_async(
             cmdlet=cmdlet, force_run=force_run, sudo=sudo, no_debug_log=no_debug_log
         )
@@ -68,7 +71,8 @@ def run_cmdlet(
             # if stdout is output already, it doesn't need to output again.
             no_debug_log=not no_debug_log,
         )
-
+        if output_json and result.stdout:
+            return json.loads(result.stdout)
         return result.stdout
 
     def wait_result(