190 lines
7.4 KiB
Python
190 lines
7.4 KiB
Python
import os
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
import util.translation as T
|
|
|
|
|
|
class CacheTestBase(unittest.TestCase):
|
|
"""Isolate the on-disk cache to a temp dir and reset the in-memory cache."""
|
|
|
|
def setUp(self):
|
|
self._tmp = tempfile.TemporaryDirectory()
|
|
tmp = Path(self._tmp.name)
|
|
self._orig_file = T.CACHE_FILE
|
|
self._orig_lock = T.CACHE_LOCK_FILE
|
|
self._orig_cache = T._cache
|
|
T.CACHE_FILE = tmp / "translation_cache.json"
|
|
T.CACHE_LOCK_FILE = tmp / "translation_cache.lock"
|
|
T._cache = {}
|
|
|
|
def tearDown(self):
|
|
T.CACHE_FILE = self._orig_file
|
|
T.CACHE_LOCK_FILE = self._orig_lock
|
|
T._cache = self._orig_cache
|
|
self._tmp.cleanup()
|
|
|
|
|
|
class CacheKeyTests(CacheTestBase):
|
|
def test_same_payload_and_language_is_stable(self):
|
|
payload = '{"Line1": "テスト"}'
|
|
self.assertEqual(
|
|
T.get_cache_key(payload, "English"),
|
|
T.get_cache_key(payload, "English"),
|
|
)
|
|
|
|
def test_language_changes_the_key(self):
|
|
payload = '{"Line1": "テスト"}'
|
|
self.assertNotEqual(
|
|
T.get_cache_key(payload, "English"),
|
|
T.get_cache_key(payload, "Spanish"),
|
|
)
|
|
|
|
def test_identical_japanese_collides_across_files(self):
|
|
# Two files producing the same Japanese-only payload share a key. This is
|
|
# exactly why the cached value must be Japanese-only (no file-specific
|
|
# neighbours), so the read path can re-expand it per file.
|
|
payload = '{"Line1": "おはよう"}'
|
|
self.assertEqual(
|
|
T.get_cache_key(payload, "English"),
|
|
T.get_cache_key(payload, "English"),
|
|
)
|
|
|
|
|
|
class CacheRoundTripTests(CacheTestBase):
|
|
def test_list_value_preserved_exactly(self):
|
|
payload = '{"Line1": "あ", "Line2": "い"}'
|
|
value = ["A", "B"]
|
|
T.cache_translation(payload, value, "English")
|
|
self.assertEqual(T.peek_cached_translation(payload, "English"), value)
|
|
|
|
def test_string_value_preserved(self):
|
|
payload = '{"Line1": "名前"}'
|
|
T.cache_translation(payload, "Name", "English")
|
|
self.assertEqual(T.peek_cached_translation(payload, "English"), "Name")
|
|
|
|
def test_miss_returns_none_from_peek(self):
|
|
self.assertIsNone(T.peek_cached_translation("nope", "English"))
|
|
|
|
def test_persists_to_disk(self):
|
|
payload = '{"Line1": "保存"}'
|
|
T.cache_translation(payload, ["Save"], "English")
|
|
self.assertTrue(T.CACHE_FILE.exists())
|
|
# Drop the in-memory copy; a fresh read must come from disk.
|
|
T._cache = {}
|
|
self.assertEqual(T.peek_cached_translation(payload, "English"), ["Save"])
|
|
|
|
|
|
class PendingMarkerTests(CacheTestBase):
|
|
def test_miss_writes_pending_then_peek_sees_it_as_none(self):
|
|
payload = '{"Line1": "待機"}'
|
|
# First lookup is a miss: returns None and leaves a pending marker.
|
|
self.assertIsNone(T.get_cached_translation(payload, "English"))
|
|
# peek treats a pending marker as "not ready" -> None (never blocks).
|
|
self.assertIsNone(T.peek_cached_translation(payload, "English"))
|
|
|
|
def test_own_pending_does_not_deadlock(self):
|
|
# A second call from the same pid/thread must not block on its own marker.
|
|
payload = '{"Line1": "再入"}'
|
|
self.assertIsNone(T.get_cached_translation(payload, "English"))
|
|
self.assertIsNone(T.get_cached_translation(payload, "English"))
|
|
|
|
def test_real_value_after_pending_is_returned(self):
|
|
payload = '{"Line1": "結果"}'
|
|
self.assertIsNone(T.get_cached_translation(payload, "English"))
|
|
T.cache_translation(payload, ["Result"], "English")
|
|
self.assertEqual(T.get_cached_translation(payload, "English"), ["Result"])
|
|
|
|
|
|
class MergeTests(CacheTestBase):
|
|
def test_pending_never_overwrites_real_translation(self):
|
|
real = {"k": ["done"]}
|
|
overlay = {"k": T._pending_cache_entry()}
|
|
merged = T._merge_translation_caches(real, overlay)
|
|
self.assertEqual(merged["k"], ["done"])
|
|
|
|
def test_real_translation_overwrites_pending(self):
|
|
base = {"k": T._pending_cache_entry()}
|
|
overlay = {"k": ["done"]}
|
|
merged = T._merge_translation_caches(base, overlay)
|
|
self.assertEqual(merged["k"], ["done"])
|
|
|
|
def test_new_keys_are_added(self):
|
|
merged = T._merge_translation_caches({"a": [1]}, {"b": [2]})
|
|
self.assertEqual(merged, {"a": [1], "b": [2]})
|
|
|
|
def test_stale_pending_can_be_replaced_by_pending(self):
|
|
stale = T._pending_cache_entry()
|
|
stale["time"] = 0 # far in the past -> stale
|
|
fresh = T._pending_cache_entry()
|
|
merged = T._merge_translation_caches({"k": stale}, {"k": fresh})
|
|
self.assertIs(merged["k"], fresh)
|
|
|
|
|
|
class ExpandCleanToBatchTests(unittest.TestCase):
|
|
"""The core of the cache fix: cached values are Japanese-only and must be
|
|
re-expanded with the *current* batch's skipped originals."""
|
|
|
|
def test_no_skips_returns_clean_values_in_order(self):
|
|
tItem = ["あ", "い", "う"]
|
|
out = T.expand_clean_to_batch(["A", "B", "C"], tItem, {}, {})
|
|
self.assertEqual(out, ["A", "B", "C"])
|
|
|
|
def test_no_japanese_originals_reinserted_per_file(self):
|
|
# Same Japanese-only cache value ["A", "C"], but each file supplies its
|
|
# own English neighbour at index 1 — proving no cross-file leakage.
|
|
tItem = ["あ", "Hello", "う"]
|
|
clean = ["A", "C"]
|
|
out = T.expand_clean_to_batch(clean, tItem, {}, {1: "Hello"})
|
|
self.assertEqual(out, ["A", "Hello", "C"])
|
|
|
|
tItem2 = ["あ", "World", "う"]
|
|
out2 = T.expand_clean_to_batch(clean, tItem2, {}, {1: "World"})
|
|
self.assertEqual(out2, ["A", "World", "C"])
|
|
|
|
def test_corrupted_originals_reinserted(self):
|
|
tItem = ["あ", "\ufffd bad", "う"]
|
|
out = T.expand_clean_to_batch(["A", "C"], tItem, {1: "\ufffd bad"}, {})
|
|
self.assertEqual(out, ["A", "\ufffd bad", "C"])
|
|
|
|
def test_mixed_corrupted_and_no_japanese(self):
|
|
tItem = ["あ", "\ufffd", "Name", "え"]
|
|
out = T.expand_clean_to_batch(
|
|
["A", "E"], tItem, {1: "\ufffd"}, {2: "Name"}
|
|
)
|
|
self.assertEqual(out, ["A", "\ufffd", "Name", "E"])
|
|
|
|
def test_result_length_always_matches_batch(self):
|
|
tItem = ["あ", "x", "う", "y"]
|
|
out = T.expand_clean_to_batch(["A", "C"], tItem, {}, {1: "x", 3: "y"})
|
|
self.assertEqual(len(out), len(tItem))
|
|
|
|
def test_short_clean_values_falls_back_to_source(self):
|
|
# Defensive: a stale/short cache value must not raise; it falls back to
|
|
# the source line for any positions it can't fill.
|
|
tItem = ["あ", "い", "う"]
|
|
out = T.expand_clean_to_batch(["A"], tItem, {}, {})
|
|
self.assertEqual(out, ["A", "い", "う"])
|
|
|
|
|
|
class SaveLoadTests(CacheTestBase):
|
|
def test_save_preserves_entries_from_other_workers(self):
|
|
# Simulate another worker having written an entry to disk.
|
|
T._write_cache_to_disk({"other": ["kept"]})
|
|
T._cache = {"mine": ["added"]}
|
|
T.save_cache()
|
|
on_disk = T._read_cache_from_disk()
|
|
self.assertEqual(on_disk.get("other"), ["kept"])
|
|
self.assertEqual(on_disk.get("mine"), ["added"])
|
|
|
|
def test_load_merges_disk_and_memory(self):
|
|
T._write_cache_to_disk({"disk": ["d"]})
|
|
T._cache = {"mem": ["m"]}
|
|
loaded = T.load_cache()
|
|
self.assertEqual(loaded.get("disk"), ["d"])
|
|
self.assertEqual(loaded.get("mem"), ["m"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|