"""Requisition-owner job/candidate scope — unit + live DB checks. Run from the API container (PYTHONPATH=/app): python tests/test_requisition_scope.py """ from __future__ import annotations import asyncio import sys from sqlalchemy import select from job.candidate.views import owned_job_ids_for_candidate_scope from job.job_post.models import JobPosts from job.job_post.views import JobPost from role.models import Roles from users.models import Users from users.permissions import ( scopes_to_own_requisitions, sees_all_candidates, is_hiring_manager, is_admin, ) def _ok(name: str, cond: bool, extra: str = "") -> None: global failed if cond: print(f"ok {name}" + (f" {extra}" if extra else "")) else: failed += 1 print(f"FAIL {name}" + (f" {extra}" if extra else "")) failed = 0 def test_helper_unit() -> None: recruiter = { "id": "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", "role_name": "recruiter", "permissions": ["candidates.view", "jobs.view"], } _ok("recruiter is not requisition-scoped", not scopes_to_own_requisitions(recruiter)) _ok("recruiter does not see all candidates", not sees_all_candidates(recruiter)) custom = { **recruiter, "role_name": "AI_TEAM_MANAGER", "permissions": ["candidates.view", "jobs.view", "requisitions.create"], } _ok("requisitions.create alone does not scope jobs/candidates", not scopes_to_own_requisitions(custom)) _ok("custom role is not hiring-manager portal", not is_hiring_manager(custom)) _ok("custom role is not admin", not is_admin(custom)) _ok("custom role does not see all candidates", not sees_all_candidates(custom)) configured = { **custom, "permissions": ["candidates.view", "jobs.view", "requisitions.create", "requisitions.configure"], } _ok("Access Control requisitions.configure enables the scope", scopes_to_own_requisitions(configured)) manage = { **custom, "permissions": ["candidates.view", "candidates.manage", "requisitions.configure"], } _ok( "candidates.manage wins over requisitions.configure", not scopes_to_own_requisitions(manage) and sees_all_candidates(manage), ) admin = { "id": "bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", "role_name": "admin", "permissions": ["requisitions.create", "requisitions.manage"], } _ok("admin is not requisition-scoped", not scopes_to_own_requisitions(admin) and is_admin(admin)) hm = { "id": "cccccccc-cccc-cccc-cccc-cccccccccccc", "role_name": "hiring_manager", "permissions": ["candidates.view"], } _ok("hiring_manager is requisition-scoped", scopes_to_own_requisitions(hm) and is_hiring_manager(hm)) manager_named = {**hm, "role_name": "Manager"} _ok("Manager role name is requisition-scoped", scopes_to_own_requisitions(manager_named)) req_manage = { "id": "dddddddd-dddd-dddd-dddd-dddddddddddd", "role_name": "ops_lead", "permissions": ["requisitions.manage", "candidates.view"], } _ok( "requisitions.manage is admin, not this scope", is_admin(req_manage) and not scopes_to_own_requisitions(req_manage), ) async def test_live_db() -> None: from candidate_forms.models import Requisition from db_setup import session_scope from inbox.models import Inbox_Messages from job.candidate.views import CandidateView async with session_scope() as session: # Job opened from a requisition this user created, but job_posts.created_by # is someone else: recruiter/creator scope hides it; requisition scope # must still show it (recruiter assignment on the job is irrelevant). stmt = ( select(JobPosts, Requisition) .join(Requisition, JobPosts.requisition_id == Requisition.id) .where( JobPosts.is_deleted == False, # noqa: E712 Requisition.is_deleted == False, # noqa: E712 Requisition.created_by != JobPosts.created_by, ) .limit(1) ) row = (await session.execute(stmt)).first() if not row: print("skip live job whose requisition creator is not job_posts.created_by") else: job, req = row owner = { "id": str(req.created_by), "role_name": "AI_TEAM_MANAGER", "permissions": ["candidates.view", "jobs.view", "requisitions.configure"], } recruiter_only = { "id": str(req.created_by), "role_name": "recruiter", "permissions": ["candidates.view", "jobs.view"], } manage_all = { "id": str(req.created_by), "role_name": "AI_TEAM_MANAGER", "permissions": ["candidates.view", "candidates.manage"], } owned = await owned_job_ids_for_candidate_scope(session, owner) creator = await owned_job_ids_for_candidate_scope(session, recruiter_only) unscoped = await owned_job_ids_for_candidate_scope(session, manage_all) _ok( "requisition owner sees job they did not create as job_posts.created_by", job.id in set(owned or []), f"job={job.title!r} req_owner={req.created_by} job_created_by={job.created_by} recruiter={job.current_recruiter_id}", ) _ok( "without requisitions.create, recruiter/creator filter hides that job", job.id not in set(creator or []), ) _ok("candidates.manage leaves the list unscoped", unscoped is None) applicants = await Inbox_Messages.counts_by_job_post_ids(session, [job.id]) n = applicants.get(str(job.id), 0) print(f"info applicants on that job: {n}") service = JobPost(session) jobs_rows, jobs_total = await service.fetch_jobs( active_only=False, top=500, skip=0, current_user=owner ) job_ids = {str(r.get("id") or "") for r in jobs_rows} _ok( "/jobs/fetch for requisition owner includes the job", str(job.id) in job_ids, f"total={jobs_total}", ) _, all_total = await service.fetch_jobs( active_only=False, top=500, skip=0, current_user=manage_all ) _ok( "candidates.manage /jobs/fetch is wider than requisition scope", all_total > jobs_total, f"scoped={jobs_total} unscoped={all_total}", ) cand_view = CandidateView(session) scoped_rows = await cand_view.get_candidate( current_user=owner, limit=100, offset=0 ) all_cand = await cand_view.get_candidate( current_user=manage_all, limit=100, offset=0 ) scoped_n = len(scoped_rows) if isinstance(scoped_rows, list) else 0 all_n = len(all_cand) if isinstance(all_cand, list) else 0 _ok( "candidate list for requisition owner is non-empty when the job has applicants", (n == 0) or scoped_n > 0, f"scoped_candidates={scoped_n}", ) _ok( "candidates.manage sees at least the requisition-scoped rows", all_n >= scoped_n, f"scoped={scoped_n} unscoped={all_n}", ) user = await Users.get_user_by_email(session, "new@utopiabrands.com") if not user: print("skip live new@utopiabrands.com (user missing)") return tags = await Roles.resolve_tags(session, user.role) role = getattr(user, "role", None) live = { "id": str(user.id), "role_name": getattr(role.role_name, "value", role.role_name) if role is not None else None, "permissions": list(tags), } print( f"info live user {user.email} role={live['role_name']!r} " f"requisitions.create={'requisitions.create' in tags} " f"requisitions.configure={'requisitions.configure' in tags} " f"candidates.manage={'candidates.manage' in tags}" ) _ok( "live AI Team Manager is not hiring-manager portal", not is_hiring_manager(live), f"role={live['role_name']!r}", ) scoped = scopes_to_own_requisitions(live) print(f"info scopes_to_own_requisitions={scoped}") owned = await owned_job_ids_for_candidate_scope(session, live) if owned is None: print("info live candidate list is unscoped (admin or candidates.manage)") else: print(f"info live owned job ids: {len(owned)}") if "requisitions.configure" not in tags and not scoped: print( "info tick Requisitions → Configure on this role in Access Control " "to enable requisition-owner job/candidate scope" ) def main() -> int: test_helper_unit() asyncio.run(test_live_db()) if failed: print(f"\n{failed} failed") return 1 print("\nall passed") return 0 if __name__ == "__main__": sys.exit(main())