Kyle Isom
0c441f5c4f
- clients/README.md: canonical API surface and error type reference - clients/testdata/: shared JSON response fixtures - clients/go/: mciasgoclient package; net/http + TLS 1.2+; sync.RWMutex token state; DisallowUnknownFields on all decoders; 25 tests pass - clients/rust/: async mcias-client crate; reqwest+rustls (no OpenSSL); thiserror MciasError enum; Arc<RwLock> token state; 22+1 tests pass; cargo clippy -D warnings clean - clients/lisp/: ASDF mcias-client; dexador HTTP, yason JSON; mcias-error condition hierarchy; Hunchentoot mock-dispatcher; 37 fiveam checks pass on SBCL 2.6.1; yason boolean normalisation in validate-token - clients/python/: mcias_client package (Python 3.11+); httpx sync; py.typed; dataclasses; 32 pytest tests; mypy --strict + ruff clean - test/mock/mockserver.go: in-memory mock server for Go client tests - ARCHITECTURE.md §19: updated per-language notes to match implementation - PROGRESS.md: Phase 9 marked complete - .gitignore: exclude clients/rust/target/, python .venv, .pytest_cache, .fasl files Security: token never logged or exposed in error messages in any library; TLS enforced in all four languages; token stored under lock/mutex/RwLock
321 lines
11 KiB
Python
321 lines
11 KiB
Python
"""Tests for the MCIAS Python client using respx to mock httpx."""
|
|
from __future__ import annotations
|
|
|
|
import httpx
|
|
import pytest
|
|
import respx
|
|
|
|
from mcias_client import (
|
|
Client,
|
|
MciasAuthError,
|
|
MciasConflictError,
|
|
MciasError,
|
|
MciasForbiddenError,
|
|
MciasInputError,
|
|
MciasNotFoundError,
|
|
MciasServerError,
|
|
)
|
|
from mcias_client._models import Account, PGCreds, PublicKey, TokenClaims
|
|
|
|
BASE_URL = "https://auth.example.com"
|
|
SAMPLE_ACCOUNT: dict[str, object] = {
|
|
"id": "acc-001",
|
|
"username": "alice",
|
|
"account_type": "user",
|
|
"status": "active",
|
|
"created_at": "2024-01-01T00:00:00Z",
|
|
"updated_at": "2024-01-01T00:00:00Z",
|
|
"totp_enabled": False,
|
|
}
|
|
SAMPLE_PK: dict[str, object] = {
|
|
"kty": "OKP",
|
|
"crv": "Ed25519",
|
|
"x": "base64urlpublickey",
|
|
"use": "sig",
|
|
"alg": "EdDSA",
|
|
}
|
|
@pytest.fixture
|
|
def client() -> Client:
|
|
return Client(BASE_URL)
|
|
@pytest.fixture
|
|
def admin_client() -> Client:
|
|
return Client(BASE_URL, token="admin-token")
|
|
def test_client_init() -> None:
|
|
c = Client(BASE_URL)
|
|
assert c.token is None
|
|
assert c._base_url == BASE_URL
|
|
c.close()
|
|
def test_client_strips_trailing_slash() -> None:
|
|
c = Client(BASE_URL + "/")
|
|
assert c._base_url == BASE_URL
|
|
c.close()
|
|
def test_client_init_with_token() -> None:
|
|
c = Client(BASE_URL, token="mytoken")
|
|
assert c.token == "mytoken"
|
|
c.close()
|
|
@respx.mock
|
|
def test_health_ok(client: Client) -> None:
|
|
respx.get(f"{BASE_URL}/v1/health").mock(return_value=httpx.Response(200))
|
|
client.health() # should not raise
|
|
@respx.mock
|
|
def test_health_server_error(client: Client) -> None:
|
|
respx.get(f"{BASE_URL}/v1/health").mock(
|
|
return_value=httpx.Response(503, json={"error": "service unavailable"})
|
|
)
|
|
with pytest.raises(MciasServerError) as exc_info:
|
|
client.health()
|
|
assert exc_info.value.status_code == 503
|
|
@respx.mock
|
|
def test_get_public_key(client: Client) -> None:
|
|
respx.get(f"{BASE_URL}/v1/keys/public").mock(
|
|
return_value=httpx.Response(200, json=SAMPLE_PK)
|
|
)
|
|
pk = client.get_public_key()
|
|
assert isinstance(pk, PublicKey)
|
|
assert pk.kty == "OKP"
|
|
assert pk.crv == "Ed25519"
|
|
assert pk.alg == "EdDSA"
|
|
@respx.mock
|
|
def test_login_success(client: Client) -> None:
|
|
respx.post(f"{BASE_URL}/v1/auth/login").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={"token": "jwt-token-abc", "expires_at": "2099-01-01T00:00:00Z"},
|
|
)
|
|
)
|
|
token, expires_at = client.login("alice", "s3cr3t")
|
|
assert token == "jwt-token-abc"
|
|
assert expires_at == "2099-01-01T00:00:00Z"
|
|
assert client.token == "jwt-token-abc"
|
|
@respx.mock
|
|
def test_login_unauthorized(client: Client) -> None:
|
|
respx.post(f"{BASE_URL}/v1/auth/login").mock(
|
|
return_value=httpx.Response(
|
|
401, json={"error": "invalid credentials"}
|
|
)
|
|
)
|
|
with pytest.raises(MciasAuthError) as exc_info:
|
|
client.login("alice", "wrong")
|
|
assert exc_info.value.status_code == 401
|
|
@respx.mock
|
|
def test_logout_clears_token(admin_client: Client) -> None:
|
|
respx.post(f"{BASE_URL}/v1/auth/logout").mock(
|
|
return_value=httpx.Response(204)
|
|
)
|
|
assert admin_client.token == "admin-token"
|
|
admin_client.logout()
|
|
assert admin_client.token is None
|
|
@respx.mock
|
|
def test_renew_token(admin_client: Client) -> None:
|
|
respx.post(f"{BASE_URL}/v1/auth/renew").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={"token": "new-jwt-token", "expires_at": "2099-06-01T00:00:00Z"},
|
|
)
|
|
)
|
|
token, expires_at = admin_client.renew_token()
|
|
assert token == "new-jwt-token"
|
|
assert expires_at == "2099-06-01T00:00:00Z"
|
|
assert admin_client.token == "new-jwt-token"
|
|
@respx.mock
|
|
def test_validate_token_valid(admin_client: Client) -> None:
|
|
respx.post(f"{BASE_URL}/v1/token/validate").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={
|
|
"valid": True,
|
|
"sub": "acc-001",
|
|
"roles": ["admin"],
|
|
"expires_at": "2099-01-01T00:00:00Z",
|
|
},
|
|
)
|
|
)
|
|
claims = admin_client.validate_token("some-token")
|
|
assert isinstance(claims, TokenClaims)
|
|
assert claims.valid is True
|
|
assert claims.sub == "acc-001"
|
|
assert claims.roles == ["admin"]
|
|
@respx.mock
|
|
def test_validate_token_invalid(admin_client: Client) -> None:
|
|
"""valid=False in the response body is NOT an exception — just a falsy claim."""
|
|
respx.post(f"{BASE_URL}/v1/token/validate").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={"valid": False, "sub": "", "roles": []},
|
|
)
|
|
)
|
|
claims = admin_client.validate_token("expired-token")
|
|
assert claims.valid is False
|
|
@respx.mock
|
|
def test_create_account(admin_client: Client) -> None:
|
|
respx.post(f"{BASE_URL}/v1/accounts").mock(
|
|
return_value=httpx.Response(201, json=SAMPLE_ACCOUNT)
|
|
)
|
|
acc = admin_client.create_account("alice", "user", password="pass123")
|
|
assert isinstance(acc, Account)
|
|
assert acc.id == "acc-001"
|
|
assert acc.username == "alice"
|
|
@respx.mock
|
|
def test_create_account_conflict(admin_client: Client) -> None:
|
|
respx.post(f"{BASE_URL}/v1/accounts").mock(
|
|
return_value=httpx.Response(409, json={"error": "username already exists"})
|
|
)
|
|
with pytest.raises(MciasConflictError) as exc_info:
|
|
admin_client.create_account("alice", "user")
|
|
assert exc_info.value.status_code == 409
|
|
@respx.mock
|
|
def test_list_accounts(admin_client: Client) -> None:
|
|
second = {**SAMPLE_ACCOUNT, "id": "acc-002"}
|
|
respx.get(f"{BASE_URL}/v1/accounts").mock(
|
|
return_value=httpx.Response(
|
|
200, json={"accounts": [SAMPLE_ACCOUNT, second]}
|
|
)
|
|
)
|
|
accounts = admin_client.list_accounts()
|
|
assert len(accounts) == 2
|
|
assert all(isinstance(a, Account) for a in accounts)
|
|
@respx.mock
|
|
def test_get_account(admin_client: Client) -> None:
|
|
respx.get(f"{BASE_URL}/v1/accounts/acc-001").mock(
|
|
return_value=httpx.Response(200, json=SAMPLE_ACCOUNT)
|
|
)
|
|
acc = admin_client.get_account("acc-001")
|
|
assert acc.id == "acc-001"
|
|
@respx.mock
|
|
def test_update_account(admin_client: Client) -> None:
|
|
updated = {**SAMPLE_ACCOUNT, "status": "suspended"}
|
|
respx.patch(f"{BASE_URL}/v1/accounts/acc-001").mock(
|
|
return_value=httpx.Response(200, json=updated)
|
|
)
|
|
acc = admin_client.update_account("acc-001", status="suspended")
|
|
assert acc.status == "suspended"
|
|
@respx.mock
|
|
def test_delete_account(admin_client: Client) -> None:
|
|
respx.delete(f"{BASE_URL}/v1/accounts/acc-001").mock(
|
|
return_value=httpx.Response(204)
|
|
)
|
|
admin_client.delete_account("acc-001") # should not raise
|
|
@respx.mock
|
|
def test_get_roles(admin_client: Client) -> None:
|
|
respx.get(f"{BASE_URL}/v1/accounts/acc-001/roles").mock(
|
|
return_value=httpx.Response(200, json={"roles": ["admin", "viewer"]})
|
|
)
|
|
roles = admin_client.get_roles("acc-001")
|
|
assert roles == ["admin", "viewer"]
|
|
@respx.mock
|
|
def test_set_roles(admin_client: Client) -> None:
|
|
respx.put(f"{BASE_URL}/v1/accounts/acc-001/roles").mock(
|
|
return_value=httpx.Response(204)
|
|
)
|
|
admin_client.set_roles("acc-001", ["viewer"]) # should not raise
|
|
@respx.mock
|
|
def test_issue_service_token(admin_client: Client) -> None:
|
|
respx.post(f"{BASE_URL}/v1/accounts/acc-001/token").mock(
|
|
return_value=httpx.Response(
|
|
200,
|
|
json={"token": "svc-token-xyz", "expires_at": "2099-12-31T00:00:00Z"},
|
|
)
|
|
)
|
|
token, expires_at = admin_client.issue_service_token("acc-001")
|
|
assert token == "svc-token-xyz"
|
|
assert expires_at == "2099-12-31T00:00:00Z"
|
|
@respx.mock
|
|
def test_revoke_token(admin_client: Client) -> None:
|
|
jti = "some-jti-uuid"
|
|
respx.delete(f"{BASE_URL}/v1/token/{jti}").mock(
|
|
return_value=httpx.Response(204)
|
|
)
|
|
admin_client.revoke_token(jti) # should not raise
|
|
SAMPLE_PG_CREDS: dict[str, object] = {
|
|
"host": "db.example.com",
|
|
"port": 5432,
|
|
"database": "myapp",
|
|
"username": "appuser",
|
|
"password": "s3cr3t",
|
|
}
|
|
@respx.mock
|
|
def test_get_pg_creds(admin_client: Client) -> None:
|
|
respx.get(f"{BASE_URL}/v1/accounts/acc-001/pgcreds").mock(
|
|
return_value=httpx.Response(200, json=SAMPLE_PG_CREDS)
|
|
)
|
|
creds = admin_client.get_pg_creds("acc-001")
|
|
assert isinstance(creds, PGCreds)
|
|
assert creds.host == "db.example.com"
|
|
assert creds.port == 5432
|
|
assert creds.database == "myapp"
|
|
@respx.mock
|
|
def test_set_pg_creds(admin_client: Client) -> None:
|
|
respx.put(f"{BASE_URL}/v1/accounts/acc-001/pgcreds").mock(
|
|
return_value=httpx.Response(204)
|
|
)
|
|
admin_client.set_pg_creds(
|
|
"acc-001",
|
|
host="db.example.com",
|
|
port=5432,
|
|
database="myapp",
|
|
username="appuser",
|
|
password="s3cr3t",
|
|
) # should not raise
|
|
@pytest.mark.parametrize(
|
|
("status_code", "exc_class"),
|
|
[
|
|
(400, MciasInputError),
|
|
(401, MciasAuthError),
|
|
(403, MciasForbiddenError),
|
|
(404, MciasNotFoundError),
|
|
(409, MciasConflictError),
|
|
(500, MciasServerError),
|
|
],
|
|
)
|
|
@respx.mock
|
|
def test_error_types(
|
|
client: Client,
|
|
status_code: int,
|
|
exc_class: type,
|
|
) -> None:
|
|
respx.get(f"{BASE_URL}/v1/health").mock(
|
|
return_value=httpx.Response(
|
|
status_code, json={"error": "test error"}
|
|
)
|
|
)
|
|
with pytest.raises(exc_class) as exc_info:
|
|
client.health()
|
|
err = exc_info.value
|
|
assert isinstance(err, MciasError)
|
|
assert err.status_code == status_code
|
|
@respx.mock
|
|
def test_context_manager() -> None:
|
|
respx.get(f"{BASE_URL}/v1/health").mock(return_value=httpx.Response(200))
|
|
with Client(BASE_URL) as c:
|
|
c.health()
|
|
assert c._http.is_closed
|
|
@respx.mock
|
|
def test_integration_login_validate_logout() -> None:
|
|
"""Full flow: login, validate the issued token, then logout."""
|
|
login_resp = httpx.Response(
|
|
200,
|
|
json={"token": "flow-token-abc", "expires_at": "2099-01-01T00:00:00Z"},
|
|
)
|
|
validate_resp = httpx.Response(
|
|
200,
|
|
json={
|
|
"valid": True,
|
|
"sub": "acc-001",
|
|
"roles": ["admin"],
|
|
"expires_at": "2099-01-01T00:00:00Z",
|
|
},
|
|
)
|
|
logout_resp = httpx.Response(204)
|
|
respx.post(f"{BASE_URL}/v1/auth/login").mock(return_value=login_resp)
|
|
respx.post(f"{BASE_URL}/v1/token/validate").mock(return_value=validate_resp)
|
|
respx.post(f"{BASE_URL}/v1/auth/logout").mock(return_value=logout_resp)
|
|
with Client(BASE_URL) as c:
|
|
token, _ = c.login("alice", "password")
|
|
assert token == "flow-token-abc"
|
|
assert c.token == "flow-token-abc"
|
|
claims = c.validate_token(token)
|
|
assert claims.valid is True
|
|
assert "admin" in claims.roles
|
|
c.logout()
|
|
assert c.token is None
|