Jens
faeb58ef6d
GeoIntel release gates / Compile, test, contracts and builds (push) Successful in 1m49s
GeoIntel release gates / Python and npm vulnerability policy (push) Successful in 21s
GeoIntel release gates / Production AI image, SBOM and container scan (push) Successful in 5m39s
GeoIntel release gates / Deploy exact gated revision to Unraid (push) Failing after 58m43s
43 lines
1.7 KiB
Python
43 lines
1.7 KiB
Python
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
from pathlib import Path
|
|
|
|
|
|
SCRIPT = Path(__file__).parents[2] / "scripts" / "supervise_container_training_loop.py"
|
|
SPEC = importlib.util.spec_from_file_location("loop_supervisor", SCRIPT)
|
|
assert SPEC and SPEC.loader
|
|
MODULE = importlib.util.module_from_spec(SPEC)
|
|
SPEC.loader.exec_module(MODULE)
|
|
|
|
|
|
def test_loop_status_is_fail_closed_and_recognises_completion(tmp_path: Path) -> None:
|
|
state = tmp_path / "state.json"
|
|
assert MODULE.read_loop_status(state) is None
|
|
state.write_text("not-json", encoding="utf-8")
|
|
assert MODULE.read_loop_status(state) == "invalid"
|
|
state.write_text(json.dumps({"status": "training_complete"}), encoding="utf-8")
|
|
assert MODULE.read_loop_status(state) == "training_complete"
|
|
|
|
|
|
def test_process_detection_requires_exact_marker(monkeypatch) -> None:
|
|
class Result:
|
|
returncode = 0
|
|
stdout = "12 python api.py\n13 python run_belgium_building_training_loop.py --output-dir /runs/v37\n"
|
|
|
|
monkeypatch.setattr(MODULE.subprocess, "run", lambda *args, **kwargs: Result())
|
|
assert MODULE.process_active("geointel", "run_belgium_building_training_loop.py")
|
|
assert not MODULE.process_active("geointel", "other_loop.py")
|
|
|
|
|
|
def test_write_state_creates_output_directory_atomically(tmp_path: Path) -> None:
|
|
state_path = tmp_path / "new-run" / "supervisor-state.json"
|
|
|
|
MODULE.write_state(state_path, {"schema_version": 1, "status": "monitoring"})
|
|
|
|
written = json.loads(state_path.read_text(encoding="utf-8"))
|
|
assert written["status"] == "monitoring"
|
|
assert written["updated_at"]
|
|
assert not state_path.with_suffix(".json.tmp").exists()
|