Files
geointel/backend/tests/test_belgium_training_loop.py
T
Jens 2316ff28f5
GeoIntel release gates / Compile, test, contracts and builds (push) Canceled after 0s
GeoIntel release gates / Python and npm vulnerability policy (push) Canceled after 0s
GeoIntel release gates / GIS image, SBOM and container scan (push) Canceled after 0s
Expand regional corpus and aerial finetuning controls
2026-07-27 06:19:11 +02:00

95 lines
2.8 KiB
Python

from __future__ import annotations
import importlib.util
import json
import subprocess
import sys
from pathlib import Path
SCRIPT = Path(__file__).parents[2] / "scripts" / "run_belgium_building_training_loop.py"
SPEC = importlib.util.spec_from_file_location("training_loop", SCRIPT)
assert SPEC and SPEC.loader
MODULE = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(MODULE)
def test_training_command_is_cuda_deterministic_and_bound_to_frozen_inputs(tmp_path: Path) -> None:
command = MODULE.training_command(
"yolo",
model=tmp_path / "base.pt",
data=tmp_path / "dataset.yaml",
project=tmp_path / "runs",
name="iteration-001",
epochs=160,
seed=42,
batch=2,
workers=4,
)
assert command[:2] == ["yolo", "train"]
assert "device=0" in command
assert "deterministic=True" in command
assert "seed=42" in command
assert "epochs=160" in command
assert "max_det=1000" in command
assert "imgsz=640" in command
assert "optimizer=auto" in command
assert "mosaic=1.0" in command
def test_training_command_supports_conservative_aerial_finetuning(tmp_path: Path) -> None:
command = MODULE.training_command(
"yolo",
model=tmp_path / "base.pt",
data=tmp_path / "dataset.yaml",
project=tmp_path / "runs",
name="aerial",
epochs=50,
seed=42,
batch=2,
workers=0,
optimizer="AdamW",
lr0=0.0001,
mosaic=0.0,
scale=0.2,
translate=0.05,
)
assert "optimizer=AdamW" in command
assert "lr0=0.0001" in command
assert "mosaic=0.0" in command
assert "scale=0.2" in command
assert "translate=0.05" in command
assert f"data={tmp_path / 'dataset.yaml'}" in command
def test_loop_refuses_failed_dataset_audit(tmp_path: Path) -> None:
audit = tmp_path / "audit.json"
audit.write_text(json.dumps({"status": "needs_attention", "low_variance_positive_tile_count": 4}))
result = subprocess.run(
[
sys.executable,
str(SCRIPT),
"--initial-model",
str(tmp_path / "base.pt"),
"--train-yaml",
str(tmp_path / "dataset.yaml"),
"--dataset-audit",
str(audit),
"--calibration-summary",
str(tmp_path / "cal.json"),
"--test-summary",
str(tmp_path / "test.json"),
"--background-summary",
str(tmp_path / "background.json"),
"--corpus-manifest",
str(tmp_path / "manifest.json"),
"--output-dir",
str(tmp_path / "output"),
],
capture_output=True,
text=True,
check=False,
)
assert result.returncode != 0
assert "Dataset audit is not ok" in result.stderr