zlwu
0bddf38e52
Separate cross-team selection from attribution maintenance and apply the same scope to options, history filters and writes. Add an opt-in tenant permission without granting existing roles. Protect authorization mutation paths against self-grants, validate tenant targets and links, and include menu ancestors atomically. Cover role union, revocation, mobile writes and administration boundaries with regression tests.
149 lines
9.1 KiB
Python
149 lines
9.1 KiB
Python
"""Existing order write paths retain project attribution without changing business fields."""
|
|
import unittest
|
|
from sqlalchemy import create_engine
|
|
from sqlalchemy.orm import Session
|
|
from app.models.base import Base
|
|
from app.models.models import BizCollectionProtocol, BizPresignOrder
|
|
from app.crud import collection_protocol, presign_order
|
|
|
|
|
|
class ProjectAssignmentIntegrationTest(unittest.TestCase):
|
|
def setUp(self):
|
|
self.engine = create_engine('sqlite:///:memory:')
|
|
Base.metadata.create_all(self.engine)
|
|
self.db = Session(self.engine)
|
|
|
|
def tearDown(self):
|
|
self.db.close()
|
|
self.engine.dispose()
|
|
|
|
def test_protocol_creation_persists_assignment(self):
|
|
row = collection_protocol.create_collection_protocol(self.db, protocol_no='CP1', tenant_no='t', project_id='p')
|
|
self.assertEqual(self.db.get(BizCollectionProtocol, row.id).project_id, 'p')
|
|
|
|
def test_presign_creation_persists_assignment_and_region(self):
|
|
row = presign_order.create_presign_order(self.db, tenant_no='t', user_id='u', fields={'protocol_no':'PS1', 'project_id':'p', 'region_id':'region'})
|
|
self.assertEqual(self.db.get(BizPresignOrder, row.id).project_id, 'p')
|
|
self.assertEqual(row.region_id, 'region')
|
|
|
|
def test_edit_rejects_assigning_legacy_order_and_preserves_omitted(self):
|
|
from app.services.project_assignment import guard_project_edit
|
|
from fastapi import HTTPException
|
|
old = BizPresignOrder(project_id=None)
|
|
guard_project_edit(old, {})
|
|
with self.assertRaises(HTTPException):
|
|
guard_project_edit(old, {'projectId':'p'})
|
|
old.project_id = 'p'
|
|
guard_project_edit(old, {'project_id':'p'})
|
|
with self.assertRaises(HTTPException):
|
|
guard_project_edit(old, {'project_id':None})
|
|
|
|
def test_conversions_inherit_existing_assignment(self):
|
|
from app.crud import research_donation
|
|
from app.models.research_donation import BizResearchDonation
|
|
row = collection_protocol.create_collection_protocol(self.db, protocol_no='CONVERT1', tenant_no='t', project_id='p', protocol_status='sxz')
|
|
donation = research_donation.convert_from_storage(self.db, row.id, tenant_no='t')
|
|
self.assertEqual(donation.project_id, 'p')
|
|
# A direct donation may convert into a new storage protocol.
|
|
direct = BizResearchDonation(id='direct', tenant_no='t', donation_no='DIRECT1', donation_status='sxz', project_id='p')
|
|
self.db.add(direct)
|
|
self.db.commit()
|
|
research_donation.convert_to_storage(self.db, direct.id, tenant_no='t')
|
|
converted = self.db.query(BizCollectionProtocol).filter_by(protocol_no='DIRECT1').one()
|
|
self.assertEqual(converted.project_id, 'p')
|
|
|
|
def test_basic_experiment_creation_and_read_preserve_project_title(self):
|
|
from app.crud import pilot_application
|
|
data = dict(bizType='basic_experiment', projectId='p', projectTitle='课题', customerName='客户', customerContact='联系人', customerPhone='13800000000')
|
|
result = pilot_application.create_application(self.db, data=data, tenant_no='t', user_id='u', dept_id='')
|
|
self.assertEqual(result['projectId'], 'p')
|
|
self.assertEqual(result['projectTitle'], '课题')
|
|
|
|
def test_web_schema_keeps_explicit_null_for_edit_guard(self):
|
|
from app.api.presign_order import PresignOrderBody
|
|
from app.api.collection_protocol import ProtocolUpdateBody
|
|
from app.api.research_donation import DonationUpdateBody
|
|
from app.api.pilot_application import ApplicationSaveBody
|
|
for cls in (PresignOrderBody, ProtocolUpdateBody, DonationUpdateBody, ApplicationSaveBody):
|
|
self.assertNotIn('project_id', cls().model_dump(exclude_unset=True))
|
|
self.assertIn('project_id', cls(projectId=None).model_dump(exclude_unset=True))
|
|
|
|
def test_statistics_aggregates_before_loading_project_metadata(self):
|
|
from sqlalchemy import event
|
|
from app.models.project_management import Project, ProjectTeam
|
|
from app.services.project_management import statistics
|
|
self.db.add(ProjectTeam(id='g', tenant_no='t', team_name='组', leader_user_id='u'))
|
|
self.db.add(Project(id='p', tenant_no='t', project_no='P', project_name='项目', team_id='g', leader_user_id='u', biz_types='cell_storage', status='active'))
|
|
self.db.add_all([BizCollectionProtocol(id=str(i), tenant_no='t', protocol_no=str(i), project_id='p') for i in range(30)])
|
|
self.db.commit()
|
|
selects = []
|
|
def count(conn, cursor, statement, params, context, many):
|
|
if statement.lstrip().upper().startswith('SELECT'):
|
|
selects.append(statement)
|
|
event.listen(self.engine, 'before_cursor_execute', count)
|
|
result = statistics(self.db, 't', {'user_id':'u', 'role':'super_admin'})
|
|
self.assertEqual(result['totals']['cell_storage'], 30)
|
|
self.assertLessEqual(len(selects), 10)
|
|
|
|
def test_ordinary_edit_cannot_rewire_conversion_source(self):
|
|
from app.crud import research_donation
|
|
from app.models.research_donation import BizResearchDonation
|
|
row = collection_protocol.create_collection_protocol(self.db, protocol_no='LINK1', tenant_no='t', source_presign_order_id='original')
|
|
with self.assertRaises(ValueError):
|
|
collection_protocol.update_collection_protocol(self.db, row.id, tenant_no='t', source_presign_order_id='other')
|
|
self.db.rollback()
|
|
donation = BizResearchDonation(id='rd', tenant_no='t', donation_no='LINK2', source_type='from_storage', source_protocol_id='original')
|
|
self.db.add(donation)
|
|
self.db.commit()
|
|
with self.assertRaises(ValueError):
|
|
research_donation.update_donation(self.db, donation.id, tenant_no='t', data={'sourceProtocolId':'other'})
|
|
self.db.rollback()
|
|
with self.assertRaises(ValueError):
|
|
research_donation.update_donation(self.db, donation.id, tenant_no='t', data={'sourceType':'direct'})
|
|
|
|
def test_management_initial_empty_filters_do_not_hide_existing_rows(self):
|
|
from app.models.project_management import ProjectTeam
|
|
from app.services.project_management import list_entities
|
|
self.db.add(ProjectTeam(id='g', tenant_no='t', team_name='组', leader_user_id='u'))
|
|
self.db.commit()
|
|
self.assertEqual(list_entities(self.db, 't', ProjectTeam, keyword='', leaderUserId='', enabled=None)['total'], 1)
|
|
self.assertEqual(list_entities(self.db, 't', ProjectTeam, enabled=False)['total'], 0)
|
|
|
|
def test_history_filters_keep_stopped_projects_within_membership_and_business_access(self):
|
|
from app.models.project_management import Project, ProjectTeam
|
|
from app.models.models import SysRole, SysUserRole
|
|
from app.services.project_management import filter_options, options
|
|
self.db.add_all([
|
|
ProjectTeam(id='g', tenant_no='t', team_name='停用组', leader_user_id='u', enabled=0),
|
|
Project(id='p', tenant_no='t', project_no='P', project_name='历史项目', team_id='g', leader_user_id='other', biz_types='cell_storage', status='disabled'),
|
|
Project(id='hidden', tenant_no='t', project_no='H', project_name='不可见项目', team_id='g', leader_user_id='other', biz_types='cell_storage', status='active'),
|
|
SysRole(id='role', role_name='member', tenant_no='t', data_scope='4'),
|
|
SysUserRole(id='ur', user_id='u', role_id='role', tenant_no='t'),
|
|
BizCollectionProtocol(id='mine', protocol_no='M', tenant_no='t', create_by='u', project_id='p'),
|
|
BizCollectionProtocol(id='other', protocol_no='O', tenant_no='t', create_by='other', project_id='hidden'),
|
|
])
|
|
self.db.commit()
|
|
user = {'user_id':'u', 'role':'admin'}
|
|
self.assertEqual(options(self.db, 't', user, 'cell_storage')['list'], [])
|
|
self.assertEqual([r['id'] for r in filter_options(self.db, 't', user, 'cell_storage')['list']], ['p'])
|
|
|
|
def test_generic_history_options_require_existing_web_role_guard(self):
|
|
from app.api.project_management import options
|
|
from fastapi import HTTPException
|
|
with self.assertRaises(HTTPException) as caught:
|
|
options(bizType='cell_storage', usage='filter', keyword=None, ctx=('t', {'user_id':'u', 'role':'user'}), db=self.db)
|
|
self.assertEqual(caught.exception.status_code, 403)
|
|
|
|
def test_presign_conversion_inherits_assignment_after_project_stops(self):
|
|
from app.models.project_management import Project, ProjectTeam
|
|
self.db.add_all([
|
|
ProjectTeam(id='g', tenant_no='t', team_name='组', leader_user_id='other', enabled=0),
|
|
Project(id='p', tenant_no='t', project_no='P', project_name='历史项目', team_id='g', leader_user_id='other', biz_types='cell_storage', status='disabled'),
|
|
BizPresignOrder(id='ps', tenant_no='t', protocol_no='PSCONVERT', project_id='p', order_status='active'),
|
|
])
|
|
self.db.commit()
|
|
result = presign_order.convert_to_protocol(self.db, order_id='ps', tenant_no='t', user_id=None)
|
|
protocol = self.db.get(BizCollectionProtocol, result['protocolId'])
|
|
self.assertEqual(protocol.project_id, 'p')
|
|
self.assertEqual(protocol.source_presign_order_id, 'ps')
|