84 lines
2.9 KiB
Python
84 lines
2.9 KiB
Python
"""Integration test for sync.sql_writer.SqlWriter.
|
|
|
|
Validates the dedup INSERT (same SourceLogID is only ever enqueued once), the
|
|
``applied_log_ids`` query (rows flipped to ``applied`` come back in order), and
|
|
that ``call_apply`` invokes the proc without raising.
|
|
|
|
conn_str is read from the gitignored ``config.yaml`` via ``load_config``; no
|
|
credentials are hardcoded here. The test self-cleans using a throwaway
|
|
``SourceFile='sqlw_test.accdb'`` marker so no residue is left on SyncQueue.
|
|
"""
|
|
import os
|
|
import pytest
|
|
from unittest.mock import MagicMock
|
|
|
|
from sync.sql_writer import SqlWriter, QueueRow
|
|
from sync.config import load_config
|
|
|
|
|
|
@pytest.mark.integration
|
|
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")
|
|
try:
|
|
cur = w._conn.cursor()
|
|
cur.execute("DELETE dbo.SyncQueue WHERE SourceFile='sqlw_test.accdb'")
|
|
qr = QueueRow(
|
|
source_file="sqlw_test.accdb",
|
|
source_table="T",
|
|
record_id="7",
|
|
target_schema="sync_test",
|
|
target_table="T_YEAR2026",
|
|
source_log_id=100,
|
|
operate_type="Insert",
|
|
row_data='{"ID":7}',
|
|
)
|
|
w.insert_queue_row(qr)
|
|
w.insert_queue_row(qr) # duplicate must be deduped (ignored)
|
|
cur.execute(
|
|
"SELECT COUNT(*) FROM dbo.SyncQueue "
|
|
"WHERE SourceFile='sqlw_test.accdb' AND SourceLogID=100"
|
|
)
|
|
assert cur.fetchone()[0] == 1
|
|
|
|
# call_apply must run without error (proc logic validated elsewhere).
|
|
w.call_apply(max_retries=5)
|
|
|
|
cur.execute(
|
|
"UPDATE dbo.SyncQueue 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'")
|
|
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")
|