♻️ refactor(sync): extract shared target-table resolution
This commit is contained in:
53
tests/test_targets.py
Normal file
53
tests/test_targets.py
Normal file
@@ -0,0 +1,53 @@
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from sync.config import FileMapping
|
||||
from sync.targets import is_synced_table, resolve_synced_tables
|
||||
|
||||
|
||||
def _fm(**kw):
|
||||
base = dict(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026")
|
||||
base.update(kw)
|
||||
return FileMapping(**base)
|
||||
|
||||
|
||||
def test_is_synced_table_excludes_listed():
|
||||
fm = _fm(exclude_tables=["TableChangeLog", "一车间每日催货落实记录_停"])
|
||||
assert is_synced_table(fm, "TableChangeLog") is False
|
||||
assert is_synced_table(fm, "一车间每日催货落实记录_停") is False
|
||||
assert is_synced_table(fm, "表壳焊接记录") is True
|
||||
|
||||
|
||||
def test_is_synced_table_include_restricts():
|
||||
fm = _fm(exclude_tables=["TableChangeLog"], include_tables=["检验合格记录表"])
|
||||
assert is_synced_table(fm, "检验合格记录表") is True
|
||||
assert is_synced_table(fm, "其它表") is False
|
||||
|
||||
|
||||
def test_is_synced_table_exclude_beats_include():
|
||||
fm = _fm(exclude_tables=["TableChangeLog"],
|
||||
include_tables=["TableChangeLog", "检验合格记录表"])
|
||||
assert is_synced_table(fm, "TableChangeLog") is False
|
||||
assert is_synced_table(fm, "检验合格记录表") is True
|
||||
|
||||
|
||||
def test_is_synced_table_no_filters_includes_all():
|
||||
fm = _fm()
|
||||
assert is_synced_table(fm, "任意表") is True
|
||||
|
||||
|
||||
def test_resolve_synced_tables_filters_user_tables():
|
||||
fm = _fm(exclude_tables=["TableChangeLog"])
|
||||
reader = MagicMock()
|
||||
reader.list_user_tables.return_value = [
|
||||
"TableChangeLog", "表壳焊接记录", "超压", "氩弧焊每日催货落实记录_停",
|
||||
]
|
||||
assert resolve_synced_tables(fm, reader) == [
|
||||
"表壳焊接记录", "超压", "氩弧焊每日催货落实记录_停",
|
||||
]
|
||||
|
||||
|
||||
def test_resolve_synced_tables_with_include():
|
||||
fm = _fm(exclude_tables=["TableChangeLog"], include_tables=["检验合格记录表"])
|
||||
reader = MagicMock()
|
||||
reader.list_user_tables.return_value = ["TableChangeLog", "检验合格记录表", "其它表"]
|
||||
assert resolve_synced_tables(fm, reader) == ["检验合格记录表"]
|
||||
Reference in New Issue
Block a user