javadandCursor
b93ecf1691
Keep source, config, and deploy files. Leave runtime data, uploads, and logs out of git. Co-authored-by: Cursor <cursoragent@cursor.com>
128 lines
4.7 KiB
Python
128 lines
4.7 KiB
Python
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
from fastapi import FastAPI
|
|
from fastapi.testclient import TestClient
|
|
|
|
from kamidso.api.deps import KamidsoAuthMiddleware
|
|
from kamidso.api.routes import auth as auth_routes
|
|
from kamidso.auth.otp_service import OtpService
|
|
from kamidso.contracts.errors import KamidsoError, KamidsoErrorCode
|
|
from kamidso.repositories.store import RepositoryStore
|
|
from kamidso.settings.flags import load_kamidso_settings
|
|
from kamidso.workspace.service import WorkspaceService
|
|
|
|
|
|
def _make_app():
|
|
settings = load_kamidso_settings()
|
|
store = RepositoryStore()
|
|
workspace = WorkspaceService(store, settings)
|
|
app = FastAPI()
|
|
app.add_middleware(KamidsoAuthMiddleware)
|
|
app.include_router(auth_routes.router, prefix="/api/kamidso/v1")
|
|
app.state.kamidso = SimpleNamespace(settings=settings, workspace=workspace, store=store)
|
|
return app, store, settings, workspace
|
|
|
|
|
|
def test_otp_send_and_verify_creates_isolated_workspaces():
|
|
app, _store, _settings, _workspace = _make_app()
|
|
client = TestClient(app)
|
|
|
|
first_send = client.post("/api/kamidso/v1/auth/otp/send", json={"mobile": "09121111111"})
|
|
assert first_send.status_code == 200
|
|
first_code = first_send.json()["dev_code"]
|
|
first_verify = client.post(
|
|
"/api/kamidso/v1/auth/otp/verify",
|
|
json={"mobile": "09121111111", "code": first_code},
|
|
)
|
|
assert first_verify.status_code == 200
|
|
first_user = first_verify.json()["user"]
|
|
assert first_user["role"] == "admin"
|
|
assert first_user["mobile"] == "09121111111"
|
|
|
|
other = TestClient(app)
|
|
second_send = other.post("/api/kamidso/v1/auth/otp/send", json={"mobile": "09122222222"})
|
|
second_code = second_send.json()["dev_code"]
|
|
second_verify = other.post(
|
|
"/api/kamidso/v1/auth/otp/verify",
|
|
json={"mobile": "09122222222", "code": second_code},
|
|
)
|
|
assert second_verify.status_code == 200
|
|
second_user = second_verify.json()["user"]
|
|
assert first_user["workspace_id"] != second_user["workspace_id"]
|
|
assert first_user["id"] != second_user["id"]
|
|
|
|
|
|
def test_otp_verify_bearer_token_opens_session():
|
|
app, _store, _settings, _workspace = _make_app()
|
|
sender = TestClient(app)
|
|
sent = sender.post("/api/kamidso/v1/auth/otp/send", json={"mobile": "09123333333"})
|
|
verified = sender.post(
|
|
"/api/kamidso/v1/auth/otp/verify",
|
|
json={"mobile": "09123333333", "code": sent.json()["dev_code"]},
|
|
)
|
|
token = verified.json()["access_token"]
|
|
bare = TestClient(app)
|
|
session = bare.get(
|
|
"/api/kamidso/v1/auth/session",
|
|
headers={"Authorization": f"Bearer {token}"},
|
|
)
|
|
assert session.status_code == 200
|
|
assert session.json()["user"]["mobile"] == "09123333333"
|
|
|
|
|
|
def test_otp_invalid_mobile_and_wrong_code():
|
|
app, _store, _settings, _workspace = _make_app()
|
|
client = TestClient(app)
|
|
bad = client.post("/api/kamidso/v1/auth/otp/send", json={"mobile": "123"})
|
|
assert bad.status_code == 400
|
|
assert bad.json()["code"] == "INVALID_MOBILE"
|
|
|
|
sent = client.post("/api/kamidso/v1/auth/otp/send", json={"mobile": "09124444444"})
|
|
assert sent.status_code == 200
|
|
wrong = client.post(
|
|
"/api/kamidso/v1/auth/otp/verify",
|
|
json={"mobile": "09124444444", "code": "000000"},
|
|
)
|
|
assert wrong.status_code == 401
|
|
assert wrong.json()["code"] == "OTP_INVALID"
|
|
|
|
|
|
def test_otp_send_cooldown():
|
|
app, _store, _settings, _workspace = _make_app()
|
|
client = TestClient(app)
|
|
first = client.post("/api/kamidso/v1/auth/otp/send", json={"mobile": "09125555555"})
|
|
assert first.status_code == 200
|
|
second = client.post("/api/kamidso/v1/auth/otp/send", json={"mobile": "09125555555"})
|
|
assert second.status_code == 429
|
|
assert second.json()["code"] == "OTP_COOLDOWN"
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_email_login_still_works_for_bootstrap_admin():
|
|
app, _store, settings, workspace = _make_app()
|
|
await workspace.bootstrap_identity()
|
|
client = TestClient(app)
|
|
response = client.post(
|
|
"/api/kamidso/v1/auth/login",
|
|
json={
|
|
"email": settings.bootstrap_admin_email,
|
|
"password": settings.bootstrap_admin_password,
|
|
},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()["user"]["email"] == settings.bootstrap_admin_email
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_otp_service_verify_consumes_code():
|
|
settings = load_kamidso_settings()
|
|
store = RepositoryStore()
|
|
service = OtpService(store, settings)
|
|
sent = await service.send("09126666666")
|
|
mobile = await service.verify("09126666666", sent["dev_code"])
|
|
assert mobile == "09126666666"
|
|
with pytest.raises(KamidsoError) as exc:
|
|
await service.verify("09126666666", sent["dev_code"])
|
|
assert exc.value.code == KamidsoErrorCode.OTP_INVALID
|