Robotail/Corp
Robotail/Corp
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(api): enforce single active department membership

Browse Source
This commit is contained in:
lizerk 2026-04-14 01:18:32 +03:00
parent 2d00699831
commit 2574b47923
3 changed files with 142 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

31
backend/api/v1/employees/views/departments.py
View File

@ -27,7 +27,11 @@ from employees.models import (
EmployeeDepartment,
RoleAssignment,
)
from employees.utils import _build_links_for_dept, _validate_head_active
from employees.utils import (
_build_links_for_dept,
_other_active_department_link,
_validate_head_active,
)
from rest_framework import status, viewsets
from rest_framework.decorators import action
from rest_framework.permissions import IsAuthenticated
@ -473,7 +477,30 @@ class DepartmentViewSet(viewsets.ModelViewSet):
emp_id = payload.validated_data["employee_id"]
employee_model = Department._meta.get_field("head").remote_field.model
get_object_or_404(employee_model, id=emp_id)
employee = get_object_or_404(employee_model, id=emp_id)
if getattr(employee, "is_active", True) is False:
return Response(
{"employee_id": ["Employee is inactive."]},
status=status.HTTP_400_BAD_REQUEST,
)
other_active_link = _other_active_department_link(
emp_id,
exclude_department_id=dept.id,
)
if other_active_link is not None:
return Response(
{
"employee_id": [
(
"Employee already belongs to another active "
f"department: {other_active_link.department.name}."
)
]
},
status=status.HTTP_400_BAD_REQUEST,
)
link, created = EmployeeDepartment.objects.get_or_create(
employee_id=emp_id,

32
backend/employees/utils.py
View File

@ -17,6 +17,24 @@ from phonenumbers import PhoneNumberFormat
Employee = get_user_model()
def _other_active_department_link(
employee_id: int,
*,
exclude_department_id: int | None = None,
) -> EmployeeDepartment | None:
qs = (
EmployeeDepartment.objects.filter(
employee_id=employee_id,
is_active=True,
)
.select_related("department")
.order_by("id")
)
if exclude_department_id is not None:
qs = qs.exclude(department_id=exclude_department_id)
return qs.first()
def _to_bool(val: str | None) -> bool | None:
if val is None:
return None
@ -87,6 +105,20 @@ def _validate_head_active(
if require_email_verified and getattr(emp, "email_verified", True) is False:
return False, {"head_id": ["Employee is inactive."]}
other_active_link = _other_active_department_link(
employee_id,
exclude_department_id=dept.id,
)
if other_active_link is not None:
return False, {
"head_id": [
(
"Employee already belongs to another active department: "
f"{other_active_link.department.name}."
)
]
}
return True, None

81
backend/tests/api/v1/employees/test_departments.py
View File

@ -514,6 +514,58 @@ def test_set_head_requires_perm(api_client: APIClient):
== status.HTTP_403_FORBIDDEN
)
def test_set_head_rejects_employee_from_another_active_department(
api_client: APIClient,
):
d1 = Department.objects.create(name="Dept 1")
d2 = Department.objects.create(name="Dept 2")
manager = make_user("head-manager@example.com")
api_client.force_authenticate(user=manager)
r_change = make_role(d1, "head-assigner", ["change_department_head"])
EmployeeDepartment.objects.create(
employee=manager,
department=d1,
is_active=True,
role=r_change,
)
candidate = make_user("foreign-head@example.com")
EmployeeDepartment.objects.create(
employee=candidate,
department=d2,
is_active=True,
)
url = reverse("api:v1:departments-set-head", args=[d1.pk])
resp = api_client.post(url, {"head_id": candidate.id}, format="json")
assert resp.status_code == status.HTTP_400_BAD_REQUEST
assert "another active department" in resp.json()["head_id"][0]
def test_set_head_makes_head_active_department_member(api_client: APIClient):
d = Department.objects.create(name="Dept Head Link")
manager = make_user("head-manager-2@example.com")
api_client.force_authenticate(user=manager)
r_change = make_role(d, "head-assigner", ["change_department_head"])
EmployeeDepartment.objects.create(
employee=manager,
department=d,
is_active=True,
role=r_change,
)
candidate = make_user("new-head@example.com")
url = reverse("api:v1:departments-set-head", args=[d.pk])
resp = api_client.post(url, {"head_id": candidate.id}, format="json")
assert resp.status_code == status.HTTP_200_OK
link = EmployeeDepartment.objects.get(employee=candidate, department=d)
assert link.is_active is True
# ---------- tests: action set_member_role ----------
def test_set_member_role_happy_path(api_client: APIClient):
@ -592,3 +644,32 @@ def test_set_member_role_missing_employee_id(api_client: APIClient):
url = reverse("api:v1:departments-set-member-role", args=[d.pk])
resp = api_client.post(url, {"role_id": None}, format="json")
assert resp.status_code == status.HTTP_400_BAD_REQUEST
def test_add_member_rejects_employee_from_another_active_department(
api_client: APIClient,
):
d1 = Department.objects.create(name="Target Dept")
d2 = Department.objects.create(name="Other Dept")
manager = make_user("dept-manager@example.com")
api_client.force_authenticate(user=manager)
r_manage = make_role(d1, "manager", ["manage_department"])
EmployeeDepartment.objects.create(
employee=manager,
department=d1,
is_active=True,
role=r_manage,
)
employee = make_user("busy-member@example.com")
EmployeeDepartment.objects.create(
employee=employee,
department=d2,
is_active=True,
)
url = reverse("api:v1:departments-add-member", args=[d1.pk])
resp = api_client.post(url, {"employee_id": employee.id}, format="json")
assert resp.status_code == status.HTTP_400_BAD_REQUEST
assert "another active department" in resp.json()["employee_id"][0]