266 lines
12 KiB
Python
266 lines
12 KiB
Python
"""来源派车与现场日历的匹配、关联生命周期和事务回归。"""
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import event
|
|
from sqlalchemy.exc import IntegrityError, OperationalError
|
|
|
|
from test_pl_dispatch_calendar import db, create, modules, payload, user
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def production_session_options(db):
|
|
db.autoflush = False
|
|
db.expire_on_commit = False
|
|
|
|
|
|
def site_rows(db, tenant='t1', caller=None, month='2026-10'):
|
|
return modules()[3].list_records(db, tenant, caller or user(tenant=tenant),
|
|
month=month, calendar_kind='site')['list']
|
|
|
|
|
|
@pytest.mark.parametrize('kind', ['acid', 'alkali', 'other'])
|
|
@pytest.mark.parametrize('origin,destination,expected', [
|
|
('山峡化工仓库', '武汉', '出库计划'),
|
|
('云栋厂区', '武汉', '出库计划'),
|
|
('武汉', '山峡化工仓库', '入库计划'),
|
|
('武汉', '云栋厂区', '入库计划'),
|
|
('山峡仓库', '云栋厂区', '出入库计划'),
|
|
('云栋一库', '云栋二库', '出入库计划'),
|
|
('武汉', '安庆', None),
|
|
('', '', None),
|
|
])
|
|
def test_new_source_match_matrix(db, kind, origin, destination, expected):
|
|
source = create(db, calendarKind=kind, origin=origin, destination=destination,
|
|
quantity='32.50', remark='保持备注')
|
|
rows = site_rows(db)
|
|
assert len(rows) == (1 if expected else 0)
|
|
if expected:
|
|
target = rows[0]
|
|
assert target['id'] != source.id
|
|
assert target['sourceRecordId'] == source.id
|
|
assert target['documentType'] == expected
|
|
original = modules()[3]._row_to_dict(source)
|
|
for key in ('dispatchDate', 'vehicleId', 'licensePlate', 'productId', 'productName',
|
|
'origin', 'destination', 'quantity', 'remark'):
|
|
assert target[key] == original[key]
|
|
|
|
|
|
def test_source_update_preserves_identity_and_soft_delete_lifecycle(db):
|
|
model, _, Update, svc = modules()
|
|
source = create(db, origin='山峡')
|
|
assert len(site_rows(db)) == 1
|
|
target = db.get(model, site_rows(db)[0]['id'])
|
|
identity = (target.id, target.create_by, target.create_dept_id, target.create_time)
|
|
# 不变保存以及更换月份、主数据和类型，均为原地更新。
|
|
for _ in range(2):
|
|
svc.update_record(db, 't1', user(), source.id, Update(**payload(origin='山峡')))
|
|
assert len(site_rows(db)) == 1
|
|
svc.update_record(db, 't1', user('u2'), source.id, Update(**payload(
|
|
dispatchDate='2026-11-01', vehicleId='v2', productId='p2',
|
|
origin='武汉', destination='云栋', quantity='20.00', remark='改备注')))
|
|
assert site_rows(db) == []
|
|
changed = site_rows(db, month='2026-11')[0]
|
|
assert changed['id'] == target.id and changed['documentType'] == '入库计划'
|
|
assert changed['licensePlate'] == '鄂E67890' and changed['productName'] == '盐酸'
|
|
assert changed['quantity'] == '20.00' and changed['remark'] == '改备注'
|
|
assert (target.id, target.create_by, target.create_dept_id, target.create_time) == identity
|
|
assert target.update_by == 'u2'
|
|
svc.update_record(db, 't1', user(), source.id, Update(**payload()))
|
|
assert site_rows(db) == [] and target.del_flag == 1
|
|
svc.update_record(db, 't1', user(), source.id,
|
|
Update(**payload(origin='山峡', destination='云栋')))
|
|
assert site_rows(db)[0]['id'] == target.id
|
|
assert target.del_flag == 0 and target.document_type == '出入库计划'
|
|
assert (target.id, target.create_by, target.create_dept_id, target.create_time) == identity
|
|
svc.delete_record(db, 't1', user(), source.id)
|
|
assert source.del_flag == target.del_flag == 1
|
|
assert site_rows(db) == []
|
|
assert db.query(model).count() == 2
|
|
|
|
|
|
def test_new_nonmatching_source_can_match_later_but_history_never_backfills(db):
|
|
model, _, Update, svc = modules()
|
|
source = create(db)
|
|
assert site_rows(db) == []
|
|
svc.update_record(db, 't1', user(), source.id, Update(**payload(destination='云栋')))
|
|
assert len(site_rows(db)) == 1
|
|
# 模拟迁移前已有记录，数据库新字段使用默认值；不依赖派车日期判断历史。
|
|
historical = model(id='legacy', tenant_no='t1', calendar_kind='acid',
|
|
dispatch_date=source.dispatch_date, vehicle_id='v1', license_plate='鄂E12345',
|
|
product_id='p1', product_name='液碱', origin='山峡', destination='云栋',
|
|
create_by='u1', create_dept_id='d1')
|
|
db.add(historical)
|
|
db.commit()
|
|
svc.update_record(db, 't1', user(), historical.id, Update(**payload(origin='山峡', destination='云栋')))
|
|
assert len(site_rows(db)) == 1
|
|
assert all(row['sourceRecordId'] != historical.id for row in site_rows(db))
|
|
svc.delete_record(db, 't1', user(), historical.id)
|
|
assert len(site_rows(db)) == 1
|
|
|
|
|
|
def test_manual_site_is_independent_and_generated_site_is_read_only(db):
|
|
model, _, Update, svc = modules()
|
|
manual = create(db, calendarKind='site', origin='山峡', destination='云栋', documentType='自定义')
|
|
assert len(site_rows(db)) == 1
|
|
create(db, origin='山峡')
|
|
assert len(site_rows(db)) == 2
|
|
generated = next(row for row in site_rows(db) if row['id'] != manual.id)
|
|
for action in [lambda: svc.update_record(db, 't1', user(), generated['id'],
|
|
Update(**payload(calendarKind='site'))),
|
|
lambda: svc.delete_record(db, 't1', user(), generated['id'])]:
|
|
with pytest.raises(HTTPException) as exc:
|
|
action()
|
|
assert exc.value.status_code == 409 and '来源' in exc.value.detail
|
|
svc.update_record(db, 't1', user(), manual.id, Update(**payload(calendarKind='site', documentType='出库计划')))
|
|
svc.delete_record(db, 't1', user(), manual.id)
|
|
assert len(site_rows(db)) == 1 and db.get(model, generated['id']).del_flag == 0
|
|
|
|
|
|
def test_generated_record_inherits_scope_and_cannot_cross_tenant(db, monkeypatch):
|
|
_, Create, Update, svc = modules()
|
|
own = create(db, origin='山峡')
|
|
other = create(db, caller=user('u2'), origin='山峡')
|
|
foreign = svc.create_record(db, 't2', user(tenant='t2'), Create(**payload(
|
|
vehicleId='v3', productId='p3', origin='山峡')))
|
|
assert len(site_rows(db)) == 2
|
|
assert site_rows(db, tenant='t2')[0]['sourceRecordId'] == foreign.id
|
|
import app.core.data_scope as scope
|
|
monkeypatch.setattr(scope, '_resolve_user_role_scopes', lambda *a: [{'data_scope': '4', 'details': []}])
|
|
limited = user(role='admin')
|
|
assert [r['sourceRecordId'] for r in site_rows(db, caller=limited)] == [own.id]
|
|
for source in (other, foreign):
|
|
for action in [lambda: svc.update_record(db, 't1', limited, source.id, Update(**payload())),
|
|
lambda: svc.delete_record(db, 't1', limited, source.id)]:
|
|
with pytest.raises(HTTPException) as exc:
|
|
action()
|
|
assert exc.value.status_code == 404
|
|
svc.update_record(db, 't1', limited, own.id, Update(**payload(origin='山峡', destination='云栋')))
|
|
assert site_rows(db, caller=limited)[0]['documentType'] == '出入库计划'
|
|
svc.delete_record(db, 't1', limited, own.id)
|
|
assert len(site_rows(db)) == 1 and len(site_rows(db, tenant='t2')) == 1
|
|
|
|
|
|
@pytest.mark.parametrize('operation', ['create', 'update', 'delete'])
|
|
def test_source_and_site_rollback_together(db, operation):
|
|
model, _, Update, svc = modules()
|
|
source = None
|
|
if operation != 'create':
|
|
source = create(db, origin='山峡')
|
|
assert len(site_rows(db)) == 1
|
|
def reject_site(mapper, connection, target):
|
|
if target.calendar_kind == 'site':
|
|
raise RuntimeError('模拟现场写入失败')
|
|
hook = 'before_insert' if operation == 'create' else 'before_update'
|
|
event.listen(model, hook, reject_site)
|
|
try:
|
|
with pytest.raises(RuntimeError, match='模拟现场写入失败'):
|
|
if operation == 'create':
|
|
create(db, origin='山峡')
|
|
elif operation == 'update':
|
|
svc.update_record(db, 't1', user(), source.id, Update(**payload(origin='云栋')))
|
|
else:
|
|
svc.delete_record(db, 't1', user(), source.id)
|
|
finally:
|
|
event.remove(model, hook, reject_site)
|
|
# 服务已回滚，会话可直接继续查询。
|
|
if source is None:
|
|
assert db.query(model).count() == 0
|
|
else:
|
|
assert source.origin == '山峡' and source.del_flag == 0
|
|
assert len(site_rows(db)) == 1 and site_rows(db)[0]['origin'] == '山峡'
|
|
|
|
|
|
def test_database_unique_source_including_soft_deleted_rows(db):
|
|
model, _, Update, svc = modules()
|
|
source = create(db, origin='山峡')
|
|
assert len(site_rows(db)) == 1
|
|
target = db.get(model, site_rows(db)[0]['id'])
|
|
svc.update_record(db, 't1', user(), source.id, Update(**payload()))
|
|
assert target.del_flag == 1
|
|
db.add(model(id='duplicate', tenant_no='t1', source_record_id=source.id,
|
|
calendar_kind='site', dispatch_date=source.dispatch_date,
|
|
vehicle_id='v1', license_plate='鄂E12345', product_id='p1', product_name='液碱',
|
|
origin='山峡', destination='武汉', create_dept_id='d1'))
|
|
with pytest.raises(IntegrityError):
|
|
db.commit()
|
|
db.rollback()
|
|
|
|
|
|
def test_identical_source_trips_remain_independent_of_manual_site(db):
|
|
_, _, _, svc = modules()
|
|
manual = create(db, calendarKind='site', origin='山峡')
|
|
first = create(db, origin='山峡')
|
|
second = create(db, origin='山峡')
|
|
assert len(site_rows(db)) == 3
|
|
assert {row['sourceRecordId'] for row in site_rows(db)} == {None, first.id, second.id}
|
|
svc.delete_record(db, 't1', user(), first.id)
|
|
assert len(site_rows(db)) == 2
|
|
assert {row['sourceRecordId'] for row in site_rows(db)} == {None, second.id}
|
|
assert svc.get_record(db, 't1', user(), manual.id).del_flag == 0
|
|
|
|
|
|
@pytest.mark.parametrize('operation', ['create', 'update', 'delete'])
|
|
def test_mysql_deadlock_retries_entire_transaction(db, operation):
|
|
model, _, Update, svc = modules()
|
|
source = create(db, origin='山峡') if operation != 'create' else None
|
|
attempts = []
|
|
def fail_once(mapper, connection, target):
|
|
if target.calendar_kind == 'site':
|
|
attempts.append(target.id)
|
|
if len(attempts) == 1:
|
|
raise OperationalError('test', {}, Exception(1213, 'Deadlock found'))
|
|
hook = 'before_insert' if operation == 'create' else 'before_update'
|
|
event.listen(model, hook, fail_once)
|
|
try:
|
|
if operation == 'create':
|
|
source = create(db, origin='山峡')
|
|
elif operation == 'update':
|
|
svc.update_record(db, 't1', user(), source.id, Update(**payload(origin='云栋')))
|
|
else:
|
|
svc.delete_record(db, 't1', user(), source.id)
|
|
finally:
|
|
event.remove(model, hook, fail_once)
|
|
assert len(attempts) == 2
|
|
assert db.query(model).count() == 2
|
|
if operation == 'delete':
|
|
assert source.del_flag == 1 and site_rows(db) == []
|
|
else:
|
|
assert len(site_rows(db)) == 1
|
|
assert site_rows(db)[0]['sourceRecordId'] == source.id
|
|
assert site_rows(db)[0]['origin'] == ('云栋' if operation == 'update' else '山峡')
|
|
|
|
|
|
@pytest.mark.parametrize('error_code,expected_attempts', [(1213, 3), (1205, 1)])
|
|
def test_database_retry_is_bounded_and_only_for_deadlocks(db, error_code, expected_attempts):
|
|
model, _, _, _ = modules()
|
|
attempts = []
|
|
def reject_site(mapper, connection, target):
|
|
if target.calendar_kind == 'site':
|
|
attempts.append(target.id)
|
|
raise OperationalError('test', {}, Exception(error_code, 'database failure'))
|
|
event.listen(model, 'before_insert', reject_site)
|
|
try:
|
|
with pytest.raises(OperationalError):
|
|
create(db, origin='山峡')
|
|
finally:
|
|
event.remove(model, 'before_insert', reject_site)
|
|
assert len(attempts) == expected_attempts
|
|
assert db.query(model).count() == 0
|
|
|
|
|
|
def test_post_commit_refresh_failure_does_not_replay_a_committed_create(db, monkeypatch):
|
|
model, _, _, _ = modules()
|
|
refresh = db.refresh
|
|
calls = []
|
|
def fail_first_refresh(*args, **kwargs):
|
|
calls.append(1)
|
|
if len(calls) == 1:
|
|
raise OperationalError('test', {}, Exception(1213, 'post-commit read failed'))
|
|
return refresh(*args, **kwargs)
|
|
monkeypatch.setattr(db, 'refresh', fail_first_refresh)
|
|
try:
|
|
create(db, origin='山峡')
|
|
except OperationalError:
|
|
pass
|
|
assert db.query(model).count() == 2
|