robot-u-site/tests/test_app.py

809 lines
32 KiB
Python
Raw Normal View History

2026-04-08 06:03:48 -04:00
from __future__ import annotations
2026-04-13 18:19:50 -04:00
import hmac
import os
2026-04-08 06:03:48 -04:00
import unittest
2026-04-13 18:19:50 -04:00
from hashlib import sha256
from urllib.parse import parse_qs, urlparse
2026-04-08 06:03:48 -04:00
from unittest.mock import AsyncMock, patch
from fastapi.testclient import TestClient
2026-04-08 06:03:48 -04:00
from app import create_app
from settings import get_settings
class AppTestCase(unittest.TestCase):
def setUp(self) -> None:
self.env_patcher = patch.dict(
os.environ,
{
"APP_BASE_URL": "http://testserver",
2026-04-12 22:02:47 -04:00
"AUTH_SECRET_KEY": "test-auth-secret-key-that-is-long-enough",
"FORGEJO_TOKEN": "",
"FORGEJO_OAUTH_CLIENT_ID": "client-id",
"FORGEJO_OAUTH_CLIENT_SECRET": "client-secret",
},
)
self.env_patcher.start()
2026-04-08 06:03:48 -04:00
get_settings.cache_clear()
self.app = create_app()
self.client = TestClient(self.app)
2026-04-08 06:03:48 -04:00
def tearDown(self) -> None:
get_settings.cache_clear()
self.env_patcher.stop()
2026-04-08 06:03:48 -04:00
def test_health(self) -> None:
response = self.client.get("/health")
2026-04-08 06:03:48 -04:00
self.assertEqual(response.status_code, 200)
self.assertEqual(response.json(), {"status": "ok"})
2026-04-08 06:03:48 -04:00
def test_prototype_payload_shape(self) -> None:
payload = {
"hero": {"title": "Robot U"},
"auth": {
"authenticated": False,
"login": None,
"source": "none",
"can_reply": False,
"oauth_configured": True,
},
"featured_courses": [],
"recent_posts": [],
"recent_discussions": [],
"upcoming_events": [],
"source_of_truth": [],
}
builder = AsyncMock(return_value=payload)
2026-04-13 18:19:50 -04:00
with patch("prototype_cache.build_live_prototype_payload", new=builder):
response = self.client.get("/api/prototype")
response_payload = response.json()
self.assertEqual(response.status_code, 200)
self.assertIn("hero", response_payload)
self.assertIn("auth", response_payload)
self.assertIn("featured_courses", response_payload)
self.assertIn("recent_posts", response_payload)
self.assertIn("recent_discussions", response_payload)
self.assertIn("upcoming_events", response_payload)
self.assertIn("source_of_truth", response_payload)
self.assertEqual(builder.call_args.kwargs["forgejo_token"], None)
self.assertEqual(builder.call_args.kwargs["auth_source"], "none")
2026-04-13 18:19:50 -04:00
def test_prototype_reuses_cached_public_payload(self) -> None:
payload = {
"hero": {"title": "Robot U"},
"auth": {
"authenticated": False,
"login": None,
"source": "none",
"can_reply": False,
"oauth_configured": True,
},
"featured_courses": [],
"recent_posts": [],
"recent_discussions": [],
"upcoming_events": [],
"source_of_truth": [],
}
builder = AsyncMock(return_value=payload)
with patch("prototype_cache.build_live_prototype_payload", new=builder):
first_response = self.client.get("/api/prototype")
second_response = self.client.get("/api/prototype")
self.assertEqual(first_response.status_code, 200)
self.assertEqual(second_response.status_code, 200)
self.assertEqual(builder.await_count, 1)
def test_forgejo_webhook_invalidates_prototype_cache(self) -> None:
initial_payload = {
"hero": {"title": "Before"},
"auth": {
"authenticated": False,
"login": None,
"source": "none",
"can_reply": False,
"oauth_configured": True,
},
"featured_courses": [],
"recent_posts": [],
"recent_discussions": [],
"upcoming_events": [],
"source_of_truth": [],
}
refreshed_payload = {
**initial_payload,
"hero": {"title": "After"},
}
builder = AsyncMock(side_effect=[initial_payload, refreshed_payload])
with patch("prototype_cache.build_live_prototype_payload", new=builder):
first_response = self.client.get("/api/prototype")
webhook_response = self.client.post("/api/forgejo/webhook", json={"ref": "main"})
second_response = self.client.get("/api/prototype")
self.assertEqual(first_response.json()["hero"]["title"], "Before")
self.assertEqual(webhook_response.status_code, 200)
self.assertEqual(second_response.json()["hero"]["title"], "After")
self.assertEqual(builder.await_count, 2)
def test_forgejo_webhook_validates_signature_when_secret_is_configured(self) -> None:
get_settings.cache_clear()
body = b'{"ref":"main"}'
signature = hmac.new(b"webhook-secret", body, sha256).hexdigest()
with patch.dict(os.environ, {"FORGEJO_WEBHOOK_SECRET": "webhook-secret"}):
bad_response = self.client.post(
"/api/forgejo/webhook",
content=body,
headers={"X-Forgejo-Signature": "bad-signature"},
)
good_response = self.client.post(
"/api/forgejo/webhook",
content=body,
headers={"X-Forgejo-Signature": signature},
)
self.assertEqual(bad_response.status_code, 401)
self.assertEqual(good_response.status_code, 200)
def test_prototype_accepts_authorization_token(self) -> None:
payload = {
"hero": {"title": "Robot U"},
"auth": {
2026-04-13 18:19:50 -04:00
"authenticated": False,
"login": None,
"source": "none",
"can_reply": False,
"oauth_configured": True,
},
"featured_courses": [],
"recent_posts": [],
"recent_discussions": [],
"upcoming_events": [],
"source_of_truth": [],
}
builder = AsyncMock(return_value=payload)
2026-04-13 18:19:50 -04:00
fake_client = _FakeForgejoClient(user={"login": "kacper"})
with (
patch("prototype_cache.build_live_prototype_payload", new=builder),
2026-04-14 20:17:29 -04:00
patch("app.ForgejoClient", return_value=fake_client) as client_factory,
2026-04-13 18:19:50 -04:00
):
response = self.client.get(
"/api/prototype",
headers={"Authorization": "token test-token"},
)
2026-04-13 18:19:50 -04:00
response_payload = response.json()
self.assertEqual(response.status_code, 200)
2026-04-13 18:19:50 -04:00
self.assertEqual(builder.call_args.kwargs["forgejo_token"], None)
self.assertEqual(builder.call_args.kwargs["auth_source"], "none")
self.assertEqual(response_payload["auth"]["authenticated"], True)
self.assertEqual(response_payload["auth"]["login"], "kacper")
self.assertEqual(response_payload["auth"]["source"], "authorization")
2026-04-14 20:17:29 -04:00
self.assertEqual(client_factory.call_args.kwargs["auth_scheme"], "token")
def test_prototype_preserves_bearer_authorization_scheme(self) -> None:
payload = {
"hero": {"title": "Robot U"},
"auth": {
"authenticated": False,
"login": None,
"source": "none",
"can_reply": False,
"oauth_configured": True,
},
"featured_courses": [],
"recent_posts": [],
"recent_discussions": [],
"upcoming_events": [],
"source_of_truth": [],
}
builder = AsyncMock(return_value=payload)
fake_client = _FakeForgejoClient(user={"login": "kacper"})
with (
patch("prototype_cache.build_live_prototype_payload", new=builder),
patch("app.ForgejoClient", return_value=fake_client) as client_factory,
):
response = self.client.get(
"/api/prototype",
headers={"Authorization": "Bearer oauth-token"},
)
self.assertEqual(response.status_code, 200)
self.assertEqual(client_factory.call_args.kwargs["forgejo_token"], "oauth-token")
self.assertEqual(client_factory.call_args.kwargs["auth_scheme"], "Bearer")
def test_prototype_can_use_server_token_without_user_session(self) -> None:
payload = {
"hero": {"title": "Robot U"},
"auth": {
"authenticated": False,
"login": None,
"source": "server",
"can_reply": False,
"oauth_configured": True,
},
"featured_courses": [],
"recent_posts": [],
"recent_discussions": [],
"upcoming_events": [],
"source_of_truth": [],
}
builder = AsyncMock(return_value=payload)
get_settings.cache_clear()
with (
patch.dict(os.environ, {"FORGEJO_TOKEN": "server-token"}),
2026-04-13 18:19:50 -04:00
patch("prototype_cache.build_live_prototype_payload", new=builder),
):
response = self.client.get("/api/prototype")
self.assertEqual(response.status_code, 200)
self.assertEqual(builder.call_args.kwargs["forgejo_token"], "server-token")
self.assertEqual(builder.call_args.kwargs["auth_source"], "server")
def test_auth_session_ignores_server_token_fallback(self) -> None:
get_settings.cache_clear()
with patch.dict(os.environ, {"FORGEJO_TOKEN": "server-token"}):
response = self.client.get("/api/auth/session")
payload = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(payload["authenticated"], False)
self.assertEqual(payload["login"], None)
self.assertEqual(payload["source"], "none")
self.assertEqual(payload["can_reply"], False)
def test_forgejo_auth_start_redirects_to_authorization_endpoint(self) -> None:
fake_client = _FakeForgejoClient()
with patch("app.ForgejoClient", return_value=fake_client):
response = self.client.get(
"/api/auth/forgejo/start?return_to=/discussions/7",
follow_redirects=False,
)
location = response.headers["location"]
query = parse_qs(urlparse(location).query)
self.assertEqual(response.status_code, 303)
self.assertTrue(location.startswith("https://aksal.cloud/login/oauth/authorize?"))
self.assertEqual(query["client_id"], ["client-id"])
self.assertEqual(query["redirect_uri"], ["http://testserver/api/auth/forgejo/callback"])
self.assertEqual(query["response_type"], ["code"])
self.assertEqual(query["scope"], ["openid profile"])
self.assertEqual(query["code_challenge_method"], ["S256"])
self.assertIn("state", query)
self.assertIn("code_challenge", query)
def test_forgejo_auth_callback_sets_session_cookie(self) -> None:
fake_client = _FakeForgejoClient(user={"login": "kacper"}, access_token="oauth-token")
with patch("app.ForgejoClient", return_value=fake_client):
start_response = self.client.get(
"/api/auth/forgejo/start?return_to=/discussions/7",
follow_redirects=False,
)
state = parse_qs(urlparse(start_response.headers["location"]).query)["state"][0]
callback_response = self.client.get(
f"/api/auth/forgejo/callback?code=auth-code&state={state}",
follow_redirects=False,
)
self.assertEqual(callback_response.status_code, 303)
self.assertEqual(callback_response.headers["location"], "/discussions/7")
self.assertIn("robot_u_session", callback_response.cookies)
2026-04-12 22:02:47 -04:00
self.assertNotIn("oauth-token", callback_response.cookies["robot_u_session"])
self.assertEqual(fake_client.exchanged_code, "auth-code")
session_response = self.client.get("/api/auth/session")
session_payload = session_response.json()
self.assertEqual(session_response.status_code, 200)
self.assertEqual(session_payload["authenticated"], True)
self.assertEqual(session_payload["login"], "kacper")
self.assertEqual(session_payload["source"], "session")
self.assertEqual(session_payload["can_reply"], True)
def test_prototype_uses_signed_in_forgejo_identity_token(self) -> None:
fake_client = _FakeForgejoClient(user={"login": "kacper"}, access_token="oauth-token")
with patch("app.ForgejoClient", return_value=fake_client):
start_response = self.client.get(
"/api/auth/forgejo/start",
follow_redirects=False,
)
state = parse_qs(urlparse(start_response.headers["location"]).query)["state"][0]
self.client.get(
f"/api/auth/forgejo/callback?code=auth-code&state={state}",
follow_redirects=False,
)
payload = {
"hero": {"title": "Robot U"},
"auth": {
"authenticated": True,
"login": "kacper",
"source": "session",
"can_reply": True,
"oauth_configured": True,
},
2026-04-08 06:03:48 -04:00
"featured_courses": [],
"recent_posts": [],
"recent_discussions": [],
"upcoming_events": [],
"source_of_truth": [],
}
builder = AsyncMock(return_value=payload)
2026-04-13 18:19:50 -04:00
with patch("prototype_cache.build_live_prototype_payload", new=builder):
response = self.client.get("/api/prototype")
2026-04-13 18:19:50 -04:00
response_payload = response.json()
2026-04-08 06:03:48 -04:00
self.assertEqual(response.status_code, 200)
2026-04-13 18:19:50 -04:00
self.assertEqual(builder.call_args.kwargs["forgejo_token"], None)
self.assertEqual(builder.call_args.kwargs["auth_source"], "none")
self.assertIsNone(builder.call_args.kwargs["session_user"])
self.assertEqual(response_payload["auth"]["authenticated"], True)
self.assertEqual(response_payload["auth"]["login"], "kacper")
self.assertEqual(response_payload["auth"]["source"], "session")
2026-04-12 22:02:47 -04:00
def test_encrypted_session_cookie_survives_new_app_instance(self) -> None:
fake_client = _FakeForgejoClient(user={"login": "kacper"}, access_token="oauth-token")
with patch("app.ForgejoClient", return_value=fake_client):
start_response = self.client.get(
"/api/auth/forgejo/start",
follow_redirects=False,
)
state = parse_qs(urlparse(start_response.headers["location"]).query)["state"][0]
callback_response = self.client.get(
f"/api/auth/forgejo/callback?code=auth-code&state={state}",
follow_redirects=False,
)
fresh_client = TestClient(create_app())
fresh_client.cookies.set("robot_u_session", callback_response.cookies["robot_u_session"])
session_response = fresh_client.get("/api/auth/session")
self.assertEqual(session_response.status_code, 200)
self.assertEqual(session_response.json()["authenticated"], True)
self.assertEqual(session_response.json()["login"], "kacper")
self.assertEqual(session_response.json()["source"], "session")
def test_create_discussion_reply(self) -> None:
fake_client = _FakeForgejoClient(
comment={
"id": 123,
"body": "Thanks, this helped.",
"created_at": "2026-04-11T12:00:00Z",
"html_url": "https://aksal.cloud/Robot-U/RobotClass/issues/2#issuecomment-123",
"user": {
"login": "Kacper",
"avatar_url": "https://aksal.cloud/avatar.png",
},
},
)
with patch("app.ForgejoClient", return_value=fake_client) as client_factory:
response = self.client.post(
"/api/discussions/replies",
json={
"owner": "Robot-U",
"repo": "RobotClass",
"number": 2,
"body": "Thanks, this helped.",
},
headers={"Authorization": "token test-token"},
)
payload = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(client_factory.call_args.kwargs["forgejo_token"], "test-token")
self.assertEqual(
fake_client.created_comment, ("Robot-U", "RobotClass", 2, "Thanks, this helped.")
)
self.assertEqual(payload["id"], 123)
self.assertEqual(payload["author"], "Kacper")
self.assertEqual(payload["body"], "Thanks, this helped.")
2026-04-13 18:19:50 -04:00
def test_discussion_detail_fetches_public_issue(self) -> None:
fake_client = _FakeForgejoClient(
issue={
"id": 456,
"number": 9,
"title": "Encoder math question",
"body": "Canonical URL: http://testserver/posts/Robot-U/robot-u-site/building-robot-u-site",
"comments": 1,
"updated_at": "2026-04-11T12:00:00Z",
"html_url": "https://aksal.cloud/Robot-U/robot-u-site/issues/9",
"user": {"login": "Kacper", "avatar_url": ""},
2026-04-14 20:17:29 -04:00
"labels": [{"name": "discussion"}],
2026-04-13 18:19:50 -04:00
"state": "open",
},
comments=[
{
"id": 777,
"body": "Reply body",
"created_at": "2026-04-11T12:30:00Z",
"html_url": "https://aksal.cloud/Robot-U/robot-u-site/issues/9#issuecomment-777",
"user": {"login": "Ada", "avatar_url": ""},
}
],
)
with patch("app.ForgejoClient", return_value=fake_client):
response = self.client.get("/api/discussions/Robot-U/robot-u-site/9")
payload = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(payload["title"], "Encoder math question")
self.assertEqual(payload["comments"][0]["body"], "Reply body")
self.assertEqual(payload["links"][0]["kind"], "post")
2026-04-14 20:17:29 -04:00
def test_discussion_detail_rejects_issue_without_discussion_label(self) -> None:
fake_client = _FakeForgejoClient(
issue={
"id": 456,
"number": 9,
"title": "Encoder math question",
"body": "Regular issue",
"comments": 0,
"updated_at": "2026-04-11T12:00:00Z",
"html_url": "https://aksal.cloud/Robot-U/robot-u-site/issues/9",
"user": {"login": "Kacper", "avatar_url": ""},
"labels": [],
"state": "open",
},
)
with patch("app.ForgejoClient", return_value=fake_client):
response = self.client.get("/api/discussions/Robot-U/robot-u-site/9")
self.assertEqual(response.status_code, 404)
2026-04-13 18:19:50 -04:00
def test_create_discussion_reply_invalidates_prototype_cache(self) -> None:
initial_payload = {
"hero": {"title": "Before"},
"auth": {
"authenticated": False,
"login": None,
"source": "none",
"can_reply": False,
"oauth_configured": True,
},
"featured_courses": [],
"recent_posts": [],
"recent_discussions": [],
"upcoming_events": [],
"source_of_truth": [],
}
refreshed_payload = {
**initial_payload,
"hero": {"title": "After"},
}
builder = AsyncMock(side_effect=[initial_payload, refreshed_payload])
fake_client = _FakeForgejoClient(
comment={
"id": 123,
"body": "Thanks, this helped.",
"created_at": "2026-04-11T12:00:00Z",
"html_url": "https://aksal.cloud/Robot-U/RobotClass/issues/2#issuecomment-123",
"user": {"login": "Kacper", "avatar_url": ""},
},
)
with (
patch("prototype_cache.build_live_prototype_payload", new=builder),
patch("app.ForgejoClient", return_value=fake_client),
):
first_response = self.client.get("/api/prototype")
reply_response = self.client.post(
"/api/discussions/replies",
json={
"owner": "Robot-U",
"repo": "RobotClass",
"number": 2,
"body": "Thanks, this helped.",
},
headers={"Authorization": "token test-token"},
)
second_response = self.client.get("/api/prototype")
self.assertEqual(first_response.json()["hero"]["title"], "Before")
self.assertEqual(reply_response.status_code, 200)
self.assertEqual(second_response.json()["hero"]["title"], "After")
self.assertEqual(builder.await_count, 2)
def test_create_linked_discussion(self) -> None:
fake_client = _FakeForgejoClient(
issue={
"id": 456,
"number": 9,
"title": "Encoder math question",
"body": "How should I debounce this?\n\n---\nCanonical URL: http://testserver/posts/Robot-U/robot-u-site/building-robot-u-site",
"comments": 0,
"updated_at": "2026-04-11T12:00:00Z",
"html_url": "https://aksal.cloud/Robot-U/robot-u-site/issues/9",
"user": {"login": "Kacper", "avatar_url": ""},
"labels": [],
"state": "open",
},
)
with patch("app.ForgejoClient", return_value=fake_client) as client_factory:
response = self.client.post(
"/api/discussions",
json={
"owner": "Robot-U",
"repo": "robot-u-site",
"title": "Encoder math question",
"body": "How should I debounce this?",
"context_url": "http://testserver/posts/Robot-U/robot-u-site/building-robot-u-site",
"context_path": "blogs/building-robot-u-site/index.md",
"context_title": "Building Robot U",
},
headers={"Authorization": "token test-token"},
)
payload = response.json()
self.assertEqual(response.status_code, 200)
self.assertEqual(client_factory.call_args.kwargs["forgejo_token"], "test-token")
self.assertIsNotNone(fake_client.created_issue)
assert fake_client.created_issue is not None
self.assertEqual(
fake_client.created_issue[:3], ("Robot-U", "robot-u-site", "Encoder math question")
)
self.assertIn(
"Canonical URL: http://testserver/posts/Robot-U/robot-u-site/building-robot-u-site",
fake_client.created_issue[3],
)
self.assertEqual(payload["id"], 456)
self.assertEqual(payload["repo"], "Robot-U/robot-u-site")
2026-04-14 20:17:29 -04:00
self.assertEqual(payload["labels"], ["discussion"])
2026-04-13 18:19:50 -04:00
self.assertEqual(payload["links"][0]["kind"], "post")
def test_create_general_discussion_uses_configured_repo(self) -> None:
get_settings.cache_clear()
fake_client = _FakeForgejoClient(
issue={
"id": 457,
"number": 10,
"title": "General project help",
"body": "I need help choosing motors.",
"comments": 0,
"updated_at": "2026-04-11T12:00:00Z",
"html_url": "https://aksal.cloud/Robot-U/community/issues/10",
"user": {"login": "Kacper", "avatar_url": ""},
"labels": [],
"state": "open",
},
)
with (
patch.dict(os.environ, {"FORGEJO_GENERAL_DISCUSSION_REPO": "Robot-U/community"}),
patch("app.ForgejoClient", return_value=fake_client),
):
response = self.client.post(
"/api/discussions",
json={
"title": "General project help",
"body": "I need help choosing motors.",
},
headers={"Authorization": "token test-token"},
)
self.assertEqual(response.status_code, 200)
self.assertEqual(
fake_client.created_issue,
("Robot-U", "community", "General project help", "I need help choosing motors."),
)
2026-04-14 20:17:29 -04:00
self.assertEqual(fake_client.created_issue_labels, [123])
2026-04-13 18:19:50 -04:00
def test_create_discussion_rejects_server_token_fallback(self) -> None:
get_settings.cache_clear()
with patch.dict(os.environ, {"FORGEJO_TOKEN": "server-token"}):
response = self.client.post(
"/api/discussions",
json={
"owner": "Robot-U",
"repo": "robot-u-site",
"title": "General project help",
"body": "I need help choosing motors.",
},
)
self.assertEqual(response.status_code, 401)
def test_create_discussion_reply_uses_signed_in_identity(self) -> None:
sign_in_client = _FakeForgejoClient(user={"login": "kacper"}, access_token="oauth-token")
with patch("app.ForgejoClient", return_value=sign_in_client):
start_response = self.client.get(
"/api/auth/forgejo/start",
follow_redirects=False,
)
state = parse_qs(urlparse(start_response.headers["location"]).query)["state"][0]
self.client.get(
f"/api/auth/forgejo/callback?code=auth-code&state={state}",
follow_redirects=False,
)
reply_client = _FakeForgejoClient(
comment={
"id": 123,
"body": "Thanks, this helped.",
"created_at": "2026-04-11T12:00:00Z",
"html_url": "https://aksal.cloud/Robot-U/RobotClass/issues/2#issuecomment-123",
"user": {"login": "Kacper", "avatar_url": ""},
},
)
with patch("app.ForgejoClient", return_value=reply_client) as client_factory:
response = self.client.post(
"/api/discussions/replies",
json={
"owner": "Robot-U",
"repo": "RobotClass",
"number": 2,
"body": "Thanks, this helped.",
},
)
self.assertEqual(response.status_code, 200)
self.assertEqual(client_factory.call_args.kwargs["forgejo_token"], "oauth-token")
2026-04-14 20:17:29 -04:00
self.assertEqual(client_factory.call_args.kwargs["auth_scheme"], "Bearer")
self.assertEqual(
reply_client.created_comment, ("Robot-U", "RobotClass", 2, "Thanks, this helped.")
)
def test_create_discussion_reply_rejects_private_repo(self) -> None:
fake_client = _FakeForgejoClient(repo_private=True)
with patch("app.ForgejoClient", return_value=fake_client):
response = self.client.post(
"/api/discussions/replies",
json={
"owner": "Robot-U",
"repo": "PrivateRobotClass",
"number": 2,
"body": "Thanks, this helped.",
},
headers={"Authorization": "token test-token"},
)
self.assertEqual(response.status_code, 403)
self.assertIsNone(fake_client.created_comment)
def test_create_discussion_reply_rejects_server_token_fallback(self) -> None:
get_settings.cache_clear()
with patch.dict(os.environ, {"FORGEJO_TOKEN": "server-token"}):
response = self.client.post(
"/api/discussions/replies",
json={
"owner": "Robot-U",
"repo": "RobotClass",
"number": 2,
"body": "Thanks, this helped.",
},
)
self.assertEqual(response.status_code, 401)
class _FakeForgejoClient:
def __init__(
self,
comment: dict[str, object] | None = None,
2026-04-13 18:19:50 -04:00
comments: list[dict[str, object]] | None = None,
issue: dict[str, object] | None = None,
user: dict[str, object] | None = None,
access_token: str = "test-oauth-token",
repo_private: bool = False,
) -> None:
self._comment = comment
2026-04-13 18:19:50 -04:00
self._comments = comments or []
self._issue = issue
self._user = user or {"login": "test-user"}
self._access_token = access_token
self._repo_private = repo_private
self.created_comment: tuple[str, str, int, str] | None = None
2026-04-13 18:19:50 -04:00
self.created_issue: tuple[str, str, str, str] | None = None
2026-04-14 20:17:29 -04:00
self.created_issue_labels: list[int] | None = None
self.exchanged_code: str | None = None
async def __aenter__(self) -> _FakeForgejoClient:
return self
async def __aexit__(self, _exc_type: object, _exc: object, _tb: object) -> None:
return None
async def create_issue_comment(
self,
owner: str,
repo: str,
issue_number: int,
body: str,
) -> dict[str, object]:
self.created_comment = (owner, repo, issue_number, body)
if self._comment is None:
raise AssertionError("Fake comment was not configured.")
return self._comment
2026-04-13 18:19:50 -04:00
async def create_issue(
self,
owner: str,
repo: str,
title: str,
body: str,
2026-04-14 20:17:29 -04:00
label_ids: list[int] | None = None,
2026-04-13 18:19:50 -04:00
) -> dict[str, object]:
self.created_issue = (owner, repo, title, body)
2026-04-14 20:17:29 -04:00
self.created_issue_labels = label_ids
2026-04-13 18:19:50 -04:00
if self._issue is None:
raise AssertionError("Fake issue was not configured.")
return self._issue
2026-04-14 20:17:29 -04:00
async def ensure_repo_label(
self,
_owner: str,
_repo: str,
_name: str,
*,
color: str,
description: str,
) -> int:
assert color
assert description
return 123
2026-04-13 18:19:50 -04:00
async def fetch_issue(self, _owner: str, _repo: str, _issue_number: int) -> dict[str, object]:
if self._issue is None:
raise AssertionError("Fake issue was not configured.")
return self._issue
async def fetch_current_user(self) -> dict[str, object]:
return self._user
async def fetch_repository(self, owner: str, repo: str) -> dict[str, object]:
return {
"owner": {"login": owner},
"name": repo,
"full_name": f"{owner}/{repo}",
"private": self._repo_private,
}
2026-04-13 18:19:50 -04:00
async def list_issue_comments(
self,
_owner: str,
_repo: str,
_issue_number: int,
) -> list[dict[str, object]]:
return self._comments
async def fetch_openid_configuration(self) -> dict[str, object]:
return {
"authorization_endpoint": "https://aksal.cloud/login/oauth/authorize",
"token_endpoint": "https://aksal.cloud/login/oauth/access_token",
"userinfo_endpoint": "https://aksal.cloud/login/oauth/userinfo",
}
async def exchange_oauth_code(
self,
*,
token_endpoint: str,
client_id: str,
client_secret: str,
code: str,
redirect_uri: str,
code_verifier: str,
) -> dict[str, object]:
self.exchanged_code = code
self.exchanged_token_request = {
"token_endpoint": token_endpoint,
"client_id": client_id,
"client_secret": client_secret,
"redirect_uri": redirect_uri,
"code_verifier": code_verifier,
}
return {"access_token": self._access_token}
async def fetch_userinfo(self, userinfo_endpoint: str, access_token: str) -> dict[str, object]:
self.fetched_userinfo = {
"userinfo_endpoint": userinfo_endpoint,
"access_token": access_token,
}
return {
"preferred_username": self._user["login"],
"picture": self._user.get("avatar_url", ""),
"email": self._user.get("email", ""),
}
2026-04-08 06:03:48 -04:00
if __name__ == "__main__":
unittest.main()