from __future__ import annotations import io import subprocess import tarfile from pathlib import Path import pytest from scripts import image_release COMMIT = "a" * 40 IMAGE_ID = "sha256:" + "b" * 64 DIGEST = "sha256:" + "c" * 64 LOCAL = f"portfolio-hub-ci:{COMMIT}" class RecordingRunner: def __init__(self, fail_on=None): self.commands = []; self.fail_on = fail_on; self.push_count = 0 def run(self, command, *, input_text=None, timeout=None): command = list(command); self.commands.append((command, input_text, timeout)) if self.fail_on and command[:len(self.fail_on)] == list(self.fail_on): raise subprocess.TimeoutExpired(command, timeout or 0) if command[:2] == ["docker", "load"]: return image_release.Result(f"Loaded image: {LOCAL}\n") if command[:3] == ["docker", "image", "inspect"] and "{{.Id}}" in command: return image_release.Result(IMAGE_ID + "\n") if command[:3] == ["docker", "image", "inspect"] and "{{.Config.User}}" in command: return image_release.Result("10001:10001\n") if command[:3] == ["docker", "image", "inspect"] and "{{.Architecture}}" in command: return image_release.Result("amd64\n") if command[:3] == ["docker", "container", "inspect"] and "{{.State.ExitCode}}" in command: return image_release.Result("0\n") if command[:3] == ["docker", "image", "inspect"] and any("org.opencontainers.image.version" in part for part in command): return image_release.Result("v1.2.3\n") if command[:3] == ["docker", "image", "inspect"]: return image_release.Result(COMMIT + "\n") if command[:2] == ["docker", "push"]: self.push_count += 1; return image_release.Result(f"digest: {DIGEST} size: 1\n") if command[:2] == ["docker", "create"]: return image_release.Result("created-container\n") if command[:3] == ["docker", "run", "--rm"] and "urllib.request" in command[-1]: enabled = "TARGET:8000" not in command[-1] and "plain" not in command[-1] return image_release.Result('{"status":"ok","db_enabled":' + str(enabled).lower() + '}') if command[:3] == ["docker", "run", "--rm"] and "alembic_version" in command[-1]: return image_release.Result("20260820_0030\n") if command[:3] == ["docker", "run", "--rm"] and "health-check" in command: return image_release.Result('{"database_revision":"20260820_0030"}\n') return image_release.Result() def artifacts(tmp_path: Path): archive = tmp_path / "portfolio-hub-image.tar"; archive.write_bytes(b"tested") checksum = tmp_path / "portfolio-hub-image.tar.sha256"; image_release.write_checksum(archive, checksum) image_id = tmp_path / "portfolio-hub-image.id"; image_id.write_text(IMAGE_ID + "\n") return archive, checksum, image_id def publish(tmp_path, runner, **extra): archive, checksum, image_id = artifacts(tmp_path) values = dict(archive=archive, checksum=checksum, image_id=image_id, local_image=LOCAL, commit=COMMIT, event_name="push", ref="refs/tags/v1.2.3", actor="actor", token="synthetic", runner=runner); values.update(extra) return image_release.publish_image(**values) def test_strict_tag_and_only_two_non_latest_selectors(): assert image_release.release_tags("push", "refs/tags/v1.2.3", COMMIT) == ( f"{image_release.IMAGE}:{COMMIT[:12]}", f"{image_release.IMAGE}:v1.2.3") @pytest.mark.parametrize("event,ref", [("pull_request","refs/tags/v1.2.3"),("push","refs/heads/main"),("push","refs/tags/v1.2"),("workflow_dispatch","refs/tags/v1.2.3")]) def test_non_strict_event_rejected(event, ref): with pytest.raises(ValueError, match="strict vX.Y.Z"): image_release.release_tags(event, ref, COMMIT) def test_checkout_identity_is_exact(): assert image_release.validate_checkout_identity(COMMIT, COMMIT) == COMMIT with pytest.raises(ValueError): image_release.validate_checkout_identity("d"*40, COMMIT) def test_project_version_must_match_release_tag(tmp_path): project = tmp_path / "pyproject.toml" project.write_text('[project]\nversion = "0.1.3"\n') assert image_release.project_release_version(project) == "v0.1.3" assert image_release.validate_release_event("push", "refs/tags/v0.1.3", COMMIT, "v0.1.3") == "v0.1.3" with pytest.raises(ValueError, match="project version"): image_release.validate_release_event("push", "refs/tags/v0.1.4", COMMIT, "v0.1.3") def test_installed_smoke_has_import_cli_exact_inventory_and_cleanup(): inventory = tuple( [f"202601{i:02d}_{i:04d}_migration.py" for i in range(1, 30)] + ["20260820_0030_portfolio_ledger_keyset_index.py"] ) runner = RecordingRunner(); image_release.smoke_installed_image("candidate", inventory, runner, 17) runs = [x[0] for x in runner.commands if x[0][:2] == ["docker","run"]] assert len(runs) == 3 and all("--read-only" in x and "none" in x for x in runs) assert "migration_inventory" in runs[-1][-1] and "20260820_0030" in runs[-1][-1] assert len([x for x in runner.commands if x[0][:3] == ["docker","rm","--force"]]) == 3 def _assert_exact_startup_cleanup(commands: list[list[str]]) -> None: attempted_names = { command[command.index("--name") + 1] for command in commands if command[:2] == ["docker", "run"] and "--name" in command } removed_names = { command[-1] for command in commands if command[:3] == ["docker", "rm", "--force"] } created_networks = { command[-1] for command in commands if command[:3] == ["docker", "network", "create"] } removed_networks = { command[-1] for command in commands if command[:3] == ["docker", "network", "rm"] } created_volumes = { command[-1] for command in commands if command[:3] == ["docker", "volume", "create"] } removed_volumes = { command[-1] for command in commands if command[:4] == ["docker", "volume", "rm", "--force"] } assert attempted_names assert removed_names == attempted_names assert removed_networks == created_networks assert removed_volumes == created_volumes def test_wrong_shaped_health_cleans_exact_attempted_resources(): class WrongShapeRunner(RecordingRunner): def run(self, command, **kwargs): if list(command)[:3] == ["docker", "run", "--rm"] and "urllib.request" in list(command)[-1]: self.commands.append((list(command), kwargs.get("input_text"), kwargs.get("timeout"))) return image_release.Result("[]") return super().run(command, **kwargs) runner = WrongShapeRunner() with pytest.raises(RuntimeError, match="startup smoke"): image_release.startup_smoke("candidate", runner, 17) commands = [x[0] for x in runner.commands] _assert_exact_startup_cleanup(commands) def test_timeout_cleans_exact_attempted_resources(): runner = RecordingRunner(fail_on=("docker","run","-d")) with pytest.raises(RuntimeError): image_release.startup_smoke("candidate", runner, 17) commands = [x[0] for x in runner.commands] _assert_exact_startup_cleanup(commands) def test_scan_create_canary_failure_removes_exact_created_container(tmp_path): class CreateCanaryRunner(RecordingRunner): def run(self, command, **kwargs): command = list(command) if command[:2] == ["docker", "create"]: self.commands.append((command, kwargs.get("input_text"), kwargs.get("timeout"))) return image_release.Result( "created-container\n", image_release.CANARY.decode(), ) return super().run(command, **kwargs) archive = tmp_path / "portfolio-hub-image.tar" archive.write_bytes(b"unused") runner = CreateCanaryRunner() with pytest.raises(ValueError, match="canary"): image_release.scan_image("candidate", archive, runner, 17) removals = [ command for command, _input, _timeout in runner.commands if command[:3] == ["docker", "rm", "--force"] ] assert removals == [["docker", "rm", "--force", "created-container"]] def test_corrupt_archive_stops_before_registry_write(tmp_path): archive, checksum, image_id = artifacts(tmp_path); archive.write_bytes(b"corrupt"); runner = RecordingRunner() with pytest.raises(ValueError, match="checksum mismatch"): image_release.publish_image(archive=archive, checksum=checksum, image_id=image_id, local_image=LOCAL, commit=COMMIT,event_name="push",ref="refs/tags/v1.2.3",actor="a",token="t",runner=runner) assert runner.commands == [] def test_publish_loads_without_rebuild_and_records_digest(tmp_path): runner = RecordingRunner(); output = tmp_path/"output"; assert publish(tmp_path, runner, output_path=output) == DIGEST commands = [x[0][:2] for x in runner.commands] assert ["docker","build"] not in commands and commands.count(["docker","push"]) == 2 assert COMMIT in output.read_text() and DIGEST in output.read_text() and ":latest" not in output.read_text() def test_post_push_bad_digest_emits_no_success_evidence(tmp_path): class Bad(RecordingRunner): def run(self, command, **kwargs): result = super().run(command, **kwargs) return image_release.Result("pushed without digest") if list(command)[:2] == ["docker","push"] else result output=tmp_path/"output" with pytest.raises(ValueError, match="valid digest"): publish(tmp_path, Bad(), output_path=output) assert not output.exists() @pytest.mark.parametrize("kind", ["payload","symlink","pax"]) def test_canary_scan_covers_archive_payload_and_metadata(kind): data=io.BytesIO() with tarfile.open(fileobj=data, mode="w", format=tarfile.PAX_FORMAT) as archive: member=tarfile.TarInfo("item") if kind=="payload": payload=image_release.CANARY; member.size=len(payload); archive.addfile(member,io.BytesIO(payload)) elif kind=="symlink": member.type=tarfile.SYMTYPE; member.linkname=image_release.CANARY.decode(); archive.addfile(member) else: member.pax_headers={"key":image_release.CANARY.decode()}; archive.addfile(member,io.BytesIO()) with pytest.raises(ValueError, match="canary"): image_release._scan_tar(data.getvalue(), True)