tests/test_evaluate_temp_paths.py

import json
import os
import subprocess
import sys
import tempfile
import unittest
from pathlib import Path


REPO_ROOT = Path(__file__).resolve().parents[1]
EVAL_DIR = REPO_ROOT / "evaluation"

PYCOCOTOOLS_STUB = """
import sys
import types
pycocotools = types.ModuleType("pycocotools")
mask = types.ModuleType("pycocotools._mask")
def _not_used(*args, **kwargs):
    raise AssertionError("pycocotools._mask stub should not be exercised in this test")
for _name in ["iou", "merge", "frPyObjects", "encode", "decode", "area", "toBbox"]:
    setattr(mask, _name, _not_used)
pycocotools._mask = mask
sys.modules["pycocotools"] = pycocotools
sys.modules["pycocotools._mask"] = mask
"""


def run_eval_import(script: str, env_overrides: dict[str, str]) -> str:
    env = os.environ.copy()
    env.update(env_overrides)
    env.pop("ORIENTER_TMP_ANN_PATH", None)
    env.pop("ORIENTER_GT_CAT_MATCH_PATH", None)
    pythonpath = [str(EVAL_DIR), str(REPO_ROOT)]
    if env.get("PYTHONPATH"):
        pythonpath.append(env["PYTHONPATH"])
    env["PYTHONPATH"] = os.pathsep.join(pythonpath)
    return subprocess.check_output(
        [sys.executable, "-c", PYCOCOTOOLS_STUB + script],
        cwd=EVAL_DIR,
        env=env,
        text=True,
    ).strip()


class EvaluationTempPathTests(unittest.TestCase):
    def test_default_temp_paths_are_process_isolated(self):
        script = """
import json
import evaluate_coco
import pycocotools_ovod.semantic_matching as semantic_matching
print(json.dumps({
    "tmp_ann": evaluate_coco.tmp_ann_path,
    "gt_match": semantic_matching.gt_cat_match_path,
}))
"""
        with tempfile.TemporaryDirectory() as tmpdir:
            env = {"ORIENTER_EVALUATION_TMPDIR": tmpdir}
            first = json.loads(run_eval_import(script, env))
            second = json.loads(run_eval_import(script, env))

        self.assertNotEqual(first["tmp_ann"], second["tmp_ann"])
        self.assertNotEqual(first["gt_match"], second["gt_match"])
        self.assertIn("tmp_ann.", Path(first["tmp_ann"]).name)
        self.assertIn("tmp_gt_cat_match.", Path(first["gt_match"]).name)

    def test_cleanup_removes_configured_temp_files_and_resets_match_cache(self):
        script = """
import json
from pathlib import Path
import evaluate_coco
import pycocotools_ovod.semantic_matching as semantic_matching

semantic_matching.gt_cat_match_dict = {"Button": ["Button"]}
for path in [evaluate_coco.tmp_ann_path, semantic_matching.gt_cat_match_path]:
    Path(path).parent.mkdir(parents=True, exist_ok=True)
    Path(path).write_text("{}")

evaluate_coco.cleanup_temp_outputs()
print(json.dumps({
    "tmp_ann_exists": Path(evaluate_coco.tmp_ann_path).exists(),
    "gt_match_exists": Path(semantic_matching.gt_cat_match_path).exists(),
    "match_cache": semantic_matching.gt_cat_match_dict,
}))
"""
        with tempfile.TemporaryDirectory() as tmpdir:
            tmp = Path(tmpdir)
            env = os.environ.copy()
            env.update(
                {
                    "ORIENTER_TMP_ANN_PATH": str(tmp / "custom_tmp_ann.json"),
                    "ORIENTER_GT_CAT_MATCH_PATH": str(tmp / "custom_gt_match.json"),
                    "PYTHONPATH": os.pathsep.join([str(EVAL_DIR), str(REPO_ROOT), env.get("PYTHONPATH", "")]),
                }
            )
            result = subprocess.check_output(
                [sys.executable, "-c", PYCOCOTOOLS_STUB + script],
                cwd=EVAL_DIR,
                env=env,
                text=True,
            ).strip()

        state = json.loads(result)
        self.assertFalse(state["tmp_ann_exists"])
        self.assertFalse(state["gt_match_exists"])
        self.assertIsNone(state["match_cache"])


if __name__ == "__main__":
    unittest.main()