feat(categorizer): implement load_registry, merge_registry, categorize_all
This commit is contained in:
@@ -1,4 +1,5 @@
|
|||||||
import ast
|
import ast
|
||||||
|
import tomllib
|
||||||
from dataclasses import dataclass, field
|
from dataclasses import dataclass, field
|
||||||
from enum import Enum
|
from enum import Enum
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
@@ -141,3 +142,40 @@ def auto_classify(path: Path, durations: dict[str, float] | None = None) -> Cate
|
|||||||
batch_group=batch_group,
|
batch_group=batch_group,
|
||||||
source="auto",
|
source="auto",
|
||||||
)
|
)
|
||||||
|
|
||||||
|
def load_registry(toml_path: Path) -> dict[str, dict]:
|
||||||
|
if not toml_path.exists():
|
||||||
|
return {}
|
||||||
|
with toml_path.open("rb") as f:
|
||||||
|
data = tomllib.load(f)
|
||||||
|
return data.get("files", {})
|
||||||
|
|
||||||
|
def merge_registry(auto: CategoryRecord, entry: dict) -> CategoryRecord:
|
||||||
|
warnings = list(auto.warnings)
|
||||||
|
if "fixture_class" in entry and entry["fixture_class"] != auto.fixture_class.value:
|
||||||
|
warnings.append(f"fixture_class-override: {auto.fixture_class.value} -> {entry['fixture_class']}")
|
||||||
|
if "subsystems" in entry and set(entry["subsystems"]) != set(auto.subsystems):
|
||||||
|
warnings.append(f"subsystems-override: {auto.subsystems} -> {entry['subsystems']}")
|
||||||
|
return CategoryRecord(
|
||||||
|
filename=auto.filename,
|
||||||
|
fixture_class=FixtureClass(entry.get("fixture_class", auto.fixture_class.value)),
|
||||||
|
subsystems=list(entry.get("subsystems", auto.subsystems)),
|
||||||
|
speed=Speed(entry.get("speed", auto.speed.value)),
|
||||||
|
batch_group=entry.get("batch_group", auto.batch_group),
|
||||||
|
notes=entry.get("notes", auto.notes),
|
||||||
|
test_order=dict(auto.test_order),
|
||||||
|
source="registry",
|
||||||
|
warnings=warnings,
|
||||||
|
)
|
||||||
|
|
||||||
|
def categorize_all(tests_dir: Path, registry_path: Path) -> list[CategoryRecord]:
|
||||||
|
registry = load_registry(registry_path)
|
||||||
|
records: list[CategoryRecord] = []
|
||||||
|
for path in sorted(tests_dir.glob("test_*.py")):
|
||||||
|
auto = auto_classify(path)
|
||||||
|
entry = registry.get(path.name, {})
|
||||||
|
if entry:
|
||||||
|
records.append(merge_registry(auto, entry))
|
||||||
|
else:
|
||||||
|
records.append(auto)
|
||||||
|
return records
|
||||||
|
|||||||
Reference in New Issue
Block a user