94 lines
3.5 KiB
Python
94 lines
3.5 KiB
Python
import json
|
|
import os
|
|
import shutil
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
|
from core import vision_cache as vc # noqa: E402
|
|
|
|
PARAMS = {"temperature": 0.0, "max_tokens": 4000}
|
|
|
|
|
|
class FakeVision:
|
|
def __init__(self):
|
|
self.calls = 0
|
|
|
|
def __call__(self):
|
|
self.calls += 1
|
|
return f"text-{self.calls}"
|
|
|
|
|
|
class VisionCacheTests(unittest.TestCase):
|
|
def setUp(self):
|
|
self.dir = tempfile.mkdtemp(prefix="ttvc-vc-")
|
|
self.addCleanup(shutil.rmtree, self.dir, True)
|
|
|
|
def env(self, mode):
|
|
return mock.patch.dict(os.environ, {vc.ENV_MODE: mode, vc.ENV_DIR: self.dir})
|
|
|
|
def call(self, fake, model="m1", prompt="p", img=b"img", params=PARAMS):
|
|
return vc.cached_completion(model, prompt, img, dict(params), fake)
|
|
|
|
def test_live_default_never_touches_the_cache(self):
|
|
fake = FakeVision()
|
|
with mock.patch.dict(os.environ, {vc.ENV_DIR: self.dir}, clear=False):
|
|
os.environ.pop(vc.ENV_MODE, None)
|
|
self.assertEqual(self.call(fake), "text-1")
|
|
self.assertEqual(self.call(fake), "text-2")
|
|
self.assertEqual(list(Path(self.dir).iterdir()), [])
|
|
|
|
def test_record_then_replay_hit_returns_frozen_text_without_calling(self):
|
|
fake = FakeVision()
|
|
with self.env("record"):
|
|
self.assertEqual(self.call(fake), "text-1")
|
|
entry = json.loads(next(Path(self.dir).glob("[0-9a-f]*.json")).read_text("utf-8"))
|
|
self.assertEqual(entry["key_inputs"]["params"], PARAMS)
|
|
self.assertEqual(entry["key_inputs"]["model"], "m1")
|
|
with self.env("replay"):
|
|
self.assertEqual(self.call(fake), "text-1")
|
|
self.assertEqual(self.call(fake), "text-1")
|
|
self.assertEqual(vc.run_summary()["mode"], "replay")
|
|
self.assertEqual(fake.calls, 1)
|
|
self.assertEqual(json.loads((Path(self.dir) / "last_run.json").read_text())["mode"], "replay")
|
|
|
|
def test_replay_miss_fails_loudly_and_does_not_call(self):
|
|
fake = FakeVision()
|
|
with self.env("replay"), self.assertRaises(vc.VisionCacheMiss):
|
|
self.call(fake)
|
|
self.assertEqual(fake.calls, 0)
|
|
|
|
def test_failed_fetch_is_not_recorded(self):
|
|
with self.env("record"):
|
|
self.assertIsNone(vc.cached_completion("m", "p", b"i", PARAMS, lambda: None))
|
|
self.assertEqual(list(Path(self.dir).glob("[0-9a-f]*.json")), [])
|
|
|
|
def test_key_is_stable_and_independent_of_dict_order(self):
|
|
k1, _ = vc.make_key("m", "p", b"x", {"a": 1, "b": 2})
|
|
k2, _ = vc.make_key("m", "p", b"x", {"b": 2, "a": 1})
|
|
self.assertEqual(k1, k2)
|
|
|
|
def test_any_input_change_invalidates_the_key(self):
|
|
base, _ = vc.make_key("m", "p", b"x", PARAMS)
|
|
for args in (("m2", "p", b"x", PARAMS), ("m", "p2", b"x", PARAMS), ("m", "p", b"y", PARAMS),
|
|
("m", "p", b"x", {**PARAMS, "temperature": 0.7}),
|
|
("m", "p", b"x", {**PARAMS, "seed": 1})):
|
|
self.assertNotEqual(base, vc.make_key(*args)[0], args)
|
|
|
|
def test_changed_parameters_miss_in_replay(self):
|
|
with self.env("record"):
|
|
self.call(FakeVision())
|
|
with self.env("replay"), self.assertRaises(vc.VisionCacheMiss):
|
|
self.call(FakeVision(), params={**PARAMS, "temperature": 0.5})
|
|
|
|
def test_invalid_mode_is_rejected(self):
|
|
with self.env("bogus"), self.assertRaises(ValueError):
|
|
vc.get_mode()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|