zlwu
29f9866a24
Add four business tabs and identifying fields so users can locate orders before assigning projects. Introduce a permission-scoped detail query with child EXISTS searches and batched enrichment, preserving statistics and assignment writes. Approved Figma v2; independent source review completed. Tests authored but not executed under the user self-test arrangement. Related to issue #13.
153 lines
8.8 KiB
Python
153 lines
8.8 KiB
Python
"""归属维护业务检索的权限、分页及关联资料边界。"""
|
|
import unittest
|
|
from datetime import datetime
|
|
|
|
from fastapi import HTTPException
|
|
from sqlalchemy import create_engine, event
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.models.base import Base
|
|
from app.models.models import (
|
|
BizChannelServiceOrder, BizCollectionHospital, BizCollectionProtocol,
|
|
BizPilotApplication, BizPilotPatient, BizProtocolCollectionInfo,
|
|
BizProtocolPartyCustomer, SysMenu, SysRole, SysRoleMenu, SysUser, SysUserRole,
|
|
)
|
|
from app.models.research_donation import BizResearchDonation, BizResearchDonationParty
|
|
from app.services import project_assignment_records as svc
|
|
from app.services import project_management
|
|
|
|
|
|
class ProjectAssignmentRecordsTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.engine = create_engine('sqlite:///:memory:')
|
|
Base.metadata.create_all(self.engine)
|
|
self.db = Session(self.engine)
|
|
self.user = dict(user_id='u', role='super_admin', tenant_no='t')
|
|
|
|
def tearDown(self):
|
|
self.db.close()
|
|
self.engine.dispose()
|
|
|
|
def records(self, kind='cell_storage', **filters):
|
|
return svc.records(self.db, 't', self.user, kind, **filters)
|
|
|
|
def grant(self, code):
|
|
self.db.add_all([
|
|
SysRole(id=code, tenant_no='t', role_name=code, data_scope='4'),
|
|
SysMenu(id=code, tenant_no='t', menu_name=code, menu_type='F', perms=code),
|
|
SysRoleMenu(id=code, tenant_no='t', role_id=code, menu_id=code),
|
|
SysUserRole(id=code, tenant_no='t', role_id=code, user_id='u'),
|
|
])
|
|
self.db.commit()
|
|
|
|
def test_assignment_permission_and_parent_data_scope(self):
|
|
self.user['role'] = 'admin'
|
|
self.grant('project:statistics:list')
|
|
with self.assertRaises(HTTPException) as denied:
|
|
self.records()
|
|
self.assertEqual(denied.exception.status_code, 403)
|
|
self.grant('project:assignment:edit')
|
|
self.grant('project:project:all')
|
|
for row_id, tenant, owner, deleted in [
|
|
('own', 't', 'u', 0), ('other', 't', 'other', 0),
|
|
('foreign', 'foreign', 'u', 0), ('deleted', 't', 'u', 1),
|
|
]:
|
|
self.db.add(BizCollectionProtocol(id=row_id, tenant_no=tenant, protocol_no=row_id,
|
|
create_by=owner, del_flag=deleted))
|
|
self.db.commit()
|
|
self.assertEqual([r['id'] for r in self.records()['list']], ['own'])
|
|
|
|
def test_children_do_not_duplicate_counts_or_leak_foreign_deleted_data(self):
|
|
self.db.add_all([
|
|
BizCollectionProtocol(id='a', tenant_no='t', protocol_no='A', protocol_status='sxz'),
|
|
BizCollectionProtocol(id='b', tenant_no='t', protocol_no='B', protocol_status='dsx'),
|
|
BizProtocolPartyCustomer(id='p1', tenant_no='t', collection_protocol_id='a', mother_name='Alice', mother_phone='13800000001'),
|
|
BizProtocolPartyCustomer(id='p2', tenant_no='t', collection_protocol_id='a', mother_name='Alice'),
|
|
BizProtocolPartyCustomer(id='foreign', tenant_no='foreign', collection_protocol_id='a', mother_name='ForeignOnly'),
|
|
BizProtocolPartyCustomer(id='deleted', tenant_no='t', collection_protocol_id='a', mother_name='DeletedOnly', del_flag=1),
|
|
BizCollectionHospital(id='h', tenant_no='t', region_id='region', hospital_name='Central'),
|
|
SysUser(id='sales', tenant_no='t', user_no='sales', nick_name='Salesperson'),
|
|
BizProtocolCollectionInfo(id='c', tenant_no='t', collection_protocol_id='a', hospital_id='h', sales_user_id='sales', sample_collect_time=datetime(2026, 11, 4, 10)),
|
|
])
|
|
self.db.commit()
|
|
for keyword in ('Alice', '13800000001', 'Central', 'Salesperson'):
|
|
result = self.records(keyword=keyword, pageSize=1)
|
|
self.assertEqual(result['total'], 1)
|
|
self.assertEqual(result['statusCounts'], {'sxz': 1})
|
|
self.assertEqual(result['list'][0]['motherName'], 'Alice')
|
|
self.assertEqual(result['list'][0]['collectionDate'], '2026-11-04')
|
|
for keyword in ('ForeignOnly', 'DeletedOnly', '%', '_'):
|
|
self.assertEqual(self.records(keyword=keyword)['total'], 0)
|
|
self.assertEqual(self.records(status='sxz')['statusCounts'], {'sxz': 1, 'dsx': 1})
|
|
self.assertEqual(self.records(status='sxz')['total'], 1)
|
|
self.assertEqual(self.records(keyword='Alice', page=2, pageSize=1)['list'], [])
|
|
self.db.get(BizCollectionHospital, 'h').tenant_no = 'foreign'
|
|
self.db.get(SysUser, 'sales').del_flag = 1
|
|
self.db.commit()
|
|
row = self.records(keyword='Alice')['list'][0]
|
|
self.assertEqual(row['hospitalName'], '')
|
|
self.assertEqual(row['salesUserName'], '')
|
|
self.assertEqual(self.records(keyword='Central')['total'], 0)
|
|
self.assertEqual(self.records(keyword='Salesperson')['total'], 0)
|
|
|
|
def test_donation_fields_and_statistics_contract(self):
|
|
self.db.add_all([
|
|
BizResearchDonation(id='d', tenant_no='t', donation_no='D1', donation_date='20261104'),
|
|
BizResearchDonationParty(id='p', tenant_no='t', research_donation_id='d', mother_name='Donor', mother_phone='13900000000'),
|
|
])
|
|
self.db.commit()
|
|
detail = self.records('research_donation', keyword='13900000000')['list'][0]
|
|
self.assertEqual(detail['donationDate'], '2026-11-04')
|
|
self.assertEqual(detail['motherName'], 'Donor')
|
|
old = project_management.records(self.db, 't', self.user, 'research_donation')['list'][0]
|
|
self.assertNotIn('motherName', old)
|
|
self.assertNotIn('motherPhone', old)
|
|
with self.assertRaises(HTTPException) as unsupported:
|
|
self.records('presign')
|
|
self.assertEqual(unsupported.exception.status_code, 400)
|
|
|
|
def test_pilot_and_basic_search_and_relation_boundaries(self):
|
|
self.db.add_all([
|
|
BizPilotApplication(id='pilot', tenant_no='t', application_no='P1', application_name='Trial', channel_name='Channel', channel_service_order_id='order', biz_type='pilot_service'),
|
|
BizPilotApplication(id='basic', tenant_no='t', application_no='B1', channel_service_order_id='', biz_type='basic_experiment', project_title='Study', customer_name='Customer', customer_contact='Contact', customer_phone='13600000000'),
|
|
BizChannelServiceOrder(id='order', tenant_no='t', order_name='Service', channel_id='channel', order_type='gm'),
|
|
BizPilotPatient(id='p1', tenant_no='t', pilot_application_id='pilot', patient_name='Patient', phone_no='13700000000'),
|
|
BizPilotPatient(id='p2', tenant_no='t', pilot_application_id='pilot', patient_name='Patient'),
|
|
BizPilotPatient(id='foreign', tenant_no='foreign', pilot_application_id='pilot', patient_name='ForeignPatient'),
|
|
BizPilotPatient(id='deleted', tenant_no='t', pilot_application_id='pilot', patient_name='DeletedPatient', del_flag=1),
|
|
])
|
|
self.db.commit()
|
|
for keyword in ('P1', 'Trial', 'Channel', 'Service', 'Patient', '13700000000'):
|
|
result = self.records('pilot_service', keyword=keyword)
|
|
self.assertEqual(result['total'], 1)
|
|
self.assertEqual(result['list'][0]['patientName'], 'Patient')
|
|
self.assertEqual(result['list'][0]['serviceOrderName'], 'Service')
|
|
for keyword in ('B1', 'Study', 'Customer', 'Contact', '13600000000'):
|
|
self.assertEqual(self.records('basic_experiment', keyword=keyword)['total'], 1)
|
|
self.assertEqual(self.records('pilot_service', keyword=keyword)['total'], 0)
|
|
for keyword in ('ForeignPatient', 'DeletedPatient'):
|
|
self.assertEqual(self.records('pilot_service', keyword=keyword)['total'], 0)
|
|
self.db.get(BizChannelServiceOrder, 'order').tenant_no = 'foreign'
|
|
self.db.commit()
|
|
self.assertEqual(self.records('pilot_service', keyword='Service')['total'], 0)
|
|
self.assertIsNone(self.records('pilot_service')['list'][0]['serviceOrderName'])
|
|
|
|
def test_batch_enrichment_does_not_query_per_row(self):
|
|
for index in range(20):
|
|
row_id = str(index)
|
|
self.db.add(BizCollectionProtocol(id=row_id, tenant_no='t', protocol_no=row_id))
|
|
self.db.add(BizProtocolPartyCustomer(id=row_id, tenant_no='t', collection_protocol_id=row_id, mother_name='Mother'))
|
|
self.db.commit()
|
|
statements = []
|
|
def count_statement(conn, cursor, statement, parameters, context, executemany):
|
|
statements.append(statement)
|
|
event.listen(self.engine, 'before_cursor_execute', count_statement)
|
|
try:
|
|
self.records(pageSize=1)
|
|
small = len(statements)
|
|
statements.clear()
|
|
self.assertEqual(len(self.records(pageSize=20)['list']), 20)
|
|
self.assertEqual(len(statements), small)
|
|
finally:
|
|
event.remove(self.engine, 'before_cursor_execute', count_statement)
|