feat: import 6 MCP services + common workspace
This commit is contained in:
@@ -0,0 +1,84 @@
|
||||
import pytest
|
||||
from fastapi import Depends, FastAPI
|
||||
from fastapi.testclient import TestClient
|
||||
from option_mcp_common.auth import (
|
||||
Principal,
|
||||
TokenStore,
|
||||
acl_requires,
|
||||
require_principal,
|
||||
)
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def token_store():
|
||||
return TokenStore(tokens={
|
||||
"token-core-123": Principal(name="core", capabilities={"core"}),
|
||||
"token-obs-456": Principal(name="observer", capabilities={"observer"}),
|
||||
})
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def app(token_store):
|
||||
app = FastAPI()
|
||||
app.state.token_store = token_store
|
||||
|
||||
@app.get("/public")
|
||||
def public():
|
||||
return {"ok": True}
|
||||
|
||||
@app.get("/private")
|
||||
def private(principal: Principal = Depends(require_principal)):
|
||||
return {"name": principal.name}
|
||||
|
||||
@app.post("/core-only")
|
||||
@acl_requires(core=True, observer=False)
|
||||
def core_only(principal: Principal = Depends(require_principal)):
|
||||
return {"who": principal.name}
|
||||
|
||||
@app.post("/observer-only")
|
||||
@acl_requires(core=False, observer=True)
|
||||
def observer_only(principal: Principal = Depends(require_principal)):
|
||||
return {"who": principal.name}
|
||||
|
||||
return app
|
||||
|
||||
|
||||
def test_public_endpoint_no_auth(app):
|
||||
client = TestClient(app)
|
||||
assert client.get("/public").status_code == 200
|
||||
|
||||
|
||||
def test_private_without_header_401(app):
|
||||
client = TestClient(app)
|
||||
assert client.get("/private").status_code == 401
|
||||
|
||||
|
||||
def test_private_bad_token_403(app):
|
||||
client = TestClient(app)
|
||||
r = client.get("/private", headers={"Authorization": "Bearer nope"})
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
def test_private_good_token_200(app):
|
||||
client = TestClient(app)
|
||||
r = client.get("/private", headers={"Authorization": "Bearer token-core-123"})
|
||||
assert r.status_code == 200
|
||||
assert r.json() == {"name": "core"}
|
||||
|
||||
|
||||
def test_acl_core_token_on_core_only_endpoint(app):
|
||||
client = TestClient(app)
|
||||
r = client.post("/core-only", headers={"Authorization": "Bearer token-core-123"})
|
||||
assert r.status_code == 200
|
||||
|
||||
|
||||
def test_acl_observer_on_core_only_rejected(app):
|
||||
client = TestClient(app)
|
||||
r = client.post("/core-only", headers={"Authorization": "Bearer token-obs-456"})
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
def test_acl_observer_on_observer_only_ok(app):
|
||||
client = TestClient(app)
|
||||
r = client.post("/observer-only", headers={"Authorization": "Bearer token-obs-456"})
|
||||
assert r.status_code == 200
|
||||
Reference in New Issue
Block a user