138 lines
6.3 KiB
Python
138 lines
6.3 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_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
|