b427ee9f49
Multi-tenancy isolation (P0):
- MT-3: Add get_job_or_403 (org membership check) to all 19+ job action endpoints
- MT-6: Same gate added to all review_notes (5) and vtt_versions (4) handlers
- MT-7: WebSocket /ws/jobs/{job_id} closes with 4403 on org mismatch;
/ws/jobs passes accessible_org_ids to ConnectionManager; server-side
keepalive at 20 s (asyncio.wait_for timeout) prevents proxy idle drops
- MT-8: list_users scoped to org memberships for non-platform-admins
WebSocket fixes (Mod Comms 2026-03-18 incident):
- Frontend heartbeat lowered 30 000 → 20 000 ms (was at Apache timeout edge)
- Terminal close codes 4001/4003/4004/4403 no longer trigger reconnect loop
- Silently discard server "keepalive" frames alongside existing "pong"
English-first QC (P1):
- _assert_can_approve blocks target language approval until source is APPROVED
- PRODUCTION/ADMIN roles bypass the gate
- Source VTT edits reset stale APPROVED/PENDING_REVIEW/IN_REVIEW target states
Tests (all passing):
- backend/tests/unit/test_language_qc_english_first.py (15 cases)
- backend/tests/unit/test_routes_jobs_org_isolation.py (12 cases)
- backend/tests/unit/test_review_notes_org_isolation.py (16 parametrized cases)
- backend/tests/unit/test_vtt_versions_org_isolation.py (16 parametrized cases)
- backend/tests/unit/test_websocket_org_isolation.py (11 cases)
- backend/tests/unit/test_admin_users_org_filter.py (7 cases)
- frontend: useJobStatusWebSocket.terminal.test.ts (9 cases)
- frontend: useJobStatusWebSocket.heartbeat.test.ts (9 cases)
Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
97 lines
3.9 KiB
Python
97 lines
3.9 KiB
Python
"""
|
|
Review notes org isolation tests (MT-6).
|
|
|
|
All 5 review_notes handlers delegate to get_job_or_403 for org enforcement.
|
|
These tests verify cross-org access is blocked and same-org access is allowed
|
|
by calling get_job_or_403 directly — the same mechanism used by the handlers.
|
|
"""
|
|
|
|
from unittest.mock import AsyncMock, MagicMock
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from app.core.authz import MembershipContext, get_job_or_403
|
|
from app.models.organization import OrgRole
|
|
from app.models.user import User, UserRole
|
|
|
|
# ── Helpers ────────────────────────────────────────────────────────────────────
|
|
|
|
def _make_user(role: UserRole, user_id: str = "user-a") -> User:
|
|
return User(
|
|
**{
|
|
"_id": user_id,
|
|
"email": f"{user_id}@example.com",
|
|
"full_name": "Test User",
|
|
"role": role,
|
|
"is_active": True,
|
|
}
|
|
)
|
|
|
|
|
|
def _make_ctx(user: User, memberships: dict[str, OrgRole]) -> MembershipContext:
|
|
return MembershipContext(user=user, is_platform_admin=False, memberships=memberships)
|
|
|
|
|
|
def _make_db(job: dict | None) -> MagicMock:
|
|
db = MagicMock()
|
|
db.jobs.find_one = AsyncMock(return_value=job)
|
|
db.projects.find_one = AsyncMock(return_value=None)
|
|
return db
|
|
|
|
|
|
# ── Handler gate coverage ──────────────────────────────────────────────────────
|
|
|
|
@pytest.mark.parametrize("handler_name", [
|
|
"list_notes (GET /jobs/{job_id}/notes)",
|
|
"create_note (POST /jobs/{job_id}/notes)",
|
|
"get_note (GET /jobs/{job_id}/notes/{note_id})",
|
|
"update_note (PATCH /jobs/{job_id}/notes/{note_id})",
|
|
"delete_note (DELETE /jobs/{job_id}/notes/{note_id})",
|
|
])
|
|
class TestReviewNotesOrgGate:
|
|
"""
|
|
All 5 review_notes handlers call get_job_or_403(job_id, ctx, db) as their first
|
|
action. These parametrized tests confirm the gate blocks cross-org access and
|
|
allows same-org access — mirroring what each handler would encounter.
|
|
"""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_same_org_passes(self, handler_name: str):
|
|
user = _make_user(UserRole.REVIEWER)
|
|
ctx = _make_ctx(user, {"org-a": OrgRole.MEMBER})
|
|
db = _make_db({"_id": "job-1", "organization_id": "org-a"})
|
|
|
|
result = await get_job_or_403("job-1", ctx, db)
|
|
assert result["_id"] == "job-1", f"{handler_name}: same-org should pass"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cross_org_raises_404(self, handler_name: str):
|
|
user = _make_user(UserRole.REVIEWER)
|
|
ctx = _make_ctx(user, {"org-a": OrgRole.MEMBER})
|
|
db = _make_db({"_id": "job-1", "organization_id": "org-b"})
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
await get_job_or_403("job-1", ctx, db)
|
|
assert exc.value.status_code == 404, (
|
|
f"{handler_name}: cross-org must return 404 (not 403) to avoid leaking job existence"
|
|
)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_missing_job_raises_404(self, handler_name: str):
|
|
user = _make_user(UserRole.REVIEWER)
|
|
ctx = _make_ctx(user, {"org-a": OrgRole.MEMBER})
|
|
db = _make_db(None)
|
|
|
|
with pytest.raises(HTTPException) as exc:
|
|
await get_job_or_403("job-missing", ctx, db)
|
|
assert exc.value.status_code == 404, f"{handler_name}: missing job must 404"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_platform_admin_bypasses_org_check(self, handler_name: str):
|
|
user = _make_user(UserRole.ADMIN)
|
|
ctx = MembershipContext(user=user, is_platform_admin=True, memberships={})
|
|
db = _make_db({"_id": "job-1", "organization_id": "org-x"})
|
|
|
|
result = await get_job_or_403("job-1", ctx, db)
|
|
assert result["_id"] == "job-1", f"{handler_name}: platform admin should bypass"
|