File size: 4,509 Bytes
3f3265f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 | 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()
|