video-accessibility/backend/tests/unit/test_websocket_org_isolation.py

"""
WebSocket org isolation tests (MT-7).

Tests _can_access_org helper and the org-check logic baked into the
websocket_job_status handler. Uses unit-level testing of the helpers
rather than spinning up a full ASGI server.
"""

from unittest.mock import AsyncMock, MagicMock

import pytest

from app.api.v1.routes_websockets import _can_access_org

# ── _can_access_org helper ─────────────────────────────────────────────────────

class TestCanAccessOrg:
    def test_platform_admin_none_memberships_always_true(self):
        assert _can_access_org("org-x", None) is True

    def test_none_org_id_always_true(self):
        """Legacy job with no org_id — pass through (further checks done in handler)."""
        assert _can_access_org(None, {}) is True
        assert _can_access_org(None, {"org-a": "member"}) is True

    def test_org_in_memberships_returns_true(self):
        from app.models.organization import OrgRole
        memberships = {"org-a": OrgRole.MEMBER, "org-b": OrgRole.OWNER}
        assert _can_access_org("org-a", memberships) is True
        assert _can_access_org("org-b", memberships) is True

    def test_org_not_in_memberships_returns_false(self):
        from app.models.organization import OrgRole
        memberships = {"org-a": OrgRole.MEMBER}
        assert _can_access_org("org-b", memberships) is False
        assert _can_access_org("org-c", memberships) is False

    def test_empty_memberships_returns_false_for_any_org(self):
        assert _can_access_org("org-x", {}) is False

    def test_string_memberships_dict_accepted(self):
        """_can_access_org only uses `in` operator — any dict-like works."""
        memberships = {"org-a": "member"}
        assert _can_access_org("org-a", memberships) is True
        assert _can_access_org("org-b", memberships) is False


# ── Org close-code logic ───────────────────────────────────────────────────────

class TestWebSocketOrgCloseCode:
    """
    Verify that the handler emits 4403 for org-denied jobs by simulating
    the relevant logic path with mocks.
    """

    @pytest.mark.asyncio
    async def test_org_denied_closes_with_4403(self):
        """When _can_access_org returns False, handler must close with code 4403."""
        # Simulate the decision: user is in org-a, job is in org-b
        job_org = "org-b"
        memberships = {"org-a": "member"}  # org-b not in memberships

        can_access = _can_access_org(job_org, memberships)
        assert can_access is False

        # Simulate what the handler does:
        mock_websocket = MagicMock()
        mock_websocket.close = AsyncMock()

        if not can_access:
            await mock_websocket.close(code=4403, reason="Org access denied")

        mock_websocket.close.assert_called_once_with(code=4403, reason="Org access denied")

    @pytest.mark.asyncio
    async def test_org_allowed_does_not_close(self):
        """When _can_access_org returns True, handler proceeds to connect."""
        job_org = "org-a"
        memberships = {"org-a": "member"}

        can_access = _can_access_org(job_org, memberships)
        assert can_access is True

        mock_websocket = MagicMock()
        mock_websocket.close = AsyncMock()

        if not can_access:
            await mock_websocket.close(code=4403, reason="Org access denied")

        mock_websocket.close.assert_not_called()

    @pytest.mark.asyncio
    async def test_platform_admin_none_memberships_not_denied(self):
        """Platform admin (memberships=None) always passes org check."""
        job_org = "org-any"
        memberships = None  # platform admin

        can_access = _can_access_org(job_org, memberships)
        assert can_access is True

        mock_websocket = MagicMock()
        mock_websocket.close = AsyncMock()

        if not can_access:
            await mock_websocket.close(code=4403, reason="Org access denied")

        mock_websocket.close.assert_not_called()


# ── Keepalive constants ────────────────────────────────────────────────────────

class TestWebSocketKeepAliveInterval:
    def test_keepalive_interval_is_20_seconds(self):
        """
        Interval must be ≤20s to stay under Apache mod_proxy_wstunnel idle timeout.
        Mod Comms 2026-03-18: 25s was insufficient; 20s is safe.
        """
        from app.api.v1.routes_websockets import _KEEPALIVE_INTERVAL_S
        assert _KEEPALIVE_INTERVAL_S == 20

    def test_terminal_close_codes_include_4403(self):
        from app.api.v1.routes_websockets import _TERMINAL_CLOSE_CODES
        assert 4403 in _TERMINAL_CLOSE_CODES
        assert 4001 in _TERMINAL_CLOSE_CODES
        assert 4003 in _TERMINAL_CLOSE_CODES
        assert 4004 in _TERMINAL_CLOSE_CODES


# ── Job list accessible_org_ids filter ────────────────────────────────────────

class TestJobListOrgFilter:
    def test_non_admin_gets_accessible_org_ids_list(self):
        """
        For /ws/jobs, accessible_org_ids must be the list of the user's org IDs.
        Platform admins pass None (unrestricted).
        """
        from app.models.organization import OrgRole

        memberships = {"org-a": OrgRole.MEMBER, "org-b": OrgRole.OWNER}
        accessible_org_ids = list(memberships.keys())

        assert set(accessible_org_ids) == {"org-a", "org-b"}

    def test_platform_admin_gets_none_accessible_org_ids(self):
        """Platform admin memberships=None → accessible_org_ids=None (unrestricted)."""
        memberships = None
        accessible_org_ids = None if memberships is None else list(memberships.keys())
        assert accessible_org_ids is None