62 lines
3.0 KiB
Python
62 lines
3.0 KiB
Python
from unittest.mock import MagicMock
|
|
from sync.config import FileMapping, SyncConfig, AccessConfig, RuntimeConfig, SqlServerConfig
|
|
from sync.access_reader import LogRow
|
|
from sync.capture import capture_file
|
|
|
|
def _cfg():
|
|
return SyncConfig(sql_server=SqlServerConfig(conn_str="x"),
|
|
access=AccessConfig(driver="d", roots={"2026":"r"}),
|
|
runtime=RuntimeConfig(),
|
|
files=[])
|
|
|
|
def test_capture_insert_reads_row_and_queues():
|
|
cfg = _cfg()
|
|
fm = FileMapping(file="氩弧焊.accdb", root="2026", schema="TIGWelding", year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"])
|
|
reader = MagicMock()
|
|
reader.read_log.return_value = [LogRow(10, "表壳焊接记录", "34041", "Insert", None)]
|
|
reader.read_row.return_value = {"ID": 34041, "订单号": "X1"}
|
|
writer = MagicMock()
|
|
n = capture_file(fm, reader, writer, cfg)
|
|
assert n == 1
|
|
args = writer.insert_queue_row.call_args[0][0]
|
|
assert args.target_schema == "TIGWelding"
|
|
assert args.target_table == "表壳焊接记录_YEAR2026"
|
|
assert args.operate_type == "Insert"
|
|
assert '"订单号": "X1"' in args.row_data
|
|
assert args.source_log_id == 10
|
|
|
|
def test_capture_update_missing_row_downgrades_to_delete():
|
|
cfg = _cfg()
|
|
fm = FileMapping(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026", exclude_tables=["TableChangeLog"])
|
|
reader = MagicMock()
|
|
reader.read_log.return_value = [LogRow(11, "T", "5", "Update", None)]
|
|
reader.read_row.return_value = None # 行已删
|
|
writer = MagicMock()
|
|
n = capture_file(fm, reader, writer, cfg)
|
|
assert n == 1
|
|
args = writer.insert_queue_row.call_args[0][0]
|
|
assert args.operate_type == "Delete"
|
|
assert args.row_data is None
|
|
|
|
def test_capture_skips_excluded_tables():
|
|
cfg = _cfg()
|
|
fm = FileMapping(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026", exclude_tables=["TableChangeLog", "氩弧焊每日催货落实记录_停"])
|
|
reader = MagicMock()
|
|
reader.read_log.return_value = [LogRow(1, "TableChangeLog", "1", "Insert", None),
|
|
LogRow(2, "氩弧焊每日催货落实记录_停", "1", "Insert", None)]
|
|
writer = MagicMock()
|
|
assert capture_file(fm, reader, writer, cfg) == 0
|
|
writer.insert_queue_row.assert_not_called()
|
|
|
|
def test_capture_include_tables_filter():
|
|
cfg = _cfg()
|
|
fm = FileMapping(file="x.accdb", root="2026", schema="inspectionRecords", year_suffix="_YEAR2026",
|
|
exclude_tables=["TableChangeLog"], include_tables=["检验合格记录表"])
|
|
reader = MagicMock()
|
|
reader.read_log.return_value = [LogRow(1, "检验合格记录表", "1", "Insert", None),
|
|
LogRow(2, "其它表", "1", "Insert", None)]
|
|
reader.read_row.return_value = {"ID": 1}
|
|
writer = MagicMock()
|
|
assert capture_file(fm, reader, writer, cfg) == 1
|
|
assert writer.insert_queue_row.call_args[0][0].target_table == "检验合格记录表_YEAR2026"
|