from __future__ import annotations import hashlib import importlib.util import json import tempfile import unittest from pathlib import Path from types import SimpleNamespace from datetime import datetime, timedelta from unittest import mock from shared.runtime import material_name_cache as cache from shared.runtime import install_material_name_refresh as installer from shared.runtime import kingdee_materials as kingdee 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_updates_are_seen_without_restart_and_integrity_is_checked(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"], "编码器") refresh.write_cache([{"FNumber": "A", "FName": "新编码器"}, {"FNumber": "B", "FName": "轴承"}], "2026-10-11T10:00:00+08:00", self.path) self.assertEqual(cache.load_material_names(self.path), {"A": "新编码器", "B": "轴承"}) payload = json.loads(self.path.read_text()) payload["materials"]["A"] = "篡改名称" self.path.write_text(json.dumps(payload)) 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")) def test_due_refresh_queries_after_48_hours_and_merges_without_deleting_names(self) -> None: start = datetime.fromisoformat("2026-10-09T10:00:00+08:00") state = Path(self.temp.name) / "state.json" rows = [{"FNumber": "A", "FName": "编码器"}, {"FNumber": "B", "FName": "轴承"}] args = dict(output=self.path, state_file=state, due_only=True, interval_hours=48, client_path=Path("client.py")) with mock.patch.object(refresh, "query_materials", return_value=(rows, start.isoformat())) as query: first = refresh.refresh_once(**args, current_time=start) self.assertEqual(first["added_count"], 2) result = refresh.refresh_once(**args, current_time=start + timedelta(hours=47)) self.assertEqual(result["status"], "NOT_DUE") query.assert_called_once() query.return_value = ([{"FNumber": "A", "FName": "新编码器"}, {"FNumber": "C", "FName": "减速器"}], start.isoformat()) result = refresh.refresh_once(**args, current_time=start + timedelta(hours=48)) self.assertEqual(result["added_count"], 1) self.assertEqual(result["renamed_count"], 1) self.assertEqual(result["retained_missing_count"], 1) self.assertEqual(cache.load_material_names(self.path), {"A": "新编码器", "B": "轴承", "C": "减速器"}) def test_unchanged_refresh_does_not_rewrite_git_cache_but_advances_due_state(self) -> None: start = datetime.fromisoformat("2026-10-09T10:00:00+08:00") state = Path(self.temp.name) / "state.json" rows = [{"FNumber": "A", "FName": "编码器"}] refresh.write_cache(rows, start.isoformat(), self.path) before = self.path.read_bytes() with mock.patch.object(refresh, "query_materials", return_value=(rows, start.isoformat())): result = refresh.refresh_once(output=self.path, state_file=state, due_only=True, interval_hours=48, client_path=Path("client.py"), current_time=start) self.assertEqual(result["status"], "UNCHANGED") self.assertEqual(self.path.read_bytes(), before) self.assertEqual(json.loads(state.read_text())["last_success_at"], start.isoformat()) def test_failed_due_refresh_preserves_cache_and_last_success(self) -> None: start = datetime.fromisoformat("2026-10-09T10:00:00+08:00") state = Path(self.temp.name) / "state.json" refresh.write_cache([{"FNumber": "A", "FName": "编码器"}], start.isoformat(), self.path) state.write_text(json.dumps({"last_success_at": start.isoformat()})) previous = self.path.read_bytes(), state.read_bytes() with mock.patch.object(refresh, "query_materials", side_effect=cache.MaterialNameCacheError("网络失败")): with self.assertRaises(cache.MaterialNameCacheError): refresh.refresh_once(output=self.path, state_file=state, due_only=True, interval_hours=48, client_path=Path("client.py"), current_time=start + timedelta(hours=48)) self.assertEqual((self.path.read_bytes(), state.read_bytes()), previous) def test_launch_agent_uses_host_paths_due_gate_and_no_credentials(self) -> None: job = installer.launch_agent(Path("/another-machine/AM516"), Path("/usr/bin/python3"), Path("/user/logs")) self.assertEqual(job["StartInterval"], 3600) self.assertTrue(job["RunAtLoad"]) self.assertIn("--due", job["ProgramArguments"]) self.assertIn("48", job["ProgramArguments"]) self.assertIn("/another-machine/AM516", job["WorkingDirectory"]) self.assertNotIn("KINGDEE_PASSWORD", job["EnvironmentVariables"]) self.assertNotIn("codex", " ".join(job["ProgramArguments"]).lower()) def test_bundled_kingdee_client_rejects_other_forms_or_fields_before_network(self) -> None: for form, fields, filter_string in (("STK_INVENTORY", "FNumber,FName", ""), ("BD_MATERIAL", "FNumber,FName,FSpecification", ""), ("BD_MATERIAL", "FNumber,FName", "FNumber='A'")): with self.subTest(form=form, fields=fields), mock.patch.object(kingdee, "post") as request: with self.assertRaises(kingdee.KingdeeMaterialError): kingdee.execute_list_query(object(), SimpleNamespace(form_id=form, fields=fields, filter=filter_string)) request.assert_not_called() if __name__ == "__main__": unittest.main()