from __future__ import annotations from pathlib import Path ROOT = Path(__file__).resolve().parents[2] def test_runner_fleet_upgrade_action_is_fixed_and_registered(): registry = (ROOT / "backend/app/services/action_registry.py").read_text(encoding="utf-8") execution = (ROOT / "backend/app/services/action_execution.py").read_text(encoding="utf-8") assert "'runner.fleet.upgrade_v2'" in registry assert "return 'tower/runner-fleet-v2'" in registry assert 'action == "runner.fleet.upgrade_v2"' in execution assert 'target.strip().lower() != "tower/runner-fleet-v2"' in execution def test_runner_fleet_upgrade_uses_signed_fixed_broker_path(): broker = (ROOT / "backend/app/services/deployment_broker.py").read_text(encoding="utf-8") assert '"/v1/appops/runner-fleet/upgrade-v2"' in broker assert "runner_fleet_upgrade_v2" in broker def test_root_controller_upgrade_route_is_exact_revision_bound(): status = (ROOT / "ops/gitea-deploy-broker/status_server.py").read_text(encoding="utf-8") assert 'RUNNER_FLEET_UPGRADE_PATH = "/v1/appops/runner-fleet/upgrade-v2"' in status assert "_self_upgrade_signature_is_valid" in status assert "_self_upgrade_preflight(revision)" in status assert 'f"{revision}:scripts/upgrade_ci_runner_fleet_v2.sh"' in status assert "execute_runner_fleet_v2(script_result.stdout)" in status def test_controller_artifact_carries_fixed_fleet_executor(): status = (ROOT / "ops/gitea-deploy-broker/status_server.py").read_text(encoding="utf-8") prepare = (ROOT / "scripts/prepare_controller_artifact.py").read_text(encoding="utf-8") upgrade = (ROOT / "scripts/upgrade_deployment_controller.sh").read_text(encoding="utf-8") for text in (status, prepare, upgrade): assert "runner_capacity_controller.py" in text assert "runner_capacity_policy.py" in text assert "runner_fleet_upgrade_v2.py" in text def test_root_controller_sources_compile(): for relative in ( "ops/gitea-deploy-broker/status_server.py", "ops/gitea-deploy-broker/runner_fleet_upgrade_v2.py", "ops/gitea-deploy-broker/runner_capacity_controller.py", "ops/gitea-deploy-broker/runner_capacity_policy.py", ): path = ROOT / relative compile(path.read_text(encoding="utf-8"), str(path), "exec")