Polish utility file I/O and add isolated contract tests

src/utils.py

@@ -1,32 +1,41 @@
-from pathlib import Path
+from __future__ import annotations
+
 import json
-import yaml
+from pathlib import Path
+
 import pandas as pd
+import yaml
 
 ROOT = Path(__file__).resolve().parents[1]
 
-def project_path(*parts):
+
+def project_path(*parts: str | Path) -> Path:
+    """Resolve paths against the repository root, retaining absolute overrides."""
     return ROOT.joinpath(*parts)
 
-def ensure_dirs():
-    for d in ['data/public', 'data/synthetic', 'data/processed', 'outputs', 'reports']:
-        project_path(d).mkdir(parents=True, exist_ok=True)
 
-def load_yaml(rel):
-    with open(project_path(rel), 'r', encoding='utf-8') as f:
-        return yaml.safe_load(f)
+def ensure_dirs() -> None:
+    for directory in ("data/public", "data/synthetic", "data/processed", "outputs", "reports"):
+        project_path(directory).mkdir(parents=True, exist_ok=True)
+
+
+def load_yaml(rel: str | Path):
+    return yaml.safe_load(project_path(rel).read_text(encoding="utf-8"))
+
+
+def write_json(rel: str | Path, obj: object) -> None:
+    path = project_path(rel)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(json.dumps(obj, indent=2), encoding="utf-8")
+
 
-def write_json(rel, obj):
-    p = project_path(rel)
-    p.parent.mkdir(parents=True, exist_ok=True)
-    p.write_text(json.dumps(obj, indent=2), encoding='utf-8')
+def write_md(rel: str | Path, content: str) -> None:
+    path = project_path(rel)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    path.write_text(content, encoding="utf-8")
 
-def write_md(rel, content):
-    p = project_path(rel)
-    p.parent.mkdir(parents=True, exist_ok=True)
-    p.write_text(content, encoding='utf-8')
 
-def write_csv(rel, df):
-    p = project_path(rel)
-    p.parent.mkdir(parents=True, exist_ok=True)
-    df.to_csv(p, index=False)
+def write_csv(rel: str | Path, df: pd.DataFrame) -> None:
+    path = project_path(rel)
+    path.parent.mkdir(parents=True, exist_ok=True)
+    df.to_csv(path, index=False)

tests/test_utils_contract.py

@@ -0,0 +1,29 @@
+"""File helper contracts use temporary directories and preserve path semantics."""
+
+import json
+
+import pandas as pd
+
+from utils import load_yaml, project_path, write_csv, write_json, write_md
+
+
+def test_project_path_resolves_relative_and_absolute(tmp_path):
+    assert project_path("reports", "result.json") == project_path() / "reports/result.json"
+    assert project_path(tmp_path, "absolute.json") == tmp_path / "absolute.json"
+
+
+def test_file_helpers_create_parent_directories(tmp_path):
+    payload = {"status": "ok", "count": 2}
+    output = tmp_path / "nested"
+    write_json(output / "result.json", payload)
+    write_md(output / "summary.md", "# Summary\n")
+    write_csv(output / "scores.csv", pd.DataFrame({"score": [0.2, 0.7]}))
+    assert json.loads((output / "result.json").read_text(encoding="utf-8")) == payload
+    assert (output / "summary.md").read_text(encoding="utf-8") == "# Summary\n"
+    assert pd.read_csv(output / "scores.csv")["score"].tolist() == [0.2, 0.7]
+
+
+def test_load_yaml_from_absolute_path(tmp_path):
+    config = tmp_path / "configuration.yaml"
+    config.write_text("enabled: true\nthreshold: 0.25\n", encoding="utf-8")
+    assert load_yaml(config) == {"enabled": True, "threshold": 0.25}