from types import SimpleNamespace
from unittest.mock import patch
import asyncio

import pytest
from fastapi import HTTPException

from hermes_cli import web_server


def _request(email: str | None):
    session = SimpleNamespace(email=email) if email else None
    return SimpleNamespace(state=SimpleNamespace(session=session))


class TestProfilesSessionsAccess:
    def test_non_admin_forbidden(self, monkeypatch):
        monkeypatch.setattr(web_server, "_dashboard_identity_scope", lambda request: {"admin": False})
        with pytest.raises(HTTPException) as exc:
            web_server.get_profiles_sessions(_request("user@bremen.com.tw"))
        assert exc.value.status_code == 403

    def test_non_admin_may_access_own_principal_profile(self):
        scope = {"admin": False, "principal_ids": ["p-123"]}
        web_server._enforce_dashboard_profile_access(scope, "principal_p123")

    def test_non_admin_cannot_access_other_principal_profile(self):
        scope = {"admin": False, "principal_ids": ["p-123"]}
        with pytest.raises(HTTPException) as exc:
            web_server._enforce_dashboard_profile_access(scope, "principal_other999")
        assert exc.value.status_code == 403

    def test_admin_allowed_to_reach_listing(self, monkeypatch, tmp_path):
        monkeypatch.setattr(web_server, "_dashboard_identity_scope", lambda request: {"admin": True})
        monkeypatch.setattr(web_server, "_cron_profile_home", lambda profile: (profile or "default", tmp_path))

        import hermes_cli.profiles as profiles_mod
        monkeypatch.setattr(profiles_mod, "list_profiles", lambda: [])
        monkeypatch.setattr(profiles_mod, "get_profile_dir", lambda name: tmp_path)

        result = web_server.get_profiles_sessions(_request("admin@bremen.com.tw"))
        assert result["sessions"] == []
        assert result["total"] == 0


class TestApiServerSessionVisibility:
    def test_api_server_session_visible_to_principal_scope(self):
        scope = {"admin": False, "email": "dk96@bremen.com.tw", "principal_ids": ["p-123"], "external_user_ids": ["openwebui-42"]}
        session = {"source": "api_server", "user_id": "principal:p-123"}
        assert web_server._session_visible_to_scope(session, scope) is True

    def test_api_server_session_visible_to_email_scope(self):
        scope = {"admin": False, "email": "dk96@bremen.com.tw", "principal_ids": [], "external_user_ids": []}
        session = {"source": "api_server", "user_id": "email:dk96@bremen.com.tw"}
        assert web_server._session_visible_to_scope(session, scope) is True

    def test_api_server_session_visible_to_external_user_scope(self):
        scope = {"admin": False, "email": "", "principal_ids": [], "external_user_ids": ["openwebui-42"]}
        assert web_server._session_visible_to_scope({"source": "api_server", "user_id": "user:openwebui-42"}, scope) is True
        assert web_server._session_visible_to_scope({"source": "api_server", "user_id": "key:openwebui-42"}, scope) is True


class TestExportEndpointDeidentify:
    def test_export_endpoint_forbids_other_users_session(self, monkeypatch):
        class _DB:
            def resolve_session_id(self, session_id):
                return session_id

            def get_session(self, session_id):
                return {"id": session_id, "source": "email", "user_id": "other@bremen.com.tw", "chat_id": "other@bremen.com.tw"}

            def close(self):
                return None

        monkeypatch.setattr(web_server, "_dashboard_identity_scope", lambda request: {"admin": False, "email": "dk96@bremen.com.tw", "platforms": [], "external_user_ids": [], "principal_ids": []})
        monkeypatch.setattr(web_server, "_enforce_dashboard_profile_access", lambda scope, profile: None)
        monkeypatch.setattr(web_server, "_open_session_db_for_profile", lambda profile: _DB())

        with pytest.raises(HTTPException) as exc:
            asyncio.run(web_server.export_session_endpoint(_request("dk96@bremen.com.tw"), "sess-1"))
        assert exc.value.status_code == 403

    def test_non_admin_export_masks_payload(self, monkeypatch):
        class _DB:
            def resolve_session_id(self, session_id):
                return session_id

            def get_session(self, session_id):
                return {"id": session_id, "source": "email", "user_id": "dk96@bremen.com.tw", "chat_id": "dk96@bremen.com.tw"}

            def export_session(self, session_id):
                return {"messages": [{"content": "john.doe@bremen.com.tw 0912-345-678"}]}

            def close(self):
                return None

        monkeypatch.setattr(web_server, "_dashboard_identity_scope", lambda request: {"admin": False, "email": "dk96@bremen.com.tw", "platforms": [], "external_user_ids": [], "principal_ids": []})
        monkeypatch.setattr(web_server, "_enforce_dashboard_profile_access", lambda scope, profile: None)
        monkeypatch.setattr(web_server, "_open_session_db_for_profile", lambda profile: _DB())

        out = asyncio.run(web_server.export_session_endpoint(_request("dk96@bremen.com.tw"), "sess-1"))
        assert "john.doe@bremen.com.tw" not in str(out)
        assert "0912-345-678" not in str(out)


class TestAuthorizedSessionHelper:
    def test_helper_allows_visible_session(self):
        class _DB:
            def resolve_session_id(self, session_id):
                return session_id

            def get_session(self, session_id):
                return {"id": session_id, "source": "telegram", "user_id": "alice"}

        sid, session = web_server._require_authorized_session(
            _DB(),
            "sess-1",
            {"admin": False, "platforms": ["telegram"], "external_user_ids": ["alice"]},
        )
        assert sid == "sess-1"
        assert session["id"] == "sess-1"

    def test_helper_rejects_foreign_session(self):
        class _DB:
            def resolve_session_id(self, session_id):
                return session_id

            def get_session(self, session_id):
                return {"id": session_id, "source": "telegram", "user_id": "bob"}

        with pytest.raises(HTTPException) as exc:
            web_server._require_authorized_session(
                _DB(),
                "sess-1",
                {"admin": False, "platforms": ["telegram"], "external_user_ids": ["alice"]},
            )
        assert exc.value.status_code == 403
