"""Project choices and writes share explicit team/all-project permission boundaries.""" import unittest from fastapi import HTTPException from sqlalchemy import create_engine from sqlalchemy.orm import Session from app.models.base import Base from app.models.models import (BizCollectionProtocol, SysMenu, SysRole, SysRoleMenu, SysUserRole, SysPost, SysPostRole, SysUserPost) from app.models.project_management import Project, ProjectTeam, ProjectTeamMember from app.services import project_management as svc from app.services.project_assignment import create_assignment class ProjectSelectionPermissionsTest(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='member', role='admin', tenant_no='t') kinds = ','.join(svc.COUNT_TYPES) self.db.add_all([ ProjectTeam(id='own', tenant_no='t', team_name='Own', leader_user_id='leader'), ProjectTeam(id='other', tenant_no='t', team_name='Other', leader_user_id='other'), ProjectTeam(id='foreign', tenant_no='foreign', team_name='Foreign', leader_user_id='member'), ProjectTeamMember(id='member', tenant_no='t', team_id='own', user_id='member'), Project(id='own-p', tenant_no='t', project_no='OWN', project_name='Own', team_id='own', leader_user_id='other', status='active', biz_types=kinds), Project(id='other-p', tenant_no='t', project_no='OTHER', project_name='Other', team_id='other', leader_user_id='other', status='active', biz_types=kinds), Project(id='foreign-p', tenant_no='foreign', project_no='FOREIGN', project_name='Foreign', team_id='foreign', leader_user_id='member', status='active', biz_types=kinds), ]) self.db.commit() def tearDown(self): self.db.close() self.engine.dispose() def grant(self, code, suffix='all', *, inherited=False): self.db.add_all([ SysRole(id=suffix, tenant_no='t', role_name=suffix, data_scope='4'), SysMenu(id=suffix, tenant_no='t', menu_name=suffix, menu_type='F', visible='0', perms=code), SysRoleMenu(id=suffix, tenant_no='t', role_id=suffix, menu_id=suffix), ]) if inherited: self.db.add_all([ SysPost(id=suffix, tenant_no='t', post_name=suffix, post_code=suffix), SysPostRole(id=suffix, tenant_no='t', post_id=suffix, role_id=suffix), SysUserPost(id=suffix, tenant_no='t', user_id='member', post_id=suffix), ]) else: self.db.add(SysUserRole(id=suffix, tenant_no='t', user_id='member', role_id=suffix)) self.db.commit() def choices(self, user=None, kind='cell_storage'): return {r['id'] for r in svc.options(self.db, 't', user or self.user, kind)['list']} def assert_forbidden(self, project_id='other-p', kind='cell_storage'): with self.assertRaises(HTTPException) as caught: create_assignment(self.db, 't', self.user, project_id, kind) self.assertEqual(caught.exception.status_code, 403) def test_project_leader_alone_does_not_bypass_team_membership(self): self.db.get(Project, 'other-p').leader_user_id = 'member' self.db.commit() self.assertEqual(self.choices(), {'own-p'}) self.assert_forbidden() def test_team_leader_and_membership_union(self): self.assertEqual(self.choices(dict(user_id='leader', role='admin')), {'own-p'}) self.db.add(ProjectTeamMember(id='second', tenant_no='t', team_id='other', user_id='member')) self.db.commit() self.assertEqual(self.choices(), {'own-p', 'other-p'}) self.db.get(ProjectTeamMember, 'second').del_flag = 1 self.db.commit() self.assertEqual(self.choices(), {'own-p'}) self.assert_forbidden() def test_admin_with_no_team_has_no_choices(self): self.assertEqual(self.choices(dict(user_id='outsider', role='admin')), set()) def test_assignment_edit_does_not_grant_all_projects(self): self.grant('project:assignment:edit', 'assignment') self.assertEqual(self.choices(), {'own-p'}) self.assert_forbidden() def test_all_permission_in_role_union_allows_all_business_types(self): self.grant('project:assignment:edit', 'assignment') self.grant('project:project:all') for kind in svc.BUSINESS: with self.subTest(kind=kind): self.assertEqual(self.choices(kind=kind), {'own-p', 'other-p'}) self.assertEqual(create_assignment(self.db, 't', self.user, 'other-p', kind), 'other-p') with self.assertRaises(HTTPException) as caught: create_assignment(self.db, 't', self.user, 'foreign-p', 'cell_storage') self.assertEqual(caught.exception.status_code, 404) def test_all_permission_does_not_bypass_project_state_or_business_type(self): self.grant('project:project:all') self.db.get(Project, 'other-p').status = 'disabled' self.db.commit() self.assertEqual(self.choices(), {'own-p'}) with self.assertRaises(HTTPException): create_assignment(self.db, 't', self.user, 'other-p', 'cell_storage') self.db.get(Project, 'other-p').status = 'active' self.db.get(Project, 'other-p').biz_types = 'basic_experiment' self.db.commit() self.assertEqual(self.choices(), {'own-p'}) with self.assertRaises(HTTPException): create_assignment(self.db, 't', self.user, 'other-p', 'cell_storage') def test_permission_revocation_is_effective_on_next_request(self): self.grant('project:project:all') self.assertEqual(self.choices(), {'own-p', 'other-p'}) for model, field, disabled, enabled in [ (SysRole, 'status', '0', '1'), (SysRoleMenu, 'del_flag', 1, 0), (SysMenu, 'status', '0', '1'), (SysUserRole, 'del_flag', 1, 0), ]: with self.subTest(model=model.__name__): setattr(self.db.get(model, 'all'), field, disabled) self.db.commit() self.assertEqual(self.choices(), {'own-p'}) self.assert_forbidden() setattr(self.db.get(model, 'all'), field, enabled) self.db.commit() def test_post_inherited_all_permission_and_disabled_post(self): self.grant('project:project:all', inherited=True) self.assertEqual(self.choices(), {'own-p', 'other-p'}) self.assertEqual(create_assignment(self.db, 't', self.user, 'other-p', 'presign'), 'other-p') self.db.get(SysPost, 'all').status = '0' self.db.commit() self.assertEqual(self.choices(), {'own-p'}) self.assert_forbidden() def test_filter_options_require_team_scope_and_preserve_business_scope(self): self.grant('project:assignment:edit', 'assignment') self.db.add_all([ BizCollectionProtocol(id='own-record', protocol_no='OWN', tenant_no='t', create_by='member', project_id='own-p'), BizCollectionProtocol(id='other-record', protocol_no='OTHER', tenant_no='t', create_by='member', project_id='other-p'), BizCollectionProtocol(id='hidden-record', protocol_no='HIDDEN', tenant_no='t', create_by='other', project_id='own-p'), ]) self.db.commit() def filters(**kwargs): return {r['id'] for r in svc.filter_options(self.db, 't', self.user, 'cell_storage', **kwargs)['list']} self.assertEqual(filters(), {'own-p'}) self.grant('project:project:all') self.assertEqual(filters(), {'own-p', 'other-p'}) self.assertEqual(svc.records(self.db, 't', self.user, 'cell_storage')['total'], 2) # Mobile passes its stricter business query to the same options service. mobile_scope = self.db.query(BizCollectionProtocol).filter(BizCollectionProtocol.id == 'own-record') self.assertEqual(filters(business_scope=mobile_scope), {'own-p'}) def test_cross_group_reassignment_needs_both_permissions(self): self.grant('project:assignment:edit', 'assignment') self.db.add(BizCollectionProtocol(id='record', protocol_no='R', tenant_no='t', create_by='member', project_id='own-p')) self.db.commit() with self.assertRaises(HTTPException) as caught: svc.preview_assignment(self.db, 't', self.user, 'cell_storage', 'record', 'other-p') self.assertEqual(caught.exception.status_code, 403) self.grant('project:project:all') preview = svc.preview_assignment(self.db, 't', self.user, 'cell_storage', 'record', 'other-p') self.db.get(SysRoleMenu, 'all').del_flag = 1 self.db.commit() with self.assertRaises(HTTPException) as caught: svc.assign(self.db, 't', self.user, 'cell_storage', 'record', 'other-p', 'reason', preview['records']) self.assertEqual(caught.exception.status_code, 403) self.assertEqual(self.db.get(BizCollectionProtocol, 'record').project_id, 'own-p') def test_all_permission_does_not_grant_maintenance_or_management_actions(self): self.grant('project:project:all') for code in ('project:assignment:edit', 'project:project:edit', 'project:team:edit'): with self.subTest(code=code), self.assertRaises(HTTPException): svc.check_permission(self.db, 't', self.user, code) def test_super_admin_retains_tenant_boundary(self): self.assertEqual(self.choices(dict(user_id='root', role='super_admin')), {'own-p', 'other-p'})