Codex
317b0e6feb
Add order_staff normalization, salesperson query templates, guarded read-only SQL fallback, and incremental Kingdee sync cursor handling. Add a user-level systemd timer manager for daily Kingdee business sync and document the new workflow. Validation: python3 -m unittest discover -v; python3 -m compileall obsidian_wiki tests scripts; git diff --check.
85 lines
3.4 KiB
Python
85 lines
3.4 KiB
Python
import os
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import call, patch
|
|
|
|
|
|
SCRIPTS_DIR = Path(__file__).resolve().parent.parent / "scripts"
|
|
if str(SCRIPTS_DIR) not in sys.path:
|
|
sys.path.insert(0, str(SCRIPTS_DIR))
|
|
|
|
import manage_business_sync
|
|
|
|
|
|
class BusinessSyncServiceTests(unittest.TestCase):
|
|
def test_render_service_and_timer_units_for_incremental_sync(self):
|
|
service = manage_business_sync.render_service_unit(
|
|
project_root=Path("/opt/obsidian-wiki"),
|
|
vault_path=Path("/data/bjd-vault"),
|
|
config_path=Path("/opt/obsidian-wiki/.runtime/kingdee.env"),
|
|
python_path=Path("/usr/bin/python3"),
|
|
lookback_hours=48,
|
|
initial_days=7,
|
|
)
|
|
timer = manage_business_sync.render_timer_unit(
|
|
on_calendar="*-*-* 07:10:00",
|
|
randomized_delay_sec="15m",
|
|
)
|
|
|
|
self.assertIn("WorkingDirectory=/opt/obsidian-wiki", service)
|
|
self.assertIn("EnvironmentFile=/opt/obsidian-wiki/.runtime/kingdee.env", service)
|
|
self.assertIn("/usr/bin/python3 -m obsidian_wiki business sync kingdee", service)
|
|
self.assertIn("--vault /data/bjd-vault", service)
|
|
self.assertIn("--config /opt/obsidian-wiki/.runtime/kingdee.env", service)
|
|
self.assertIn("--incremental", service)
|
|
self.assertIn("--lookback-hours 48", service)
|
|
self.assertIn("--initial-days 7", service)
|
|
self.assertNotIn("KINGDEE_CLIENT_SECRET=", service)
|
|
self.assertNotIn("KINGDEE_ACCESS_TOKEN=", service)
|
|
self.assertIn("OnCalendar=*-*-* 07:10:00", timer)
|
|
self.assertIn("Persistent=true", timer)
|
|
self.assertIn("RandomizedDelaySec=15m", timer)
|
|
|
|
def test_start_timer_writes_units_and_enables_timer(self):
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
project_root = Path(tmp) / "project"
|
|
vault = Path(tmp) / "vault"
|
|
config_home = Path(tmp) / "config"
|
|
project_root.mkdir()
|
|
vault.mkdir()
|
|
(project_root / ".runtime").mkdir()
|
|
(project_root / ".runtime" / "kingdee.env").write_text("KINGDEE_ACGW_IDENTITY=test\n", encoding="utf-8")
|
|
|
|
with (
|
|
patch.dict("os.environ", {"XDG_CONFIG_HOME": str(config_home)}),
|
|
patch.object(manage_business_sync, "run_systemctl") as run_systemctl,
|
|
):
|
|
output = manage_business_sync.start_timer(
|
|
project_root=project_root,
|
|
vault_path=vault,
|
|
python_path=Path("/usr/bin/python3"),
|
|
)
|
|
|
|
paths = manage_business_sync.resolve_service_paths(project_root, config_home=config_home)
|
|
self.assertTrue(paths.service_unit_path.is_file())
|
|
self.assertTrue(paths.timer_unit_path.is_file())
|
|
self.assertIn("Business sync timer is enabled.", output)
|
|
run_systemctl.assert_has_calls(
|
|
[
|
|
call(["daemon-reload"]),
|
|
call(["enable", "--now", manage_business_sync.TIMER_NAME]),
|
|
]
|
|
)
|
|
|
|
def test_run_now_starts_service_unit(self):
|
|
with patch.object(manage_business_sync, "run_systemctl") as run_systemctl:
|
|
manage_business_sync.run_now()
|
|
|
|
run_systemctl.assert_called_once_with(["start", manage_business_sync.SERVICE_NAME])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|