"""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_follow_business_access_not_membership(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='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'), 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')