| from pathlib import Path | |
| ROOT = Path(__file__).resolve().parents[1] | |
| def test_safe_restart_space_wrapper_uses_hf_api_with_factory_reboot_guardrails(): | |
| worker = (ROOT / "src" / "worker_payload.py").read_text(encoding="utf-8") | |
| assert "def safe_restart_space" in worker | |
| assert "api.restart_space(repo_id=target_space_id, token=token, factory_reboot=factory_reboot)" in worker | |
| assert "restart_guardrail_allows" in worker | |
| assert "require_logs_checked" in worker | |
| assert "Space is already busy" in worker | |
| assert "factory_reboot=True" in worker | |
| def test_same_code_reupload_is_guarded_against_loops(): | |
| worker = (ROOT / "src" / "worker_payload.py").read_text(encoding="utf-8") | |
| assert "def safe_same_code_reupload" in worker | |
| assert "same_code_upload" in worker | |
| assert "RESTART_GUARDRAIL_COOLDOWN_SECONDS" in worker | |
| assert "Skipped same-code upload: Space is already building/starting" in worker | |
| def test_recovery_uses_safe_reupload_not_raw_upload_for_factory_rebuild_actions(): | |
| worker = (ROOT / "src" / "worker_payload.py").read_text(encoding="utf-8") | |
| recovery = worker.split("def recover_after_live_validation_failure", 1)[1].split("def upload_workspace", 1)[0] | |
| assert "safe_same_code_reupload(api, workspace, target_space_id, token, run_dir, events_path, reason=\"dependency_guardrail_repair\")" in recovery | |
| assert "safe_same_code_reupload(api, workspace, target_space_id, token, run_dir, events_path, reason=\"pi_decision_factory_rebuild_same_code\")" in recovery | |
| assert "same_code_upload_blocked_by_restart_guardrail" in recovery | |