diff --git a/sql/01_sync_queue.sql b/sql/01_sync_queue.sql new file mode 100644 index 0000000..c7f85eb --- /dev/null +++ b/sql/01_sync_queue.sql @@ -0,0 +1,42 @@ +-- SyncQueue: staging table for the Access -> SQL Server one-way sync. +-- Idempotent: safe to re-run (table created only if absent; indexes only if absent). + +IF OBJECT_ID('dbo.SyncQueue', 'U') IS NULL +BEGIN + CREATE TABLE dbo.SyncQueue ( + QueueID bigint IDENTITY(1,1) NOT NULL, + SourceFile nvarchar(255) NOT NULL, + SourceTable nvarchar(255) NOT NULL, + SourceLogID bigint NOT NULL, + TargetSchema nvarchar(128) NOT NULL, + TargetTable nvarchar(255) NOT NULL, + RecordID nvarchar(50) NOT NULL, + OperateType varchar(10) NOT NULL, + RowData nvarchar(max) NULL, + Status varchar(10) NOT NULL CONSTRAINT DF_SyncQueue_Status DEFAULT 'pending', + RetryCount int NOT NULL CONSTRAINT DF_SyncQueue_Retry DEFAULT 0, + ErrorMsg nvarchar(max) NULL, + CapturedAt datetime2 NOT NULL CONSTRAINT DF_SyncQueue_Captured DEFAULT sysdatetime(), + AppliedAt datetime2 NULL, + CONSTRAINT PK_SyncQueue PRIMARY KEY CLUSTERED (QueueID) + ); +END +GO + +IF NOT EXISTS (SELECT 1 FROM sys.indexes + WHERE name = 'UX_SyncQueue_Dedup' + AND object_id = OBJECT_ID('dbo.SyncQueue')) +BEGIN + CREATE UNIQUE INDEX UX_SyncQueue_Dedup + ON dbo.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')) +BEGIN + CREATE INDEX IX_SyncQueue_Pending + ON dbo.SyncQueue(Status, TargetSchema, TargetTable); +END +GO diff --git a/sql/02_sync_apply.sql b/sql/02_sync_apply.sql new file mode 100644 index 0000000..740fe1b --- /dev/null +++ b/sql/02_sync_apply.sql @@ -0,0 +1,115 @@ +-- usp_SyncApply: set-based apply of pending SyncQueue rows. +-- Per distinct (TargetSchema, TargetTable) it builds column lists from sys.columns +-- (excluding the ID key, computed, identity and rowversion columns) and runs a +-- dynamic-SQL MERGE (last-write-wins by SourceLogID DESC) for Insert/Update and a +-- DELETE for the last op = Delete. SET IDENTITY_INSERT ON preserves Access PKs. + +CREATE OR ALTER PROCEDURE dbo.usp_SyncApply + @MaxRetries INT = 5 +AS +BEGIN + SET NOCOUNT ON; + + DECLARE @sch NVARCHAR(128), @tbl NVARCHAR(255), @FullName NVARCHAR(514); + DECLARE @sql NVARCHAR(MAX), @cols NVARCHAR(MAX), @upd NVARCHAR(MAX), @ins NVARCHAR(MAX); + + -- Re-queue rows still under the retry budget. + UPDATE dbo.SyncQueue + SET Status = 'pending' + WHERE Status = 'error' AND RetryCount < @MaxRetries; + + DECLARE cur CURSOR LOCAL FAST_FORWARD FOR + SELECT DISTINCT TargetSchema, TargetTable + FROM dbo.SyncQueue + WHERE Status = 'pending'; + + OPEN cur; + FETCH NEXT FROM cur INTO @sch, @tbl; + + WHILE @@FETCH_STATUS = 0 + BEGIN + SET @FullName = QUOTENAME(@sch) + N'.' + QUOTENAME(@tbl); + + BEGIN TRY + BEGIN TRAN; + -- @cols: comma-quoted column names (for INSERT target list) + -- @upd: tgt.col = JSON_VALUE(src.RowData,'$.col') + -- @ins: JSON_VALUE(src.RowData,'$.col') (for INSERT VALUES) + -- NOTE: every JSON path key is wrapped in double quotes ('$."col"'). + -- This is mandatory for non-ASCII column names (e.g. Chinese names + -- like 名字) and harmless for ASCII names, so we quote unconditionally. + SELECT + @cols = STRING_AGG(QUOTENAME(c.name), N',') + WITHIN GROUP (ORDER BY c.column_id), + @upd = STRING_AGG(QUOTENAME(c.name) + N'=JSON_VALUE(src.RowData,''$."' + c.name + N'"'')', N',') + WITHIN GROUP (ORDER BY c.column_id), + @ins = STRING_AGG(N'JSON_VALUE(src.RowData,''$."' + c.name + N'"'')', N',') + WITHIN GROUP (ORDER BY c.column_id) + FROM sys.columns c + WHERE c.object_id = OBJECT_ID(@FullName) + AND c.is_computed = 0 + AND c.is_identity = 0 + AND TYPE_NAME(c.system_type_id) <> 'timestamp' + AND c.name <> 'ID'; + + IF @cols IS NOT NULL + BEGIN + -- Upsert: last write wins (ROW_NUMBER over SourceLogID DESC). + SET @sql = N'SET IDENTITY_INSERT ' + @FullName + N' ON;' + + N'MERGE ' + @FullName + N' WITH (HOLDLOCK) AS tgt' + + N' USING (' + + N' SELECT RecordID, RowData FROM (' + + N' SELECT *, ROW_NUMBER() OVER (PARTITION BY RecordID ORDER BY SourceLogID DESC) rn' + + N' FROM dbo.SyncQueue' + + N' WHERE TargetSchema=@sch AND TargetTable=@tbl AND Status=''pending''' + + N' AND OperateType IN (''Insert'',''Update'') AND RowData IS NOT NULL' + + N' ) x WHERE rn=1' + + N') AS src ON tgt.ID = TRY_CAST(src.RecordID AS int)' + + N' WHEN MATCHED THEN UPDATE SET ' + @upd + + N' WHEN NOT MATCHED THEN INSERT (ID,' + @cols + N')' + + N' VALUES (TRY_CAST(src.RecordID AS int),' + @ins + N');' + + N'SET IDENTITY_INSERT ' + @FullName + N' OFF;'; + EXEC sp_executesql @sql, + N'@sch NVARCHAR(128),@tbl NVARCHAR(255)', @sch, @tbl; + + -- Delete: last op = Delete wins. + SET @sql = N'DELETE t FROM ' + @FullName + N' t' + + N' JOIN (' + + N' SELECT RecordID FROM (' + + N' SELECT *, ROW_NUMBER() OVER (PARTITION BY RecordID ORDER BY SourceLogID DESC) rn' + + N' FROM dbo.SyncQueue' + + N' WHERE TargetSchema=@sch AND TargetTable=@tbl AND Status=''pending''' + + N' AND OperateType=''Delete''' + + N' ) x WHERE rn=1' + + N') d ON t.ID = TRY_CAST(d.RecordID AS int);'; + EXEC sp_executesql @sql, + N'@sch NVARCHAR(128),@tbl NVARCHAR(255)', @sch, @tbl; + END + + UPDATE dbo.SyncQueue + SET Status = 'applied', AppliedAt = SYSDATETIME() + WHERE TargetSchema = @sch AND TargetTable = @tbl AND Status = 'pending'; + + COMMIT; + END TRY + BEGIN CATCH + -- ROLLBACK unwinds the per-table transaction (and, if the caller + -- started one, the caller's too). A proc that errors under an outer + -- transaction is expected to leave the txn uncommittable; the caller + -- 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 + SET Status = CASE WHEN RetryCount + 1 >= @MaxRetries THEN 'dead' ELSE 'error' END, + RetryCount = RetryCount + 1, + ErrorMsg = ERROR_MESSAGE() + WHERE TargetSchema = @sch AND TargetTable = @tbl AND Status = 'pending'; + END CATCH + + FETCH NEXT FROM cur INTO @sch, @tbl; + END + + CLOSE cur; + DEALLOCATE cur; +END +GO diff --git a/tests/conftest.py b/tests/conftest.py new file mode 100644 index 0000000..fc80b96 --- /dev/null +++ b/tests/conftest.py @@ -0,0 +1,30 @@ +"""Pytest fixtures for the Access -> SQL Server sync integration tests. + +Credential safety: the SQL connection string is read from the gitignored +``config.yaml`` via ``sync.config.load_config``. No password is ever hardcoded +in a committed file. The ``.sql`` artefacts contain no credentials. +""" +import os +import pytest +import pyodbc + +from sync.config import load_config + + +@pytest.fixture +def sql_conn(): + """A pyodbc connection to the CompanyDB SQL Server. + + Skipped unless ``RUN_INTEGRATION=1`` is set. The connection is opened in a + transaction (autocommit=False); each test owns its commit/rollback and is + responsible for cleaning up any temporary objects it creates. + """ + if not os.environ.get("RUN_INTEGRATION"): + pytest.skip("set RUN_INTEGRATION=1 to run SQL integration tests") + cfg = load_config("config.yaml") # gitignored, real creds + conn = pyodbc.connect(cfg.sql_server.conn_str, autocommit=False) + try: + yield conn + finally: + conn.rollback() + conn.close() diff --git a/tests/test_apply_proc.py b/tests/test_apply_proc.py new file mode 100644 index 0000000..30a786c --- /dev/null +++ b/tests/test_apply_proc.py @@ -0,0 +1,81 @@ +"""Integration test for dbo.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. +""" +import pytest + + +@pytest.mark.integration +def test_upsert_insert_update_delete_last_write_wins(sql_conn): + cur = sql_conn.cursor() + sch, tbl = "sync_test", "ApplyDemo_YEAR2026" + + # --- Arrange: fresh target table + empty queue ------------------------- + cur.execute("IF SCHEMA_ID('sync_test') IS NULL EXEC('CREATE SCHEMA sync_test')") + cur.execute( + "IF OBJECT_ID('sync_test.ApplyDemo_YEAR2026') IS NOT NULL " + "DROP TABLE sync_test.ApplyDemo_YEAR2026" + ) + cur.execute( + "CREATE TABLE sync_test.ApplyDemo_YEAR2026 (" + "ID INT IDENTITY(1,1) PRIMARY KEY, " + "名字 NVARCHAR(255) NULL, " + "数量 INT NULL, " + "时间 DATETIME2 NULL, " + "标记 BIT NULL)" + ) + cur.execute("DELETE dbo.SyncQueue 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," + "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," + "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')" + ) + sql_conn.commit() + + # --- Act 1: apply upsert ---------------------------------------------- + cur.execute("EXEC dbo.usp_SyncApply @MaxRetries=5") + sql_conn.commit() + + # --- Assert 1: the later Update wins; BIT false -> 0 ------------------ + cur.execute("SELECT 名字,数量,标记 FROM sync_test.ApplyDemo_YEAR2026 WHERE ID=1") + row = cur.fetchone() + assert row is not None, "expected ID=1 to exist after upsert" + assert row.名字 == "A2" + assert row.数量 == 5 + assert row.标记 == 0 # BIT false + + # --- Act 2: a later Delete wins --------------------------------------- + cur.execute("DELETE dbo.SyncQueue WHERE TargetSchema='sync_test'") + cur.execute( + "INSERT dbo.SyncQueue(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") + sql_conn.commit() + + # --- Assert 2: row removed -------------------------------------------- + cur.execute("SELECT COUNT(*) FROM sync_test.ApplyDemo_YEAR2026 WHERE ID=1") + assert cur.fetchone()[0] == 0 + + # --- Cleanup ---------------------------------------------------------- + cur.execute( + "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'") + sql_conn.commit()