tests/test_semantic_matching_cache.py

import importlib.util
import json
import os
import sys
import tempfile
import types
import unittest
from pathlib import Path
from unittest import mock

import numpy  # noqa: F401 - preload C extension before dynamic module reloads.


MODULE_PATH = (
    Path(__file__).resolve().parents[1]
    / "evaluation"
    / "pycocotools_ovod"
    / "semantic_matching.py"
)


class _FakeEmbeddings:
    def __init__(self):
        self.calls = []

    def create(self, input, model):
        self.calls.append((list(input), model))

        class Response:
            def model_dump(self_inner):
                return {
                    "data": [
                        {"embedding": [1.0, 0.0]},
                        {"embedding": [1.0, 0.0]},
                    ]
                }

        return Response()


def load_semantic_matching(cache_path, *, offline=False, readonly=False, api_key=None, fake_openai=None):
    fake_openai = fake_openai or types.SimpleNamespace(embeddings=_FakeEmbeddings())

    with mock.patch.dict(
        os.environ,
        {
            "ORIENTER_EMBEDDING_CACHE": str(cache_path),
            "ORIENTER_EMBEDDING_OFFLINE": "1" if offline else "",
            "ORIENTER_EMBEDDING_READONLY": "1" if readonly else "",
            "ZHIPU_API_KEY": api_key or "",
        },
        clear=False,
    ), mock.patch.dict(sys.modules, {"openai": fake_openai}):
        name = f"semantic_matching_under_test_{id(cache_path)}_{offline}_{readonly}"
        spec = importlib.util.spec_from_file_location(name, MODULE_PATH)
        module = importlib.util.module_from_spec(spec)
        spec.loader.exec_module(module)
        return module, fake_openai


class SemanticMatchingCacheTests(unittest.TestCase):
    def test_offline_cache_hit_does_not_call_api(self):
        with tempfile.TemporaryDirectory() as tmpdir:
            cache_path = Path(tmpdir) / "embedding_dict.json"
            cache_path.write_text(json.dumps({"Button": [1.0, 0.0]}))

            module, fake_openai = load_semantic_matching(cache_path, offline=True)

            self.assertTrue(module.is_semantic_match("button", "button", eval_dimension="s"))
            self.assertEqual(fake_openai.embeddings.calls, [])

    def test_offline_cache_miss_fails_before_api_call(self):
        with tempfile.TemporaryDirectory() as tmpdir:
            cache_path = Path(tmpdir) / "embedding_dict.json"
            cache_path.write_text(json.dumps({"Known": [1.0, 0.0]}))

            module, fake_openai = load_semantic_matching(cache_path, offline=True)

            with self.assertRaisesRegex(RuntimeError, "ORIENTER_EMBEDDING_OFFLINE=1"):
                module.is_semantic_match("known", "missing", eval_dimension="s")

            self.assertEqual(fake_openai.embeddings.calls, [])
            self.assertFalse((Path(tmpdir) / "embedding_dict.json.save").exists())

    def test_invalid_cache_is_rejected_at_load_time(self):
        with tempfile.TemporaryDirectory() as tmpdir:
            cache_path = Path(tmpdir) / "embedding_dict.json"
            cache_path.write_text(json.dumps({"A": [1.0, 0.0], "B": [1.0]}))

            with self.assertRaisesRegex(ValueError, "dimension"):
                load_semantic_matching(cache_path, offline=True)

    def test_non_finite_cache_value_is_rejected_at_load_time(self):
        with tempfile.TemporaryDirectory() as tmpdir:
            cache_path = Path(tmpdir) / "embedding_dict.json"
            cache_path.write_text(json.dumps({"A": [1.0, float("nan")]}))

            with self.assertRaisesRegex(ValueError, "non-finite"):
                load_semantic_matching(cache_path, offline=True)

    def test_readonly_cache_does_not_write_after_api_fill(self):
        with tempfile.TemporaryDirectory() as tmpdir:
            cache_path = Path(tmpdir) / "embedding_dict.json"
            cache_path.write_text(json.dumps({}))
            original = cache_path.read_text()

            module, fake_openai = load_semantic_matching(
                cache_path,
                readonly=True,
                api_key="fake-key",
            )
            module.SAVE_INTERVAL = 1

            self.assertTrue(module.is_semantic_match("alpha", "beta", eval_dimension="s"))
            self.assertEqual(fake_openai.embeddings.calls, [(["Alpha", "Beta"], "embedding-3")])
            self.assertEqual(cache_path.read_text(), original)
            self.assertFalse((Path(tmpdir) / "embedding_dict.json.save").exists())


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