forked from retoor/devplacepy
feat: add DevPlace agent, auth token service, and form-data dependency modules
Add DevPlace agent configuration with dynamic OpenAPI schema fetching, implement access token issuance/resolution/revocation with configurable expiry, and create generic FastAPI dependency for JSON or form-encoded data validation against Pydantic models. Include comprehensive unit tests for form-data parsing and token lifecycle operations.
This commit is contained in:
@@ -0,0 +1,280 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
from starlette.datastructures import FormData
|
||||
|
||||
from devplacepy.dependencies import _formdata_to_dict, _JsonOrForm, json_or_form
|
||||
from devplacepy.models import (
|
||||
PostForm,
|
||||
CommentForm,
|
||||
LoginForm,
|
||||
SignupForm,
|
||||
VoteForm,
|
||||
IssueStatusForm,
|
||||
)
|
||||
|
||||
|
||||
# ── _formdata_to_dict unit tests ────────────────────────────────────────────
|
||||
|
||||
def test_formdata_scalar_fields():
|
||||
"""Plain string fields map directly."""
|
||||
fd = FormData([("content", "hello world"), ("title", "My Title")])
|
||||
body = _formdata_to_dict(fd, PostForm)
|
||||
assert body == {"content": "hello world", "title": "My Title"}
|
||||
|
||||
|
||||
def test_formdata_skips_absent_scalar():
|
||||
"""Fields not present in the form are omitted so Pydantic applies defaults."""
|
||||
fd = FormData([("content", "enough content here")])
|
||||
body = _formdata_to_dict(fd, PostForm)
|
||||
assert "content" in body
|
||||
assert "title" not in body
|
||||
assert "topic" not in body
|
||||
|
||||
|
||||
def test_formdata_list_field_with_values():
|
||||
"""Multi-value list fields become Python lists."""
|
||||
fd = FormData([
|
||||
("poll_options", "Red"),
|
||||
("poll_options", "Green"),
|
||||
("poll_options", "Blue"),
|
||||
])
|
||||
body = _formdata_to_dict(fd, PostForm)
|
||||
assert body["poll_options"] == ["Red", "Green", "Blue"]
|
||||
|
||||
|
||||
def test_formdata_list_field_single_value():
|
||||
"""A single value for a list field becomes a single-element list."""
|
||||
fd = FormData([("poll_options", "Only")])
|
||||
body = _formdata_to_dict(fd, PostForm)
|
||||
assert body["poll_options"] == ["Only"]
|
||||
|
||||
|
||||
def test_formdata_list_field_absent():
|
||||
"""Absent list fields are omitted."""
|
||||
fd = FormData([("content", "x" * 20)])
|
||||
body = _formdata_to_dict(fd, PostForm)
|
||||
assert "poll_options" not in body
|
||||
assert "attachment_uids" not in body
|
||||
|
||||
|
||||
def test_formdata_list_field_empty_string_dropped():
|
||||
"""A single empty string (browser hidden input) is treated as absent."""
|
||||
fd = FormData([("attachment_uids", ""), ("poll_options", "")])
|
||||
body = _formdata_to_dict(fd, PostForm)
|
||||
assert "attachment_uids" not in body
|
||||
assert "poll_options" not in body
|
||||
|
||||
|
||||
def test_formdata_list_field_mixed_empty_and_real():
|
||||
"""Real values survive even when some are empty."""
|
||||
fd = FormData([
|
||||
("attachment_uids", ""),
|
||||
("attachment_uids", "uid-real"),
|
||||
])
|
||||
body = _formdata_to_dict(fd, PostForm)
|
||||
assert body["attachment_uids"] == ["uid-real"]
|
||||
|
||||
|
||||
def test_formdata_literal_fields_are_scalar():
|
||||
"""Literal-typed fields (e.g. target_type in CommentForm) are treated as
|
||||
scalars — the origin check must not crash on typing.Literal."""
|
||||
fd = FormData([
|
||||
("content", "Great comment here!"),
|
||||
("target_type", "post"),
|
||||
("post_uid", "abc123"),
|
||||
])
|
||||
body = _formdata_to_dict(fd, CommentForm)
|
||||
assert body["target_type"] == "post"
|
||||
assert body["post_uid"] == "abc123"
|
||||
|
||||
|
||||
def test_formdata_literal_form_from_browser():
|
||||
"""CommentForm from a real browser sending empty hidden inputs."""
|
||||
fd = FormData([
|
||||
("content", "Nice post! Great stuff."),
|
||||
("target_type", "post"),
|
||||
("target_uid", ""),
|
||||
("post_uid", "abc123"),
|
||||
("parent_uid", ""),
|
||||
("attachment_uids", ""),
|
||||
])
|
||||
body = _formdata_to_dict(fd, CommentForm)
|
||||
assert body["content"] == "Nice post! Great stuff."
|
||||
assert body["target_type"] == "post"
|
||||
assert body["target_uid"] == ""
|
||||
assert body["post_uid"] == "abc123"
|
||||
assert body["parent_uid"] == ""
|
||||
assert "attachment_uids" not in body # empty → dropped
|
||||
|
||||
|
||||
def test_formdata_vote_form():
|
||||
"""value: int field with form data — string '1' is kept for Pydantic coerce."""
|
||||
fd = FormData([("value", "1")])
|
||||
body = _formdata_to_dict(fd, VoteForm)
|
||||
assert body["value"] == "1" # Pydantic coerces str→int
|
||||
|
||||
|
||||
def test_formdata_issue_status_literal():
|
||||
"""IssueStatusForm has Literal['open','closed'] — must not crash."""
|
||||
fd = FormData([("status", "open")])
|
||||
body = _formdata_to_dict(fd, IssueStatusForm)
|
||||
assert body["status"] == "open"
|
||||
|
||||
|
||||
def test_formdata_empty_form():
|
||||
"""An empty FormData produces an empty dict."""
|
||||
fd = FormData()
|
||||
body = _formdata_to_dict(fd, PostForm)
|
||||
assert body == {}
|
||||
|
||||
|
||||
def test_formdata_unknown_keys_ignored():
|
||||
"""Keys not in the model are simply not included."""
|
||||
fd = FormData([("nosuchfield", "val"), ("content", "x" * 20)])
|
||||
body = _formdata_to_dict(fd, PostForm)
|
||||
assert "nosuchfield" not in body
|
||||
assert body["content"] == "x" * 20
|
||||
|
||||
|
||||
# ── model_validate integration tests (without running server) ──────────────
|
||||
|
||||
def test_postform_validate_minimal_form():
|
||||
"""A minimal form submission validates successfully."""
|
||||
fd = FormData([("content", "Hello world post"), ("title", "Hi")])
|
||||
body = _formdata_to_dict(fd, PostForm)
|
||||
inst = PostForm.model_validate(body)
|
||||
assert inst.content == "Hello world post"
|
||||
assert inst.title == "Hi"
|
||||
assert inst.topic == "random" # default
|
||||
assert inst.attachment_uids == [] # default
|
||||
assert inst.poll_options == [] # default
|
||||
|
||||
|
||||
def test_postform_validate_full_browser_form():
|
||||
"""Browser form with all fields (including empty hidden inputs)."""
|
||||
fd = FormData([
|
||||
("content", "A full browser form post"),
|
||||
("title", "Full Form"),
|
||||
("topic", "random"),
|
||||
("project_uid", ""),
|
||||
("attachment_uids", ""),
|
||||
("poll_question", ""),
|
||||
("poll_options", ""),
|
||||
])
|
||||
body = _formdata_to_dict(fd, PostForm)
|
||||
inst = PostForm.model_validate(body)
|
||||
assert inst.content == "A full browser form post"
|
||||
assert inst.title == "Full Form"
|
||||
assert inst.project_uid == ""
|
||||
assert inst.poll_question == ""
|
||||
assert inst.attachment_uids == []
|
||||
assert inst.poll_options == []
|
||||
|
||||
|
||||
def test_postform_validate_json_body():
|
||||
"""A JSON-style dict (no FormData) validates correctly."""
|
||||
body = {
|
||||
"content": "JSON body post",
|
||||
"title": "JSON Post",
|
||||
"topic": "random",
|
||||
"attachment_uids": ["uid1", "uid2"],
|
||||
"poll_question": "Question?",
|
||||
"poll_options": ["A", "B"],
|
||||
}
|
||||
inst = PostForm.model_validate(body)
|
||||
assert inst.content == "JSON body post"
|
||||
assert inst.attachment_uids == ["uid1", "uid2"]
|
||||
assert inst.poll_options == ["A", "B"]
|
||||
|
||||
|
||||
def test_postform_validate_multi_value_poll():
|
||||
"""Multi-value form poll options become a list."""
|
||||
fd = FormData([
|
||||
("content", "Poll post here"),
|
||||
("poll_question", "Q?"),
|
||||
("poll_options", "Opt 1"),
|
||||
("poll_options", "Opt 2"),
|
||||
("poll_options", "Opt 3"),
|
||||
])
|
||||
body = _formdata_to_dict(fd, PostForm)
|
||||
inst = PostForm.model_validate(body)
|
||||
assert inst.poll_options == ["Opt 1", "Opt 2", "Opt 3"]
|
||||
assert inst.content == "Poll post here"
|
||||
|
||||
|
||||
def test_commentform_validate_browser():
|
||||
"""CommentForm validates with browser-style form data."""
|
||||
fd = FormData([
|
||||
("content", "Nice post! Great stuff here."),
|
||||
("target_type", "post"),
|
||||
("target_uid", ""),
|
||||
("post_uid", "abc123"),
|
||||
("parent_uid", ""),
|
||||
("attachment_uids", ""),
|
||||
])
|
||||
body = _formdata_to_dict(fd, CommentForm)
|
||||
inst = CommentForm.model_validate(body)
|
||||
assert inst.content == "Nice post! Great stuff here."
|
||||
assert inst.target_type == "post"
|
||||
assert inst.post_uid == "abc123"
|
||||
assert inst.attachment_uids == []
|
||||
|
||||
|
||||
def test_commentform_validate_json():
|
||||
"""CommentForm validates JSON dict."""
|
||||
body = {
|
||||
"content": "JSON comment",
|
||||
"target_type": "post",
|
||||
"post_uid": "abc123",
|
||||
"attachment_uids": ["a1"],
|
||||
}
|
||||
inst = CommentForm.model_validate(body)
|
||||
assert inst.content == "JSON comment"
|
||||
assert inst.attachment_uids == ["a1"]
|
||||
|
||||
|
||||
def test_loginform_validate_json():
|
||||
"""LoginForm with JSON body."""
|
||||
body = {
|
||||
"email": "user@test.com",
|
||||
"password": "secret123",
|
||||
"remember_me": "",
|
||||
"next": "",
|
||||
}
|
||||
inst = LoginForm.model_validate(body)
|
||||
assert inst.email == "user@test.com"
|
||||
|
||||
|
||||
def test_signup_form_rejected_short_password():
|
||||
"""SignupForm raises ValidationError for short passwords."""
|
||||
from pydantic import ValidationError
|
||||
import pytest
|
||||
|
||||
fd = FormData([
|
||||
("username", "valid_user"),
|
||||
("email", "u@t.com"),
|
||||
("password", "short"),
|
||||
("confirm_password", "short"),
|
||||
])
|
||||
body = _formdata_to_dict(fd, SignupForm)
|
||||
with pytest.raises(ValidationError):
|
||||
SignupForm.model_validate(body)
|
||||
|
||||
|
||||
def test_postform_rejected_content_too_short():
|
||||
"""PostForm requires content >= 10 chars."""
|
||||
from pydantic import ValidationError
|
||||
import pytest
|
||||
|
||||
fd = FormData([("content", "short")])
|
||||
body = _formdata_to_dict(fd, PostForm)
|
||||
with pytest.raises(ValidationError):
|
||||
PostForm.model_validate(body)
|
||||
|
||||
|
||||
# ── json_or_form dependency factory ─────────────────────────────────────────
|
||||
|
||||
def test_json_or_form_returns_callable():
|
||||
dep = json_or_form(PostForm)
|
||||
assert isinstance(dep, _JsonOrForm)
|
||||
assert dep.model is PostForm
|
||||
@@ -0,0 +1,374 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
import secrets
|
||||
from datetime import datetime, timedelta, timezone
|
||||
|
||||
import pytest
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
from devplacepy.database import get_table
|
||||
from devplacepy.utils import generate_uid, _resolve_user, hash_password
|
||||
from devplacepy.services.access_tokens import (
|
||||
issue_token,
|
||||
resolve_token,
|
||||
revoke_token,
|
||||
revoke_all,
|
||||
prune_expired,
|
||||
)
|
||||
from devplacepy.main import app
|
||||
|
||||
|
||||
# ── mock helpers ─────────────────────────────────────────────────────────────
|
||||
|
||||
class _MockHeaders:
|
||||
def __init__(self, data: dict):
|
||||
self._data = data
|
||||
|
||||
def get(self, key: str, default=None):
|
||||
return self._data.get(key, default)
|
||||
|
||||
|
||||
class _MockCookies:
|
||||
def __init__(self, data: dict):
|
||||
self._data = data
|
||||
|
||||
def get(self, key: str):
|
||||
return self._data.get(key)
|
||||
|
||||
|
||||
class _MockRequest:
|
||||
def __init__(self, headers=None, cookies=None):
|
||||
self.headers = _MockHeaders(headers or {})
|
||||
self.cookies = _MockCookies(cookies or {})
|
||||
|
||||
|
||||
# ── seed helpers ─────────────────────────────────────────────────────────────
|
||||
|
||||
def _seed_user(username, email, is_active=True, **extra):
|
||||
uid = generate_uid()
|
||||
get_table("users").insert(
|
||||
{
|
||||
"uid": uid,
|
||||
"username": username,
|
||||
"email": email,
|
||||
"is_active": is_active,
|
||||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||||
**extra,
|
||||
}
|
||||
)
|
||||
return uid
|
||||
|
||||
|
||||
# ── service tests ────────────────────────────────────────────────────────────
|
||||
|
||||
def test_issue_token_returns_expected_fields(local_db):
|
||||
"""issue_token returns all expected fields with correct types/values."""
|
||||
uid = _seed_user("issuer", "issuer@test.dev")
|
||||
user = get_table("users").find_one(uid=uid)
|
||||
|
||||
result = issue_token(user)
|
||||
|
||||
assert isinstance(result, dict)
|
||||
assert "access_token" in result
|
||||
assert len(result["access_token"]) == 64 # 32 bytes hex
|
||||
assert result["token_type"] == "bearer"
|
||||
assert isinstance(result["expires_in"], int)
|
||||
assert result["expires_in"] > 0
|
||||
assert "expires_at" in result
|
||||
assert "uid" in result
|
||||
assert len(result["uid"]) > 0
|
||||
|
||||
|
||||
def test_resolve_token_valid(local_db):
|
||||
"""Issue a token then resolve it — returns the correct user."""
|
||||
uid = _seed_user("resolver", "resolver@test.dev")
|
||||
user = get_table("users").find_one(uid=uid)
|
||||
|
||||
result = issue_token(user)
|
||||
resolved = resolve_token(result["access_token"])
|
||||
|
||||
assert resolved is not None
|
||||
assert resolved["uid"] == uid
|
||||
assert resolved["username"] == "resolver"
|
||||
|
||||
|
||||
def test_resolve_token_not_found(local_db):
|
||||
"""A random 64-char hex string that has no matching token returns None."""
|
||||
random_token = secrets.token_hex(32)
|
||||
assert resolve_token(random_token) is None
|
||||
|
||||
|
||||
def test_resolve_token_expired(local_db):
|
||||
"""Issue a token with max_age_days=0 (expires immediately) → resolve returns None."""
|
||||
uid = _seed_user("expired_tok", "expiredtok@test.dev")
|
||||
user = get_table("users").find_one(uid=uid)
|
||||
|
||||
result = issue_token(user, max_age_days=0)
|
||||
resolved = resolve_token(result["access_token"])
|
||||
|
||||
assert resolved is None
|
||||
|
||||
|
||||
def test_resolve_token_inactive_user(local_db):
|
||||
"""Issue a token for an inactive user → resolve returns None."""
|
||||
uid = _seed_user("inactive_tok", "inactivetok@test.dev", is_active=False)
|
||||
user = get_table("users").find_one(uid=uid)
|
||||
|
||||
result = issue_token(user)
|
||||
resolved = resolve_token(result["access_token"])
|
||||
|
||||
assert resolved is None
|
||||
|
||||
|
||||
def test_revoke_token_valid(local_db):
|
||||
"""Issue a token, revoke by uid, resolve returns None."""
|
||||
uid = _seed_user("revoker", "revoker@test.dev")
|
||||
user = get_table("users").find_one(uid=uid)
|
||||
|
||||
result = issue_token(user)
|
||||
token_uid = result["uid"]
|
||||
token_str = result["access_token"]
|
||||
|
||||
# Token resolves before revoke
|
||||
assert resolve_token(token_str) is not None
|
||||
|
||||
revoked = revoke_token(token_uid)
|
||||
assert revoked is True
|
||||
|
||||
# Token no longer resolves
|
||||
assert resolve_token(token_str) is None
|
||||
|
||||
|
||||
def test_revoke_token_not_found(local_db):
|
||||
"""Revoking a random uid returns False."""
|
||||
random_uid = generate_uid()
|
||||
assert revoke_token(random_uid) is False
|
||||
|
||||
|
||||
def test_revoke_all(local_db):
|
||||
"""Issue 3 tokens, revoke_all, all resolve to None."""
|
||||
uid = _seed_user("revokeall", "revokeall@test.dev")
|
||||
user = get_table("users").find_one(uid=uid)
|
||||
|
||||
tokens = [issue_token(user) for _ in range(3)]
|
||||
for t in tokens:
|
||||
assert resolve_token(t["access_token"]) is not None
|
||||
|
||||
count = revoke_all(uid)
|
||||
assert count == 3
|
||||
|
||||
for t in tokens:
|
||||
assert resolve_token(t["access_token"]) is None
|
||||
|
||||
|
||||
def test_prune_expired(local_db):
|
||||
"""Issue expired tokens, prune, all resolve to None."""
|
||||
uid = _seed_user("pruner", "pruner@test.dev")
|
||||
user = get_table("users").find_one(uid=uid)
|
||||
|
||||
# max_age_days=-1 → expires_at is 1 day in the past
|
||||
tokens = [issue_token(user, max_age_days=-1) for _ in range(2)]
|
||||
|
||||
# They're already expired, resolve returns None without pruning
|
||||
for t in tokens:
|
||||
assert resolve_token(t["access_token"]) is None
|
||||
|
||||
# But they still exist in the table (deleted_at is None)
|
||||
for t in tokens:
|
||||
row = get_table("access_tokens").find_one(uid=t["uid"])
|
||||
assert row is not None
|
||||
assert row["deleted_at"] is None
|
||||
|
||||
pruned = prune_expired()
|
||||
assert pruned >= 2
|
||||
|
||||
# Now they're soft-deleted
|
||||
for t in tokens:
|
||||
row = get_table("access_tokens").find_one(uid=t["uid"])
|
||||
assert row["deleted_at"] is not None
|
||||
|
||||
|
||||
def test_issue_token_with_label(local_db):
|
||||
"""Label is stored in the access_tokens table."""
|
||||
uid = _seed_user("labeler", "labeler@test.dev")
|
||||
user = get_table("users").find_one(uid=uid)
|
||||
|
||||
result = issue_token(user, label="my-cli-token")
|
||||
row = get_table("access_tokens").find_one(uid=result["uid"])
|
||||
|
||||
assert row is not None
|
||||
assert row["label"] == "my-cli-token"
|
||||
|
||||
|
||||
# ── auth chain integration tests ─────────────────────────────────────────────
|
||||
|
||||
def test_resolve_user_from_access_token_x_api_key(local_db):
|
||||
"""_resolve_user finds a user from X-API-KEY header containing an access token."""
|
||||
uid = _seed_user("acctok_xapikey", "acctokxapikey@test.dev")
|
||||
user = get_table("users").find_one(uid=uid)
|
||||
|
||||
result = issue_token(user)
|
||||
token_str = result["access_token"]
|
||||
|
||||
request = _MockRequest(headers={"X-API-KEY": token_str})
|
||||
resolved = _resolve_user(request)
|
||||
|
||||
assert resolved is not None
|
||||
assert resolved["uid"] == uid
|
||||
assert resolved["username"] == "acctok_xapikey"
|
||||
|
||||
|
||||
def test_resolve_user_from_access_token_bearer(local_db):
|
||||
"""_resolve_user finds a user from Authorization: Bearer with access token."""
|
||||
uid = _seed_user("acctok_bearer", "acctokbearer@test.dev")
|
||||
user = get_table("users").find_one(uid=uid)
|
||||
|
||||
result = issue_token(user)
|
||||
token_str = result["access_token"]
|
||||
|
||||
request = _MockRequest(headers={"Authorization": f"Bearer {token_str}"})
|
||||
resolved = _resolve_user(request)
|
||||
|
||||
assert resolved is not None
|
||||
assert resolved["uid"] == uid
|
||||
assert resolved["username"] == "acctok_bearer"
|
||||
|
||||
|
||||
def test_resolve_user_access_token_priority_after_api_key(local_db):
|
||||
"""api_key user wins over access token when the same string matches both."""
|
||||
shared_key = secrets.token_hex(32) # 64-char hex
|
||||
|
||||
# User A — owns the api_key
|
||||
uid_a = _seed_user("apikey_winner", "apikeywinner@test.dev", api_key=shared_key)
|
||||
|
||||
# User B — owns an access token with the same string as the key
|
||||
uid_b = _seed_user("acctok_loser", "acctokloser@test.dev")
|
||||
get_table("access_tokens").insert(
|
||||
{
|
||||
"uid": generate_uid(),
|
||||
"token": shared_key,
|
||||
"user_uid": uid_b,
|
||||
"label": "",
|
||||
"expires_at": (datetime.now(timezone.utc) + timedelta(days=1)).isoformat(),
|
||||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||||
"deleted_at": None,
|
||||
"deleted_by": None,
|
||||
}
|
||||
)
|
||||
|
||||
# X-API-KEY path — api_key user wins
|
||||
request = _MockRequest(headers={"X-API-KEY": shared_key})
|
||||
resolved = _resolve_user(request)
|
||||
assert resolved is not None
|
||||
assert resolved["uid"] == uid_a
|
||||
assert resolved["username"] == "apikey_winner"
|
||||
|
||||
# Bearer path — api_key user wins
|
||||
request2 = _MockRequest(headers={"Authorization": f"Bearer {shared_key}"})
|
||||
resolved2 = _resolve_user(request2)
|
||||
assert resolved2 is not None
|
||||
assert resolved2["uid"] == uid_a
|
||||
assert resolved2["username"] == "apikey_winner"
|
||||
|
||||
|
||||
def test_access_token_wrong_length_skipped(local_db):
|
||||
"""Non-64-char strings never reach the access_tokens DB lookup."""
|
||||
# Empty / falsy
|
||||
request = _MockRequest(headers={"X-API-KEY": ""})
|
||||
assert _resolve_user(request) is None
|
||||
|
||||
# Too short
|
||||
request = _MockRequest(headers={"X-API-KEY": "abc123"})
|
||||
assert _resolve_user(request) is None
|
||||
|
||||
# 63 chars (one short)
|
||||
request = _MockRequest(headers={"X-API-KEY": "a" * 63})
|
||||
assert _resolve_user(request) is None
|
||||
|
||||
# 65 chars (one too many)
|
||||
request = _MockRequest(headers={"X-API-KEY": "a" * 65})
|
||||
assert _resolve_user(request) is None
|
||||
|
||||
# 40-char hex reaches devrant token lookup (not access token)
|
||||
# but with no DB match it returns None
|
||||
request = _MockRequest(headers={"X-API-KEY": "a" * 40})
|
||||
assert _resolve_user(request) is None
|
||||
|
||||
|
||||
# ── endpoint tests (TestClient) ──────────────────────────────────────────────
|
||||
|
||||
@pytest.fixture
|
||||
def _token_test_user(local_db):
|
||||
"""Seed a user with a known password for the token endpoint tests."""
|
||||
pw_hash = hash_password("test123456")
|
||||
uid = _seed_user("tokentest", "tokentest@test.dev", password_hash=pw_hash)
|
||||
return uid
|
||||
|
||||
|
||||
def test_token_endpoint_valid_credentials(local_db, _token_test_user):
|
||||
"""POST /auth/token with valid email+password → 200 with access_token."""
|
||||
client = TestClient(app)
|
||||
resp = client.post(
|
||||
"/auth/token",
|
||||
data={"email": "tokentest@test.dev", "password": "test123456"},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert "access_token" in body
|
||||
assert len(body["access_token"]) == 64
|
||||
assert body["token_type"] == "bearer"
|
||||
assert isinstance(body["expires_in"], int)
|
||||
assert body["expires_in"] > 0
|
||||
|
||||
|
||||
def test_token_endpoint_valid_username(local_db, _token_test_user):
|
||||
"""POST /auth/token with username instead of email → 200."""
|
||||
client = TestClient(app)
|
||||
resp = client.post(
|
||||
"/auth/token",
|
||||
data={"email": "tokentest", "password": "test123456"},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
body = resp.json()
|
||||
assert "access_token" in body
|
||||
|
||||
|
||||
def test_token_endpoint_bad_password(local_db, _token_test_user):
|
||||
"""POST /auth/token with wrong password → 401."""
|
||||
client = TestClient(app)
|
||||
resp = client.post(
|
||||
"/auth/token",
|
||||
data={"email": "tokentest@test.dev", "password": "wrongpassword"},
|
||||
)
|
||||
|
||||
assert resp.status_code == 401
|
||||
body = resp.json()
|
||||
assert "error" in body
|
||||
|
||||
|
||||
def test_token_endpoint_missing_fields(local_db):
|
||||
"""POST /auth/token with whitespace-only fields → 400.
|
||||
|
||||
Pydantic rejects truly missing/empty fields (422), but the endpoint's
|
||||
own guard catches values that pass Pydantic's min_length yet become
|
||||
empty after strip(), e.g. a single space.
|
||||
"""
|
||||
client = TestClient(app)
|
||||
|
||||
# Whitespace-only email passes Pydantic (len >= 1), stripped to empty → 400
|
||||
resp = client.post(
|
||||
"/auth/token",
|
||||
json={"email": " ", "password": "secret"},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert resp.json()["error"] == "email and password are required"
|
||||
|
||||
# Both whitespace-only → 400
|
||||
resp = client.post(
|
||||
"/auth/token",
|
||||
json={"email": " ", "password": " "},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
assert resp.json()["error"] == "email and password are required"
|
||||
@@ -0,0 +1,184 @@
|
||||
# retoor <retoor@molodetz.nl>
|
||||
|
||||
import secrets
|
||||
from datetime import datetime, timezone
|
||||
|
||||
from devplacepy.database import get_table
|
||||
from devplacepy.utils import generate_uid, _user_from_devrant_token, _resolve_user
|
||||
from devplacepy.services.devrant.tokens import resolve_user_by_key
|
||||
|
||||
|
||||
# ── mock helpers ─────────────────────────────────────────────────────────────
|
||||
|
||||
class _MockHeaders:
|
||||
def __init__(self, data: dict):
|
||||
self._data = data
|
||||
|
||||
def get(self, key: str, default=None):
|
||||
return self._data.get(key, default)
|
||||
|
||||
|
||||
class _MockCookies:
|
||||
def __init__(self, data: dict):
|
||||
self._data = data
|
||||
|
||||
def get(self, key: str):
|
||||
return self._data.get(key)
|
||||
|
||||
|
||||
class _MockRequest:
|
||||
def __init__(self, headers=None, cookies=None):
|
||||
self.headers = _MockHeaders(headers or {})
|
||||
self.cookies = _MockCookies(cookies or {})
|
||||
|
||||
|
||||
# ── seed helpers ─────────────────────────────────────────────────────────────
|
||||
|
||||
def _seed_user(username, email, is_active=True, **extra):
|
||||
uid = generate_uid()
|
||||
get_table("users").insert(
|
||||
{
|
||||
"uid": uid,
|
||||
"username": username,
|
||||
"email": email,
|
||||
"is_active": is_active,
|
||||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||||
**extra,
|
||||
}
|
||||
)
|
||||
return uid
|
||||
|
||||
|
||||
def _seed_token(user_uid, key, expire_time):
|
||||
get_table("devrant_tokens").insert(
|
||||
{
|
||||
"uid": generate_uid(),
|
||||
"key": key,
|
||||
"user_uid": user_uid,
|
||||
"user_id": 1,
|
||||
"expire_time": expire_time,
|
||||
"created_at": datetime.now(timezone.utc).isoformat(),
|
||||
"deleted_at": None,
|
||||
"deleted_by": None,
|
||||
}
|
||||
)
|
||||
|
||||
|
||||
# ── resolve_user_by_key tests ────────────────────────────────────────────────
|
||||
|
||||
def test_resolve_user_by_key_valid(local_db):
|
||||
"""Create a devrant token for a user, then resolve it by key — the right user is returned."""
|
||||
uid = _seed_user("dr_valid", "drvalid@test.dev")
|
||||
token_key = secrets.token_hex(20)
|
||||
expire_time = int(datetime.now(timezone.utc).timestamp()) + 86400
|
||||
_seed_token(uid, token_key, expire_time)
|
||||
|
||||
user = resolve_user_by_key(token_key)
|
||||
assert user is not None
|
||||
assert user["uid"] == uid
|
||||
assert user["username"] == "dr_valid"
|
||||
|
||||
|
||||
def test_resolve_user_by_key_not_found(local_db):
|
||||
"""Look up a random key that has no token — verify None."""
|
||||
random_key = secrets.token_hex(20)
|
||||
assert resolve_user_by_key(random_key) is None
|
||||
|
||||
|
||||
def test_resolve_user_by_key_expired(local_db):
|
||||
"""Create an expired token — resolve returns None."""
|
||||
uid = _seed_user("dr_expired", "drexpired@test.dev")
|
||||
token_key = secrets.token_hex(20)
|
||||
expire_time = int(datetime.now(timezone.utc).timestamp()) - 86400
|
||||
_seed_token(uid, token_key, expire_time)
|
||||
|
||||
assert resolve_user_by_key(token_key) is None
|
||||
|
||||
|
||||
def test_resolve_user_by_key_inactive_user(local_db):
|
||||
"""Create a token for an inactive user — resolve returns None."""
|
||||
uid = _seed_user("dr_inactive", "drinactive@test.dev", is_active=False)
|
||||
token_key = secrets.token_hex(20)
|
||||
expire_time = int(datetime.now(timezone.utc).timestamp()) + 86400
|
||||
_seed_token(uid, token_key, expire_time)
|
||||
|
||||
assert resolve_user_by_key(token_key) is None
|
||||
|
||||
|
||||
# ── _user_from_devrant_token hex-format guard tests ──────────────────────────
|
||||
|
||||
def test_user_from_devrant_token_hex_format(local_db):
|
||||
"""_user_from_devrant_token only tries DB for 40-char hex strings;
|
||||
non-hex or wrong-length keys return None without a DB hit."""
|
||||
# empty / falsy
|
||||
assert _user_from_devrant_token("") is None
|
||||
assert _user_from_devrant_token(None) is None
|
||||
|
||||
# wrong length
|
||||
assert _user_from_devrant_token("abc123") is None
|
||||
assert _user_from_devrant_token("a" * 39) is None
|
||||
assert _user_from_devrant_token("a" * 41) is None
|
||||
|
||||
# right length but non-hex
|
||||
assert _user_from_devrant_token("z" * 40) is None
|
||||
assert _user_from_devrant_token(("g" * 39) + "x") is None
|
||||
|
||||
# valid 40-char hex that simply has no matching token in the DB
|
||||
assert _user_from_devrant_token("a" * 40) is None
|
||||
|
||||
|
||||
# ── _resolve_user bridge tests ───────────────────────────────────────────────
|
||||
|
||||
def test_resolve_user_falls_back_to_devrant_token(local_db):
|
||||
"""_resolve_user finds a user from X-API-KEY header containing a DevRant token key."""
|
||||
uid = _seed_user("dr_fallback", "drfallback@test.dev")
|
||||
token_key = secrets.token_hex(20)
|
||||
expire_time = int(datetime.now(timezone.utc).timestamp()) + 86400
|
||||
_seed_token(uid, token_key, expire_time)
|
||||
|
||||
request = _MockRequest(headers={"X-API-KEY": token_key})
|
||||
user = _resolve_user(request)
|
||||
assert user is not None
|
||||
assert user["uid"] == uid
|
||||
assert user["username"] == "dr_fallback"
|
||||
|
||||
|
||||
def test_resolve_user_falls_back_to_devrant_token_bearer(local_db):
|
||||
"""_resolve_user finds a user from Authorization: Bearer header containing a DevRant token key."""
|
||||
uid = _seed_user("dr_bearer", "drbearer@test.dev")
|
||||
token_key = secrets.token_hex(20)
|
||||
expire_time = int(datetime.now(timezone.utc).timestamp()) + 86400
|
||||
_seed_token(uid, token_key, expire_time)
|
||||
|
||||
request = _MockRequest(headers={"Authorization": f"Bearer {token_key}"})
|
||||
user = _resolve_user(request)
|
||||
assert user is not None
|
||||
assert user["uid"] == uid
|
||||
assert user["username"] == "dr_bearer"
|
||||
|
||||
|
||||
def test_resolve_user_prefers_api_key_over_devrant(local_db):
|
||||
"""If a key matches both an api_key and a devrant token, the api_key user wins."""
|
||||
shared_key = secrets.token_hex(20)
|
||||
|
||||
# User A — owns the api_key
|
||||
uid_a = _seed_user("api_key_owner", "apikey@test.dev", api_key=shared_key)
|
||||
|
||||
# User B — owns the devrant token with the same key
|
||||
uid_b = _seed_user("devrant_owner", "drowner@test.dev")
|
||||
expire_time = int(datetime.now(timezone.utc).timestamp()) + 86400
|
||||
_seed_token(uid_b, shared_key, expire_time)
|
||||
|
||||
# ── X-API-KEY path ──
|
||||
request = _MockRequest(headers={"X-API-KEY": shared_key})
|
||||
user = _resolve_user(request)
|
||||
assert user is not None
|
||||
assert user["uid"] == uid_a
|
||||
assert user["username"] == "api_key_owner"
|
||||
|
||||
# ── Bearer path ──
|
||||
request2 = _MockRequest(headers={"Authorization": f"Bearer {shared_key}"})
|
||||
user2 = _resolve_user(request2)
|
||||
assert user2 is not None
|
||||
assert user2["uid"] == uid_a
|
||||
assert user2["username"] == "api_key_owner"
|
||||
Reference in New Issue
Block a user