106 lines
3.4 KiB
Python
106 lines
3.4 KiB
Python
from unittest.mock import MagicMock
|
|
|
|
from sync.config import FileMapping
|
|
from sync.compare import compare_file, any_mismatch, format_report, TableResult
|
|
|
|
|
|
def _fm(**kw):
|
|
base = dict(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026")
|
|
base.update(kw)
|
|
return FileMapping(**base)
|
|
|
|
|
|
def _reader_with_tables(tables):
|
|
r = MagicMock()
|
|
r.list_user_tables.return_value = tables
|
|
return r
|
|
|
|
|
|
def test_count_match():
|
|
fm = _fm(exclude_tables=["TableChangeLog"])
|
|
reader = _reader_with_tables(["T1", "TableChangeLog"])
|
|
reader.count_rows.return_value = 5
|
|
writer = MagicMock()
|
|
writer.table_exists.return_value = True
|
|
writer.count_target.return_value = 5
|
|
res = compare_file(fm, reader, writer, "count")
|
|
assert len(res) == 1
|
|
assert res[0].access_table == "T1"
|
|
assert res[0].status == "match"
|
|
assert res[0].access_count == 5 and res[0].sql_count == 5
|
|
|
|
|
|
def test_count_mismatch():
|
|
fm = _fm()
|
|
reader = _reader_with_tables(["T1"])
|
|
reader.count_rows.return_value = 5
|
|
writer = MagicMock()
|
|
writer.table_exists.return_value = True
|
|
writer.count_target.return_value = 7
|
|
res = compare_file(fm, reader, writer, "count")
|
|
assert res[0].status == "mismatch"
|
|
|
|
|
|
def test_skip_when_mirror_missing():
|
|
fm = _fm()
|
|
reader = _reader_with_tables(["T1"])
|
|
writer = MagicMock()
|
|
writer.table_exists.return_value = False
|
|
res = compare_file(fm, reader, writer, "count")
|
|
assert res[0].status == "skipped"
|
|
writer.count_target.assert_not_called()
|
|
writer.read_target_ids.assert_not_called()
|
|
|
|
|
|
def test_ids_match():
|
|
fm = _fm()
|
|
reader = _reader_with_tables(["T1"])
|
|
reader.read_ids.return_value = [1, 2, 3]
|
|
writer = MagicMock()
|
|
writer.table_exists.return_value = True
|
|
writer.read_target_ids.return_value = [1, 2, 3]
|
|
res = compare_file(fm, reader, writer, "ids")
|
|
assert res[0].status == "match"
|
|
assert res[0].missing_in_sql == []
|
|
assert res[0].extra_in_sql == []
|
|
|
|
|
|
def test_ids_reports_missing_and_extra():
|
|
fm = _fm()
|
|
reader = _reader_with_tables(["T1"])
|
|
reader.read_ids.return_value = [1, 2, 3]
|
|
writer = MagicMock()
|
|
writer.table_exists.return_value = True
|
|
writer.read_target_ids.return_value = [2, 3, 4]
|
|
res = compare_file(fm, reader, writer, "ids")
|
|
assert res[0].status == "mismatch"
|
|
assert res[0].missing_in_sql == [1] # in Access, not in SQL
|
|
assert res[0].extra_in_sql == [4] # in SQL, not in Access
|
|
|
|
|
|
def test_excluded_tables_not_compared():
|
|
fm = _fm(exclude_tables=["TableChangeLog"])
|
|
reader = _reader_with_tables(["T1", "TableChangeLog"])
|
|
reader.count_rows.return_value = 1
|
|
writer = MagicMock()
|
|
writer.table_exists.return_value = True
|
|
writer.count_target.return_value = 1
|
|
res = compare_file(fm, reader, writer, "count")
|
|
assert [r.access_table for r in res] == ["T1"]
|
|
|
|
|
|
def test_any_mismatch_detects_mismatch_only():
|
|
r_match = TableResult("f", "T", "s", "T_YEAR2026", "match", 1, 1)
|
|
r_skip = TableResult("f", "T2", "s", "T2_YEAR2026", "skipped")
|
|
r_mis = TableResult("f", "T3", "s", "T3_YEAR2026", "mismatch", 1, 2)
|
|
assert any_mismatch([r_match, r_skip]) is False
|
|
assert any_mismatch([r_match, r_mis]) is True
|
|
|
|
|
|
def test_format_report_count():
|
|
r = TableResult("x.accdb", "T1", "s", "T1_YEAR2026", "mismatch", 5, 7)
|
|
rep = format_report([r], "count")
|
|
assert "x.accdb: T1 -> s.T1_YEAR2026" in rep
|
|
assert "access=5 sql=7" in rep
|
|
assert "[MISMATCH]" in rep
|