Compare commits
4 Commits
706b9c33db
...
7d11eddc7a
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
7d11eddc7a | ||
|
|
1b44ca28e4 | ||
|
|
e9f012eab5 | ||
|
|
5ca0715801 |
191
README.md
191
README.md
@@ -1,50 +1,199 @@
|
|||||||
# ProductionDataBaseSync_DataMacro
|
# ProductionDataBaseSync_DataMacro
|
||||||
|
|
||||||
Access → SQL Server 增量同步(数据宏驱动)。设计详见 `docs/superpowers/specs/2026-07-14-access-datamacro-sync-design.md`。
|
Access → SQL Server 单向增量同步(数据宏驱动)。把各 Access 库的业务数据周期性同步到 SQL Server 镜像表,作为 Access → SQL Server 迁移期的过渡数据层。
|
||||||
|
|
||||||
|
设计详见 `docs/superpowers/specs/2026-07-14-access-datamacro-sync-design.md`。
|
||||||
|
|
||||||
|
## 背景
|
||||||
|
|
||||||
|
生产数据实际承载在网络共享下的多个 Access `.accdb`(按车间/年份分库)。旧机制靠客户端前端 VBA 写变更日志,但 VBA 只在特定表单事件触发,批量改表、直接改表等路径会绕过 → 漏数据。
|
||||||
|
|
||||||
|
改用 Access **数据宏**(表级引擎触发器,任何写路径必触发):每张业务表挂 `After Insert/Update/Delete`,变更写入各库本地 `TableChangeLog`。本程序就是「读各库日志 → 增量同步到 SQL」的搬运器,理论上 100% 捕获、对客户端零侵入。
|
||||||
|
|
||||||
## 架构
|
## 架构
|
||||||
|
|
||||||
每个 Access 库通过数据宏把变更写入本地 `TableChangeLog`;同步服务周期性地把这些变更搬运到 SQL Server 镜像表。单库一轮分三段:
|
每个 Access 库通过数据宏把变更写入本地 `TableChangeLog`;同步服务周期性把这些变更搬运到 SQL Server 镜像表。单库一轮分三段:
|
||||||
|
|
||||||
| 阶段 | 动作 | 说明 |
|
| 阶段 | 动作 | 说明 |
|
||||||
| --- | --- | --- |
|
| --- | --- | --- |
|
||||||
| Capture | `SELECT` 读取 `TableChangeLog` 中最旧的 N 条 | 只读 Access |
|
| Capture | `SELECT` 读取 `TableChangeLog` 最旧的 N 条,I/U 按 `RecordID` 回读整行 | 只读 Access |
|
||||||
| Apply | 调用 `dbo.usp_SyncApply` 写入 SQL 镜像表 | 按 `ID` 精确落库 |
|
| Apply | 调用 `dbo.usp_SyncApply` 写入 SQL 镜像表 | 按 `ID` 精确落库,保序「最后操作胜」 |
|
||||||
| Cleanup | `DELETE` 已应用的 `TableChangeLog` 行 | 按 `ID` 列表精确删除,遇锁自动退避重试 |
|
| Cleanup | `DELETE` 已应用的 `TableChangeLog` 行 | 按 `ID` 列表精确删除,遇锁自动退避重试 |
|
||||||
|
|
||||||
## 命令
|
关键设计点:
|
||||||
|
- **无水位线表**——Access 日志「应用成功即删」,日志本身就是待处理队列;`dbo.SyncQueue` 的唯一索引 `(SourceFile, SourceTable, SourceLogID)` 兜底去重,重复捕获幂等。
|
||||||
|
- **保序「最后操作胜」**——同一 `RecordID` 多次操作(先 Insert 后 Delete 等)按日志顺序取最后一条,保证最终态与 Access 一致。
|
||||||
|
- **每表一个事务**——单表失败只回滚该表,失败行标 `error` 重试,超限标 `dead` 待人工。
|
||||||
|
- **`SyncQueue` 长期保留**作审计/重试日志;`applied` 行清理后标 `cleaned`,超保留期再 purge,控制表增长。
|
||||||
|
|
||||||
| 用途 | 命令 |
|
## 环境要求
|
||||||
| --- | --- |
|
|
||||||
| 增量同步服务(常驻,由 nssm 托管 `DataMacroSync`) | `.venv/Scripts/python.exe -m sync.service` |
|
|
||||||
| 一次性全量同步(TRUNCATE + 全量 INSERT,所有库所有表) | `.venv/Scripts/python.exe -m sync.fullsync config.yaml` |
|
|
||||||
|
|
||||||
> 上述命令均使用项目自带的 `.venv`。同步类命令由 nssm 以服务方式运行,无需手动设置环境变量。
|
- **Python 3.10+**(实测 3.13)。代码用 `X | None` 等新语法。
|
||||||
|
- **ODBC 驱动**(系统级,非 pip 安装,需预先装好):
|
||||||
|
- `Microsoft Access Driver (*.accdb, *.mdb)`(ACE Redist 2016)
|
||||||
|
- `ODBC Driver 17 for SQL Server`
|
||||||
|
- **SQL Server ≥ 2017**(存储过程用 `STRING_AGG ... WITHIN GROUP`)。
|
||||||
|
- 执行账号需对目标表有 `ALTER` 权限(`SET IDENTITY_INSERT` 要求)。
|
||||||
|
|
||||||
|
## 安装
|
||||||
|
|
||||||
|
```bash
|
||||||
|
python -m venv .venv
|
||||||
|
.venv/Scripts/python.exe -m pip install --upgrade pip
|
||||||
|
.venv/Scripts/python.exe -m pip install -r requirements.txt
|
||||||
|
```
|
||||||
|
|
||||||
|
依赖(`requirements.txt`):`pyodbc`、`PyYAML`、`pydantic`、`pytest`。
|
||||||
|
|
||||||
|
## SQL 端部署
|
||||||
|
|
||||||
|
首次需在目标库建好暂存表与 apply 存储过程(两个脚本都幂等,可重复执行):
|
||||||
|
|
||||||
|
```bash
|
||||||
|
sqlcmd -S <SERVER>,1433 -U <USER> -P <PASSWORD> -d <DB> -C -N o -i sql/01_sync_queue.sql
|
||||||
|
sqlcmd -S <SERVER>,1433 -U <USER> -P <PASSWORD> -d <DB> -C -N o -i sql/02_sync_apply.sql
|
||||||
|
```
|
||||||
|
|
||||||
|
- `sql/01_sync_queue.sql`:建 `dbo.SyncQueue` + 去重/清理索引 + `CleanedAt` 列。
|
||||||
|
- `sql/02_sync_apply.sql`:`dbo.usp_SyncApply` 集合化 apply 存储过程。
|
||||||
|
|
||||||
|
> 连接串/凭据以 `config.yaml` 为准,README 不硬编码。
|
||||||
|
|
||||||
## 配置
|
## 配置
|
||||||
|
|
||||||
编辑 `config.yaml`(从 `config.example.yaml` 复制并填入真实凭据)。关键段:
|
编辑 `config.yaml`(从 `config.example.yaml` 复制并填入真实凭据;该文件 gitignored)。关键段:
|
||||||
|
|
||||||
- `sql_server`:SQL Server 连接串与 `SyncQueue` 表名。
|
- **`sql_server`**:`conn_str`(ODBC 连接串)与 `sync_queue_table`(默认 `dbo.SyncQueue`)。
|
||||||
- `access`:ACE ODBC 驱动名与各根目录(`roots`)映射。
|
- **`access`**:`driver`(ACE 驱动名)与 `roots`(年份→根目录映射,如 `2026: "\\\\srv\\生产进度表\\2026年数据"`)。
|
||||||
- `runtime`:轮询间隔、批大小、重试与保留策略。
|
- **`runtime`**:`poll_interval_seconds`(轮询间隔)、`capture_batch_size`/`apply_batch_size`/`cleanup_batch_size`(各段批大小)、`max_retries`/`retry_backoff_seconds`(重试)、`cleanup_lock_retries`(Access 锁重试次数)、`cleaned_retention_hours`(`cleaned` 行保留多久后 purge)。
|
||||||
- `files`:每个 Access 文件一条映射(`file` / `root` / `schema` / `year_suffix` / `exclude_tables` / `include_tables`)。
|
- **`files`**:每个 Access 文件一条映射:
|
||||||
|
- `file` / `root`(对应 `access.roots` 的 key)/ `schema`(SQL 目标 schema)。
|
||||||
|
- `year_suffix`:拼到表名后(`2026年数据` 用 `_YEAR2026`,`2025年数据`/合同表用 `""`)。
|
||||||
|
- `exclude_tables` / `include_tables`:排除/包含规则,**exclude 优先于 include**。`TableChangeLog` 必须排除。
|
||||||
|
|
||||||
|
## 命令行
|
||||||
|
|
||||||
|
统一入口 `main.py`(仓库根目录),三个功能块都用它调用。配置固定读取同目录的 `config.yaml`,不在命令中指定:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
.venv/Scripts/python.exe main.py fullsync [--db FILE] [--table NAME] [--clear-change-log]
|
||||||
|
.venv/Scripts/python.exe main.py incremental [--loop] [--poll-interval N]
|
||||||
|
.venv/Scripts/python.exe main.py compare [--granularity count|ids] [--db FILE] [--table NAME] [--report PATH]
|
||||||
|
```
|
||||||
|
|
||||||
|
| 子命令 | 说明 | 退出码 |
|
||||||
|
| --- | --- | --- |
|
||||||
|
| `fullsync` | 一次性全量同步:TRUNCATE + 批量 INSERT,绕开增量队列。 | 0 |
|
||||||
|
| `incremental` | 增量同步一轮(capture→apply→cleanup)。`--loop` 切持续轮询(服务模式)。 | 0 |
|
||||||
|
| `compare` | 数据一致性核对:默认行数总量,`--granularity ids` 精确到 ID 集合差异。 | 全一致 0 / 有不一致 1 |
|
||||||
|
|
||||||
|
> 三个块共用同一份 `config.yaml`,目标表集合完全一致(由 `sync.targets` 统一解析)。`main.py` 在根目录、自行把 `src/` 加入 `sys.path`,无需 `-m`、无需设环境变量;控制台强制 UTF-8,中文表名不乱码。
|
||||||
|
>
|
||||||
|
> 旧入口 `-m sync.service` / `-m sync.fullsync` 保留为兼容,行为不变。
|
||||||
|
|
||||||
## 全量同步
|
## 全量同步
|
||||||
|
|
||||||
用于从零重建镜像表或修复 Access 与 SQL 之间的漂移。会**清空目标表再全量写入**,绕开增量队列:
|
用于从零重建镜像表或修复 Access 与 SQL 之间的漂移。会**清空目标表再全量写入**,绕开增量队列:
|
||||||
|
|
||||||
```bash
|
```bash
|
||||||
.venv/Scripts/python.exe -m sync.fullsync config.yaml # 全部库、全部表
|
.venv/Scripts/python.exe main.py fullsync # 全部库、全部表
|
||||||
.venv/Scripts/python.exe -m sync.fullsync config.yaml --db OEM.accdb # 仅单个库
|
.venv/Scripts/python.exe main.py fullsync --db OEM.accdb # 仅单个库
|
||||||
.venv/Scripts/python.exe -m sync.fullsync config.yaml --table 表壳焊接记录 # 仅单表(作用于所有库)
|
.venv/Scripts/python.exe main.py fullsync --table 表壳焊接记录 # 仅单表(作用于所有库)
|
||||||
.venv/Scripts/python.exe -m sync.fullsync config.yaml --clear-change-log # 同步后同时清空 TableChangeLog(谨慎)
|
.venv/Scripts/python.exe main.py fullsync --clear-change-log # 同步后同时清空 TableChangeLog(谨慎)
|
||||||
```
|
```
|
||||||
|
|
||||||
- `year_suffix` 通过 `FileMapping` 拼到表名后(如 `表壳焊接记录` → `表壳焊接记录_YEAR2026`)。
|
- `year_suffix` 通过 `FileMapping` 拼到表名后(如 `表壳焊接记录` → `表壳焊接记录_YEAR2026`)。
|
||||||
|
- 写入时 `SET IDENTITY_INSERT ON`,保留 Access 原 ID,保证后续增量的 `RecordID` 匹配不错位。
|
||||||
- 无镜像表的目标表按设计跳过(`target table missing`),不报错。
|
- 无镜像表的目标表按设计跳过(`target table missing`),不报错。
|
||||||
|
|
||||||
|
## 增量同步
|
||||||
|
|
||||||
|
以 Access 数据宏日志为唯一变更源,每轮跑一遍 capture → apply → cleanup(三段说明见上面「架构」)。这是**主用模式**,生产上常驻运行。两种调用方式:
|
||||||
|
|
||||||
|
```bash
|
||||||
|
.venv/Scripts/python.exe main.py incremental # 跑一轮就退出(手动/按需补跑)
|
||||||
|
.venv/Scripts/python.exe main.py incremental --loop # 持续轮询(服务模式,不退出)
|
||||||
|
.venv/Scripts/python.exe main.py incremental --loop --poll-interval 30 # 覆盖 runtime.poll_interval_seconds
|
||||||
|
```
|
||||||
|
|
||||||
|
- 生产环境以 nssm 服务 `DataMacroSync` 常驻(即 `--loop` 模式),见下文「NSSM 服务」;手动单轮适合验证或临时补跑积压。
|
||||||
|
- `--loop` 持续轮询直到进程被停(`nssm stop` 或 Ctrl+C);默认单轮跑完即退出。
|
||||||
|
- 单轮一次最多处理每库 `capture_batch_size` 条日志;积压多时连续跑几轮或用 `--loop` 直到清空。
|
||||||
|
- 每个文件的 capture/cleanup 独立隔离,单文件失败不影响其它;apply 失败不阻塞 cleanup;失败行 `error` 下轮重试、超 `max_retries` 标 `dead` 待人工。
|
||||||
|
- 幂等:`SyncQueue` 唯一索引去重,重复 capture、中断续跑都不会重写或漏写。
|
||||||
|
- 与全量同步共用同一份 `config.yaml` 和 `sync.targets`,目标表集合完全一致;全量是「从零重建」的补充手段,不替代增量。
|
||||||
|
|
||||||
|
## 数据对比
|
||||||
|
|
||||||
|
核对 Access 源表与 SQL 镜像表是否一致。两种粒度:
|
||||||
|
|
||||||
|
- **行数总量**(默认):逐表比对 `COUNT(*)`。
|
||||||
|
- **ID 集合**(`--granularity ids`):逐表比对两边 `ID` 集合,报告「Access 有 / SQL 无」与「SQL 有 / Access 无」的 ID(每表前 50 个 + 总数)。
|
||||||
|
|
||||||
|
```bash
|
||||||
|
.venv/Scripts/python.exe main.py compare # 全部库、全部表,行数总量
|
||||||
|
.venv/Scripts/python.exe main.py compare --db 氩弧焊.accdb # 仅单个库
|
||||||
|
.venv/Scripts/python.exe main.py compare --granularity ids --table 表壳焊接记录 # 单表 ID 级
|
||||||
|
.venv/Scripts/python.exe main.py compare --report report.txt # 同时写入报告文件(UTF-8)
|
||||||
|
```
|
||||||
|
|
||||||
|
- 无镜像表按设计跳过(`[SKIPPED no mirror]`),不计为不一致——这类表多半是该排除却没排除(如 `*_停` 停用表、`USysApplicationLog`),可作为配置清理的线索。
|
||||||
|
- 任一表不一致时退出码 `1`(便于脚本化);全部一致为 `0`。
|
||||||
|
- 实时增量同步存在秒级延迟窗口,刚写入 Access 的行可能尚未到 SQL,属正常(稍后再核或对照 `SyncQueue` 的 pending 行)。
|
||||||
|
|
||||||
|
## NSSM 服务(114)
|
||||||
|
|
||||||
|
增量同步在 host 114 上以 nssm 服务 `DataMacroSync` 常驻运行。常用操作(经 `ssh 114`):
|
||||||
|
|
||||||
|
```bash
|
||||||
|
ssh 114 "nssm status DataMacroSync" # 查状态(SERVICE_RUNNING / SERVICE_STOPPED)
|
||||||
|
ssh 114 "nssm stop DataMacroSync" # 停
|
||||||
|
ssh 114 "nssm start DataMacroSync" # 起
|
||||||
|
ssh 114 "nssm restart DataMacroSync" # 重启
|
||||||
|
ssh 114 "nssm list" # 列出所有 nssm 服务
|
||||||
|
```
|
||||||
|
|
||||||
|
## 测试
|
||||||
|
|
||||||
|
```bash
|
||||||
|
.venv/Scripts/python.exe -m pytest # 仅单元测试(默认)
|
||||||
|
RUN_INTEGRATION=1 .venv/Scripts/python.exe -m pytest # 含集成测试(需能连真实 Access + SQL Server)
|
||||||
|
```
|
||||||
|
|
||||||
|
- 单元测试用 mock,不依赖数据库;集成测试(`@pytest.mark.integration`)连 `config.yaml` 里的真实库,且自带清理。
|
||||||
|
- `pyproject.toml` 仅用于配置 pytest(`pythonpath = ["src", "."]`、`testpaths`、`integration` 标记)。
|
||||||
|
|
||||||
|
## 项目结构
|
||||||
|
|
||||||
|
```
|
||||||
|
main.py 统一命令行入口(fullsync / incremental / compare)
|
||||||
|
config.yaml 真实配置(gitignored);config.example.yaml 是模板
|
||||||
|
requirements.txt 依赖
|
||||||
|
pyproject.toml pytest 配置
|
||||||
|
sql/
|
||||||
|
01_sync_queue.sql dbo.SyncQueue 建表 + 索引(幂等)
|
||||||
|
02_sync_apply.sql dbo.usp_SyncApply 存储过程
|
||||||
|
src/sync/
|
||||||
|
config.py Pydantic 配置模型 + load_config
|
||||||
|
targets.py 共享目标表解析(exclude/include,全量/增量/对比共用)
|
||||||
|
serialize.py Access 值 → JSON 可序列化
|
||||||
|
access_reader.py 读 Access(日志/整行/计数/ID/删除日志)
|
||||||
|
sql_writer.py 写 SQL(SyncQueue/apply/计数/ID/全量灌表)
|
||||||
|
capture.py 增量编排:读日志→回读整行→入队
|
||||||
|
cleanup.py 清理编排:回删已应用日志
|
||||||
|
service.py 主循环 cycle() / run()
|
||||||
|
fullsync.py 一次性全量同步
|
||||||
|
compare.py 数据一致性对比(count / ids)
|
||||||
|
logging_setup.py 日志配置(滚动文件 + 控制台)
|
||||||
|
tests/ 单元 + 集成测试
|
||||||
|
docs/superpowers/ 设计文档与实现计划
|
||||||
|
```
|
||||||
|
|
||||||
## 常见问题
|
## 常见问题
|
||||||
|
|
||||||
- **cleanup 报 `-1102 无法更新;当前被锁定`**:Access 是文件型数据库,cleanup 反写 `DELETE` 与生产客户端数据宏写日志争用页级锁。服务已对锁冲突自动退避重试(`access_reader.delete_log_ids` 捕获 `pyodbc.Error` 并判断 `-1102`/“被锁定”)。偶发属正常,持续刷错再排查。
|
- **cleanup 报 `-1102 无法更新;当前被锁定`**:Access 是文件型数据库,cleanup 反写 `DELETE` 与生产客户端数据宏写日志争用页级锁。服务已对锁冲突自动退避重试(`access_reader.delete_log_ids` 捕获 `pyodbc.Error` 并判断 `-1102`/「被锁定」)。偶发属正常,持续刷错再排查。
|
||||||
- **`No module named 'pydantic_core'` / `pyodbc`**:venv 解释器与轮子 ABI 不匹配(常见于 Python 3.13 装到 cp310 轮子)。修复:` .venv/Scripts/python.exe -m pip install --force-reinstall --no-cache-dir pyodbc pydantic`。
|
- **`No module named 'pydantic_core'` / `pyodbc`**:venv 解释器与轮子 ABI 不匹配(常见于 Python 3.13 装到 cp310 轮子)。修复:`.venv/Scripts/python.exe -m pip install --force-reinstall --no-cache-dir pyodbc pydantic`。
|
||||||
|
- **compare/fullsync 报 `[SKIPPED no mirror]` / `target table missing`**:该表在 Access 里但 SQL 端没有镜像表(多为 `*_停` 停用表、`USysApplicationLog` 等系统表,或尚未建镜像的新表)。若是该停用的表,加进对应 `exclude_tables`;若该同步,先在 SQL 建表再 fullsync。
|
||||||
|
- **`SyncQueue` 出现 `error`/`dead` 行**:`error` 会在下轮自动重试(未超 `max_retries`);`dead` 是超限放弃,需人工看 `ErrorMsg` 排查后处理。
|
||||||
|
- **`UserWarning: Field name "schema" ... shadows ... BaseModel`**:`FileMapping.schema` 字段名与 Pydantic 基类属性重名,仅告警、不影响功能。
|
||||||
|
- **控制台中文乱码**:`main.py` 已强制 stdout/stderr 为 UTF-8;若仍乱码,设环境变量 `PYTHONIOENCODING=utf-8`,或用 `compare --report` 输出 UTF-8 文件。
|
||||||
|
|||||||
113
main.py
Normal file
113
main.py
Normal file
@@ -0,0 +1,113 @@
|
|||||||
|
"""Unified command-line entry point for the Access -> SQL Server sync toolkit.
|
||||||
|
|
||||||
|
Run from the project root (no ``-m`` needed)::
|
||||||
|
|
||||||
|
python main.py fullsync [--db FILE] [--table NAME] [--clear-change-log]
|
||||||
|
python main.py incremental [--loop] [--poll-interval N]
|
||||||
|
python main.py compare [--granularity count|ids] [--db FILE] [--table NAME] [--report PATH]
|
||||||
|
|
||||||
|
Configuration is hard-coded to ``config.yaml`` next to this script -- it is not
|
||||||
|
a command-line argument, so all three blocks always use the same config (and
|
||||||
|
therefore the same target tables).
|
||||||
|
|
||||||
|
This file lives at the repo root, outside the ``src/`` package, so it puts
|
||||||
|
``src`` on ``sys.path`` itself to import ``sync.*`` regardless of how Python
|
||||||
|
was launched or whether the venv already has ``src`` on its path.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
import argparse
|
||||||
|
import os
|
||||||
|
import sys
|
||||||
|
|
||||||
|
# Make the src/ package importable when running this root script directly.
|
||||||
|
sys.path.insert(0, os.path.join(os.path.dirname(os.path.abspath(__file__)), "src"))
|
||||||
|
|
||||||
|
from sync.config import load_config
|
||||||
|
from sync.logging_setup import setup_logging
|
||||||
|
from sync.fullsync import full_sync
|
||||||
|
from sync import service
|
||||||
|
from sync.compare import compare, any_mismatch, format_report
|
||||||
|
|
||||||
|
CONFIG_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "config.yaml")
|
||||||
|
|
||||||
|
|
||||||
|
def _parse_args(argv):
|
||||||
|
p = argparse.ArgumentParser(
|
||||||
|
prog="python main.py",
|
||||||
|
description="Access -> SQL Server sync toolkit (fullsync / incremental / compare).",
|
||||||
|
)
|
||||||
|
sub = p.add_subparsers(dest="command", required=True)
|
||||||
|
|
||||||
|
pf = sub.add_parser("fullsync", help="one-shot TRUNCATE + bulk INSERT")
|
||||||
|
pf.add_argument("--db", help="limit to one Access file (by FileMapping.file)")
|
||||||
|
pf.add_argument("--table", help="limit to one table (applies to all matched files)")
|
||||||
|
pf.add_argument("--clear-change-log", action="store_true",
|
||||||
|
help="after loading, clear TableChangeLog on the synced files")
|
||||||
|
|
||||||
|
pi = sub.add_parser("incremental", help="capture -> apply -> cleanup")
|
||||||
|
pi.add_argument("--loop", action="store_true",
|
||||||
|
help="run continuously (service mode); default is a single pass")
|
||||||
|
pi.add_argument("--poll-interval", type=int, dest="poll_interval",
|
||||||
|
help="override runtime.poll_interval_seconds (with --loop)")
|
||||||
|
|
||||||
|
pc = sub.add_parser("compare", help="compare Access vs SQL Server data")
|
||||||
|
pc.add_argument("--granularity", choices=["count", "ids"], default="count",
|
||||||
|
help="count = row totals (default); ids = ID-set membership diff")
|
||||||
|
pc.add_argument("--db", help="limit to one Access file")
|
||||||
|
pc.add_argument("--table", help="limit to one table")
|
||||||
|
pc.add_argument("--report", help="write the report to this file as well as stdout")
|
||||||
|
|
||||||
|
return p.parse_args(argv)
|
||||||
|
|
||||||
|
|
||||||
|
def _force_utf8_console():
|
||||||
|
"""Render Chinese table names correctly on a Windows GBK console.
|
||||||
|
|
||||||
|
compare prints to stdout by default; without this the default console
|
||||||
|
codepage mojibakes non-ASCII. No-op when stdout is already UTF-8 or when it
|
||||||
|
does not support reconfigure (e.g. some test-capture streams).
|
||||||
|
"""
|
||||||
|
for stream in (sys.stdout, sys.stderr):
|
||||||
|
try:
|
||||||
|
stream.reconfigure(encoding="utf-8", errors="replace")
|
||||||
|
except (AttributeError, ValueError):
|
||||||
|
pass
|
||||||
|
|
||||||
|
|
||||||
|
def main(argv=None) -> int:
|
||||||
|
"""Parse argv, load config, dispatch to the chosen block. Returns exit code."""
|
||||||
|
_force_utf8_console()
|
||||||
|
args = _parse_args(argv)
|
||||||
|
cfg = load_config(CONFIG_PATH)
|
||||||
|
setup_logging(cfg.logging)
|
||||||
|
|
||||||
|
if args.command == "fullsync":
|
||||||
|
full_sync(cfg, db_filter=args.db, table_filter=args.table,
|
||||||
|
clear_change_log=args.clear_change_log)
|
||||||
|
return 0
|
||||||
|
|
||||||
|
if args.command == "incremental":
|
||||||
|
if args.poll_interval is not None:
|
||||||
|
cfg.runtime.poll_interval_seconds = args.poll_interval
|
||||||
|
if args.loop:
|
||||||
|
service.run(cfg)
|
||||||
|
else:
|
||||||
|
service.cycle(cfg)
|
||||||
|
return 0
|
||||||
|
|
||||||
|
if args.command == "compare":
|
||||||
|
results = compare(cfg, granularity=args.granularity,
|
||||||
|
db_filter=args.db, table_filter=args.table)
|
||||||
|
report = format_report(results, args.granularity)
|
||||||
|
print(report)
|
||||||
|
if args.report:
|
||||||
|
with open(args.report, "w", encoding="utf-8") as f:
|
||||||
|
f.write(report + "\n")
|
||||||
|
return 1 if any_mismatch(results) else 0
|
||||||
|
|
||||||
|
return 2 # unreachable: argparse requires a subcommand
|
||||||
|
|
||||||
|
|
||||||
|
if __name__ == "__main__":
|
||||||
|
sys.exit(main())
|
||||||
@@ -1,4 +1,4 @@
|
|||||||
[tool.pytest.ini_options]
|
[tool.pytest.ini_options]
|
||||||
pythonpath = ["src"]
|
pythonpath = ["src", "."]
|
||||||
testpaths = ["tests"]
|
testpaths = ["tests"]
|
||||||
markers = ["integration: marks tests requiring real Access/SQL Server"]
|
markers = ["integration: marks tests requiring real Access/SQL Server"]
|
||||||
|
|||||||
@@ -163,6 +163,18 @@ class AccessReader:
|
|||||||
cur.execute("SELECT ID FROM TableChangeLog ORDER BY ID")
|
cur.execute("SELECT ID FROM TableChangeLog ORDER BY ID")
|
||||||
return [r[0] for r in cur.fetchall()]
|
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):
|
def close(self):
|
||||||
if self._conn:
|
if self._conn:
|
||||||
self._conn.close()
|
self._conn.close()
|
||||||
|
|||||||
@@ -3,18 +3,15 @@ import json, logging
|
|||||||
from .access_reader import AccessReader
|
from .access_reader import AccessReader
|
||||||
from .sql_writer import SqlWriter, QueueRow
|
from .sql_writer import SqlWriter, QueueRow
|
||||||
from .config import FileMapping, SyncConfig
|
from .config import FileMapping, SyncConfig
|
||||||
|
from .targets import is_synced_table
|
||||||
|
|
||||||
log = logging.getLogger(__name__)
|
log = logging.getLogger(__name__)
|
||||||
|
|
||||||
def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter, cfg: SyncConfig) -> int:
|
def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter, cfg: SyncConfig) -> int:
|
||||||
exclude = set(fm.exclude_tables or [])
|
|
||||||
include = set(fm.include_tables) if fm.include_tables else None
|
|
||||||
rows = reader.read_log(cfg.runtime.capture_batch_size)
|
rows = reader.read_log(cfg.runtime.capture_batch_size)
|
||||||
n = 0
|
n = 0
|
||||||
for lr in rows:
|
for lr in rows:
|
||||||
if lr.table_name in exclude:
|
if not is_synced_table(fm, lr.table_name):
|
||||||
continue
|
|
||||||
if include is not None and lr.table_name not in include:
|
|
||||||
continue
|
continue
|
||||||
op = lr.operate_type
|
op = lr.operate_type
|
||||||
row_data = None
|
row_data = None
|
||||||
|
|||||||
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)
|
||||||
@@ -22,6 +22,7 @@ from .config import load_config, FileMapping, SyncConfig
|
|||||||
from .access_reader import AccessReader
|
from .access_reader import AccessReader
|
||||||
from .sql_writer import SqlWriter
|
from .sql_writer import SqlWriter
|
||||||
from .logging_setup import setup_logging
|
from .logging_setup import setup_logging
|
||||||
|
from .targets import resolve_synced_tables
|
||||||
|
|
||||||
log = logging.getLogger("sync.fullsync")
|
log = logging.getLogger("sync.fullsync")
|
||||||
|
|
||||||
@@ -29,21 +30,11 @@ log = logging.getLogger("sync.fullsync")
|
|||||||
def resolve_tables(reader: AccessReader, fm: FileMapping) -> list[str]:
|
def resolve_tables(reader: AccessReader, fm: FileMapping) -> list[str]:
|
||||||
"""Tables to fully sync for one file, after exclude/include rules.
|
"""Tables to fully sync for one file, after exclude/include rules.
|
||||||
|
|
||||||
Mirrors the precedence used by ``capture.capture_file``: a table in
|
Thin wrapper over ``targets.resolve_synced_tables`` so fullsync, capture
|
||||||
``exclude_tables`` is dropped even if it also appears in ``include_tables``.
|
and compare share one resolution path. System tables (``MSys*`` / ``~*``)
|
||||||
System tables (``MSys*`` / ``~*``) are already filtered by
|
are already filtered by ``AccessReader.list_user_tables``.
|
||||||
``AccessReader.list_user_tables``.
|
|
||||||
"""
|
"""
|
||||||
exclude = set(fm.exclude_tables or [])
|
return resolve_synced_tables(fm, reader)
|
||||||
include = set(fm.include_tables) if fm.include_tables else None
|
|
||||||
out = []
|
|
||||||
for t in reader.list_user_tables():
|
|
||||||
if t in exclude:
|
|
||||||
continue
|
|
||||||
if include is not None and t not in include:
|
|
||||||
continue
|
|
||||||
out.append(t)
|
|
||||||
return out
|
|
||||||
|
|
||||||
|
|
||||||
def full_sync_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter) -> dict:
|
def full_sync_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter) -> dict:
|
||||||
|
|||||||
@@ -143,6 +143,18 @@ class SqlWriter:
|
|||||||
)
|
)
|
||||||
return cur.fetchone() is not None
|
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:
|
def _has_identity(self, schema: str, table: str) -> bool:
|
||||||
"""True if ``[schema].[table]`` has an IDENTITY column (the ``ID`` PK)."""
|
"""True if ``[schema].[table]`` has an IDENTITY column (the ``ID`` PK)."""
|
||||||
cur = self._conn.cursor()
|
cur = self._conn.cursor()
|
||||||
|
|||||||
37
src/sync/targets.py
Normal file
37
src/sync/targets.py
Normal file
@@ -0,0 +1,37 @@
|
|||||||
|
"""Shared target-table resolution for fullsync, incremental capture, and compare.
|
||||||
|
|
||||||
|
All three pipelines must agree on which Access tables are in scope and how each
|
||||||
|
maps to its SQL Server target. Centralising the exclude/include rule here makes
|
||||||
|
that guarantee structural instead of duplicated across three files.
|
||||||
|
|
||||||
|
- ``is_synced_table`` applies the per-file exclude/include rule (exclude wins).
|
||||||
|
- ``resolve_synced_tables`` returns the in-scope user tables in
|
||||||
|
``list_user_tables`` order; system tables (``MSys*`` / ``~*``) are already
|
||||||
|
filtered by ``AccessReader.list_user_tables``.
|
||||||
|
|
||||||
|
Target *naming* is shared via ``FileMapping.target_table`` (name + year_suffix)
|
||||||
|
and ``FileMapping.schema``, so a given Access table resolves to the same
|
||||||
|
``(schema, table)`` everywhere.
|
||||||
|
"""
|
||||||
|
from __future__ import annotations
|
||||||
|
|
||||||
|
from .config import FileMapping
|
||||||
|
|
||||||
|
|
||||||
|
def is_synced_table(fm: FileMapping, table_name: str) -> bool:
|
||||||
|
"""True if ``table_name`` is in sync scope for ``fm``.
|
||||||
|
|
||||||
|
``exclude_tables`` wins over ``include_tables``: a table listed in both is
|
||||||
|
excluded. When ``include_tables`` is None, every non-excluded table is in
|
||||||
|
scope.
|
||||||
|
"""
|
||||||
|
if table_name in (fm.exclude_tables or []):
|
||||||
|
return False
|
||||||
|
if fm.include_tables is not None:
|
||||||
|
return table_name in fm.include_tables
|
||||||
|
return True
|
||||||
|
|
||||||
|
|
||||||
|
def resolve_synced_tables(fm: FileMapping, reader) -> list[str]:
|
||||||
|
"""In-scope user tables for ``fm``, in ``list_user_tables`` order."""
|
||||||
|
return [t for t in reader.list_user_tables() if is_synced_table(fm, t)]
|
||||||
@@ -6,6 +6,7 @@ No real log rows are deleted.
|
|||||||
"""
|
"""
|
||||||
import os
|
import os
|
||||||
import pytest
|
import pytest
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
from sync.access_reader import AccessReader
|
from sync.access_reader import AccessReader
|
||||||
from sync.config import load_config
|
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
|
r.delete_log_ids([], 100, 3) # empty list -> no-op, must not raise
|
||||||
finally:
|
finally:
|
||||||
r.close()
|
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
|
||||||
70
tests/test_main.py
Normal file
70
tests/test_main.py
Normal file
@@ -0,0 +1,70 @@
|
|||||||
|
"""Routing/argparse tests for the root main.py dispatcher.
|
||||||
|
|
||||||
|
The backend functions (full_sync, service.cycle/run, compare) are mocked so
|
||||||
|
these tests verify command dispatch, arg parsing, report output and exit codes
|
||||||
|
without touching Access or SQL Server.
|
||||||
|
"""
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
|
import main
|
||||||
|
from sync.compare import TableResult
|
||||||
|
|
||||||
|
|
||||||
|
def test_fullsync_routes_with_filters():
|
||||||
|
with patch("main.load_config"), patch("main.setup_logging"), \
|
||||||
|
patch("main.full_sync") as fs, patch("main.service") as svc:
|
||||||
|
rc = main.main(["fullsync", "--db", "OEM.accdb", "--table", "表壳焊接记录",
|
||||||
|
"--clear-change-log"])
|
||||||
|
fs.assert_called_once()
|
||||||
|
assert fs.call_args.kwargs["db_filter"] == "OEM.accdb"
|
||||||
|
assert fs.call_args.kwargs["table_filter"] == "表壳焊接记录"
|
||||||
|
assert fs.call_args.kwargs["clear_change_log"] is True
|
||||||
|
svc.cycle.assert_not_called()
|
||||||
|
assert rc == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_incremental_default_runs_single_cycle():
|
||||||
|
with patch("main.load_config"), patch("main.setup_logging"), \
|
||||||
|
patch("main.full_sync"), patch("main.service") as svc:
|
||||||
|
rc = main.main(["incremental"])
|
||||||
|
svc.cycle.assert_called_once()
|
||||||
|
svc.run.assert_not_called()
|
||||||
|
assert rc == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_incremental_loop_runs_service():
|
||||||
|
with patch("main.load_config"), patch("main.setup_logging"), \
|
||||||
|
patch("main.full_sync"), patch("main.service") as svc:
|
||||||
|
rc = main.main(["incremental", "--loop", "--poll-interval", "5"])
|
||||||
|
svc.run.assert_called_once()
|
||||||
|
svc.cycle.assert_not_called()
|
||||||
|
assert rc == 0
|
||||||
|
|
||||||
|
|
||||||
|
def test_compare_default_granularity_is_count(capsys):
|
||||||
|
with patch("main.load_config"), patch("main.setup_logging"), \
|
||||||
|
patch("main.full_sync"), patch("main.service"), patch("main.compare") as cmp:
|
||||||
|
cmp.return_value = [TableResult("x.accdb", "T", "s", "T_YEAR2026", "match", 5, 5)]
|
||||||
|
rc = main.main(["compare"])
|
||||||
|
assert cmp.call_args.kwargs["granularity"] == "count"
|
||||||
|
assert rc == 0
|
||||||
|
assert "x.accdb: T -> s.T_YEAR2026" in capsys.readouterr().out
|
||||||
|
|
||||||
|
|
||||||
|
def test_compare_ids_granularity_and_mismatch_exit_code():
|
||||||
|
with patch("main.load_config"), patch("main.setup_logging"), \
|
||||||
|
patch("main.full_sync"), patch("main.service"), patch("main.compare") as cmp:
|
||||||
|
cmp.return_value = [TableResult("x.accdb", "T", "s", "T_YEAR2026", "mismatch", 5, 7)]
|
||||||
|
rc = main.main(["compare", "--granularity", "ids"])
|
||||||
|
assert cmp.call_args.kwargs["granularity"] == "ids"
|
||||||
|
assert rc == 1
|
||||||
|
|
||||||
|
|
||||||
|
def test_compare_writes_report_file(tmp_path):
|
||||||
|
with patch("main.load_config"), patch("main.setup_logging"), \
|
||||||
|
patch("main.full_sync"), patch("main.service"), patch("main.compare") as cmp:
|
||||||
|
cmp.return_value = [TableResult("x.accdb", "T", "s", "T_YEAR2026", "match", 1, 1)]
|
||||||
|
rep = tmp_path / "report.txt"
|
||||||
|
rc = main.main(["compare", "--report", str(rep)])
|
||||||
|
assert rc == 0
|
||||||
|
assert "x.accdb: T -> s.T_YEAR2026" in rep.read_text(encoding="utf-8")
|
||||||
@@ -10,6 +10,7 @@ credentials are hardcoded here. The test self-cleans using a throwaway
|
|||||||
"""
|
"""
|
||||||
import os
|
import os
|
||||||
import pytest
|
import pytest
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
from sync.sql_writer import SqlWriter, QueueRow
|
from sync.sql_writer import SqlWriter, QueueRow
|
||||||
from sync.config import load_config
|
from sync.config import load_config
|
||||||
@@ -53,3 +54,30 @@ def test_insert_dedup_and_applied_ids():
|
|||||||
finally:
|
finally:
|
||||||
cur.execute("DELETE dbo.SyncQueue WHERE SourceFile='sqlw_test.accdb'")
|
cur.execute("DELETE dbo.SyncQueue WHERE SourceFile='sqlw_test.accdb'")
|
||||||
w.close()
|
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")
|
||||||
|
|||||||
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