54 lines
2.0 KiB
Python
54 lines
2.0 KiB
Python
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) == ["检验合格记录表"]
|