diff --git a/sql/00_schema.sql b/sql/00_schema.sql new file mode 100644 index 0000000..f033d15 --- /dev/null +++ b/sql/00_schema.sql @@ -0,0 +1,12 @@ +-- ProductionDataBaseSync: dedicated schema that groups every object owned by +-- the Access -> SQL Server sync (SyncQueue staging, SyncLogArchive audit store, +-- and the usp_SyncApply procedure). Keeping them out of dbo makes ownership and +-- housekeeping explicit. +-- Idempotent: CREATE SCHEMA must be the only statement in its batch, so it is +-- wrapped in EXEC() behind an existence check. + +IF NOT EXISTS (SELECT 1 FROM sys.schemas WHERE name = 'ProductionDataBaseSync') +BEGIN + EXEC('CREATE SCHEMA ProductionDataBaseSync'); +END +GO diff --git a/sql/01_sync_queue.sql b/sql/01_sync_queue.sql index f923636..f6a132f 100644 --- a/sql/01_sync_queue.sql +++ b/sql/01_sync_queue.sql @@ -1,9 +1,10 @@ -- SyncQueue: staging table for the Access -> SQL Server one-way sync. +-- Lives under the ProductionDataBaseSync schema (run 00_schema.sql first). -- Idempotent: safe to re-run (table created only if absent; indexes only if absent). -IF OBJECT_ID('dbo.SyncQueue', 'U') IS NULL +IF OBJECT_ID('ProductionDataBaseSync.SyncQueue', 'U') IS NULL BEGIN - CREATE TABLE dbo.SyncQueue ( + CREATE TABLE ProductionDataBaseSync.SyncQueue ( QueueID bigint IDENTITY(1,1) NOT NULL, SourceFile nvarchar(255) NOT NULL, SourceTable nvarchar(255) NOT NULL, @@ -18,6 +19,7 @@ BEGIN ErrorMsg nvarchar(max) NULL, CapturedAt datetime2 NOT NULL CONSTRAINT DF_SyncQueue_Captured DEFAULT sysdatetime(), AppliedAt datetime2 NULL, + CleanedAt datetime2 NULL, CONSTRAINT PK_SyncQueue PRIMARY KEY CLUSTERED (QueueID) ); END @@ -25,38 +27,27 @@ GO IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = 'UX_SyncQueue_Dedup' - AND object_id = OBJECT_ID('dbo.SyncQueue')) + AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncQueue')) BEGIN CREATE UNIQUE INDEX UX_SyncQueue_Dedup - ON dbo.SyncQueue(SourceFile, SourceTable, SourceLogID); + ON ProductionDataBaseSync.SyncQueue(SourceFile, SourceTable, SourceLogID); END GO IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = 'IX_SyncQueue_Pending' - AND object_id = OBJECT_ID('dbo.SyncQueue')) + AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncQueue')) BEGIN CREATE INDEX IX_SyncQueue_Pending - ON dbo.SyncQueue(Status, TargetSchema, TargetTable); -END -GO - --- Cleanup bookkeeping: track when a queue row's Access log counterpart has --- been physically deleted, so cleanup never re-deletes the same IDs and the --- table can be purged to bound its growth. -IF NOT EXISTS (SELECT 1 FROM sys.columns - WHERE object_id = OBJECT_ID('dbo.SyncQueue') - AND name = 'CleanedAt') -BEGIN - ALTER TABLE dbo.SyncQueue ADD CleanedAt datetime2 NULL; + ON ProductionDataBaseSync.SyncQueue(Status, TargetSchema, TargetTable); END GO IF NOT EXISTS (SELECT 1 FROM sys.indexes WHERE name = 'IX_SyncQueue_Cleaned' - AND object_id = OBJECT_ID('dbo.SyncQueue')) + AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncQueue')) BEGIN CREATE INDEX IX_SyncQueue_Cleaned - ON dbo.SyncQueue(Status, CleanedAt); + ON ProductionDataBaseSync.SyncQueue(Status, CleanedAt); END GO diff --git a/sql/02_sync_apply.sql b/sql/02_sync_apply.sql index 1d332cf..a23cf2c 100644 --- a/sql/02_sync_apply.sql +++ b/sql/02_sync_apply.sql @@ -11,7 +11,7 @@ -- and silently drop a row whose true last op was an Insert. Routing off the one -- winning row's OperateType guarantees only the genuine last op wins. -CREATE OR ALTER PROCEDURE dbo.usp_SyncApply +CREATE OR ALTER PROCEDURE ProductionDataBaseSync.usp_SyncApply @MaxRetries INT = 5 AS BEGIN @@ -21,13 +21,13 @@ BEGIN DECLARE @sql NVARCHAR(MAX), @cols NVARCHAR(MAX), @upd NVARCHAR(MAX), @ins NVARCHAR(MAX); -- Re-queue rows still under the retry budget. - UPDATE dbo.SyncQueue + UPDATE ProductionDataBaseSync.SyncQueue SET Status = 'pending' WHERE Status = 'error' AND RetryCount < @MaxRetries; DECLARE cur CURSOR LOCAL FAST_FORWARD FOR SELECT DISTINCT TargetSchema, TargetTable - FROM dbo.SyncQueue + FROM ProductionDataBaseSync.SyncQueue WHERE Status = 'pending'; OPEN cur; @@ -70,7 +70,7 @@ BEGIN + N';WITH ranked AS (' + N' SELECT RecordID, OperateType, RowData,' + N' ROW_NUMBER() OVER (PARTITION BY RecordID ORDER BY SourceLogID DESC) rn' - + N' FROM dbo.SyncQueue' + + N' FROM ProductionDataBaseSync.SyncQueue' + N' WHERE TargetSchema=@sch AND TargetTable=@tbl AND Status=''pending''' + N')' + N'MERGE ' + @FullName + N' WITH (HOLDLOCK) AS tgt' @@ -90,7 +90,7 @@ BEGIN SET @sql = N';WITH ranked AS (' + N' SELECT RecordID, OperateType,' + N' ROW_NUMBER() OVER (PARTITION BY RecordID ORDER BY SourceLogID DESC) rn' - + N' FROM dbo.SyncQueue' + + N' FROM ProductionDataBaseSync.SyncQueue' + N' WHERE TargetSchema=@sch AND TargetTable=@tbl AND Status=''pending''' + N')' + N'DELETE t FROM ' + @FullName + N' t' @@ -100,7 +100,7 @@ BEGIN N'@sch NVARCHAR(128),@tbl NVARCHAR(255)', @sch, @tbl; END - UPDATE dbo.SyncQueue + UPDATE ProductionDataBaseSync.SyncQueue SET Status = 'applied', AppliedAt = SYSDATETIME() WHERE TargetSchema = @sch AND TargetTable = @tbl AND Status = 'pending'; @@ -113,7 +113,7 @@ BEGIN -- issues its own rollback. We roll back here so the partial per-table -- work is discarded before marking rows as 'error'/'dead'. IF @@TRANCOUNT > 0 ROLLBACK; - UPDATE dbo.SyncQueue + UPDATE ProductionDataBaseSync.SyncQueue SET Status = CASE WHEN RetryCount + 1 >= @MaxRetries THEN 'dead' ELSE 'error' END, RetryCount = RetryCount + 1, ErrorMsg = ERROR_MESSAGE() diff --git a/sql/03_sync_log_archive.sql b/sql/03_sync_log_archive.sql new file mode 100644 index 0000000..72aaee6 --- /dev/null +++ b/sql/03_sync_log_archive.sql @@ -0,0 +1,71 @@ +-- SyncLogArchive: permanent, append-only audit store for every Access change-log +-- row consumed by the incremental sync. This is the durable evidence layer that +-- SyncQueue is NOT: SyncQueue is a transient work queue (purged 24h after a row +-- is cleaned) and it only stores the *processed* OperateType, so a downgrade +-- (e.g. capture turning an Insert into a Delete when the row is momentarily +-- unreadable) erases the original intent. This table preserves both the ORIGINAL +-- operate type recorded by the Access data macro and the PROCESSED type actually +-- sent to SQL Server, plus the original Access log timestamp and the row payload. +-- +-- With this in place, cases like the 14287 incident (a real Insert applied to SQL +-- as a Delete) stay fully reconstructible: OriginalOperateType != ProcessedOperateType +-- flags exactly where the pipeline diverged from the source. +-- +-- Retention: PERMANENT. No purge job touches this table. SyncQueue keeps its +-- short-lived queue role; this table keeps history. Lives under the +-- ProductionDataBaseSync schema (run 00_schema.sql first). +-- Idempotent: safe to re-run. + +IF OBJECT_ID('ProductionDataBaseSync.SyncLogArchive', 'U') IS NULL +BEGIN + CREATE TABLE ProductionDataBaseSync.SyncLogArchive ( + ArchiveID bigint IDENTITY(1,1) NOT NULL, + SourceFile nvarchar(255) NOT NULL, + SourceTable nvarchar(255) NOT NULL, + SourceLogID bigint NOT NULL, + RecordID nvarchar(50) NOT NULL, + TargetSchema nvarchar(128) NOT NULL, + TargetTable nvarchar(255) NOT NULL, + OriginalOperateType varchar(10) NOT NULL, -- as recorded by the Access data macro + ProcessedOperateType varchar(10) NOT NULL, -- as actually sent to SyncQueue / SQL + RowData nvarchar(max) NULL, -- captured row payload (NULL for Delete) + OriginalTime datetime2 NULL, -- Access TableChangeLog.Time (previously discarded) + CapturedAt datetime2 NOT NULL + CONSTRAINT DF_SyncLogArchive_Captured DEFAULT sysdatetime(), + CONSTRAINT PK_SyncLogArchive PRIMARY KEY CLUSTERED (ArchiveID) + ); +END +GO + +-- Dedup: one archive row per source log entry. Capture may re-run the same log +-- row if a prior cycle's apply failed (the Access log is only deleted after a +-- successful apply), so the write path uses IF NOT EXISTS on these keys. +IF NOT EXISTS (SELECT 1 FROM sys.indexes + WHERE name = 'UX_SyncLogArchive_Dedup' + AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncLogArchive')) +BEGIN + CREATE UNIQUE INDEX UX_SyncLogArchive_Dedup + ON ProductionDataBaseSync.SyncLogArchive(SourceFile, SourceTable, SourceLogID); +END +GO + +-- Evidence lookup by table + record (e.g. "show every log ever seen for ID 14287"). +IF NOT EXISTS (SELECT 1 FROM sys.indexes + WHERE name = 'IX_SyncLogArchive_Record' + AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncLogArchive')) +BEGIN + CREATE INDEX IX_SyncLogArchive_Record + ON ProductionDataBaseSync.SyncLogArchive(SourceTable, RecordID); +END +GO + +-- Fast filter for the anomaly the archive exists to catch: original != processed. +IF NOT EXISTS (SELECT 1 FROM sys.indexes + WHERE name = 'IX_SyncLogArchive_Divergence' + AND object_id = OBJECT_ID('ProductionDataBaseSync.SyncLogArchive')) +BEGIN + CREATE INDEX IX_SyncLogArchive_Divergence + ON ProductionDataBaseSync.SyncLogArchive(OriginalOperateType, ProcessedOperateType) + INCLUDE (SourceFile, SourceTable, RecordID, CapturedAt); +END +GO diff --git a/src/sync/capture.py b/src/sync/capture.py index cda759a..730f594 100644 --- a/src/sync/capture.py +++ b/src/sync/capture.py @@ -1,7 +1,7 @@ from __future__ import annotations import json, logging from .access_reader import AccessReader -from .sql_writer import SqlWriter, QueueRow +from .sql_writer import SqlWriter, QueueRow, ArchiveRow from .config import FileMapping, SyncConfig from .targets import is_synced_table @@ -24,12 +24,30 @@ def capture_file(fm: FileMapping, reader: AccessReader, writer: SqlWriter, cfg: elif op != "Delete": log.warning("unknown OperateType %r in %s log %s", op, fm.file, lr.id) continue + target_schema = fm.schema + target_table = fm.target_table(lr.table_name) + # Persist the ORIGINAL log entry to the permanent audit store BEFORE the + # queue insert (and long before cleanup deletes the Access log). This + # keeps both the source operate type (lr.operate_type) and the processed + # one (op) so a downgrade like Insert->Delete stays reconstructible. + writer.insert_archive_row(ArchiveRow( + source_file=fm.file, + source_table=lr.table_name, + source_log_id=lr.id, + record_id=lr.record_id, + target_schema=target_schema, + target_table=target_table, + original_operate_type=lr.operate_type, + processed_operate_type=op, + row_data=row_data, + original_time=lr.time, + )) qr = QueueRow( source_file=fm.file, source_table=lr.table_name, record_id=lr.record_id, - target_schema=fm.schema, - target_table=fm.target_table(lr.table_name), + target_schema=target_schema, + target_table=target_table, source_log_id=lr.id, operate_type=op, row_data=row_data, diff --git a/src/sync/cleanup.py b/src/sync/cleanup.py index 3448134..7864106 100644 --- a/src/sync/cleanup.py +++ b/src/sync/cleanup.py @@ -1,6 +1,6 @@ """Cleanup phase: delete applied Access log rows. -After ``dbo.usp_SyncApply`` flips queue rows to ``applied``, those rows' +After ``usp_SyncApply`` flips queue rows to ``applied``, those rows' ``SourceLogID`` values are no longer needed on the Access side. This module asks the writer which log IDs have been applied for a given source file and deletes them from ``TableChangeLog`` via the reader, in batches with lock diff --git a/src/sync/config.py b/src/sync/config.py index 71e8562..51025d7 100644 --- a/src/sync/config.py +++ b/src/sync/config.py @@ -5,7 +5,9 @@ from pydantic import BaseModel, ConfigDict, Field class SqlServerConfig(BaseModel): conn_str: str - sync_queue_table: str = "dbo.SyncQueue" + sync_queue_table: str = "ProductionDataBaseSync.SyncQueue" + archive_table: str = "ProductionDataBaseSync.SyncLogArchive" + apply_proc: str = "ProductionDataBaseSync.usp_SyncApply" class AccessConfig(BaseModel): model_config = ConfigDict(coerce_numbers_to_str=True) diff --git a/src/sync/service.py b/src/sync/service.py index 270996a..9e8746a 100644 --- a/src/sync/service.py +++ b/src/sync/service.py @@ -2,7 +2,7 @@ ``cycle(cfg)`` runs one full pass over every configured file: 1. capture — read each file's change log and stage rows into SyncQueue; - 2. apply — drain the queue via ``dbo.usp_SyncApply``; + 2. apply — drain the queue via ``usp_SyncApply``; 3. cleanup — delete applied log rows from each file's Access log. Each file's capture and cleanup is wrapped in its own try/except so one @@ -45,7 +45,12 @@ def cycle(cfg): Apply failure does not block cleanup. The writer is always closed in a ``finally``. Safe to call directly from tests (does not sleep or loop). """ - writer = SqlWriter(cfg.sql_server.conn_str, cfg.sql_server.sync_queue_table) + writer = SqlWriter( + cfg.sql_server.conn_str, + cfg.sql_server.sync_queue_table, + cfg.sql_server.archive_table, + cfg.sql_server.apply_proc, + ) try: total_captured = 0 for fm in cfg.files: diff --git a/src/sync/sql_writer.py b/src/sync/sql_writer.py index 833deeb..58e5d36 100644 --- a/src/sync/sql_writer.py +++ b/src/sync/sql_writer.py @@ -1,8 +1,12 @@ """SQL-side writer for the Access -> SQL Server sync. -SqlWriter owns the pyodbc connection used to (a) dedup-insert rows into -``dbo.SyncQueue``, (b) invoke ``dbo.usp_SyncApply`` to drain the queue, and -(c) report which SourceLogIDs have been applied. +SqlWriter owns the pyodbc connection used to (a) dedup-insert rows into the +staging queue (``ProductionDataBaseSync.SyncQueue`` by default), (b) invoke the +apply proc (``ProductionDataBaseSync.usp_SyncApply``) to drain the queue, +(c) report which SourceLogIDs have been applied, and (d) append every consumed +Access change-log row to the permanent audit store +(``ProductionDataBaseSync.SyncLogArchive``). The queue/archive/proc names are +injected from config so the whole sync can live under a dedicated schema. The connection is opened with ``autocommit=True`` on purpose. ``usp_SyncApply`` manages its own transaction internally (BEGIN TRAN ... ROLLBACK on error); if @@ -20,7 +24,7 @@ import pyodbc @dataclass class QueueRow: - """A single staged change to enqueue into ``dbo.SyncQueue``.""" + """A single staged change to enqueue into the SyncQueue table.""" source_file: str source_table: str @@ -32,12 +36,43 @@ class QueueRow: row_data: str | None -class SqlWriter: - """Writes to SyncQueue and drives the apply proc over a pyodbc connection.""" +@dataclass +class ArchiveRow: + """A single Access change-log row to append to ``SyncLogArchive``. - def __init__(self, conn_str: str, queue_table: str = "dbo.SyncQueue"): + Captures both the ORIGINAL operate type recorded by the Access data macro + and the PROCESSED operate type actually sent to the queue, so a downgrade + (e.g. Insert -> Delete when the row is momentarily unreadable) stays visible + forever. ``original_time`` preserves the Access log's own timestamp, which + the queue path discards. + """ + + source_file: str + source_table: str + source_log_id: int + record_id: str + target_schema: str + target_table: str + original_operate_type: str + processed_operate_type: str + row_data: str | None + original_time: object + + +class SqlWriter: + """Writes to SyncQueue/SyncLogArchive and drives the apply proc.""" + + def __init__( + self, + conn_str: str, + queue_table: str = "ProductionDataBaseSync.SyncQueue", + archive_table: str = "ProductionDataBaseSync.SyncLogArchive", + apply_proc: str = "ProductionDataBaseSync.usp_SyncApply", + ): self.conn_str = conn_str self.queue_table = queue_table + self.archive_table = archive_table + self.apply_proc = apply_proc # autocommit=True: usp_SyncApply manages its own transaction internally. # An outer pyodbc transaction would conflict on ROLLBACK (SQL error 266). self._conn = pyodbc.connect(conn_str, autocommit=True) @@ -53,9 +88,9 @@ class SqlWriter: # twice (once for the EXISTS check, once for the INSERT). cur = self._conn.cursor() cur.execute( - "IF NOT EXISTS (SELECT 1 FROM dbo.SyncQueue " + f"IF NOT EXISTS (SELECT 1 FROM {self.queue_table} " "WHERE SourceFile=? AND SourceTable=? AND SourceLogID=?) " - "INSERT dbo.SyncQueue(SourceFile,SourceTable,SourceLogID," + f"INSERT {self.queue_table}(SourceFile,SourceTable,SourceLogID," "TargetSchema,TargetTable,RecordID,OperateType,RowData,Status) " "VALUES (?,?,?,?,?,?,?,?, 'pending')", row.source_file, @@ -72,19 +107,52 @@ class SqlWriter: ) # autocommit: statement already committed. + def insert_archive_row(self, row: ArchiveRow) -> None: + """Append ``row`` to the permanent audit store (dedup on source keys). + + Written during capture, BEFORE cleanup deletes the Access log, so the + original evidence survives even after the queue row is purged. Stores + both the original and processed operate types plus the Access log time. + ``IF NOT EXISTS`` keeps the first archive record if capture re-runs the + same log id (a prior cycle's apply failed and the Access log persisted). + """ + cur = self._conn.cursor() + cur.execute( + f"IF NOT EXISTS (SELECT 1 FROM {self.archive_table} " + "WHERE SourceFile=? AND SourceTable=? AND SourceLogID=?) " + f"INSERT {self.archive_table}(SourceFile,SourceTable,SourceLogID," + "RecordID,TargetSchema,TargetTable,OriginalOperateType," + "ProcessedOperateType,RowData,OriginalTime) " + "VALUES (?,?,?,?,?,?,?,?,?,?)", + row.source_file, + row.source_table, + row.source_log_id, + row.source_file, + row.source_table, + row.source_log_id, + row.record_id, + row.target_schema, + row.target_table, + row.original_operate_type, + row.processed_operate_type, + row.row_data, + row.original_time, + ) + # autocommit: statement already committed. + def call_apply(self, max_retries: int) -> None: """Drain the pending queue via the stored procedure. ``usp_SyncApply`` flips rows to ``applied`` (or ``error`` after retries). """ cur = self._conn.cursor() - cur.execute("EXEC dbo.usp_SyncApply ?", max_retries) + cur.execute(f"EXEC {self.apply_proc} ?", max_retries) def applied_log_ids(self, source_file: str) -> list[int]: """Return applied SourceLogIDs for ``source_file`` in ascending order.""" cur = self._conn.cursor() cur.execute( - "SELECT SourceLogID FROM dbo.SyncQueue " + f"SELECT SourceLogID FROM {self.queue_table} " "WHERE SourceFile=? AND Status='applied' ORDER BY SourceLogID", source_file, ) @@ -107,7 +175,7 @@ class SqlWriter: chunk = source_log_ids[i:i + 1000] placeholders = ",".join("?" * len(chunk)) cur.execute( - f"UPDATE dbo.SyncQueue SET Status='cleaned', " + f"UPDATE {self.queue_table} SET Status='cleaned', " f"CleanedAt=sysdatetime() " f"WHERE SourceFile=? AND Status='applied' " f"AND SourceLogID IN ({placeholders})", @@ -120,12 +188,12 @@ class SqlWriter: Keeps the table bounded: cleanup marks rows ``cleaned`` every cycle, and this removes the old ones after a short audit/debug window so - ``dbo.SyncQueue`` stops growing without bound. Returns the number of + the queue table stops growing without bound. Returns the number of rows removed. """ cur = self._conn.cursor() cur.execute( - "DELETE FROM dbo.SyncQueue " + f"DELETE FROM {self.queue_table} " "WHERE Status='cleaned' " "AND CleanedAt < DATEADD(hour, -?, GETDATE())", retention_hours, diff --git a/tests/test_apply_proc.py b/tests/test_apply_proc.py index 30a786c..ebfe7ac 100644 --- a/tests/test_apply_proc.py +++ b/tests/test_apply_proc.py @@ -1,15 +1,23 @@ -"""Integration test for dbo.usp_SyncApply. +"""Integration test for usp_SyncApply. Validates: IDENTITY-preserving INSERT, last-write-wins UPDATE, BIT conversion from JSON ``true``/``false``, and DELETE of the last op. Creates a throwaway schema ``sync_test`` and table ``ApplyDemo_YEAR2026`` and cleans them up at the end so no residue is left on CompanyDB. + +Queue table and apply proc names are read from ``config.yaml`` so the test +follows whichever schema the deployment targets (currently ProductionDataBaseSync). """ import pytest +from sync.config import load_config + @pytest.mark.integration def test_upsert_insert_update_delete_last_write_wins(sql_conn): + cfg = load_config("config.yaml") + qt = cfg.sql_server.sync_queue_table + proc = cfg.sql_server.apply_proc cur = sql_conn.cursor() sch, tbl = "sync_test", "ApplyDemo_YEAR2026" @@ -27,17 +35,17 @@ def test_upsert_insert_update_delete_last_write_wins(sql_conn): "时间 DATETIME2 NULL, " "标记 BIT NULL)" ) - cur.execute("DELETE dbo.SyncQueue WHERE TargetSchema='sync_test'") + cur.execute("DELETE " + qt + " WHERE TargetSchema='sync_test'") # Insert then a later Update for the same RecordID=1 -> last write wins. cur.execute( - "INSERT dbo.SyncQueue(SourceFile,SourceTable,SourceLogID,TargetSchema," + "INSERT " + qt + "(SourceFile,SourceTable,SourceLogID,TargetSchema," "TargetTable,RecordID,OperateType,RowData,Status) " "VALUES('t.accdb','ApplyDemo',1,'sync_test','ApplyDemo_YEAR2026','1'," "'Insert','{\"ID\":1,\"名字\":\"A\",\"数量\":3,\"时间\":\"2026-01-01T00:00:00\",\"标记\":true}','pending')" ) cur.execute( - "INSERT dbo.SyncQueue(SourceFile,SourceTable,SourceLogID,TargetSchema," + "INSERT " + qt + "(SourceFile,SourceTable,SourceLogID,TargetSchema," "TargetTable,RecordID,OperateType,RowData,Status) " "VALUES('t.accdb','ApplyDemo',2,'sync_test','ApplyDemo_YEAR2026','1'," "'Update','{\"ID\":1,\"名字\":\"A2\",\"数量\":5,\"时间\":\"2026-01-02T00:00:00\",\"标记\":false}','pending')" @@ -45,7 +53,7 @@ def test_upsert_insert_update_delete_last_write_wins(sql_conn): sql_conn.commit() # --- Act 1: apply upsert ---------------------------------------------- - cur.execute("EXEC dbo.usp_SyncApply @MaxRetries=5") + cur.execute("EXEC " + proc + " @MaxRetries=5") sql_conn.commit() # --- Assert 1: the later Update wins; BIT false -> 0 ------------------ @@ -57,15 +65,15 @@ def test_upsert_insert_update_delete_last_write_wins(sql_conn): assert row.标记 == 0 # BIT false # --- Act 2: a later Delete wins --------------------------------------- - cur.execute("DELETE dbo.SyncQueue WHERE TargetSchema='sync_test'") + cur.execute("DELETE " + qt + " WHERE TargetSchema='sync_test'") cur.execute( - "INSERT dbo.SyncQueue(SourceFile,SourceTable,SourceLogID,TargetSchema," + "INSERT " + qt + "(SourceFile,SourceTable,SourceLogID,TargetSchema," "TargetTable,RecordID,OperateType,RowData,Status) " "VALUES('t.accdb','ApplyDemo',3,'sync_test','ApplyDemo_YEAR2026','1'," "'Delete',NULL,'pending')" ) sql_conn.commit() - cur.execute("EXEC dbo.usp_SyncApply @MaxRetries=5") + cur.execute("EXEC " + proc + " @MaxRetries=5") sql_conn.commit() # --- Assert 2: row removed -------------------------------------------- @@ -77,5 +85,5 @@ def test_upsert_insert_update_delete_last_write_wins(sql_conn): "IF OBJECT_ID('sync_test.ApplyDemo_YEAR2026') IS NOT NULL " "DROP TABLE sync_test.ApplyDemo_YEAR2026" ) - cur.execute("DELETE dbo.SyncQueue WHERE TargetSchema='sync_test'") + cur.execute("DELETE " + qt + " WHERE TargetSchema='sync_test'") sql_conn.commit() diff --git a/tests/test_sql_writer.py b/tests/test_sql_writer.py index 4bcd2dc..77f4625 100644 --- a/tests/test_sql_writer.py +++ b/tests/test_sql_writer.py @@ -21,10 +21,16 @@ def test_insert_dedup_and_applied_ids(): if not os.environ.get("RUN_INTEGRATION"): pytest.skip("integration") cfg = load_config("config.yaml") - w = SqlWriter(cfg.sql_server.conn_str, "dbo.SyncQueue") + qt = cfg.sql_server.sync_queue_table + w = SqlWriter( + cfg.sql_server.conn_str, + qt, + cfg.sql_server.archive_table, + cfg.sql_server.apply_proc, + ) try: cur = w._conn.cursor() - cur.execute("DELETE dbo.SyncQueue WHERE SourceFile='sqlw_test.accdb'") + cur.execute(f"DELETE {qt} WHERE SourceFile='sqlw_test.accdb'") qr = QueueRow( source_file="sqlw_test.accdb", source_table="T", @@ -38,7 +44,7 @@ def test_insert_dedup_and_applied_ids(): w.insert_queue_row(qr) w.insert_queue_row(qr) # duplicate must be deduped (ignored) cur.execute( - "SELECT COUNT(*) FROM dbo.SyncQueue " + f"SELECT COUNT(*) FROM {qt} " "WHERE SourceFile='sqlw_test.accdb' AND SourceLogID=100" ) assert cur.fetchone()[0] == 1 @@ -47,12 +53,12 @@ def test_insert_dedup_and_applied_ids(): w.call_apply(max_retries=5) cur.execute( - "UPDATE dbo.SyncQueue SET Status='applied' " + f"UPDATE {qt} SET Status='applied' " "WHERE SourceFile='sqlw_test.accdb'" ) assert w.applied_log_ids("sqlw_test.accdb") == [100] finally: - cur.execute("DELETE dbo.SyncQueue WHERE SourceFile='sqlw_test.accdb'") + cur.execute(f"DELETE {qt} WHERE SourceFile='sqlw_test.accdb'") w.close()