Handle systemd 252 EnvironmentFiles readback

This commit is contained in:
fwazb 2026-07-22 23:19:13 -07:00
parent 508e60edfb
commit 77ab46b956
2 changed files with 100 additions and 1 deletions

View file

@ -694,6 +694,10 @@ def _parse_properties(raw: bytes) -> dict[str, str]:
separator == "=" and key in SERVICE_PROPERTIES and key not in value, "systemd property output was invalid"
)
value[key] = item
# systemd 252 omits EnvironmentFiles entirely when its string array is
# empty. A nonempty array is emitted and remains subject to the exact
# no-EnvironmentFile checks below.
value.setdefault("EnvironmentFiles", "")
_require(set(value) == set(SERVICE_PROPERTIES), "systemd property output was incomplete")
return value

View file

@ -117,6 +117,7 @@ class FakeHost:
self.race_kind: str | None = None
self.race_applied = False
self.property_overrides: dict[str, str] = {}
self.omitted_properties: set[str] = set()
self.not_found_security_defaults: dict[str, str] = {}
self.unstable_readbacks = False
self.show_count = 0
@ -236,7 +237,11 @@ class FakeHost:
value.update(self.property_overrides)
if self.unstable_readbacks:
value["NRestarts"] = str(self.n_restarts + self.show_count % 2)
payload = "".join(f"{name}={value[name]}\n" for name in installer.SERVICE_PROPERTIES).encode()
payload = "".join(
f"{name}={value[name]}\n"
for name in installer.SERVICE_PROPERTIES
if name not in self.omitted_properties
).encode()
return subprocess.CompletedProcess(arguments, 0, stdout=payload, stderr=b"")
def __call__(self, arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
@ -435,6 +440,96 @@ def test_first_install_accepts_production_shaped_systemd_not_found_defaults(tmp_
assert receipt["release_sha256"] == release["release_sha256"]
assert host.restart_calls == 1
repeated = installer.install_release(
bundle,
paths=paths,
source_binding=source_binding(),
runner=host,
sleeper=lambda _seconds: None,
execute_restart=True,
)
assert repeated["result"] == "already_installed"
assert host.restart_calls == 1
def test_systemd_252_omitted_environment_files_property_is_normalized(tmp_path: Path) -> None:
bundle, release, image = release_bundle(tmp_path, "2")
paths = install_paths(tmp_path)
host = FakeHost(paths)
host.add_release(release, image)
host.not_found_security_defaults = dict(PRODUCTION_SHAPED_NOT_FOUND_SECURITY)
host.omitted_properties.add("EnvironmentFiles")
receipt = installer.install_release(
bundle,
paths=paths,
source_binding=source_binding(),
runner=host,
sleeper=lambda _seconds: None,
execute_restart=True,
)
assert receipt["status"] == "pass"
assert receipt["release_sha256"] == release["release_sha256"]
assert host.restart_calls == 1
def test_systemd_252_omission_does_not_hide_a_later_dropin(tmp_path: Path) -> None:
bundle, release, image = release_bundle(tmp_path, "2")
paths = install_paths(tmp_path)
host = FakeHost(paths)
host.add_release(release, image)
host.not_found_security_defaults = dict(PRODUCTION_SHAPED_NOT_FOUND_SECURITY)
host.omitted_properties.add("EnvironmentFiles")
host.fail_dropin_release = str(release["release_sha256"])
with pytest.raises(installer.InstallError, match="later drop-in"):
installer.install_release(
bundle,
paths=paths,
source_binding=source_binding(),
runner=host,
sleeper=lambda _seconds: None,
execute_restart=True,
)
assert host.restart_calls == 1
assert not host.active
assert host.container is None
assert not paths.unit.exists()
assert not paths.state_dir.exists()
@pytest.mark.parametrize(
"property_name",
tuple(name for name in installer.SERVICE_PROPERTIES if name != "EnvironmentFiles"),
)
def test_systemd_252_compatibility_still_requires_other_properties(
tmp_path: Path,
property_name: str,
) -> None:
bundle, release, image = release_bundle(tmp_path, "2")
paths = install_paths(tmp_path)
host = FakeHost(paths)
host.add_release(release, image)
host.not_found_security_defaults = dict(PRODUCTION_SHAPED_NOT_FOUND_SECURITY)
host.omitted_properties.update({"EnvironmentFiles", property_name})
with pytest.raises(installer.InstallError, match="property output was incomplete"):
installer.install_release(
bundle,
paths=paths,
source_binding=source_binding(),
runner=host,
sleeper=lambda _seconds: None,
execute_restart=True,
)
assert host.restart_calls == 0
assert not paths.unit.exists()
assert not paths.state_dir.exists()
@pytest.mark.parametrize("property_name", installer.ABSENT_SERVICE_IDENTITY_PROPERTIES)
def test_first_install_rejects_identity_bearing_not_found_property(