feat(sync): add unified main.py CLI dispatcher

This commit is contained in:
Misaka_Company
2026-07-15 14:03:24 +08:00
parent e9f012eab5
commit 1b44ca28e4
3 changed files with 184 additions and 1 deletions

70
tests/test_main.py Normal file
View File

@@ -0,0 +1,70 @@
"""Routing/argparse tests for the root main.py dispatcher.
The backend functions (full_sync, service.cycle/run, compare) are mocked so
these tests verify command dispatch, arg parsing, report output and exit codes
without touching Access or SQL Server.
"""
from unittest.mock import patch
import main
from sync.compare import TableResult
def test_fullsync_routes_with_filters():
with patch("main.load_config"), patch("main.setup_logging"), \
patch("main.full_sync") as fs, patch("main.service") as svc:
rc = main.main(["fullsync", "--db", "OEM.accdb", "--table", "表壳焊接记录",
"--clear-change-log"])
fs.assert_called_once()
assert fs.call_args.kwargs["db_filter"] == "OEM.accdb"
assert fs.call_args.kwargs["table_filter"] == "表壳焊接记录"
assert fs.call_args.kwargs["clear_change_log"] is True
svc.cycle.assert_not_called()
assert rc == 0
def test_incremental_default_runs_single_cycle():
with patch("main.load_config"), patch("main.setup_logging"), \
patch("main.full_sync"), patch("main.service") as svc:
rc = main.main(["incremental"])
svc.cycle.assert_called_once()
svc.run.assert_not_called()
assert rc == 0
def test_incremental_loop_runs_service():
with patch("main.load_config"), patch("main.setup_logging"), \
patch("main.full_sync"), patch("main.service") as svc:
rc = main.main(["incremental", "--loop", "--poll-interval", "5"])
svc.run.assert_called_once()
svc.cycle.assert_not_called()
assert rc == 0
def test_compare_default_granularity_is_count(capsys):
with patch("main.load_config"), patch("main.setup_logging"), \
patch("main.full_sync"), patch("main.service"), patch("main.compare") as cmp:
cmp.return_value = [TableResult("x.accdb", "T", "s", "T_YEAR2026", "match", 5, 5)]
rc = main.main(["compare"])
assert cmp.call_args.kwargs["granularity"] == "count"
assert rc == 0
assert "x.accdb: T -> s.T_YEAR2026" in capsys.readouterr().out
def test_compare_ids_granularity_and_mismatch_exit_code():
with patch("main.load_config"), patch("main.setup_logging"), \
patch("main.full_sync"), patch("main.service"), patch("main.compare") as cmp:
cmp.return_value = [TableResult("x.accdb", "T", "s", "T_YEAR2026", "mismatch", 5, 7)]
rc = main.main(["compare", "--granularity", "ids"])
assert cmp.call_args.kwargs["granularity"] == "ids"
assert rc == 1
def test_compare_writes_report_file(tmp_path):
with patch("main.load_config"), patch("main.setup_logging"), \
patch("main.full_sync"), patch("main.service"), patch("main.compare") as cmp:
cmp.return_value = [TableResult("x.accdb", "T", "s", "T_YEAR2026", "match", 1, 1)]
rep = tmp_path / "report.txt"
rc = main.main(["compare", "--report", str(rep)])
assert rc == 0
assert "x.accdb: T -> s.T_YEAR2026" in rep.read_text(encoding="utf-8")