✨ feat(sync): add data consistency compare (count + ID-set)
This commit is contained in:
@@ -163,6 +163,18 @@ class AccessReader:
|
||||
cur.execute("SELECT ID FROM TableChangeLog ORDER BY ID")
|
||||
return [r[0] for r in cur.fetchall()]
|
||||
|
||||
def count_rows(self, table: str) -> int:
|
||||
"""Return ``COUNT(*)`` for ``table`` (compare count check)."""
|
||||
cur = self._connect().cursor()
|
||||
cur.execute(f'SELECT COUNT(*) FROM "{table}"')
|
||||
return cur.fetchone()[0]
|
||||
|
||||
def read_ids(self, table: str) -> list:
|
||||
"""Return every ``ID`` from ``table``, ascending (compare ID-set check)."""
|
||||
cur = self._connect().cursor()
|
||||
cur.execute(f'SELECT ID FROM "{table}" ORDER BY ID')
|
||||
return [r[0] for r in cur.fetchall()]
|
||||
|
||||
def close(self):
|
||||
if self._conn:
|
||||
self._conn.close()
|
||||
|
||||
141
src/sync/compare.py
Normal file
141
src/sync/compare.py
Normal file
@@ -0,0 +1,141 @@
|
||||
"""Data consistency check: Access source tables vs their SQL Server mirrors.
|
||||
|
||||
Two granularities:
|
||||
- ``count`` (default): row-count totals per table.
|
||||
- ``ids``: ID-set membership -- which IDs exist only in Access or only in SQL.
|
||||
|
||||
Tables whose SQL mirror does not exist are skipped (same rule fullsync uses)
|
||||
and reported as ``skipped``; they do not count as mismatches. Uses
|
||||
``targets.resolve_synced_tables`` so compare visits exactly the same table set
|
||||
as full sync -- empirically confirming the two stay aligned.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from dataclasses import dataclass, field
|
||||
|
||||
from .config import FileMapping, SyncConfig
|
||||
from .access_reader import AccessReader
|
||||
from .sql_writer import SqlWriter
|
||||
from .targets import resolve_synced_tables
|
||||
|
||||
log = logging.getLogger("sync.compare")
|
||||
|
||||
|
||||
@dataclass
|
||||
class TableResult:
|
||||
"""One table's comparison outcome."""
|
||||
file: str
|
||||
access_table: str
|
||||
target_schema: str
|
||||
target_table: str
|
||||
status: str # "match" | "mismatch" | "skipped" | "error"
|
||||
access_count: int | None = None
|
||||
sql_count: int | None = None
|
||||
missing_in_sql: list = field(default_factory=list) # IDs in Access, not SQL
|
||||
extra_in_sql: list = field(default_factory=list) # IDs in SQL, not Access
|
||||
error: str | None = None
|
||||
|
||||
|
||||
def compare_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter,
|
||||
granularity: str = "count") -> list[TableResult]:
|
||||
"""Compare every in-scope table in one Access file against its SQL mirror.
|
||||
|
||||
``granularity`` is ``"count"`` (row totals, default) or ``"ids"`` (ID-set
|
||||
membership). Skips tables with no SQL mirror. Per-table errors are caught
|
||||
so one bad table does not abort the file.
|
||||
"""
|
||||
results: list[TableResult] = []
|
||||
for access_table in resolve_synced_tables(fm, reader):
|
||||
target = fm.target_table(access_table)
|
||||
if not writer.table_exists(fm.schema, target):
|
||||
results.append(TableResult(fm.file, access_table, fm.schema, target, "skipped"))
|
||||
log.warning("skip %s -> %s.%s (target table missing)",
|
||||
access_table, fm.schema, target)
|
||||
continue
|
||||
try:
|
||||
if granularity == "ids":
|
||||
a_ids = set(reader.read_ids(access_table))
|
||||
s_ids = set(writer.read_target_ids(fm.schema, target))
|
||||
missing = sorted(a_ids - s_ids)
|
||||
extra = sorted(s_ids - a_ids)
|
||||
status = "match" if not missing and not extra else "mismatch"
|
||||
results.append(TableResult(
|
||||
fm.file, access_table, fm.schema, target, status,
|
||||
access_count=len(a_ids), sql_count=len(s_ids),
|
||||
missing_in_sql=missing, extra_in_sql=extra,
|
||||
))
|
||||
else:
|
||||
a = reader.count_rows(access_table)
|
||||
s = writer.count_target(fm.schema, target)
|
||||
status = "match" if a == s else "mismatch"
|
||||
results.append(TableResult(
|
||||
fm.file, access_table, fm.schema, target, status,
|
||||
access_count=a, sql_count=s,
|
||||
))
|
||||
except Exception as e:
|
||||
results.append(TableResult(
|
||||
fm.file, access_table, fm.schema, target, "error", error=str(e)
|
||||
))
|
||||
log.exception("compare failed for %s -> %s.%s", access_table, fm.schema, target)
|
||||
return results
|
||||
|
||||
|
||||
def compare(cfg: SyncConfig, granularity: str = "count",
|
||||
db_filter: str | None = None,
|
||||
table_filter: str | None = None) -> list[TableResult]:
|
||||
"""Compare all (optionally filtered) configured files.
|
||||
|
||||
``db_filter`` limits to one Access file; ``table_filter`` restricts every
|
||||
file to that one table (overrides include_tables), mirroring fullsync.
|
||||
"""
|
||||
files = cfg.files
|
||||
if db_filter:
|
||||
files = [f for f in files if f.file == db_filter]
|
||||
if not files:
|
||||
log.warning("no file matches --db %r", db_filter)
|
||||
return []
|
||||
if table_filter:
|
||||
files = [f.model_copy(update={"include_tables": [table_filter]}) for f in files]
|
||||
|
||||
writer = SqlWriter(cfg.sql_server.conn_str, cfg.sql_server.sync_queue_table)
|
||||
results: list[TableResult] = []
|
||||
try:
|
||||
for fm in files:
|
||||
reader = AccessReader(fm.source_path(cfg), cfg.access.driver)
|
||||
try:
|
||||
results.extend(compare_file(fm, reader, writer, granularity))
|
||||
finally:
|
||||
reader.close()
|
||||
finally:
|
||||
writer.close()
|
||||
return results
|
||||
|
||||
|
||||
def any_mismatch(results: list[TableResult]) -> bool:
|
||||
"""True if any compared table diverged (skipped/error do not count)."""
|
||||
return any(r.status == "mismatch" for r in results)
|
||||
|
||||
|
||||
def format_report(results: list[TableResult], granularity: str = "count") -> str:
|
||||
"""Render a human-readable per-table report."""
|
||||
lines = []
|
||||
for r in results:
|
||||
base = f"{r.file}: {r.access_table} -> {r.target_schema}.{r.target_table}"
|
||||
if r.status == "skipped":
|
||||
lines.append(f"{base} [SKIPPED no mirror]")
|
||||
elif r.status == "error":
|
||||
lines.append(f"{base} [ERROR {r.error}]")
|
||||
elif granularity == "ids":
|
||||
lines.append(
|
||||
f"{base} access={r.access_count} sql={r.sql_count} "
|
||||
f"missing_in_sql={len(r.missing_in_sql)} extra_in_sql={len(r.extra_in_sql)} "
|
||||
f"[{r.status.upper()}]"
|
||||
)
|
||||
if r.missing_in_sql:
|
||||
lines.append(f" missing_in_sql (first 50): {r.missing_in_sql[:50]}")
|
||||
if r.extra_in_sql:
|
||||
lines.append(f" extra_in_sql (first 50): {r.extra_in_sql[:50]}")
|
||||
else:
|
||||
lines.append(f"{base} access={r.access_count} sql={r.sql_count} [{r.status.upper()}]")
|
||||
return "\n".join(lines)
|
||||
@@ -143,6 +143,18 @@ class SqlWriter:
|
||||
)
|
||||
return cur.fetchone() is not None
|
||||
|
||||
def count_target(self, schema: str, table: str) -> int:
|
||||
"""Return ``COUNT(*)`` for ``[schema].[table]`` (compare count check)."""
|
||||
cur = self._conn.cursor()
|
||||
cur.execute(f"SELECT COUNT(*) FROM [{schema}].[{table}]")
|
||||
return cur.fetchone()[0]
|
||||
|
||||
def read_target_ids(self, schema: str, table: str) -> list:
|
||||
"""Return every ``ID`` from ``[schema].[table]``, ascending (compare IDs)."""
|
||||
cur = self._conn.cursor()
|
||||
cur.execute(f"SELECT ID FROM [{schema}].[{table}] ORDER BY ID")
|
||||
return [r[0] for r in cur.fetchall()]
|
||||
|
||||
def _has_identity(self, schema: str, table: str) -> bool:
|
||||
"""True if ``[schema].[table]`` has an IDENTITY column (the ``ID`` PK)."""
|
||||
cur = self._conn.cursor()
|
||||
|
||||
@@ -6,6 +6,7 @@ No real log rows are deleted.
|
||||
"""
|
||||
import os
|
||||
import pytest
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from sync.access_reader import AccessReader
|
||||
from sync.config import load_config
|
||||
@@ -35,3 +36,23 @@ def test_read_log_and_row_and_delete():
|
||||
r.delete_log_ids([], 100, 3) # empty list -> no-op, must not raise
|
||||
finally:
|
||||
r.close()
|
||||
|
||||
|
||||
def test_count_rows_executes_count_sql_and_returns_value():
|
||||
r = AccessReader("dummy.accdb", "{Microsoft Access Driver (*.accdb, *.mdb)}")
|
||||
cur = MagicMock()
|
||||
cur.fetchone.return_value = (42,)
|
||||
r._conn = MagicMock() # bypass lazy connect
|
||||
r._conn.cursor.return_value = cur
|
||||
assert r.count_rows("表壳焊接记录") == 42
|
||||
cur.execute.assert_called_once_with('SELECT COUNT(*) FROM "表壳焊接记录"')
|
||||
|
||||
|
||||
def test_read_ids_returns_ordered_id_list():
|
||||
r = AccessReader("dummy.accdb", "{Microsoft Access Driver (*.accdb, *.mdb)}")
|
||||
cur = MagicMock()
|
||||
cur.fetchall.return_value = [(1,), (3,), (5,)]
|
||||
r._conn = MagicMock()
|
||||
r._conn.cursor.return_value = cur
|
||||
assert r.read_ids("T") == [1, 3, 5]
|
||||
cur.execute.assert_called_once_with('SELECT ID FROM "T" ORDER BY ID')
|
||||
|
||||
105
tests/test_compare.py
Normal file
105
tests/test_compare.py
Normal file
@@ -0,0 +1,105 @@
|
||||
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
|
||||
@@ -10,6 +10,7 @@ credentials are hardcoded here. The test self-cleans using a throwaway
|
||||
"""
|
||||
import os
|
||||
import pytest
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from sync.sql_writer import SqlWriter, QueueRow
|
||||
from sync.config import load_config
|
||||
@@ -53,3 +54,30 @@ def test_insert_dedup_and_applied_ids():
|
||||
finally:
|
||||
cur.execute("DELETE dbo.SyncQueue WHERE SourceFile='sqlw_test.accdb'")
|
||||
w.close()
|
||||
|
||||
|
||||
def _writer_with_cursor(fetchone=None, fetchall=None):
|
||||
"""A SqlWriter whose pyodbc connection is a mock (no real connect)."""
|
||||
w = SqlWriter.__new__(SqlWriter)
|
||||
w.conn_str = "dummy"
|
||||
w.queue_table = "dbo.SyncQueue"
|
||||
cur = MagicMock()
|
||||
if fetchone is not None:
|
||||
cur.fetchone.return_value = fetchone
|
||||
if fetchall is not None:
|
||||
cur.fetchall.return_value = fetchall
|
||||
w._conn = MagicMock()
|
||||
w._conn.cursor.return_value = cur
|
||||
return w, cur
|
||||
|
||||
|
||||
def test_count_target_executes_count_sql_and_returns_value():
|
||||
w, cur = _writer_with_cursor(fetchone=(7,))
|
||||
assert w.count_target("s", "T_YEAR2026") == 7
|
||||
cur.execute.assert_called_once_with("SELECT COUNT(*) FROM [s].[T_YEAR2026]")
|
||||
|
||||
|
||||
def test_read_target_ids_returns_ordered_id_list():
|
||||
w, cur = _writer_with_cursor(fetchall=[(2,), (4,), (6,)])
|
||||
assert w.read_target_ids("s", "T_YEAR2026") == [2, 4, 6]
|
||||
cur.execute.assert_called_once_with("SELECT ID FROM [s].[T_YEAR2026] ORDER BY ID")
|
||||
|
||||
Reference in New Issue
Block a user