emma-hermes/tests/hermes_cli/test_web_server_session_acc...

85 lines
4.1 KiB
Python

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_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)