Compare commits

..
Author SHA1 Message Date
Alexandre Teixeira f1a45e85c5 test: remove duplicate editor cache-name assertion 2026-10-29 01:19:32 +00:00
2 changed files with 0 additions and 141 deletions
-135
View File
@@ -1,135 +0,0 @@
"""Real admin gates must run before MCP and diagnostic data access."""
import builtins
import json
from types import SimpleNamespace
from unittest.mock import AsyncMock, Mock
import pytest
from fastapi import APIRouter, FastAPI
from fastapi.testclient import TestClient
from core.auth import AuthManager
from routes.auth_routes import SESSION_COOKIE
from tests.helpers.database import disposable_database
@pytest.fixture
def authenticated_app(tmp_path, monkeypatch):
monkeypatch.setenv("AUTH_ENABLED", "true")
auth = AuthManager(str(tmp_path / "auth.json"))
for user, admin in (("member", False), ("admin", True)):
assert auth.create_user(user, "disposable-test-password", is_admin=admin)
app = FastAPI()
app.state.auth_manager = auth
@app.middleware("http")
async def fixture_session(request, call_next):
# Supply the authenticated principal, without replacing require_admin.
request.state.current_user = auth.get_username_for_token(request.cookies.get(SESSION_COOKIE))
return await call_next(request)
def client_for(user):
client = TestClient(app)
if user != "anonymous":
token = auth.create_session(user, "disposable-test-password")
assert token
client.cookies.set(SESSION_COOKIE, token)
return client
return app, client_for
@pytest.mark.parametrize("user", ["anonymous", "member", "admin"])
@pytest.mark.parametrize("path", ["/servers", "/tools", "/servers/private-server/tools"])
def test_mcp_listing_requires_real_admin_before_data_access(
tmp_path, monkeypatch, authenticated_app, user, path,
):
from core.database import McpServer
from routes.mcp import mcp_routes
app, client_for = authenticated_app
with disposable_database(tmp_path) as factory:
with factory() as db:
db.add(McpServer(
id="private-server", name="private-config-marker", transport="stdio",
command="fixture-only", args="[]", env=json.dumps({"FIXTURE": "private-config-marker"}),
))
db.commit()
session = Mock(wraps=factory)
manager = SimpleNamespace(
get_server_status=Mock(return_value={"status": "connected", "tool_count": 1}),
get_all_tools=Mock(return_value=[{
"server_id": "private-server", "name": "private-tool", "description": "private-config-marker",
}]),
)
# setup_mcp_routes otherwise appends to a shared process-global router.
monkeypatch.setattr(mcp_routes, "router", APIRouter(prefix="/api/mcp"))
monkeypatch.setattr(mcp_routes, "SessionLocal", session)
app.include_router(mcp_routes.setup_mcp_routes(manager))
with client_for(user) as client:
response = client.get("/api/mcp" + path)
if user == "admin":
assert response.status_code == 200
assert "private-config-marker" in response.text
payload = response.json()
if path == "/servers":
assert payload[0]["env"] == {"FIXTURE": "private-config-marker"}
else:
assert payload[0]["name"] == "private-tool"
else:
assert response.status_code == 403
assert response.json() == {"detail": "Admin only"}
assert "private-config-marker" not in response.text
session.assert_not_called()
manager.get_server_status.assert_not_called()
manager.get_all_tools.assert_not_called()
@pytest.mark.parametrize("user", ["anonymous", "member", "admin"])
@pytest.mark.parametrize("method,path", [
("GET", "/api/db/stats"),
("GET", "/api/rag/stats"),
("GET", "/api/test/youtube?url=https://youtu.be/fixture"),
("POST", "/api/test-research"),
("GET", "/api/diagnostics/services"),
("GET", "/api/diagnostics/logs"),
])
def test_diagnostics_require_real_admin_before_sensitive_work(
tmp_path, monkeypatch, authenticated_app, user, method, path,
):
import core.database as database
import src.service_health as health
from routes import diagnostics_routes as routes
app, client_for = authenticated_app
marker = "private-diagnostics-marker"
stats = Mock(return_value={"marker": marker})
rag = SimpleNamespace(get_stats=Mock(return_value={"marker": marker}))
research = SimpleNamespace(call_research_service=AsyncMock(return_value=marker))
extract_id = Mock(return_value="fixture")
transcript = AsyncMock(return_value={"success": True, "transcript": marker})
collect = AsyncMock(return_value={"marker": marker})
log_dir = tmp_path / "logs"
log_dir.mkdir()
(log_dir / "app.log").write_text(marker + "\n", encoding="utf-8")
read_log = Mock(wraps=builtins.open)
monkeypatch.setattr(database, "get_detailed_stats", stats)
monkeypatch.setattr(routes, "extract_youtube_id", extract_id)
monkeypatch.setattr(routes, "extract_transcript_async", transcript)
monkeypatch.setattr(health, "collect_service_health", collect)
monkeypatch.setattr(routes, "DATA_DIR", str(tmp_path))
monkeypatch.setattr(routes, "open", read_log, raising=False)
app.include_router(routes.setup_diagnostics_routes(rag, True, research))
with client_for(user) as client:
response = client.request(method, path)
if user == "admin":
assert response.status_code == 200
assert marker in response.text
else:
assert response.status_code == 403
assert response.json() == {"detail": "Admin only"}
assert marker not in response.text
for operation in (stats, rag.get_stats, research.call_research_service,
extract_id, transcript, collect, read_log):
operation.assert_not_called()
@@ -266,9 +266,3 @@ def test_more6_preferred_wind_is_present():
assert 'data-filter-action="stylize-wind"' in TOPBAR
assert "else if (action === 'stylize-wind') applyWind?.();" in WIRE
assert "export function wind(snap, v, dst)" in EFFECTS
def test_more6_sw_version_is_current():
m = re.search(r"const CACHE_NAME = '([^']+)'", SW)
assert m, "CACHE_NAME not found"
assert re.fullmatch(r"odysseus-v\d+-pr40-editor-precache", m.group(1)), m.group(1)