59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
"""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 unittest.mock import MagicMock
|
|
|
|
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()
|
|
|
|
|
|
def test_count_rows_executes_count_sql_and_returns_value():
|
|
r = AccessReader("dummy.accdb", "{Microsoft Access Driver (*.accdb, *.mdb)}")
|
|
cur = MagicMock()
|
|
cur.fetchone.return_value = (42,)
|
|
r._conn = MagicMock() # bypass lazy connect
|
|
r._conn.cursor.return_value = cur
|
|
assert r.count_rows("表壳焊接记录") == 42
|
|
cur.execute.assert_called_once_with('SELECT COUNT(*) FROM "表壳焊接记录"')
|
|
|
|
|
|
def test_read_ids_returns_ordered_id_list():
|
|
r = AccessReader("dummy.accdb", "{Microsoft Access Driver (*.accdb, *.mdb)}")
|
|
cur = MagicMock()
|
|
cur.fetchall.return_value = [(1,), (3,), (5,)]
|
|
r._conn = MagicMock()
|
|
r._conn.cursor.return_value = cur
|
|
assert r.read_ids("T") == [1, 3, 5]
|
|
cur.execute.assert_called_once_with('SELECT ID FROM "T" ORDER BY ID')
|