feat: resolve shortage names from Git-synced Kingdee material cache

tests/test_material_name_cache.py

@@ -0,0 +1,110 @@
+from __future__ import annotations
+
+import hashlib
+import importlib.util
+import json
+import tempfile
+import unittest
+from pathlib import Path
+from types import SimpleNamespace
+from unittest import mock
+
+from shared.runtime import material_name_cache as cache
+
+
+script_path = Path(__file__).resolve().parents[1] / "capabilities/am516-delivery-prediction/scripts/refresh_material_names.py"
+spec = importlib.util.spec_from_file_location("refresh_material_names", script_path)
+refresh = importlib.util.module_from_spec(spec)
+spec.loader.exec_module(refresh)
+
+
+class MaterialNameCacheTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.temp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temp.cleanup)
+        self.path = Path(self.temp.name) / "another-checkout/data/material_names.json"
+        cache.load_material_names.cache_clear()
+        self.addCleanup(cache.load_material_names.cache_clear)
+
+    def test_generated_cache_is_portable_and_contains_only_name_lookup_data(self) -> None:
+        rows = [
+            {"FNumber": "20.40.00001", "FName": "编码器", "FMaterialId": 123, "FSpecification": "PRIVATE_MARKER"},
+            {"FNumber": "20.40.00001", "FName": "编码器", "FMaterialId": 456},
+            {"FNumber": "20.50.00002", "FName": "轴承"},
+        ]
+        payload = refresh.write_cache(rows, "2026-10-09T10:13:22+08:00", self.path)
+        self.assertEqual(cache.load_material_names(self.path), {"20.40.00001": "编码器", "20.50.00002": "轴承"})
+        self.assertEqual(payload["source"]["record_count"], 3)
+        self.assertEqual(payload["material_count"], 2)
+        content = self.path.read_text()
+        for unwanted in ("FMaterialId", "FSpecification", "PRIVATE_MARKER", "/Users/"):
+            self.assertNotIn(unwanted, content)
+
+    def test_empty_invalid_or_conflicting_refresh_preserves_existing_cache(self) -> None:
+        refresh.write_cache([{"FNumber": "A", "FName": "原物料"}], "2026-10-09T10:00:00+08:00", self.path)
+        previous = self.path.read_bytes()
+        for rows in (
+            [], [{"FNumber": "A", "FName": ""}],
+            [{"FNumber": "A", "FName": "名称1"}, {"FNumber": "A", "FName": "名称2"}],
+        ):
+            with self.subTest(rows=rows):
+                with self.assertRaises(cache.MaterialNameCacheError):
+                    refresh.write_cache(rows, "2026-10-09T11:00:00+08:00", self.path)
+                self.assertEqual(self.path.read_bytes(), previous)
+
+    def test_cache_integrity_is_checked_and_loaded_once_until_restart(self) -> None:
+        refresh.write_cache([{"FNumber": "A", "FName": "编码器"}], "2026-10-09T10:00:00+08:00", self.path)
+        self.assertEqual(cache.load_material_names(self.path)["A"], "编码器")
+        payload = json.loads(self.path.read_text())
+        payload["materials"]["A"] = "篡改名称"
+        self.path.write_text(json.dumps(payload))
+        self.assertEqual(cache.load_material_names(self.path)["A"], "编码器")
+        cache.load_material_names.cache_clear()
+        with self.assertRaisesRegex(cache.MaterialNameCacheError, "完整性校验失败"):
+            cache.load_material_names(self.path)
+
+    def test_reference_export_rejects_filtered_or_incomplete_source(self) -> None:
+        rows = [{"FNumber": "A", "FName": "编码器"}]
+        payload = {
+            "form_id": "BD_MATERIAL", "filter": "", "records": rows,
+            "record_count": len(rows), "generated_at": "2026-10-09T10:00:00+08:00",
+            "content_sha256": hashlib.sha256(json.dumps(rows, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode()).hexdigest(),
+        }
+        source = Path(self.temp.name) / "source.json"
+        source.write_text(json.dumps(payload))
+        self.assertEqual(refresh.read_reference(source)[0], rows)
+        for field, value in (("filter", "FForbidStatus='A'"), ("record_count", 2), ("content_sha256", "wrong")):
+            with self.subTest(field=field):
+                source.write_text(json.dumps({**payload, field: value}))
+                with self.assertRaises(cache.MaterialNameCacheError):
+                    refresh.read_reference(source)
+
+    def test_live_refresh_pages_two_fields_only_and_rejects_repeated_pages(self) -> None:
+        full_page = [[f"code-{i:04}", "编码器"] for i in range(2000)]
+        query = mock.Mock(side_effect=[full_page, [["last-code", "轴承"]]])
+        client = SimpleNamespace(
+            LOCAL_CREDENTIALS_PATH=Path("credentials.local.json"),
+            build_opener=mock.Mock(return_value=object()), login=mock.Mock(),
+            execute_list_query=query,
+        )
+        fake_spec = SimpleNamespace(loader=SimpleNamespace(exec_module=mock.Mock()))
+        with (
+            mock.patch.object(refresh.importlib.util, "spec_from_file_location", return_value=fake_spec),
+            mock.patch.object(refresh.importlib.util, "module_from_spec", return_value=client),
+            mock.patch("builtins.print"),
+        ):
+            rows, _ = refresh.query_materials(Path("/approved/source/91_Scripts/kingdee_openapi/kingdee_query.py"))
+            self.assertEqual(len(rows), 2001)
+            self.assertEqual([call.args[1].start_row for call in query.call_args_list], [0, 2000])
+            for call in query.call_args_list:
+                args = call.args[1]
+                self.assertEqual(args.fields, "FNumber,FName")
+                self.assertEqual(args.filter, "")
+                self.assertTrue(args.no_audit_log)
+            query.side_effect = [full_page, full_page]
+            with self.assertRaisesRegex(cache.MaterialNameCacheError, "分页重复"):
+                refresh.query_materials(Path("/approved/source/91_Scripts/kingdee_openapi/kingdee_query.py"))
+
+
+if __name__ == "__main__":
+    unittest.main()