feat(tests): operator assignment — two queues for techs, assign endpoint for leads

backend/alembic/versions/b050_add_operator_assignees.py

@@ -0,0 +1,24 @@
+"""Add red_tech_assignee and blue_tech_assignee to tests.
+
+Revision ID: b050
+Revises: b049
+Create Date: 2026-06-19
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects.postgresql import UUID
+
+revision = "b050"
+down_revision = "b049"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    op.add_column("tests", sa.Column("red_tech_assignee", UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=True))
+    op.add_column("tests", sa.Column("blue_tech_assignee", UUID(as_uuid=True), sa.ForeignKey("users.id"), nullable=True))
+
+
+def downgrade() -> None:
+    op.drop_column("tests", "blue_tech_assignee")
+    op.drop_column("tests", "red_tech_assignee")

backend/app/models/test.py

@@ -129,6 +129,14 @@ class Test(Base):
     blue_validator = relationship("User", foreign_keys=[blue_validated_by])
     # Assign remediation_user = relationship("User", foreign_keys=[remediation_assignee])
     remediation_user = relationship("User", foreign_keys=[remediation_assignee])
+
+    # ── Assignment fields ──────────────────────────────────────────
+    red_tech_assignee = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
+    blue_tech_assignee = Column(UUID(as_uuid=True), ForeignKey("users.id"), nullable=True)
+
+    red_tech_assigned_user = relationship("User", foreign_keys=[red_tech_assignee])
+    blue_tech_assigned_user = relationship("User", foreign_keys=[blue_tech_assignee])
+
     # Assign original_test = relationship("Test", remote_side="Test.id", foreign_keys=[retest_of])
     original_test = relationship("Test", remote_side="Test.id", foreign_keys=[retest_of])
     # Assign retests = relationship("Test", foreign_keys=[retest_of], back_populates="orig...

backend/app/routers/tests.py

@@ -51,6 +51,7 @@ from app.models.user import User
 
 # Import  from app.schemas.test
 from app.schemas.test import (
+    TestAssign,
     TestBlueUpdate,
     TestBlueValidate,
     TestClassificationUpdate,
@@ -563,6 +564,15 @@ def update_test_red(
     Returns:
         TestOut: The updated test with refreshed red-team field values.
     """
+    # Assignee lock: red_tech cannot work a test assigned to someone else
+    _pre_test = crud_get_test_or_raise(db, test_id)
+    if (
+        _pre_test.red_tech_assignee is not None
+        and current_user.role == "red_tech"
+        and _pre_test.red_tech_assignee != current_user.id
+    ):
+        raise HTTPException(status_code=403, detail="Test is assigned to another operator")
+
     # Assign update_data = payload.model_dump(exclude_unset=True)
     update_data = payload.model_dump(exclude_unset=True)
     # Open context manager
