feat: access reader and value serialization
Add to_jsonable() (Access/pyodbc value -> JSON-safe) with strict TDD (5 tests: datetime iso, bool preserved before int, decimal->str, none/numbers, dict round-trip). bool branch precedes int because bool is a subclass of int in Python. Add AccessReader: reads TableChangeLog (read_log), reads a full source row by RecordID via to_jsonable (read_row, warns on >4000-char values that JSON_VALUE would truncate), and deletes processed log IDs in chunked, retried batches (delete_log_ids, no-op on empty list). Connects via the ACE ODBC driver from config.access.driver, shared autocommit mode to coexist with the live .laccdb client. Integration test against the real 氩弧焊.accdb (read-only; delete_log_ids only exercised with []) passes. Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
37
tests/test_access_reader.py
Normal file
37
tests/test_access_reader.py
Normal file
@@ -0,0 +1,37 @@
|
||||
"""Integration test for AccessReader against the real 氩弧焊.accdb.
|
||||
|
||||
READ-ONLY: this test never mutates production data. ``delete_log_ids`` is only
|
||||
exercised with an empty list (a guaranteed no-op) to confirm it does not raise.
|
||||
No real log rows are deleted.
|
||||
"""
|
||||
import os
|
||||
import pytest
|
||||
|
||||
from sync.access_reader import AccessReader
|
||||
from sync.config import load_config
|
||||
|
||||
|
||||
@pytest.mark.integration
|
||||
def test_read_log_and_row_and_delete():
|
||||
if not os.environ.get("RUN_INTEGRATION"):
|
||||
pytest.skip("needs RUN_INTEGRATION=1")
|
||||
cfg = load_config("config.yaml")
|
||||
# Real live database (has a .laccdb lock from the client). Shared-mode
|
||||
# read via ACE is validated to coexist with the lock.
|
||||
db = os.environ.get(
|
||||
"TEST_ACCDB",
|
||||
r"\\192.168.110.114\生产进度表\2026年数据\氩弧焊.accdb",
|
||||
)
|
||||
r = AccessReader(db, cfg.access.driver)
|
||||
try:
|
||||
rows = r.read_log(500)
|
||||
assert isinstance(rows, list)
|
||||
if rows:
|
||||
lr = rows[0]
|
||||
assert lr.table_name and lr.record_id
|
||||
assert lr.operate_type in ("Insert", "Update", "Delete")
|
||||
d = r.read_row(lr.table_name, lr.record_id)
|
||||
assert d is None or "ID" in d
|
||||
r.delete_log_ids([], 100, 3) # empty list -> no-op, must not raise
|
||||
finally:
|
||||
r.close()
|
||||
Reference in New Issue
Block a user