"""Execution integrity: ledger, dataset load, failure modes, deterministic fixture.""" from __future__ import annotations import json import struct import zlib from pathlib import Path import pytest from fastapi.testclient import TestClient # Ensure repo root imports resolve import sys ROOT = Path(__file__).resolve().parents[2] if str(ROOT) not in sys.path: sys.path.insert(0, str(ROOT)) def _png_bytes(width: int = 8, height: int = 8) -> bytes: """Minimal valid PNG.""" def chunk(tag: bytes, data: bytes) -> bytes: return struct.pack(">I", len(data)) + tag + data + struct.pack(">I", zlib.crc32(tag + data) & 0xFFFFFFFF) ihdr = struct.pack(">IIBBBBB", width, height, 8, 2, 0, 0, 0) raw = b"".join(b"\x00" + (b"\xff\x00\x00" * width) for _ in range(height)) return b"\x89PNG\r\n\x1a\n" + chunk(b"IHDR", ihdr) + chunk(b"IDAT", zlib.compress(raw)) + chunk(b"IEND", b"") @pytest.fixture() def client(tmp_path, monkeypatch): monkeypatch.chdir(tmp_path) # Copy minimal scripts into tmp workspace so precheck/verify can find them for name in ("train.py", "infer.py", "autograder.py"): src = ROOT / name if src.exists(): (tmp_path / name).write_text(src.read_text()) # Point ledger at tmp from sensorflow import execution_ledger as ledger monkeypatch.setattr(ledger, "LEDGER_ROOT", tmp_path / "runs" / "executions") monkeypatch.setattr(ledger, "INDEX_PATH", tmp_path / "runs" / "executions" / "index.json") monkeypatch.setattr(ledger, "STRICT_MODE_PATH", tmp_path / "runs" / "studio_strict_mode.json") # Import app after chdir so CONFIG_PATH etc. resolve under tmp import importlib import app_backend importlib.reload(app_backend) return TestClient(app_backend.app) def test_invalid_dataset_path_fails_with_zero_loaded(client, tmp_path): res = client.post( "/api/dataset/load", json={"source_path": str(tmp_path / "does_not_exist")}, ) assert res.status_code == 200 data = res.json() assert data["status"] == "FAILED" assert data["metrics"]["images_discovered"] == 0 assert data["metrics"]["images_readable"] == 0 assert data["execution_id"] # Must not claim 100% loaded assert data["metrics"].get("loaded_pct_of_discovered") in (None, 0, 0.0) def test_corrupt_image_partial_success(client, tmp_path): d = tmp_path / "imgs" d.mkdir() good = d / "ok.png" good.write_bytes(_png_bytes()) bad = d / "bad.png" bad.write_bytes(b"not-a-png-file!!!!!!!!!!!") res = client.post("/api/dataset/load", json={"source_path": str(d)}) data = res.json() assert data["status"] == "PARTIAL_SUCCESS" assert data["metrics"]["images_discovered"] == 2 assert data["metrics"]["images_readable"] == 1 assert data["metrics"]["images_corrupt"] == 1 assert data["execution_id"] def test_yaml_class_out_of_range_validation_failed(client, tmp_path): root = tmp_path / "ds" (root / "images" / "train").mkdir(parents=True) (root / "labels" / "train").mkdir(parents=True) img = root / "images" / "train" / "a.png" img.write_bytes(_png_bytes()) (root / "labels" / "train" / "a.txt").write_text("99 0.5 0.5 0.1 0.1\n") # class 99 invalid yaml_path = tmp_path / "bad.yaml" yaml_path.write_text( f"path: {root}\ntrain: images/train\nval: images/train\nnames:\n 0: car\n 1: truck\n 2: person\n" ) res = client.post("/api/yaml/validate", json={"path": str(yaml_path)}) data = res.json() assert data["status"] == "VALIDATION_FAILED" assert data["class_id_out_of_range"] assert data["execution_id"] def test_missing_checkpoint_inference_failed(client, tmp_path): imgs = tmp_path / "imgs" imgs.mkdir() (imgs / "a.png").write_bytes(_png_bytes()) res = client.post( "/api/infer/run", json={ "weights": str(tmp_path / "missing_weights.pt"), "source": str(imgs), "conf": 0.25, "iou": 0.45, }, ) assert res.status_code == 400 detail = res.json()["detail"] assert "execution_id" in detail assert "checkpoint" in detail["message"].lower() or "missing" in detail["message"].lower() def test_deterministic_tiny_dataset_counts(client, tmp_path): """10 images, 10 annos, 25 objects, 3 classes — exact counts.""" root = tmp_path / "tiny" img_dir = root / "images" lab_dir = root / "labels" img_dir.mkdir(parents=True) lab_dir.mkdir(parents=True) # 25 objects across 10 label files, class ids 0/1/2 object_plan = [ [0, 0, 1], # 3 [1, 2], # 2 [0], # 1 [2, 2, 1, 0], # 4 [1, 1], # 2 [0, 1, 2], # 3 [2], # 1 [0, 0], # 2 [1, 2], # 2 [0, 1, 2, 0, 1], # 5 → total 25 ] assert sum(len(x) for x in object_plan) == 25 for i, classes in enumerate(object_plan): (img_dir / f"img_{i:02d}.png").write_bytes(_png_bytes()) lines = [f"{c} 0.5 0.5 0.2 0.2" for c in classes] (lab_dir / f"img_{i:02d}.txt").write_text("\n".join(lines) + "\n") res = client.post("/api/dataset/load", json={"source_path": str(root)}) data = res.json() assert data["status"] == "SUCCEEDED" assert data["metrics"]["images_discovered"] == 10 assert data["metrics"]["images_readable"] == 10 assert data["discovery"]["annotations_discovered"] == 10 assert data["discovery"]["annotation_objects"] == 25 assert set(data["discovery"]["class_ids"]) == {0, 1, 2} assert data["execution_id"] # Ledger list contains the record listed = client.get("/api/executions").json()["executions"] assert any(e["execution_id"] == data["execution_id"] for e in listed) full = client.get(f"/api/executions/{data['execution_id']}").json() assert full["status"] == "SUCCEEDED" assert full["records_succeeded"] == 10 def test_catalog_preprocess_is_not_executed(client): res = client.post("/api/dataset/preprocess", json={"dataset_type": "local"}) data = res.json() assert data["status"] == "NOT_EXECUTED" assert data["catalog_only"] is True assert data["execution_id"] def test_strict_mode_toggle(client): assert client.get("/api/strict-mode").json()["enabled"] is False client.post("/api/strict-mode", json={"enabled": True}) assert client.get("/api/strict-mode").json()["enabled"] is True def test_health_endpoint(client): data = client.get("/api/health").json() assert data["backend_connected"] is True assert "filesystem_writable" in data assert "scripts" in data def test_grader_not_executed_without_predictions(client): data = client.get("/api/grade").json() assert data["status"] == "NOT_EXECUTED" assert data["execution_id"] assert data["quality_score"] is None