@@ -620,6 +630,15 @@ def update_test_blue(
     Returns:
         TestOut: The updated test with refreshed blue-team field values.
     """
+    # Assignee lock: blue_tech cannot work a test assigned to someone else
+    _pre_test = crud_get_test_or_raise(db, test_id)
+    if (
+        _pre_test.blue_tech_assignee is not None
+        and current_user.role == "blue_tech"
+        and _pre_test.blue_tech_assignee != current_user.id
+    ):
+        raise HTTPException(status_code=403, detail="Test is assigned to another operator")
+
     # Assign update_data = payload.model_dump(exclude_unset=True)
     update_data = payload.model_dump(exclude_unset=True)
     # Open context manager
@@ -676,10 +695,22 @@ def start_execution(
     """
     # Assign test = crud_get_test_or_raise(db, test_id)
     test = crud_get_test_or_raise(db, test_id)
+
+    # Assignee lock: red_tech cannot start a test assigned to someone else
+    if (
+        test.red_tech_assignee is not None
+        and current_user.role == "red_tech"
+        and test.red_tech_assignee != current_user.id
+    ):
+        raise HTTPException(status_code=403, detail="Test is assigned to another operator")
+
     # Open context manager
     with UnitOfWork(db) as uow:
         # Assign test = wf_start_execution(db, test, current_user)
         test = wf_start_execution(db, test, current_user)
+        # Auto-assign: if no assignee set, claim this test for the current operator
+        if test.red_tech_assignee is None:
+            test.red_tech_assignee = current_user.id
         # Call uow.commit()
         uow.commit()
     # Reload ORM object attributes from the database
@@ -715,6 +746,15 @@ def submit_red(
     """
     # Assign test = crud_get_test_or_raise(db, test_id)
     test = crud_get_test_or_raise(db, test_id)
+
+    # Assignee lock: red_tech cannot submit a test assigned to someone else
+    if (
+        test.red_tech_assignee is not None
+        and current_user.role == "red_tech"
+        and test.red_tech_assignee != current_user.id
+    ):
+        raise HTTPException(status_code=403, detail="Test is assigned to another operator")
+
     # Open context manager
     with UnitOfWork(db) as uow:
         # Assign test = wf_submit_red(db, test, current_user)
@@ -754,6 +794,15 @@ def submit_blue(
     """
     # Assign test = crud_get_test_or_raise(db, test_id)
     test = crud_get_test_or_raise(db, test_id)
+
+    # Assignee lock: blue_tech cannot submit a test assigned to someone else
+    if (
+        test.blue_tech_assignee is not None
+        and current_user.role == "blue_tech"
+        and test.blue_tech_assignee != current_user.id
+    ):
+        raise HTTPException(status_code=403, detail="Test is assigned to another operator")
+
     # Open context manager
     with UnitOfWork(db) as uow:
         # Assign test = wf_submit_blue(db, test, current_user)
@@ -779,8 +828,20 @@ def start_blue_work(
 ):
     """Blue tech picks up the test to start evaluating. Sets the Tempo timer start."""
     test = crud_get_test_or_raise(db, test_id)
+
+    # Assignee lock: blue_tech cannot pick up a test assigned to someone else
+    if (
+        test.blue_tech_assignee is not None
+        and current_user.role == "blue_tech"
+        and test.blue_tech_assignee != current_user.id
+    ):
+        raise HTTPException(status_code=403, detail="Test is assigned to another operator")
+
     with UnitOfWork(db) as uow:
         test = wf_start_blue_work(db, test, current_user)
+        # Auto-assign: if no assignee set, claim this test for the current operator
+        if test.blue_tech_assignee is None:
+            test.blue_tech_assignee = current_user.id
         uow.commit()
     db.refresh(test)
     return test
@@ -1017,6 +1078,48 @@ def reopen(
     return test
 
 
+# ---------------------------------------------------------------------------
+# POST /tests/{id}/assign  —  assign red_tech / blue_tech operators (leads + admin)
+# ---------------------------------------------------------------------------
+
+
+@router.post("/{test_id}/assign", response_model=TestOut)
+def assign_test_operators(
+    test_id: uuid.UUID,
+    payload: TestAssign,
+    db: Session = Depends(get_db),
+    current_user: User = Depends(require_any_role("admin", "red_lead", "blue_lead")),
+):
+    """Assign red_tech and/or blue_tech operators to a test. Admin/leads only."""
+    test = crud_get_test_or_raise(db, test_id)
+
+    if payload.red_tech_assignee is not None:
+        u = db.query(User).filter(User.id == payload.red_tech_assignee).first()
+        if not u or u.role not in ("red_tech", "red_lead", "admin"):
+            raise HTTPException(status_code=400, detail="Invalid red tech assignee")
+        test.red_tech_assignee = payload.red_tech_assignee
+
+    if payload.blue_tech_assignee is not None:
+        u = db.query(User).filter(User.id == payload.blue_tech_assignee).first()
+        if not u or u.role not in ("blue_tech", "blue_lead", "admin"):
+            raise HTTPException(status_code=400, detail="Invalid blue tech assignee")
+        test.blue_tech_assignee = payload.blue_tech_assignee
+
+    # Handle intentional null (clearing) — model_fields_set tracks which keys were sent
+    if "red_tech_assignee" in payload.model_fields_set and payload.red_tech_assignee is None:
+        test.red_tech_assignee = None
+    if "blue_tech_assignee" in payload.model_fields_set and payload.blue_tech_assignee is None:
+        test.blue_tech_assignee = None
+
+    log_action(db, current_user.id, "assign_test", str(test_id), {
+        "red_tech_assignee": str(payload.red_tech_assignee) if payload.red_tech_assignee else None,
+        "blue_tech_assignee": str(payload.blue_tech_assignee) if payload.blue_tech_assignee else None,
+    })
+    db.commit()
+    db.refresh(test)
+    return test
+
+
 # ---------------------------------------------------------------------------
 # PATCH /tests/{id}/remediation  —  update remediation fields
 # ---------------------------------------------------------------------------

backend/app/schemas/test.py

@@ -134,6 +134,13 @@ class TestRemediationUpdate(BaseModel):
     remediation_assignee: uuid.UUID | None = None
 
 
+class TestAssign(BaseModel):
+    """Payload for assigning operators to a test."""
+
+    red_tech_assignee: uuid.UUID | None = None
+    blue_tech_assignee: uuid.UUID | None = None
+
+
 # ── Legacy validate (kept for backwards compat) ────────────────────
 
 
@@ -221,6 +228,10 @@ class TestOut(BaseModel):
     # Assign remediation_assignee = None
     remediation_assignee: uuid.UUID | None = None
 
+    # Assignment fields
+    red_tech_assignee: uuid.UUID | None = None
+    blue_tech_assignee: uuid.UUID | None = None
+
     # Re-test fields
     retest_of: uuid.UUID | None = None
     # Assign retest_count = 0