a227aed0cb
- New `sync.fullsync` CLI: `python -m sync.fullsync config.yaml [--db F] [--table T] [--clear-change-log]` - AccessReader: list_user_tables / read_all_rows / read_all_log_ids - SqlWriter: table_exists / truncate_target (TRUNCATE w/ DELETE fallback) / bulk_insert (SET IDENTITY_INSERT + chunked fast_executemany) - Reuses FileMapping exclude/include rules (exclude beats include, same as capture) - Preserves Access IDs via IDENTITY_INSERT; target schemas have no FKs so TRUNCATE is safe - TableChangeLog NOT cleared by default (opt-in --clear-change-log) - tests/test_fullsync.py covers resolve_tables exclude/include precedence
57 lines
1.8 KiB
Python
57 lines
1.8 KiB
Python
"""Unit tests for the full-sync table-resolution rules.
|
|
|
|
``resolve_tables`` is pure (no DB) once given a reader stub, so it is safe to
|
|
exercise without Access/SQL connections. Run on a host with the project venv
|
|
(e.g. 114), where ``pyodbc`` imports cleanly:
|
|
|
|
.venv\\Scripts\\python.exe -m pytest tests\\test_fullsync.py -q
|
|
"""
|
|
from sync.config import FileMapping
|
|
from sync.fullsync import resolve_tables
|
|
|
|
|
|
class FakeReader:
|
|
"""Mirrors AccessReader.list_user_tables: system tables are already
|
|
filtered here, so resolve_tables only needs to apply the exclude/include
|
|
rules from the FileMapping."""
|
|
|
|
def __init__(self, tables):
|
|
self._tables = [
|
|
t for t in tables
|
|
if not (t.startswith("MSys") or t.startswith("~"))
|
|
]
|
|
|
|
def list_user_tables(self):
|
|
return list(self._tables)
|
|
|
|
|
|
def make_fm(**kw):
|
|
base = dict(file="x.accdb", root="2026", schema="s", year_suffix="")
|
|
base.update(kw)
|
|
return FileMapping(**base)
|
|
|
|
|
|
def test_excludes_system_and_config_tables():
|
|
fm = make_fm(exclude_tables=["TableChangeLog", "Foo"])
|
|
r = FakeReader(["T1", "TableChangeLog", "MSysSysObjects", "~tmp", "Foo", "Bar"])
|
|
assert resolve_tables(r, fm) == ["T1", "Bar"]
|
|
|
|
|
|
def test_include_tables_restricts_scope():
|
|
fm = make_fm(include_tables=["T1", "T2"])
|
|
r = FakeReader(["T1", "T2", "T3", "TableChangeLog"])
|
|
assert resolve_tables(r, fm) == ["T1", "T2"]
|
|
|
|
|
|
def test_exclude_beats_include():
|
|
# Precedence matches capture.capture_file: exclude wins over include.
|
|
fm = make_fm(include_tables=["T1"], exclude_tables=["T1"])
|
|
r = FakeReader(["T1", "T2"])
|
|
assert resolve_tables(r, fm) == []
|
|
|
|
|
|
def test_empty_when_all_excluded():
|
|
fm = make_fm(exclude_tables=["T1", "T2"])
|
|
r = FakeReader(["T1", "T2"])
|
|
assert resolve_tables(r, fm) == []
|