train_utility/tests/test_mlflow_smoke_verifier.py
2026-08-04 11:14:51 +04:00

43 lines
1.3 KiB
Python

from __future__ import annotations
import importlib.util
from pathlib import Path
from types import SimpleNamespace
import pytest
SPEC = importlib.util.spec_from_file_location(
"verify_mlflow_smoke",
Path(__file__).parents[1] / "scripts" / "verify_mlflow_smoke.py",
)
assert SPEC is not None and SPEC.loader is not None
verify_mlflow_smoke = importlib.util.module_from_spec(SPEC)
SPEC.loader.exec_module(verify_mlflow_smoke)
latest_smoke_run_group = verify_mlflow_smoke.latest_smoke_run_group
latest_task_run = verify_mlflow_smoke.latest_task_run
def _run(name: str, group: str, *, anchor: bool = False) -> SimpleNamespace:
tags = {
"mlflow.runName": name,
"yolo.run_group": group,
}
if anchor:
tags["smoke.anchor"] = "true"
return SimpleNamespace(data=SimpleNamespace(tags=tags))
def test_verifier_selects_one_complete_smoke_batch() -> None:
runs = [
_run("detect-smoke", "new"),
_run("smoke-batch-new", "new", anchor=True),
_run("obb-smoke", "old"),
_run("smoke-batch-old", "old", anchor=True),
]
group = latest_smoke_run_group(runs)
assert group == "new"
assert latest_task_run(runs, "detect", group).data.tags["yolo.run_group"] == "new"
with pytest.raises(AssertionError, match="obb-smoke"):
latest_task_run(runs, "obb", group)