247 lines
9.3 KiB
Python
247 lines
9.3 KiB
Python
"""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())
|