Merge pull request #267937 from SomeoneSerge/fix/test-driver-activestate

nixos/lib/test-driver: make wait_for_unit ask for ActiveState only
This commit is contained in:
Jacek Galowicz 2023-12-14 06:55:06 +01:00 committed by GitHub
commit 66ca9c90e2
No known key found for this signature in database
GPG key ID: 4AEE18F83AFDEB23

View file

@ -447,8 +447,7 @@ class Machine:
"""
def check_active(_: Any) -> bool:
info = self.get_unit_info(unit, user)
state = info["ActiveState"]
state = self.get_unit_property(unit, "ActiveState", user)
if state == "failed":
raise Exception(f'unit "{unit}" reached state "{state}"')
@ -491,6 +490,35 @@ class Machine:
if line_pattern.match(line)
)
def get_unit_property(
self,
unit: str,
property: str,
user: Optional[str] = None,
) -> str:
status, lines = self.systemctl(
f'--no-pager show "{unit}" --property="{property}"',
user,
)
if status != 0:
raise Exception(
f'retrieving systemctl property "{property}" for unit "{unit}"'
+ ("" if user is None else f' under user "{user}"')
+ f" failed with exit code {status}"
)
invalid_output_message = (
f'systemctl show --property "{property}" "{unit}"'
f"produced invalid output: {lines}"
)
line_pattern = re.compile(r"^([^=]+)=(.*)$")
match = line_pattern.match(lines)
assert match is not None, invalid_output_message
assert match[1] == property, invalid_output_message
return match[2]
def systemctl(self, q: str, user: Optional[str] = None) -> Tuple[int, str]:
"""
Runs `systemctl` commands with optional support for