65939f5e85
- sql/01_sync_queue.sql: idempotent DDL for dbo.SyncQueue (PK + unique
dedup index + pending lookup index), safe to re-run.
- sql/02_sync_apply.sql: dbo.usp_SyncApply (@MaxRetries INT=5). Per
distinct (TargetSchema,TargetTable) it builds column projections from
sys.columns (excludes ID key/computed/identity/rowversion) and runs a
dynamic-SQL MERGE (last-write-wins via ROW_NUMBER over SourceLogID DESC)
for Insert/Update plus a DELETE for the last op = Delete.
SET IDENTITY_INSERT ON preserves Access PKs.
- tests/conftest.py: sql_conn fixture reads conn_str from gitignored
config.yaml via load_config; skipped without RUN_INTEGRATION=1.
- tests/test_apply_proc.py: integration test covering IDENTITY-preserving
INSERT, last-write-wins UPDATE, BIT conversion, and DELETE; cleans up.
Deviation from the brief's procedure (root-cause fix, design preserved):
every JSON path key is quoted ('$."col"') so non-ASCII column names
(e.g. Chinese 名字/数量) parse correctly. Without quoting, JSON_VALUE
raises "JSON path format is not correct" on Chinese columns, which is the
real target schema for this Access->SQL Server sync.
Co-Authored-By: Claude <noreply@anthropic.com>
31 lines
1.0 KiB
Python
31 lines
1.0 KiB
Python
"""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()
|