108bf87225
补齐源快照、资料预演、正式导入及工资核算的依赖链路。 新增版本化输入、逐段 FIFO、生成锁和结构化当月调整， 并提供默认关闭的周期采集、有限重试及 Windows 托管。 原始客户资料、私有配置及首页改动不纳入本次提交。 BREAKING CHANGE: 升级前需应用新增迁移，并通过 PAYROLL_MAKER_USER_ID 绑定实际工资制作人，否则业务只读。 旧单批次核算默认关闭；历史兼容需显式启用 PAYROLL_ALLOW_LEGACY_CALCULATION。生产启用和真实金额待验收。
127 lines
5.1 KiB
Python
127 lines
5.1 KiB
Python
"""用真实事务验证 T+ staging 重跑、失败隔离和敏感信息保护。"""
|
||
|
||
from contextlib import nullcontext
|
||
from unittest.mock import Mock
|
||
|
||
import pytest
|
||
from sqlalchemy import DefaultClause, Integer, MetaData, create_engine, func, select, text
|
||
from sqlalchemy.orm import Session
|
||
|
||
from app.core.config import Settings
|
||
from app.models import TPlusSyncBatch, TPlusSyncError, TPlusSyncRecord
|
||
from app.services.tplus.sync_service import TPlusSyncService
|
||
|
||
|
||
@pytest.fixture
|
||
def target(monkeypatch):
|
||
"""以 SQLite 执行 staging 事务，MySQL 专属 DDL 仍由迁移测试覆盖。"""
|
||
engine = create_engine("sqlite:///:memory:")
|
||
metadata = MetaData()
|
||
for model in (TPlusSyncBatch, TPlusSyncError, TPlusSyncRecord):
|
||
table = model.__table__.to_metadata(metadata)
|
||
table.c.id.type = Integer()
|
||
table.c.update_time.server_default = DefaultClause(text("CURRENT_TIMESTAMP"))
|
||
metadata.create_all(engine)
|
||
settings = Settings(_env_file=None, app_secret_key="x" * 32)
|
||
monkeypatch.setattr("app.services.tplus.sync_service.get_settings", lambda: settings)
|
||
monkeypatch.setattr("app.services.tplus.sync_service.sync_lock", lambda session: nullcontext())
|
||
with Session(engine, expire_on_commit=False) as session:
|
||
yield session
|
||
engine.dispose()
|
||
|
||
|
||
def _service(target, rows):
|
||
service = TPlusSyncService(Mock(), target)
|
||
service.reader = Mock()
|
||
service.reader.iter_entity.side_effect = lambda *a, **kw: iter(rows)
|
||
return service
|
||
|
||
|
||
def test_重复全量读取不增加记录且业务表未被引用(target):
|
||
rows = [{"id": 1, "code": "A"}, {"id": 2, "code": "B"}]
|
||
service = _service(target, rows)
|
||
first = service.run(entity_types=["products"])
|
||
second = service.run(entity_types=["products"])
|
||
assert first.status == second.status == "SUCCEEDED"
|
||
assert first.staged_record_count == 2
|
||
assert second.staged_record_count == 0
|
||
assert second.source_record_count == 2
|
||
assert target.scalar(select(func.count()).select_from(TPlusSyncRecord)) == 2
|
||
assert second.watermark_to is None
|
||
|
||
|
||
def test_单条缺失主键不会回滚其他有效行(target):
|
||
service = _service(target, [{"id": 1, "code": "A"}, {"code": "bad"}, {"id": 2}])
|
||
batch = service.run(entity_types=["products"])
|
||
assert batch.status == "PARTIAL"
|
||
assert batch.staged_record_count == 2
|
||
assert batch.failed_record_count == 1
|
||
assert target.scalar(select(func.count()).select_from(TPlusSyncRecord)) == 2
|
||
|
||
|
||
def test_数据库约束失败后仍可写入下一条(target):
|
||
service = _service(target, [{"id": 1, "code": "A"}])
|
||
# 用数据库真实触发器拒绝坏行，验证 executemany 失败后的回滚和逐条隔离。
|
||
service.reader.iter_entity.side_effect = lambda *a, **kw: iter([{"id": 1}, {"id": 2}])
|
||
target.execute(
|
||
text(
|
||
"CREATE TRIGGER reject_first BEFORE INSERT ON tplus_sync_records "
|
||
"WHEN NEW.source_id='1' BEGIN SELECT RAISE(ABORT, 'invalid row'); END"
|
||
)
|
||
)
|
||
batch = service.run(entity_types=["products"])
|
||
assert batch.status == "PARTIAL"
|
||
assert batch.staged_record_count == 1
|
||
assert target.scalar(select(TPlusSyncRecord.source_id)) == "2"
|
||
|
||
|
||
def test_读取失败有审计且异常中的敏感参数不落日志(target):
|
||
service = _service(target, [])
|
||
service.reader.iter_entity.side_effect = RuntimeError("password=secret phone=13800000000")
|
||
batch = service.run(entity_types=["products"])
|
||
assert batch.status == "FAILED"
|
||
error = target.scalar(select(TPlusSyncError))
|
||
assert error.error_code == "READ_ERROR"
|
||
assert "secret" not in error.error_message
|
||
assert "13800000000" not in error.error_message
|
||
|
||
|
||
def test_成功一类后另一类读取失败应标记部分成功(target):
|
||
service = _service(target, [])
|
||
|
||
def source(entity, **kwargs):
|
||
if entity == "products":
|
||
return iter([{"id": 1}])
|
||
raise RuntimeError("读取中断")
|
||
|
||
service.reader.iter_entity.side_effect = source
|
||
batch = service.run(entity_types=["products", "customers"])
|
||
assert batch.status == "PARTIAL"
|
||
assert batch.source_record_count == batch.staged_record_count == 1
|
||
assert batch.failed_record_count == 1
|
||
|
||
|
||
def test_未确认时间增量不能创建误导性批次(target):
|
||
from datetime import datetime
|
||
|
||
service = _service(target, [])
|
||
with pytest.raises(ValueError, match="增量水位语义尚未确认"):
|
||
service.run(entity_types=["receipts"], watermark=datetime(2026, 10, 27))
|
||
assert target.scalar(select(func.count()).select_from(TPlusSyncBatch)) == 0
|
||
|
||
|
||
def test_已读数据后明确超时仍标记可重试失败(target):
|
||
service = _service(target, [])
|
||
|
||
def source(entity, **kwargs):
|
||
if entity == "products":
|
||
return iter([{"id": 1}])
|
||
raise TimeoutError("敏感驱动信息不得回显")
|
||
|
||
service.reader.iter_entity.side_effect = source
|
||
batch = service.run(entity_types=["products", "customers", "personnel"])
|
||
assert batch.status == "FAILED"
|
||
assert batch.staged_record_count == 1
|
||
assert service.reader.iter_entity.call_count == 2
|
||
assert target.scalar(select(TPlusSyncError.error_code)) == "SOURCE_TIMEOUT"
|