Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

powershell - ouput as json #3574

Merged
merged 4 commits into from
Jan 2, 2025
Merged

powershell - ouput as json #3574

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
9fe17ee
powershell - ouput as json
SRIKKANTH Dec 31, 2024
e5bd41e
Update powershell.py
SRIKKANTH Dec 31, 2024
25d54d0
mypy fixes
SRIKKANTH Jan 1, 2025
564cd43
Update hyperv.py
SRIKKANTH Jan 1, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion lisa/sut_orchestrator/hyperv/get_assignable_devices.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -104,7 +104,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)
Expand Down
18 changes: 10 additions & 8 deletions lisa/tools/hyperv.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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,
Expand All @@ -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):
Expand Down Expand Up @@ -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):
Expand Down Expand Up @@ -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
)
)
6 changes: 3 additions & 3 deletions lisa/tools/ls.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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'
Expand All @@ -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())
Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

@SRIKKANTH What is the reason for doing List() here ? It actually is not instantiate-able. It throws following error.

File "XYZ/lisa/lisa/tools/ls.py", line 109, in list
   return List(output.split())
 File "XYZ/Python/3.10.15/x64/lib/python3.10/typing.py", line 955, in __call__
   raise TypeError(f"Type {self._name} cannot be instantiated; "
TypeError: Type List cannot be instantiated; use list() instead

Copy link
Collaborator

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok, I see this is fixed in main now.

else:
return []

Expand Down
2 changes: 1 addition & 1 deletion lisa/tools/mdadm.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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):
Expand Down
10 changes: 7 additions & 3 deletions lisa/tools/powershell.py
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -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

Expand Down Expand Up @@ -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
)
Expand All @@ -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(
Expand Down
Loading