7c6cb10b91
Add Pydantic config models (SqlServerConfig, AccessConfig, RuntimeConfig, FileMapping, SyncConfig) and a YAML loader (load_config). FileMapping provides source_path() and target_table() helpers; number-typed YAML keys/values (e.g. root: 2026) are coerced to str via coerce_numbers_to_str. Includes config.example.yaml template (config.yaml with real credentials stays gitignored) and pyproject.toml pytest config (pythonpath=src). Co-Authored-By: Claude <noreply@anthropic.com>
45 lines
1.7 KiB
Python
45 lines
1.7 KiB
Python
import pathlib, textwrap
|
|
from sync.config import load_config, FileMapping
|
|
|
|
def test_load_config_parses_fields(tmp_path):
|
|
cfg_text = textwrap.dedent("""
|
|
sql_server:
|
|
conn_str: "Driver={ODBC Driver 17 for SQL Server};Server=s;Database=d;UID=u;PWD=p;"
|
|
sync_queue_table: "dbo.SyncQueue"
|
|
access:
|
|
driver: "{Microsoft Access Driver (*.accdb, *.mdb)}"
|
|
roots:
|
|
2026: "\\\\\\\\srv\\\\2026"
|
|
2025: "\\\\\\\\srv\\\\2025"
|
|
runtime:
|
|
poll_interval_seconds: 10
|
|
capture_batch_size: 500
|
|
apply_batch_size: 200
|
|
max_retries: 5
|
|
retry_backoff_seconds: 30
|
|
cleanup_batch_size: 200
|
|
cleanup_lock_retries: 3
|
|
files:
|
|
- file: "氩弧焊.accdb"
|
|
root: 2026
|
|
schema: "TIGWelding"
|
|
year_suffix: "_YEAR2026"
|
|
exclude_tables: ["TableChangeLog", "氩弧焊每日催货落实记录_停"]
|
|
""")
|
|
p = tmp_path / "config.yaml"
|
|
p.write_text(cfg_text, encoding="utf-8")
|
|
cfg = load_config(str(p))
|
|
assert cfg.sql_server.sync_queue_table == "dbo.SyncQueue"
|
|
assert cfg.runtime.poll_interval_seconds == 10
|
|
assert cfg.files[0].schema == "TIGWelding"
|
|
assert cfg.files[0].year_suffix == "_YEAR2026"
|
|
assert "2026" in cfg.files[0].source_path(cfg)
|
|
|
|
def test_file_mapping_target_table_applies_year_suffix():
|
|
fm = FileMapping(file="x.accdb", root="2026", schema="s", year_suffix="_YEAR2026",
|
|
exclude_tables=[], include_tables=None)
|
|
assert fm.target_table("一车间记录") == "一车间记录_YEAR2026"
|
|
fm2 = FileMapping(file="y.accdb", root="2025", schema="s", year_suffix="",
|
|
exclude_tables=[], include_tables=None)
|
|
assert fm2.target_table("26年压力表合同数据") == "26年压力表合同数据"
|