443 lines
16 KiB
Python
443 lines
16 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import base64
|
|
from collections.abc import Callable
|
|
|
|
import httpx
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from m2_market_web import http as web_http
|
|
from m2_market_web.main import create_app
|
|
|
|
SECRET_VALUES = {
|
|
"fleet-passcode-secret",
|
|
"session-secret-value",
|
|
"mem-secret-key",
|
|
"ledger-secret-key",
|
|
"forgejo-secret-token",
|
|
"m2gpt-password-secret",
|
|
"m2gpt-introspection-secret",
|
|
}
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def env(monkeypatch: pytest.MonkeyPatch) -> None:
|
|
monkeypatch.setenv("FLEET_PASSCODE", "fleet-passcode-secret")
|
|
monkeypatch.setenv("SESSION_SECRET", "session-secret-value")
|
|
monkeypatch.setenv("MEMORY_API_URL", "https://memory.example")
|
|
monkeypatch.setenv("MEMORY_API_KEY", "mem-secret-key")
|
|
monkeypatch.setenv("LEDGER_URL", "https://ledger.example")
|
|
monkeypatch.setenv("LEDGER_SERVICE_KEY", "ledger-secret-key")
|
|
monkeypatch.setenv("FORGEJO_URL", "https://git.example")
|
|
monkeypatch.setenv("FORGEJO_TOKEN", "forgejo-secret-token")
|
|
monkeypatch.setenv("REGISTRY_REPO", "m2/market-registry")
|
|
monkeypatch.setenv("TENANT", "m2-core")
|
|
monkeypatch.setenv("M2GPT_URL", "https://gpt.example")
|
|
monkeypatch.setenv("M2GPT_INTROSPECTION_KEY", "m2gpt-introspection-secret")
|
|
|
|
|
|
@pytest.fixture
|
|
def client(monkeypatch: pytest.MonkeyPatch) -> Callable[[httpx.MockTransport], TestClient]:
|
|
def build(transport: httpx.MockTransport) -> TestClient:
|
|
monkeypatch.setattr(web_http, "async_transport", lambda: transport)
|
|
return TestClient(create_app(), base_url="https://testserver")
|
|
|
|
return build
|
|
|
|
|
|
def _authed(test_client: TestClient) -> None:
|
|
response = test_client.post("/api/login", json={"passcode": "fleet-passcode-secret"})
|
|
assert response.status_code == 204
|
|
|
|
|
|
def _jwt(claims: dict) -> str:
|
|
def encode(part: dict) -> str:
|
|
raw = json.dumps(part, separators=(",", ":")).encode()
|
|
return base64.urlsafe_b64encode(raw).decode().rstrip("=")
|
|
|
|
return f"{encode({'alg': 'none'})}.{encode(claims)}."
|
|
|
|
|
|
def _listing() -> dict:
|
|
return {
|
|
"name": "PDF Export",
|
|
"summary": "Branded reports",
|
|
"category": "reporting",
|
|
"price": {"amount": 7, "currency": "credits", "model": "fixed"},
|
|
"seller": "m2bd",
|
|
"stats": {"installs": 3},
|
|
"tenant_visibility": ["*"],
|
|
"evidence_summary": "Used in one production workflow",
|
|
"permissions": ["filesystem:write"],
|
|
"install_ref": "lst_pdf-v1.0.0",
|
|
}
|
|
|
|
|
|
def _ok_transport(request: httpx.Request) -> httpx.Response:
|
|
if request.url.host == "memory.example":
|
|
assert request.headers["X-API-Key"] == "mem-secret-key"
|
|
body = json.loads(request.content)
|
|
assert body["agent_id"] == "market:catalog"
|
|
assert body["limit"] == 6 or body["limit"] == 30
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"results": [
|
|
{
|
|
"score": 0.9,
|
|
"metadata": {"listing_id": "lst_pdf", "listing": _listing()},
|
|
},
|
|
{
|
|
"score": 0.8,
|
|
"metadata": {
|
|
"listing_id": "lst_hidden",
|
|
"listing": {"name": "Hidden", "tenant_visibility": ["other"]},
|
|
},
|
|
},
|
|
]
|
|
},
|
|
)
|
|
if request.url.host == "ledger.example":
|
|
assert request.headers["X-API-Key"] == "ledger-secret-key"
|
|
if request.url.path == "/balance/tenant-acme":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"operator_id": "tenant-acme",
|
|
"balance": 31,
|
|
"as_of": "2026-07-02T10:00:00Z",
|
|
},
|
|
)
|
|
if request.url.path == "/balance/other-tenant":
|
|
return httpx.Response(200, json={"operator_id": "other-tenant", "balance": 13})
|
|
if request.url.path == "/balance/m2bd":
|
|
return httpx.Response(
|
|
200,
|
|
json={"operator_id": "m2bd", "balance": 42, "as_of": "2026-07-02T10:00:00Z"},
|
|
)
|
|
if request.url.path == "/balance/platform":
|
|
return httpx.Response(200, json={"operator_id": "platform", "balance": 8})
|
|
if request.url.path == "/tx" and request.url.params.get("operator_id") == "m2bd":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"transactions": [
|
|
{
|
|
"ts": "2026-07-02T09:00:00Z",
|
|
"from_id": "mint",
|
|
"to": "m2bd",
|
|
"amount": 50,
|
|
"reason": "grant",
|
|
"ref": "starter",
|
|
"private": "ignored",
|
|
}
|
|
]
|
|
},
|
|
)
|
|
if request.url.path == "/tx" and request.url.params.get("operator_id") == "tenant-acme":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"transactions": [
|
|
{
|
|
"ts": "2026-07-02T09:00:00Z",
|
|
"from_id": "mint",
|
|
"to_id": "tenant-acme",
|
|
"amount": 31,
|
|
"reason": "grant",
|
|
}
|
|
]
|
|
},
|
|
)
|
|
if request.url.path == "/tx" and request.url.params.get("operator_id") == "other-tenant":
|
|
return httpx.Response(200, json={"transactions": []})
|
|
if request.url.path == "/tx":
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"transactions": [
|
|
{"from_id": "mint", "to_id": "m2bd", "amount": 50},
|
|
{"from_id": "m2bd", "to_id": "platform", "amount": 8},
|
|
]
|
|
},
|
|
)
|
|
if request.url.host == "git.example":
|
|
assert request.headers["Authorization"] == "Basic bTI6Zm9yZ2Vqby1zZWNyZXQtdG9rZW4="
|
|
if request.url.path == "/api/v1/repos/m2/market-registry/contents/audit":
|
|
return httpx.Response(200, json=[{"name": "2026-07-01.json"}])
|
|
if request.url.path == "/api/v1/repos/m2/market-registry/pulls":
|
|
return httpx.Response(
|
|
200,
|
|
json=[
|
|
{
|
|
"number": 12,
|
|
"title": "Publish PDF Export",
|
|
"created_at": "2026-07-01T10:00:00Z",
|
|
"labels": [{"name": "approved"}],
|
|
"html_url": "https://git.example/m2/market-registry/pulls/12",
|
|
}
|
|
],
|
|
)
|
|
if request.url.path == "/api/v1/repos/m2/market-registry/contents/listings":
|
|
return httpx.Response(200, json=[{"name": "lst_pdf", "type": "dir"}])
|
|
if (
|
|
request.url.path
|
|
== "/api/v1/repos/m2/market-registry/contents/listings/lst_pdf/listing.json"
|
|
):
|
|
return httpx.Response(200, json={"content_json": {"status": "published"}})
|
|
return httpx.Response(404, json={"unexpected": str(request.url)})
|
|
|
|
|
|
def _gpt_transport(request: httpx.Request) -> httpx.Response:
|
|
if request.url.host == "gpt.example":
|
|
if request.url.path == "/admin/v1/auth/login":
|
|
body = json.loads(request.content)
|
|
if (
|
|
body == {
|
|
"email": "operator@example.com",
|
|
"password": "m2gpt-password-secret",
|
|
}
|
|
):
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"access_token": _jwt(
|
|
{
|
|
"email": "operator@example.com",
|
|
"role": "operator",
|
|
"tenant_id": "tenant-acme",
|
|
"fleet_admin": False,
|
|
}
|
|
)
|
|
},
|
|
)
|
|
if (
|
|
body == {
|
|
"email": "admin@example.com",
|
|
"password": "m2gpt-password-secret",
|
|
}
|
|
):
|
|
return httpx.Response(
|
|
200,
|
|
json={
|
|
"access_token": _jwt(
|
|
{
|
|
"email": "admin@example.com",
|
|
"role": "fleet_admin",
|
|
"tenant_id": "admin-tenant",
|
|
"fleet_admin": True,
|
|
}
|
|
)
|
|
},
|
|
)
|
|
return httpx.Response(401, json={"error": "bad_credentials"})
|
|
if request.url.path == "/admin/v1/auth/introspect":
|
|
assert request.headers["X-Introspection-Key"] == "m2gpt-introspection-secret"
|
|
return httpx.Response(200, json={"active": True})
|
|
return _ok_transport(request)
|
|
|
|
|
|
def test_login_and_auth_gate(client: Callable[[httpx.MockTransport], TestClient]) -> None:
|
|
test_client = client(httpx.MockTransport(_ok_transport))
|
|
assert test_client.get("/api/search?q=pdf").status_code == 401
|
|
assert test_client.post("/api/login", json={"passcode": "wrong"}).status_code == 401
|
|
_authed(test_client)
|
|
assert test_client.get("/api/search?q=pdf&limit=2").status_code == 200
|
|
|
|
|
|
def test_auth_config_defaults_to_both(client: Callable[[httpx.MockTransport], TestClient]) -> None:
|
|
test_client = client(httpx.MockTransport(_ok_transport))
|
|
response = test_client.get("/api/config")
|
|
assert response.status_code == 200
|
|
assert response.json() == {"auth_mode": "both"}
|
|
|
|
|
|
def test_gpt_login_binds_session_and_wallet(
|
|
client: Callable[[httpx.MockTransport], TestClient],
|
|
) -> None:
|
|
test_client = client(httpx.MockTransport(_gpt_transport))
|
|
response = test_client.post(
|
|
"/api/login",
|
|
json={"email": "operator@example.com", "password": "m2gpt-password-secret"},
|
|
)
|
|
assert response.status_code == 204
|
|
|
|
session = test_client.get("/api/session")
|
|
assert session.status_code == 200
|
|
assert session.json() == {
|
|
"email": "operator@example.com",
|
|
"role": "operator",
|
|
"tenant_id": "tenant-acme",
|
|
"fleet_admin": False,
|
|
"operator_id": "tenant-acme",
|
|
"auth_method": "gpt",
|
|
}
|
|
|
|
wallet = test_client.get("/api/wallet/tenant-acme")
|
|
assert wallet.status_code == 200
|
|
assert wallet.json()["balance"] == 31
|
|
|
|
|
|
def test_gpt_login_wrong_password_401(
|
|
client: Callable[[httpx.MockTransport], TestClient],
|
|
) -> None:
|
|
test_client = client(httpx.MockTransport(_gpt_transport))
|
|
response = test_client.post(
|
|
"/api/login",
|
|
json={"email": "operator@example.com", "password": "wrong"},
|
|
)
|
|
assert response.status_code == 401
|
|
|
|
|
|
def test_gpt_non_admin_foreign_wallet_403(
|
|
client: Callable[[httpx.MockTransport], TestClient],
|
|
) -> None:
|
|
test_client = client(httpx.MockTransport(_gpt_transport))
|
|
response = test_client.post(
|
|
"/api/login",
|
|
json={"email": "operator@example.com", "password": "m2gpt-password-secret"},
|
|
)
|
|
assert response.status_code == 204
|
|
assert test_client.get("/api/wallet/other-tenant").status_code == 403
|
|
|
|
|
|
def test_gpt_fleet_admin_can_view_foreign_wallet(
|
|
client: Callable[[httpx.MockTransport], TestClient],
|
|
) -> None:
|
|
test_client = client(httpx.MockTransport(_gpt_transport))
|
|
response = test_client.post(
|
|
"/api/login",
|
|
json={"email": "admin@example.com", "password": "m2gpt-password-secret"},
|
|
)
|
|
assert response.status_code == 204
|
|
wallet = test_client.get("/api/wallet/other-tenant")
|
|
assert wallet.status_code == 200
|
|
assert wallet.json()["balance"] == 13
|
|
|
|
|
|
def test_passcode_mode_still_allows_passcode_only(
|
|
client: Callable[[httpx.MockTransport], TestClient], monkeypatch: pytest.MonkeyPatch
|
|
) -> None:
|
|
monkeypatch.setenv("AUTH_MODE", "passcode")
|
|
test_client = client(httpx.MockTransport(_gpt_transport))
|
|
assert (
|
|
test_client.post(
|
|
"/api/login",
|
|
json={"email": "operator@example.com", "password": "m2gpt-password-secret"},
|
|
).status_code
|
|
== 401
|
|
)
|
|
_authed(test_client)
|
|
assert test_client.get("/api/wallet/m2bd").status_code == 200
|
|
|
|
|
|
def test_catalog_wallet_economy_governance_happy_paths(
|
|
client: Callable[[httpx.MockTransport], TestClient],
|
|
) -> None:
|
|
test_client = client(httpx.MockTransport(_ok_transport))
|
|
_authed(test_client)
|
|
|
|
search = test_client.get("/api/search?q=pdf&limit=2")
|
|
assert search.status_code == 200
|
|
assert search.json() == [
|
|
{
|
|
"listing_id": "lst_pdf",
|
|
"name": "PDF Export",
|
|
"summary": "Branded reports",
|
|
"category": "reporting",
|
|
"price": {"amount": 7, "currency": "credits", "model": "fixed"},
|
|
"seller": "m2bd",
|
|
"stats": {"installs": 3},
|
|
"score": 0.9,
|
|
}
|
|
]
|
|
|
|
listing = test_client.get("/api/listing/lst_pdf")
|
|
assert listing.status_code == 200
|
|
assert listing.json()["install_command"] == "m2-market install lst_pdf --yes"
|
|
assert listing.json()["permissions"] == ["filesystem:write"]
|
|
|
|
wallet = test_client.get("/api/wallet/m2bd")
|
|
assert wallet.status_code == 200
|
|
assert wallet.json()["balance"] == 42
|
|
assert wallet.json()["transactions"][0] == {
|
|
"ts": "2026-07-02T09:00:00Z",
|
|
"from": "mint",
|
|
"to": "m2bd",
|
|
"amount": 50,
|
|
"reason": "grant",
|
|
"ref": "starter",
|
|
}
|
|
|
|
economy = test_client.get("/api/economy")
|
|
assert economy.status_code == 200
|
|
assert economy.json()["operators"] == [
|
|
{"operator_id": "m2bd", "balance": 42},
|
|
{"operator_id": "platform", "balance": 8},
|
|
]
|
|
assert economy.json()["audit"] == {
|
|
"date": "2026-07-01",
|
|
"url": "https://git.example/m2/market-registry/raw/branch/main/audit/2026-07-01.json",
|
|
}
|
|
|
|
governance = test_client.get("/api/governance")
|
|
assert governance.status_code == 200
|
|
assert governance.json()["open_prs"][0]["labels"] == ["approved"]
|
|
assert governance.json()["listings"] == [
|
|
{
|
|
"listing_id": "lst_pdf",
|
|
"status": "published",
|
|
"url": "https://git.example/m2/market-registry/src/branch/main/listings/lst_pdf",
|
|
}
|
|
]
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("path", "failed_host", "panel"),
|
|
[
|
|
("/api/search?q=pdf", "memory.example", "catalog"),
|
|
("/api/wallet/m2bd", "ledger.example", "wallet"),
|
|
("/api/economy", "ledger.example", "economy"),
|
|
("/api/governance", "git.example", "governance"),
|
|
],
|
|
)
|
|
def test_upstream_failures_are_panel_502(
|
|
client: Callable[[httpx.MockTransport], TestClient],
|
|
path: str,
|
|
failed_host: str,
|
|
panel: str,
|
|
) -> None:
|
|
def handler(request: httpx.Request) -> httpx.Response:
|
|
if request.url.host == failed_host:
|
|
return httpx.Response(503, json={"secret": "upstream body ignored"})
|
|
return _ok_transport(request)
|
|
|
|
test_client = client(httpx.MockTransport(handler))
|
|
_authed(test_client)
|
|
response = test_client.get(path)
|
|
assert response.status_code == 502
|
|
assert response.json() == {"error": "upstream_failure", "panel": panel}
|
|
|
|
|
|
def test_secret_values_never_appear_in_response_bodies(
|
|
client: Callable[[httpx.MockTransport], TestClient],
|
|
) -> None:
|
|
test_client = client(httpx.MockTransport(_ok_transport))
|
|
responses = [
|
|
test_client.post("/api/login", json={"passcode": "fleet-passcode-secret"}),
|
|
]
|
|
responses.extend(
|
|
[
|
|
test_client.get("/health"),
|
|
test_client.get("/api/search?q=pdf"),
|
|
test_client.get("/api/listing/lst_pdf"),
|
|
test_client.get("/api/wallet/m2bd"),
|
|
test_client.get("/api/economy"),
|
|
test_client.get("/api/governance"),
|
|
]
|
|
)
|
|
for response in responses:
|
|
body = response.text
|
|
for secret in SECRET_VALUES:
|
|
assert secret not in body
|