846 lines
39 KiB
Python
846 lines
39 KiB
Python
"""市场移动端运费批次：作用域、脱敏与原子写规则。"""
|
|
|
|
from datetime import datetime, timedelta
|
|
from decimal import Decimal
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
from pydantic import ValidationError
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.dialects import mysql
|
|
from sqlalchemy.orm import sessionmaker
|
|
|
|
from app.models.base import Base
|
|
from app.models.models import (
|
|
BizCellSampleType, BizCollectionProtocol, BizCollectionRegionStaff, BizProtocolCollectionInfo,
|
|
BizProtocolPartyCustomer, BizProtocolSample, BizSampleFreight,
|
|
BizSampleFreightItem, SysDictData, SysDictType, SysPost, SysPostRole,
|
|
SysDept, SysRole, SysRoleDataScopeDetail, SysUser, SysUserPost, SysUserRole,
|
|
)
|
|
from app.models.research_donation import (
|
|
BizResearchDonation, BizResearchDonationCollection,
|
|
BizResearchDonationParty, BizResearchDonationSample,
|
|
)
|
|
from app.services.market_field_scope import MarketFieldContext
|
|
|
|
TABLES = [
|
|
BizCollectionProtocol.__table__, BizProtocolCollectionInfo.__table__,
|
|
BizProtocolPartyCustomer.__table__, BizProtocolSample.__table__,
|
|
BizResearchDonation.__table__, BizResearchDonationCollection.__table__,
|
|
BizResearchDonationParty.__table__, BizResearchDonationSample.__table__,
|
|
BizSampleFreight.__table__, BizSampleFreightItem.__table__,
|
|
BizCellSampleType.__table__,
|
|
BizCollectionRegionStaff.__table__, SysDictType.__table__,
|
|
SysDictData.__table__, SysUser.__table__, SysRole.__table__,
|
|
SysRoleDataScopeDetail.__table__, SysUserRole.__table__, SysPost.__table__,
|
|
SysPostRole.__table__, SysUserPost.__table__, SysDept.__table__,
|
|
]
|
|
|
|
|
|
def _session():
|
|
engine = create_engine('sqlite:///:memory:')
|
|
Base.metadata.create_all(engine, tables=TABLES)
|
|
db = sessionmaker(bind=engine)()
|
|
now = datetime(2026, 10, 11, 8, 0)
|
|
db.add_all([
|
|
SysDictType(id='dt', dict_name='运输方式', dict_type='sample_transport_mode', tenant_no='tenant-a', create_dept_id='', create_time=now, update_time=now, del_flag=0),
|
|
SysDictData(id='dd', dict_label='顺丰', dict_value='sf', dict_type='sample_transport_mode', tenant_no='tenant-a', create_dept_id='', create_time=now, update_time=now, del_flag=0),
|
|
SysUser(id='u1', dept_id='', user_no='u1', nick_name='专员', password='', tenant_no='tenant-a', create_dept_id='', create_time=now, update_time=now, del_flag=0),
|
|
SysUser(id='m1', dept_id='', user_no='m1', nick_name='经理', password='', tenant_no='tenant-a', create_dept_id='', create_time=now, update_time=now, del_flag=0),
|
|
SysRole(id='fixture-super-role', role_name='超级管理员', data_scope='1', status='1', tenant_no='tenant-a', create_dept_id='', create_time=now, update_time=now, del_flag=0),
|
|
SysUserRole(id='fixture-super-u1', user_id='u1', role_id='fixture-super-role', tenant_no='tenant-a', create_dept_id='', create_time=now, update_time=now, del_flag=0),
|
|
SysUserRole(id='fixture-super-m1', user_id='m1', role_id='fixture-super-role', tenant_no='tenant-a', create_dept_id='', create_time=now, update_time=now, del_flag=0),
|
|
])
|
|
db.commit()
|
|
return db
|
|
|
|
|
|
def _ctx(kind='specialist', user_id='u1'):
|
|
return MarketFieldContext('tenant-a', user_id, kind)
|
|
|
|
|
|
def _user(user_id='u1'):
|
|
return {'user_id': user_id, 'tenant_no': 'tenant-a', 'role': 'super_admin'}
|
|
|
|
|
|
def _sample(db, key, *, source='cell_storage', sales='u1', manager='m1', region='r1', tenant='tenant-a', deleted=0, created=None):
|
|
created = created or datetime(2026, 10, 10, 8, 0)
|
|
if source == 'cell_storage':
|
|
parent = BizCollectionProtocol(id=f'cp-{key}', protocol_no=f'CP-{key}', tenant_no=tenant, create_dept_id='', create_time=created, del_flag=deleted)
|
|
collection = BizProtocolCollectionInfo(id=f'cc-{key}', collection_protocol_id=parent.id, region_id=region, sales_user_id=sales, region_manager_user_id=manager, tenant_no=tenant, create_dept_id='', create_time=created, del_flag=deleted)
|
|
party = BizProtocolPartyCustomer(id=f'pc-{key}', collection_protocol_id=parent.id, mother_name=f'客户-{key}', tenant_no=tenant, create_dept_id='', create_time=created, del_flag=deleted)
|
|
sample = BizProtocolSample(id=f'cs-{key}', collection_protocol_id=parent.id, sample_code=f'C-{key}', tenant_no=tenant, create_dept_id='', create_time=created, del_flag=deleted)
|
|
else:
|
|
parent = BizResearchDonation(id=f'rd-{key}', donation_no=f'RD-{key}', tenant_no=tenant, create_dept_id='', create_time=created, del_flag=deleted)
|
|
collection = BizResearchDonationCollection(id=f'rc-{key}', research_donation_id=parent.id, region_id=region, sales_user_id=sales, region_manager_user_id=manager, tenant_no=tenant, create_dept_id='', create_time=created, del_flag=deleted)
|
|
party = BizResearchDonationParty(id=f'rp-{key}', research_donation_id=parent.id, mother_name=f'客户-{key}', tenant_no=tenant, create_dept_id='', create_time=created, del_flag=deleted)
|
|
sample = BizResearchDonationSample(id=f'rs-{key}', research_donation_id=parent.id, sample_code=f'R-{key}', tenant_no=tenant, create_dept_id='', create_time=created, del_flag=deleted)
|
|
db.add_all([parent, collection, party, sample])
|
|
db.commit()
|
|
return sample, parent
|
|
|
|
|
|
def _freight(db, key, samples, *, tenant='tenant-a', amount='12.30', deleted=0):
|
|
now = datetime(2026, 10, 11, 8, 0)
|
|
freight = BizSampleFreight(id=f'f-{key}', freight_no=f'YF-{key}', transport_mode='sf', freight_amount=Decimal(amount), tenant_no=tenant, create_dept_id='', create_time=now, update_time=now, del_flag=deleted)
|
|
db.add(freight)
|
|
for pos, (source, sample, parent) in enumerate(samples):
|
|
db.add(BizSampleFreightItem(id=f'fi-{key}-{pos}', freight_id=freight.id, sample_source=source, sample_id=sample.id, sample_code=sample.sample_code, protocol_id=parent.id, tenant_no=tenant, create_dept_id='', create_time=now, update_time=now, del_flag=0))
|
|
db.commit()
|
|
return freight
|
|
|
|
|
|
def _items(*pairs):
|
|
return [{'sampleSource': source, 'sampleId': sample.id} for source, sample in pairs]
|
|
|
|
|
|
def test_specialist_sees_shared_batch_but_only_own_items_and_count():
|
|
from app.services.market_freight import get_freight_detail, list_freights
|
|
db = _session()
|
|
own, own_parent = _sample(db, 'own')
|
|
other, other_parent = _sample(db, 'other', sales='u2')
|
|
freight = _freight(db, 'mixed', [('cell_storage', own, own_parent), ('cell_storage', other, other_parent)])
|
|
rows, total = list_freights(db, ctx=_ctx(), user_info=_user(), include_items=True)
|
|
detail = get_freight_detail(db, freight.id, ctx=_ctx(), user_info=_user())
|
|
assert total == 1 and rows[0]['visibleItemCount'] == 1
|
|
assert [item['sampleCode'] for item in rows[0]['items']] == ['C-own']
|
|
assert [item['sampleCode'] for item in detail['items']] == ['C-own']
|
|
assert 'linkedSampleCount' not in rows[0] and 'linkedSampleCount' not in detail
|
|
assert 'hasHiddenItems' not in detail
|
|
|
|
|
|
def test_batch_without_visible_item_is_not_listed_or_addressable():
|
|
from app.services.market_freight import get_freight_detail, list_freights
|
|
db = _session()
|
|
other, parent = _sample(db, 'other', sales='u2')
|
|
freight = _freight(db, 'hidden', [('cell_storage', other, parent)])
|
|
assert list_freights(db, ctx=_ctx(), user_info=_user())[1] == 0
|
|
with pytest.raises(HTTPException) as exc:
|
|
get_freight_detail(db, freight.id, ctx=_ctx(), user_info=_user())
|
|
assert exc.value.status_code == 404
|
|
|
|
|
|
def test_keyword_and_source_filters_use_only_visible_items():
|
|
from app.services.market_freight import list_freights
|
|
db = _session()
|
|
own, op = _sample(db, 'own')
|
|
secret, sp = _sample(db, 'SECRET', source='research_donation', sales='u2')
|
|
_freight(db, 'mixed', [('cell_storage', own, op), ('research_donation', secret, sp)])
|
|
assert list_freights(db, ctx=_ctx(), user_info=_user(), keyword='SECRET')[1] == 0
|
|
assert list_freights(db, ctx=_ctx(), user_info=_user(), sample_source='research_donation')[1] == 0
|
|
assert list_freights(db, ctx=_ctx(), user_info=_user(), sample_source='cell_storage')[1] == 1
|
|
|
|
|
|
def test_manager_scope_is_region_union_document_manager_for_both_sources():
|
|
from app.services.market_freight import get_freight_detail
|
|
db = _session()
|
|
db.add(BizCollectionRegionStaff(id='staff', region_id='managed', user_id='m1', is_leader=1, tenant_no='tenant-a', create_dept_id='', del_flag=0))
|
|
db.commit()
|
|
region_sample, rp = _sample(db, 'region', sales='u9', manager='x', region='managed')
|
|
assigned, ap = _sample(db, 'assigned', source='research_donation', sales='u9', manager='m1', region='other')
|
|
hidden, hp = _sample(db, 'hidden', sales='m1', manager='x', region='other')
|
|
freight = _freight(db, 'manager', [('cell_storage', region_sample, rp), ('research_donation', assigned, ap), ('cell_storage', hidden, hp)])
|
|
detail = get_freight_detail(db, freight.id, ctx=_ctx('manager', 'm1'), user_info=_user('m1'))
|
|
assert {item['sampleCode'] for item in detail['items']} == {'C-region', 'R-assigned'}
|
|
assert detail['visibleItemCount'] == 2
|
|
assert all('sampleTypeName' in item and 'motherName' in item and 'newbornName' in item for item in detail['items'])
|
|
|
|
|
|
def test_dual_role_uses_fixed_route_context_not_union_between_pages():
|
|
from app.services.market_freight import list_freights
|
|
db = _session()
|
|
specialist, sp = _sample(db, 'specialist', sales='m1', manager='x', region='other')
|
|
manager, mp = _sample(db, 'manager', sales='u9', manager='m1', region='other')
|
|
_freight(db, 'specialist', [('cell_storage', specialist, sp)])
|
|
_freight(db, 'manager', [('cell_storage', manager, mp)])
|
|
specialist_rows, _ = list_freights(db, ctx=_ctx('specialist', 'm1'), user_info=_user('m1'))
|
|
manager_rows, _ = list_freights(db, ctx=_ctx('manager', 'm1'), user_info=_user('m1'))
|
|
assert [row['freightNo'] for row in specialist_rows] == ['YF-specialist']
|
|
assert [row['freightNo'] for row in manager_rows] == ['YF-manager']
|
|
|
|
|
|
def test_cross_tenant_deleted_parent_and_deleted_sample_are_invisible():
|
|
from app.services.market_freight import list_freights
|
|
db = _session()
|
|
cross, cp = _sample(db, 'cross', tenant='tenant-b')
|
|
deleted, dp = _sample(db, 'deleted', deleted=1)
|
|
_freight(db, 'cross', [('cell_storage', cross, cp)], tenant='tenant-b')
|
|
_freight(db, 'deleted', [('cell_storage', deleted, dp)])
|
|
assert list_freights(db, ctx=_ctx(), user_info=_user())[1] == 0
|
|
|
|
|
|
def test_sample_options_are_scoped_sql_paginated_stable_and_derive_link_state():
|
|
from app.services.market_freight import list_sample_options
|
|
db = _session()
|
|
base = datetime(2026, 10, 10, 8, 0)
|
|
first, fp = _sample(db, 'first', created=base)
|
|
_sample(db, 'second', source='research_donation', created=base + timedelta(hours=1))
|
|
_sample(db, 'third', created=base + timedelta(hours=2))
|
|
_sample(db, 'hidden', sales='u2', created=base + timedelta(hours=3))
|
|
_freight(db, 'linked', [('cell_storage', first, fp)])
|
|
page1, total = list_sample_options(db, ctx=_ctx(), user_info=_user(), page=1, page_size=2)
|
|
page2, total2 = list_sample_options(db, ctx=_ctx(), user_info=_user(), page=2, page_size=2)
|
|
assert total == total2 == 3
|
|
assert [x['sampleCode'] for x in page1] == ['C-third', 'R-second']
|
|
assert [x['sampleCode'] for x in page2] == ['C-first']
|
|
assert page2[0]['freightId'] == 'f-linked'
|
|
|
|
|
|
def test_create_derives_snapshots_and_decimal_string_and_requires_visible_item():
|
|
from app.services.market_freight import create_freight
|
|
db = _session()
|
|
own, parent = _sample(db, 'own')
|
|
result = create_freight(db, ctx=_ctx(), user_info=_user(), transport_mode='sf', freight_amount=Decimal('12.30'), items=_items(('cell_storage', own)))
|
|
item = db.query(BizSampleFreightItem).filter_by(freight_id=result['id']).one()
|
|
assert result['freightAmount'] == '12.30'
|
|
assert item.sample_code == 'C-own' and item.protocol_id == parent.id
|
|
with pytest.raises(ValueError, match='至少'):
|
|
create_freight(db, ctx=_ctx(), user_info=_user(), transport_mode='sf', freight_amount='1.00', items=[])
|
|
|
|
|
|
def test_create_prevalidates_every_item_and_rolls_back_without_partial_batch():
|
|
from app.services.market_freight import create_freight
|
|
db = _session()
|
|
own, _ = _sample(db, 'own')
|
|
other, _ = _sample(db, 'other', sales='u2')
|
|
before = db.query(BizSampleFreight).count()
|
|
with pytest.raises(HTTPException):
|
|
create_freight(db, ctx=_ctx(), user_info=_user(), transport_mode='sf', freight_amount='1.00', items=_items(('cell_storage', own), ('cell_storage', other)))
|
|
assert db.query(BizSampleFreight).count() == before
|
|
|
|
|
|
def test_link_rejects_hidden_and_existing_other_batch_without_silent_move():
|
|
from app.services.market_freight import link_samples
|
|
db = _session()
|
|
own, op = _sample(db, 'own')
|
|
other, _ = _sample(db, 'other', sales='u2')
|
|
linked, lp = _sample(db, 'linked')
|
|
target = _freight(db, 'target', [('cell_storage', own, op)])
|
|
source = _freight(db, 'source', [('cell_storage', linked, lp)])
|
|
for sample in (other, linked):
|
|
with pytest.raises((HTTPException, ValueError)):
|
|
link_samples(db, target.id, ctx=_ctx(), user_info=_user(), items=_items(('cell_storage', sample)))
|
|
assert db.query(BizSampleFreightItem).filter_by(sample_id=linked.id, del_flag=0).one().freight_id == source.id
|
|
|
|
|
|
def test_unlink_prevalidates_all_items_and_rejects_hidden_or_nonexistent():
|
|
from app.services.market_freight import unlink_samples
|
|
db = _session()
|
|
own, op = _sample(db, 'own')
|
|
other, hp = _sample(db, 'hidden', sales='u2')
|
|
freight = _freight(db, 'mixed', [('cell_storage', own, op), ('cell_storage', other, hp)])
|
|
missing, _ = _sample(db, 'missing')
|
|
for sample in (other, missing):
|
|
with pytest.raises((HTTPException, ValueError)):
|
|
unlink_samples(db, freight.id, ctx=_ctx(), user_info=_user(), items=_items(('cell_storage', sample)))
|
|
assert db.query(BizSampleFreightItem).filter_by(freight_id=freight.id, del_flag=0).count() == 2
|
|
|
|
|
|
def test_unlink_last_visible_item_disappears_without_revealing_hidden_remaining():
|
|
from app.services.market_freight import list_freights, unlink_samples
|
|
db = _session()
|
|
own, op = _sample(db, 'own')
|
|
other, hp = _sample(db, 'hidden', sales='u2')
|
|
freight = _freight(db, 'mixed', [('cell_storage', own, op), ('cell_storage', other, hp)])
|
|
result = unlink_samples(db, freight.id, ctx=_ctx(), user_info=_user(), items=_items(('cell_storage', own)))
|
|
assert result == {'id': freight.id, 'visibleItemCount': 0, 'disappeared': True}
|
|
assert list_freights(db, ctx=_ctx(), user_info=_user())[1] == 0
|
|
assert db.query(BizSampleFreight).filter_by(id=freight.id).one().del_flag == 0
|
|
|
|
|
|
@pytest.mark.parametrize('kind,user_id', [('specialist', 'u1'), ('manager', 'm1')])
|
|
def test_shared_outside_scope_header_is_immutable_for_both_roles(kind, user_id):
|
|
from app.services.market_freight import update_freight
|
|
db = _session()
|
|
visible, vp = _sample(db, 'visible', sales=user_id if kind == 'specialist' else 'u9', manager=user_id if kind == 'manager' else 'x')
|
|
hidden, hp = _sample(db, 'hidden', sales='other', manager='other')
|
|
freight = _freight(db, 'mixed', [('cell_storage', visible, vp), ('cell_storage', hidden, hp)])
|
|
with pytest.raises(PermissionError, match='全部'):
|
|
update_freight(db, freight.id, ctx=_ctx(kind, user_id), user_info=_user(user_id), freight_amount='99.00')
|
|
assert db.query(BizSampleFreight).filter_by(id=freight.id).one().freight_amount == Decimal('12.30')
|
|
|
|
|
|
def test_header_update_allowed_when_all_active_items_are_visible():
|
|
from app.services.market_freight import update_freight
|
|
db = _session()
|
|
own, op = _sample(db, 'own')
|
|
freight = _freight(db, 'own', [('cell_storage', own, op)])
|
|
result = update_freight(db, freight.id, ctx=_ctx(), user_info=_user(), freight_amount='99.10')
|
|
assert result['freightAmount'] == '99.10'
|
|
|
|
|
|
@pytest.mark.parametrize('kind,user_id', [('specialist', 'u1'), ('manager', 'm1')])
|
|
@pytest.mark.parametrize('mixed', [False, True])
|
|
def test_detail_edit_action_requires_every_active_item_visible_without_leaking_totals(
|
|
kind, user_id, mixed
|
|
):
|
|
from app.services.market_freight import get_freight_detail
|
|
from app.services.market_support import decorate_allowed_actions
|
|
|
|
db = _session()
|
|
visible, visible_parent = _sample(
|
|
db,
|
|
f'action-visible-{kind}-{mixed}',
|
|
sales=user_id if kind == 'specialist' else 'other-sales',
|
|
manager=user_id if kind == 'manager' else 'other-manager',
|
|
)
|
|
rows = [('cell_storage', visible, visible_parent)]
|
|
if mixed:
|
|
hidden, hidden_parent = _sample(
|
|
db,
|
|
f'action-hidden-{kind}',
|
|
sales='hidden-sales',
|
|
manager='hidden-manager',
|
|
)
|
|
rows.append(('cell_storage', hidden, hidden_parent))
|
|
freight = _freight(db, f'action-{kind}-{mixed}', rows)
|
|
|
|
detail = get_freight_detail(
|
|
db,
|
|
freight.id,
|
|
ctx=_ctx(kind, user_id),
|
|
user_info=_user(user_id),
|
|
)
|
|
rendered = decorate_allowed_actions(
|
|
'freight', detail, scope_kind=kind, detail=True
|
|
)
|
|
|
|
expected = ['link', 'unlink'] if mixed else ['edit', 'link', 'unlink']
|
|
assert detail['allowedActions'] == expected
|
|
assert rendered['allowedActions'] == expected
|
|
for hidden_key in (
|
|
'allItemsVisible',
|
|
'hiddenItemCount',
|
|
'totalItemCount',
|
|
'activeItemCount',
|
|
):
|
|
assert hidden_key not in detail
|
|
|
|
|
|
def test_dto_forbids_scope_snapshots_status_and_non_decimal_amounts():
|
|
from app.schemas.market_mobile import MarketFreightCreateBody, MarketFreightItemsBody
|
|
with pytest.raises(ValidationError):
|
|
MarketFreightItemsBody.model_validate({'items': [{'sampleSource': 'cell_storage', 'sampleId': 's1', 'sampleCode': 'forged'}]})
|
|
for extra in ('tenantNo', 'scope', 'status'):
|
|
with pytest.raises(ValidationError):
|
|
MarketFreightCreateBody.model_validate({'transportMode': 'sf', 'freightAmount': '1.00', 'items': [{'sampleSource': 'cell_storage', 'sampleId': 's1'}], extra: 'x'})
|
|
with pytest.raises(ValidationError):
|
|
MarketFreightCreateBody.model_validate({'transportMode': 'sf', 'freightAmount': 'NaN', 'items': [{'sampleSource': 'cell_storage', 'sampleId': 's1'}]})
|
|
|
|
|
|
def test_fixed_routes_and_precise_permissions_are_registered():
|
|
from app.api.market_mobile import router
|
|
paths = [(route.path, route.methods) for route in router.routes]
|
|
for kind in ('specialist', 'manager'):
|
|
prefix = f'/mini/market/{kind}/freights'
|
|
assert any(path == prefix and 'GET' in methods for path, methods in paths)
|
|
assert any(path == prefix and 'POST' in methods for path, methods in paths)
|
|
assert any(path == f'{prefix}/sample-options' and 'GET' in methods for path, methods in paths)
|
|
source = (Path(__file__).parents[1] / 'app/api/market_mobile.py').read_text()
|
|
assert "f'mini:market:{scope_kind}:freight'" in source
|
|
|
|
|
|
def test_migration_has_mysql57_indexes_unique_active_link_and_duplicate_preflight():
|
|
sql = (Path(__file__).parents[2] / 'docs/sql/migrations/2026-10-11-market-freight-indexes.sql').read_text()
|
|
lowered = sql.lower()
|
|
assert 'information_schema.statistics' in lowered
|
|
assert 'generated always as' in lowered
|
|
assert 'unique' in lowered and 'active' in lowered
|
|
assert 'group by tenant_no, sample_source, sample_id' in lowered
|
|
assert 'having count(*) > 1' in lowered
|
|
assert 'create procedure' in lowered and 'signal sqlstate' in lowered
|
|
assert 'character_maximum_length' in lowered
|
|
assert 'data_type' in lowered and 'is_nullable' in lowered
|
|
assert 'normalized_generation_expression' in lowered
|
|
assert "uk_sample_freight_tenant_no" in lowered
|
|
assert 'exact_freight_unique_indexes' in lowered
|
|
|
|
|
|
def test_write_service_contains_locking_postlock_validation_and_integrity_retry():
|
|
source = (Path(__file__).parents[1] / 'app/services/market_freight.py').read_text()
|
|
assert source.count('with_for_update') >= 3
|
|
assert 'IntegrityError' in source
|
|
assert '_lock_batch_header' in source
|
|
assert '_lock_sample_graph' in source
|
|
assert '_lock_active_items' in source
|
|
|
|
|
|
def test_multiple_party_rows_do_not_duplicate_visible_item_or_count():
|
|
from app.services.market_freight import get_freight_detail
|
|
db = _session()
|
|
sample, parent = _sample(db, 'one')
|
|
db.add(BizProtocolPartyCustomer(id='pc-one-duplicate', collection_protocol_id=parent.id, mother_name='重复客户行', tenant_no='tenant-a', create_dept_id='', del_flag=0))
|
|
db.commit()
|
|
freight = _freight(db, 'one', [('cell_storage', sample, parent)])
|
|
detail = get_freight_detail(db, freight.id, ctx=_ctx(), user_info=_user())
|
|
assert detail['visibleItemCount'] == 1
|
|
assert len(detail['items']) == 1
|
|
|
|
|
|
def test_deleted_batch_is_invisible_even_with_visible_active_item():
|
|
from app.services.market_freight import list_freights
|
|
db = _session()
|
|
sample, parent = _sample(db, 'one')
|
|
_freight(db, 'deleted-batch', [('cell_storage', sample, parent)], deleted=1)
|
|
assert list_freights(db, ctx=_ctx(), user_info=_user())[1] == 0
|
|
|
|
|
|
def test_context_user_mismatch_fails_closed():
|
|
from app.services.market_freight import list_sample_options
|
|
db = _session()
|
|
_sample(db, 'one')
|
|
with pytest.raises(PermissionError):
|
|
list_sample_options(db, ctx=_ctx(), user_info=_user('u2'))
|
|
|
|
|
|
def test_manager_can_link_sample_in_managed_region():
|
|
from app.services.market_freight import link_samples
|
|
db = _session()
|
|
db.add(BizCollectionRegionStaff(id='staff-link', region_id='managed', user_id='m1', is_leader=1, tenant_no='tenant-a', create_dept_id='', del_flag=0))
|
|
db.commit()
|
|
existing, ep = _sample(db, 'existing', sales='u9', manager='m1')
|
|
new_sample, _ = _sample(db, 'new', sales='u9', manager='x', region='managed')
|
|
freight = _freight(db, 'manager-link', [('cell_storage', existing, ep)])
|
|
detail = link_samples(db, freight.id, ctx=_ctx('manager', 'm1'), user_info=_user('m1'), items=_items(('cell_storage', new_sample)))
|
|
assert {item['sampleCode'] for item in detail['items']} == {'C-existing', 'C-new'}
|
|
|
|
|
|
def test_batch_paging_is_stable_and_total_counts_only_visible_batches():
|
|
from app.services.market_freight import list_freights
|
|
db = _session()
|
|
first, fp = _sample(db, 'first')
|
|
second, sp = _sample(db, 'second')
|
|
hidden, hp = _sample(db, 'hidden-page', sales='u2')
|
|
f1 = _freight(db, 'first', [('cell_storage', first, fp)])
|
|
f2 = _freight(db, 'second', [('cell_storage', second, sp)])
|
|
_freight(db, 'hidden-page', [('cell_storage', hidden, hp)])
|
|
f1.create_time = f2.create_time
|
|
db.commit()
|
|
page1, total1 = list_freights(db, ctx=_ctx(), user_info=_user(), page=1, page_size=1)
|
|
page2, total2 = list_freights(db, ctx=_ctx(), user_info=_user(), page=2, page_size=1)
|
|
assert total1 == total2 == 2
|
|
assert [page1[0]['id'], page2[0]['id']] == ['f-second', 'f-first']
|
|
|
|
|
|
@pytest.mark.parametrize('source', ['cell_storage', 'research_donation'])
|
|
def test_sample_options_choose_latest_party_once_before_count_and_paging(source):
|
|
from app.services.market_freight import list_sample_options
|
|
db = _session()
|
|
base = datetime(2026, 10, 10, 8, 0)
|
|
target, parent = _sample(db, f'duplicate-{source}', source=source, created=base)
|
|
_sample(db, f'other-{source}', source=source, created=base + timedelta(hours=1))
|
|
if source == 'cell_storage':
|
|
db.add(BizProtocolPartyCustomer(
|
|
id='zz-latest-cell-party', collection_protocol_id=parent.id,
|
|
mother_name='最新细胞客户', tenant_no='tenant-a', create_dept_id='',
|
|
create_time=base + timedelta(hours=2), del_flag=0,
|
|
))
|
|
latest_name = '最新细胞客户'
|
|
else:
|
|
db.add(BizResearchDonationParty(
|
|
id='zz-latest-donation-party', research_donation_id=parent.id,
|
|
mother_name='最新科研客户', tenant_no='tenant-a', create_dept_id='',
|
|
create_time=base + timedelta(hours=2), del_flag=0,
|
|
))
|
|
latest_name = '最新科研客户'
|
|
db.commit()
|
|
|
|
page1, total1 = list_sample_options(
|
|
db, ctx=_ctx(), user_info=_user(), page=1, page_size=1,
|
|
sample_source=source,
|
|
)
|
|
page2, total2 = list_sample_options(
|
|
db, ctx=_ctx(), user_info=_user(), page=2, page_size=1,
|
|
sample_source=source,
|
|
)
|
|
rows = page1 + page2
|
|
|
|
assert total1 == total2 == 2
|
|
assert sum(row['sampleId'] == target.id for row in rows) == 1
|
|
target_row = next(row for row in rows if row['sampleId'] == target.id)
|
|
assert target_row['customerName'] == latest_name
|
|
|
|
|
|
def test_mysql_lock_queries_are_current_reads_with_stable_global_order():
|
|
from app.services.market_freight import (
|
|
_active_item_lock_query,
|
|
_collection_lock_query,
|
|
_leader_lock_query,
|
|
_sample_parent_lock_query,
|
|
)
|
|
db = _session()
|
|
ctx = _ctx()
|
|
sample_sql = str(
|
|
_sample_parent_lock_query(
|
|
db, 'cell_storage', ['z', 'a'], ctx
|
|
).statement.compile(dialect=mysql.dialect())
|
|
).lower()
|
|
collection_sql = str(
|
|
_collection_lock_query(
|
|
db, 'cell_storage', ['p-z', 'p-a'], ctx
|
|
).statement.compile(dialect=mysql.dialect())
|
|
).lower()
|
|
item_sql = str(
|
|
_active_item_lock_query(
|
|
db, ctx, freight_id='f1', keys=[('cell_storage', 'z')]
|
|
).statement.compile(dialect=mysql.dialect())
|
|
).lower()
|
|
leader_sql = str(
|
|
_leader_lock_query(db, _ctx('manager', 'm1')).statement.compile(
|
|
dialect=mysql.dialect()
|
|
)
|
|
).lower()
|
|
|
|
for sql in (sample_sql, collection_sql, item_sql, leader_sql):
|
|
assert ' for update' in sql
|
|
assert ' order by ' in sql
|
|
assert 'tenant_no' in sql and 'del_flag' in sql
|
|
assert 'sales_user_id' in collection_sql
|
|
assert 'biz_protocol_collection_info' in collection_sql
|
|
assert 'biz_collection_region_staff' in leader_sql
|
|
|
|
|
|
def test_all_write_paths_use_batch_graph_item_lock_order():
|
|
source = (Path(__file__).parents[1] / 'app/services/market_freight.py').read_text()
|
|
for start_name, end_name in (
|
|
('def update_freight', 'def link_samples'),
|
|
('def link_samples', 'def unlink_samples'),
|
|
('def unlink_samples', None),
|
|
):
|
|
block = source[source.index(start_name):]
|
|
if end_name:
|
|
block = block[:block.index(end_name)]
|
|
assert block.index('_lock_batch_header') < block.index('_lock_sample_graph')
|
|
assert block.index('_lock_sample_graph') < block.index('_lock_active_items')
|
|
create_block = source[source.index('def create_freight'):source.index('def update_freight')]
|
|
assert create_block.index('_lock_sample_graph') < create_block.index('_lock_active_items')
|
|
|
|
|
|
def test_header_update_rejects_assignment_changed_after_prior_visible_read():
|
|
from app.services.market_freight import get_freight_detail, update_freight
|
|
db = _session()
|
|
sample, parent = _sample(db, 'scope-race')
|
|
freight = _freight(db, 'scope-race', [('cell_storage', sample, parent)])
|
|
assert get_freight_detail(db, freight.id, ctx=_ctx(), user_info=_user())['visibleItemCount'] == 1
|
|
|
|
collection = db.query(BizProtocolCollectionInfo).filter_by(
|
|
collection_protocol_id=parent.id
|
|
).one()
|
|
db.query(BizProtocolCollectionInfo).filter_by(id=collection.id).update(
|
|
{'sales_user_id': 'u2'}, synchronize_session=False
|
|
)
|
|
assert collection.sales_user_id == 'u1'
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
update_freight(
|
|
db, freight.id, ctx=_ctx(), user_info=_user(), freight_amount='55.00'
|
|
)
|
|
assert exc.value.status_code == 404
|
|
assert db.query(BizSampleFreight).filter_by(id=freight.id).one().freight_amount == Decimal('12.30')
|
|
|
|
|
|
def test_final_item_lock_rejects_participant_snapshot_change():
|
|
from app.services.market_freight import _lock_active_items
|
|
db = _session()
|
|
first, fp = _sample(db, 'snapshot-first')
|
|
second, sp = _sample(db, 'snapshot-second')
|
|
freight = _freight(db, 'snapshot', [('cell_storage', first, fp)])
|
|
expected = [('cell_storage', first.id)]
|
|
db.add(BizSampleFreightItem(
|
|
id='fi-snapshot-race', freight_id=freight.id,
|
|
sample_source='cell_storage', sample_id=second.id,
|
|
sample_code=second.sample_code, protocol_id=sp.id,
|
|
tenant_no='tenant-a', create_dept_id='', del_flag=0,
|
|
))
|
|
db.commit()
|
|
|
|
with pytest.raises(PermissionError, match='发生变化'):
|
|
_lock_active_items(
|
|
db, _ctx(), freight_id=freight.id,
|
|
expected_batch_keys=expected,
|
|
)
|
|
|
|
|
|
def test_lock_graph_does_not_rely_on_exists_scope_for_current_field_values():
|
|
source = (Path(__file__).parents[1] / 'app/services/market_freight.py').read_text()
|
|
block = source[source.index('def _lock_sample_graph'):source.index('def _lock_active_items')]
|
|
assert 'apply_cell_protocol_scope' not in block
|
|
assert 'apply_donation_scope' not in block
|
|
assert '_current_field_visible_parent_ids' in block
|
|
assert '_current_data_scope_parent_ids' in block
|
|
|
|
|
|
def test_manager_header_update_rejects_current_leader_revocation():
|
|
from app.services.market_freight import get_freight_detail, update_freight
|
|
db = _session()
|
|
leader = BizCollectionRegionStaff(
|
|
id='leader-race', region_id='managed-race', user_id='m1', is_leader=1,
|
|
tenant_no='tenant-a', create_dept_id='', del_flag=0,
|
|
)
|
|
db.add(leader)
|
|
db.commit()
|
|
sample, parent = _sample(
|
|
db, 'leader-race', sales='u9', manager='other', region='managed-race'
|
|
)
|
|
freight = _freight(db, 'leader-race', [('cell_storage', sample, parent)])
|
|
ctx = _ctx('manager', 'm1')
|
|
assert get_freight_detail(db, freight.id, ctx=ctx, user_info=_user('m1'))['visibleItemCount'] == 1
|
|
db.query(BizCollectionRegionStaff).filter_by(id=leader.id).update(
|
|
{'is_leader': 0}, synchronize_session=False
|
|
)
|
|
assert leader.is_leader == 1
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
update_freight(
|
|
db, freight.id, ctx=ctx, user_info=_user('m1'), freight_amount='88.00'
|
|
)
|
|
assert exc.value.status_code == 404
|
|
|
|
|
|
def test_header_update_uses_current_role_data_scope_not_prior_snapshot():
|
|
from app.services.market_freight import get_freight_detail, update_freight
|
|
db = _session()
|
|
now = datetime(2026, 10, 11, 8, 0)
|
|
db.add_all([
|
|
SysUser(
|
|
id='scope-user', dept_id='dept-current', user_no='scope-user',
|
|
nick_name='范围用户', password='', tenant_no='tenant-a',
|
|
create_dept_id='', create_time=now, update_time=now, del_flag=0,
|
|
),
|
|
SysRole(
|
|
id='scope-role', role_name='范围角色', data_scope='1', status='1',
|
|
tenant_no='tenant-a', create_dept_id='', create_time=now,
|
|
update_time=now, del_flag=0,
|
|
),
|
|
SysUserRole(
|
|
id='scope-user-role', user_id='scope-user', role_id='scope-role',
|
|
tenant_no='tenant-a', create_dept_id='', create_time=now,
|
|
update_time=now, del_flag=0,
|
|
),
|
|
])
|
|
db.commit()
|
|
sample, parent = _sample(db, 'data-scope', sales='scope-user')
|
|
parent.create_by = 'another-user'
|
|
parent.create_dept_id = 'another-dept'
|
|
db.commit()
|
|
freight = _freight(db, 'data-scope', [('cell_storage', sample, parent)])
|
|
ctx = _ctx('specialist', 'scope-user')
|
|
user = {'user_id': 'scope-user', 'tenant_no': 'tenant-a', 'role': 'admin'}
|
|
assert get_freight_detail(db, freight.id, ctx=ctx, user_info=user)['visibleItemCount'] == 1
|
|
cached_role = db.query(SysRole).filter_by(id='scope-role').one()
|
|
db.query(SysRole).filter_by(id='scope-role').update(
|
|
{'data_scope': '4'}, synchronize_session=False
|
|
)
|
|
assert cached_role.data_scope == '1'
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
update_freight(
|
|
db, freight.id, ctx=ctx, user_info=user, freight_amount='77.00'
|
|
)
|
|
assert exc.value.status_code == 404
|
|
|
|
|
|
def test_current_data_scope_preserves_post_roles_custom_scope_and_union():
|
|
from app.services.market_freight import _current_data_scope_parent_ids
|
|
db = _session()
|
|
now = datetime(2026, 10, 11, 8, 0)
|
|
db.add_all([
|
|
SysUser(
|
|
id='union-user', dept_id='union-home', user_no='union-user',
|
|
nick_name='并集用户', password='', tenant_no='tenant-a',
|
|
create_dept_id='', create_time=now, update_time=now, del_flag=0,
|
|
),
|
|
SysRole(
|
|
id='custom-role', role_name='自定义范围', data_scope='5', status='1',
|
|
tenant_no='tenant-a', create_dept_id='', create_time=now,
|
|
update_time=now, del_flag=0,
|
|
),
|
|
SysRole(
|
|
id='self-role', role_name='本人范围', data_scope='4', status='1',
|
|
tenant_no='tenant-a', create_dept_id='', create_time=now,
|
|
update_time=now, del_flag=0,
|
|
),
|
|
SysUserRole(
|
|
id='union-direct', user_id='union-user', role_id='custom-role',
|
|
tenant_no='tenant-a', create_dept_id='', create_time=now,
|
|
update_time=now, del_flag=0,
|
|
),
|
|
SysRoleDataScopeDetail(
|
|
id='union-custom-detail', role_id='custom-role', scope_type='dept',
|
|
scope_value='custom-dept', tenant_no='tenant-a', create_dept_id='',
|
|
create_time=now, update_time=now, del_flag=0,
|
|
),
|
|
SysPost(
|
|
id='union-post', post_name='继承职位', post_code='union-post',
|
|
status='1', tenant_no='tenant-a', create_dept_id='',
|
|
create_time=now, update_time=now, del_flag=0,
|
|
),
|
|
SysUserPost(
|
|
id='union-user-post', user_id='union-user', post_id='union-post',
|
|
tenant_no='tenant-a', create_dept_id='', create_time=now,
|
|
update_time=now, del_flag=0,
|
|
),
|
|
SysPostRole(
|
|
id='union-post-role', post_id='union-post', role_id='self-role',
|
|
tenant_no='tenant-a', create_dept_id='', create_time=now,
|
|
update_time=now, del_flag=0,
|
|
),
|
|
])
|
|
db.commit()
|
|
_, custom_parent = _sample(db, 'union-custom')
|
|
_, self_parent = _sample(db, 'union-self')
|
|
_, denied_parent = _sample(db, 'union-denied')
|
|
custom_parent.create_dept_id = 'custom-dept'
|
|
self_parent.create_by = 'union-user'
|
|
denied_parent.create_dept_id = 'other-dept'
|
|
denied_parent.create_by = 'other-user'
|
|
db.commit()
|
|
|
|
visible = _current_data_scope_parent_ids(
|
|
db,
|
|
[custom_parent, self_parent, denied_parent],
|
|
_ctx('specialist', 'union-user'),
|
|
{'user_id': 'union-user', 'tenant_no': 'tenant-a', 'role': 'admin'},
|
|
)
|
|
|
|
assert visible == {custom_parent.id, self_parent.id}
|
|
|
|
|
|
def test_current_data_scope_preserves_department_descendant_scope():
|
|
from app.services.market_freight import _current_data_scope_parent_ids
|
|
db = _session()
|
|
now = datetime(2026, 10, 11, 8, 0)
|
|
db.add_all([
|
|
SysUser(
|
|
id='dept-user', dept_id='dept-root', user_no='dept-user',
|
|
nick_name='部门用户', password='', tenant_no='tenant-a',
|
|
create_dept_id='', create_time=now, update_time=now, del_flag=0,
|
|
),
|
|
SysRole(
|
|
id='dept-role', role_name='部门以下', data_scope='3', status='1',
|
|
tenant_no='tenant-a', create_dept_id='', create_time=now,
|
|
update_time=now, del_flag=0,
|
|
),
|
|
SysUserRole(
|
|
id='dept-user-role', user_id='dept-user', role_id='dept-role',
|
|
tenant_no='tenant-a', create_dept_id='', create_time=now,
|
|
update_time=now, del_flag=0,
|
|
),
|
|
SysDept(
|
|
id='dept-root', parent_id='0', dept_name='根部门', status='1',
|
|
tenant_no='tenant-a', create_dept_id='', create_time=now,
|
|
update_time=now, del_flag=0,
|
|
),
|
|
SysDept(
|
|
id='dept-child', parent_id='dept-root', dept_name='子部门', status='1',
|
|
tenant_no='tenant-a', create_dept_id='', create_time=now,
|
|
update_time=now, del_flag=0,
|
|
),
|
|
])
|
|
db.commit()
|
|
_, child_parent = _sample(db, 'dept-child')
|
|
_, denied_parent = _sample(db, 'dept-denied')
|
|
child_parent.create_dept_id = 'dept-child'
|
|
denied_parent.create_dept_id = 'other-dept'
|
|
db.commit()
|
|
|
|
visible = _current_data_scope_parent_ids(
|
|
db,
|
|
[child_parent, denied_parent],
|
|
_ctx('specialist', 'dept-user'),
|
|
{'user_id': 'dept-user', 'tenant_no': 'tenant-a', 'role': 'admin'},
|
|
)
|
|
|
|
assert visible == {child_parent.id}
|
|
|
|
|
|
def test_locked_data_scope_ignores_payload_superadmin_across_tenants():
|
|
from app.services.market_freight import (
|
|
_current_data_scope_parent_ids,
|
|
get_freight_detail,
|
|
)
|
|
db = _session()
|
|
now = datetime(2026, 10, 11, 8, 0)
|
|
db.add_all([
|
|
SysUser(
|
|
id='cross-user', dept_id='cross-dept', user_no='cross-user',
|
|
nick_name='跨租户用户', password='', tenant_no='tenant-a',
|
|
create_dept_id='', create_time=now, update_time=now, del_flag=0,
|
|
),
|
|
SysRole(
|
|
id='default-super-role', role_name='超级管理员', data_scope='1',
|
|
status='1', tenant_no='zkxw', create_dept_id='', create_time=now,
|
|
update_time=now, del_flag=0,
|
|
),
|
|
SysUserRole(
|
|
id='default-super-link', user_id='cross-user',
|
|
role_id='default-super-role', tenant_no='zkxw', create_dept_id='',
|
|
create_time=now, update_time=now, del_flag=0,
|
|
),
|
|
SysRole(
|
|
id='business-self-role', role_name='业务专员', data_scope='4',
|
|
status='1', tenant_no='tenant-a', create_dept_id='', create_time=now,
|
|
update_time=now, del_flag=0,
|
|
),
|
|
SysUserRole(
|
|
id='business-self-link', user_id='cross-user',
|
|
role_id='business-self-role', tenant_no='tenant-a',
|
|
create_dept_id='', create_time=now, update_time=now, del_flag=0,
|
|
),
|
|
])
|
|
db.commit()
|
|
sample, parent = _sample(
|
|
db, 'cross-tenant-scope', sales='cross-user'
|
|
)
|
|
parent.create_by = 'another-user'
|
|
parent.create_dept_id = 'another-dept'
|
|
db.commit()
|
|
ctx = _ctx('specialist', 'cross-user')
|
|
forged_default_tenant_payload = {
|
|
'user_id': 'cross-user', 'tenant_no': 'tenant-a', 'role': 'super_admin'
|
|
}
|
|
freight = _freight(
|
|
db, 'cross-tenant-scope', [('cell_storage', sample, parent)]
|
|
)
|
|
|
|
assert _current_data_scope_parent_ids(
|
|
db, [parent], ctx, forged_default_tenant_payload
|
|
) == set()
|
|
with pytest.raises(HTTPException) as exc:
|
|
get_freight_detail(
|
|
db, freight.id, ctx=ctx, user_info=forged_default_tenant_payload
|
|
)
|
|
assert exc.value.status_code == 404
|
|
|
|
db.add_all([
|
|
SysRole(
|
|
id='business-super-role', role_name='超级管理员', data_scope='4',
|
|
status='1', tenant_no='tenant-a', create_dept_id='', create_time=now,
|
|
update_time=now, del_flag=0,
|
|
),
|
|
SysPost(
|
|
id='business-super-post', post_name='业务租户超级管理职位',
|
|
post_code='business-super-post', status='1', tenant_no='tenant-a',
|
|
create_dept_id='', create_time=now, update_time=now, del_flag=0,
|
|
),
|
|
SysUserPost(
|
|
id='business-super-user-post', user_id='cross-user',
|
|
post_id='business-super-post', tenant_no='tenant-a',
|
|
create_dept_id='', create_time=now, update_time=now, del_flag=0,
|
|
),
|
|
SysPostRole(
|
|
id='business-super-post-role', post_id='business-super-post',
|
|
role_id='business-super-role', tenant_no='tenant-a',
|
|
create_dept_id='', create_time=now, update_time=now, del_flag=0,
|
|
),
|
|
])
|
|
db.commit()
|
|
|
|
assert _current_data_scope_parent_ids(
|
|
db, [parent], ctx, forged_default_tenant_payload
|
|
) == {parent.id}
|
|
assert get_freight_detail(
|
|
db, freight.id, ctx=ctx, user_info=forged_default_tenant_payload
|
|
)['visibleItemCount'] == 1
|