feat: restructure test infrastructure with unit/integration/system separation
Test Organization: - Create tests/unit/, tests/integration/, tests/system/ directories - Move existing tests into appropriate categories - Add pytest markers (@pytest.mark.unit, @pytest.mark.integration) Shared Fixtures: - Create conftest.py with SQLite engine (for unit tests) - Add PostgreSQL session fixture with transaction rollback - Add TestClient fixture for API tests Configuration: - Update pyproject.toml with asyncio_mode=auto - Add test markers and default addopts - Add aiosqlite dependency for SQLite support E2E Testing: - Initialize Playwright in e2e/ directory - Add playwright.config.ts - Create login flow E2E test Build: - Add test-unit, test-integration, test-system, test-e2e to Makefile - Update test target to run all categories - Add testing documentation to README Note: Some tests have import issues due to missing python-jose package in dev environment. This needs to be addressed separately.
This commit is contained in:
Fusion
@@ -0,0 +1,234 @@
|
||||
import uuid
|
||||
from datetime import UTC, datetime, timedelta
|
||||
import asyncio
|
||||
import importlib
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
import pytest
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.ext.asyncio import create_async_engine
|
||||
|
||||
from src.auth.jwt_service import mint_access_token
|
||||
from src.config import Settings, build_database_url
|
||||
from src.models import Base
|
||||
from src.models.user import User
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def configure_local_database(monkeypatch) -> None:
|
||||
local_url = build_database_url(
|
||||
user="headquarter",
|
||||
password="headquarter",
|
||||
host="localhost",
|
||||
port=5432,
|
||||
database="headquarter",
|
||||
)
|
||||
monkeypatch.setenv("DATABASE_URL", local_url)
|
||||
|
||||
|
||||
def _prepare_auth_test_db() -> None:
|
||||
async def _run() -> None:
|
||||
engine = create_async_engine(
|
||||
build_database_url(
|
||||
user="headquarter",
|
||||
password="headquarter",
|
||||
host="localhost",
|
||||
port=5432,
|
||||
database="headquarter",
|
||||
)
|
||||
)
|
||||
async with engine.begin() as connection:
|
||||
await connection.run_sync(Base.metadata.create_all)
|
||||
await connection.execute(text("TRUNCATE TABLE refresh_tokens, users RESTART IDENTITY CASCADE"))
|
||||
await engine.dispose()
|
||||
|
||||
asyncio.run(_run())
|
||||
|
||||
|
||||
def _load_app():
|
||||
import src.database as database_module
|
||||
import src.api.auth as auth_module
|
||||
import src.main as main_module
|
||||
|
||||
importlib.reload(database_module)
|
||||
importlib.reload(auth_module)
|
||||
importlib.reload(main_module)
|
||||
return main_module.app
|
||||
|
||||
|
||||
def _insert_user_for_refresh(user_id: str) -> None:
|
||||
async def _run() -> None:
|
||||
engine = create_async_engine(
|
||||
build_database_url(
|
||||
user="headquarter",
|
||||
password="headquarter",
|
||||
host="localhost",
|
||||
port=5432,
|
||||
database="headquarter",
|
||||
)
|
||||
)
|
||||
async with engine.begin() as connection:
|
||||
await connection.run_sync(Base.metadata.create_all)
|
||||
from sqlalchemy.ext.asyncio import async_sessionmaker
|
||||
|
||||
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
||||
async with session_factory() as session:
|
||||
user = User(
|
||||
id=uuid.UUID(user_id),
|
||||
email="refresh@headquarter.local",
|
||||
name="Refresh User",
|
||||
authentik_id="refresh-user",
|
||||
avatar_url=None,
|
||||
)
|
||||
await session.merge(user)
|
||||
await session.commit()
|
||||
await engine.dispose()
|
||||
|
||||
asyncio.run(_run())
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_login_redirects_to_authentik_authorize_endpoint() -> None:
|
||||
_prepare_auth_test_db()
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
response = client.get("/auth/login", follow_redirects=False)
|
||||
|
||||
assert response.status_code == 307
|
||||
assert "response_type=code" in response.headers["location"]
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_me_returns_401_without_access_cookie() -> None:
|
||||
_prepare_auth_test_db()
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
response = client.get("/auth/me")
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_me_returns_user_payload_with_valid_access_cookie() -> None:
|
||||
_prepare_auth_test_db()
|
||||
app = _load_app()
|
||||
|
||||
settings = Settings()
|
||||
token = mint_access_token(
|
||||
settings=settings,
|
||||
subject=str(uuid.uuid4()),
|
||||
email="dev@headquarter.local",
|
||||
name="Dev User",
|
||||
expires_at=datetime.now(UTC) + timedelta(minutes=15),
|
||||
)
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", token)
|
||||
response = client.get("/auth/me")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json()["email"] == "dev@headquarter.local"
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_logout_clears_auth_cookies() -> None:
|
||||
_prepare_auth_test_db()
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("refresh_token", "opaque-token")
|
||||
response = client.post("/auth/logout")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert "access_token=" in response.headers.get("set-cookie", "")
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_callback_rejects_mismatched_state() -> None:
|
||||
_prepare_auth_test_db()
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("auth_state", "expected")
|
||||
response = client.get("/auth/callback?code=test-code&state=wrong")
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_callback_sets_auth_cookies_after_success(monkeypatch) -> None:
|
||||
_prepare_auth_test_db()
|
||||
app = _load_app()
|
||||
|
||||
async def fake_exchange_code_for_tokens(*, settings, code, redirect_uri, client):
|
||||
return {"access_token": "provider-access", "refresh_token": "provider-refresh"}
|
||||
|
||||
def fake_verify_provider_access_token(*, settings, token, jwks):
|
||||
return {"sub": "auth-sub-1", "email": "callback@headquarter.local", "name": "Callback User"}
|
||||
|
||||
async def fake_fetch_jwks(*, settings, client):
|
||||
return {"keys": []}
|
||||
|
||||
monkeypatch.setattr("src.api.auth.exchange_code_for_tokens", fake_exchange_code_for_tokens)
|
||||
monkeypatch.setattr("src.api.auth.verify_provider_access_token", fake_verify_provider_access_token)
|
||||
monkeypatch.setattr("src.api.auth.fetch_jwks", fake_fetch_jwks)
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("auth_state", "good-state")
|
||||
response = client.get("/auth/callback?code=valid-code&state=good-state")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json()["email"] == "callback@headquarter.local"
|
||||
set_cookie_header = response.headers.get("set-cookie", "")
|
||||
assert "access_token=" in set_cookie_header
|
||||
assert "refresh_token=" in set_cookie_header
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_refresh_rotates_cookie_and_returns_user_payload(monkeypatch) -> None:
|
||||
_prepare_auth_test_db()
|
||||
_insert_user_for_refresh("7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb")
|
||||
app = _load_app()
|
||||
|
||||
async def fake_rotate_refresh_token(*, session, raw_token, user_agent, ip_address):
|
||||
class StoredToken:
|
||||
user_id = uuid.UUID("7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb")
|
||||
|
||||
return "new-refresh-token", StoredToken()
|
||||
|
||||
monkeypatch.setattr("src.api.auth.rotate_refresh_token", fake_rotate_refresh_token)
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("refresh_token", "old-refresh-token")
|
||||
response = client.post("/auth/refresh")
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json()["sub"] == "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
|
||||
assert "refresh_token=" in response.headers.get("set-cookie", "")
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_refresh_returns_401_for_invalid_refresh_token(monkeypatch) -> None:
|
||||
_prepare_auth_test_db()
|
||||
app = _load_app()
|
||||
|
||||
async def fake_rotate_refresh_token(*, session, raw_token, user_agent, ip_address):
|
||||
raise ValueError("refresh token not found")
|
||||
|
||||
monkeypatch.setattr("src.api.auth.rotate_refresh_token", fake_rotate_refresh_token)
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("refresh_token", "invalid")
|
||||
response = client.post("/auth/refresh")
|
||||
|
||||
assert response.status_code == 401
|
||||
@@ -0,0 +1,227 @@
|
||||
from datetime import UTC, datetime, timedelta
|
||||
import base64
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
|
||||
from src.auth.cookies import build_cookie_options
|
||||
from src.auth.jwt_service import decode_access_token, mint_access_token
|
||||
from src.auth.oidc import build_login_redirect_url, exchange_code_for_tokens, verify_provider_access_token
|
||||
from src.auth.refresh_store import create_refresh_token, hash_refresh_token, revoke_refresh_token, rotate_refresh_token
|
||||
from src.config import Settings, build_database_url
|
||||
from src.models import Base
|
||||
from src.models.user import User
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_cookie_options_follow_environment_defaults(monkeypatch) -> None:
|
||||
monkeypatch.setenv("APP_ENV", "development")
|
||||
dev_settings = Settings()
|
||||
dev_options = build_cookie_options(dev_settings)
|
||||
|
||||
monkeypatch.setenv("APP_ENV", "production")
|
||||
prod_settings = Settings()
|
||||
prod_options = build_cookie_options(prod_settings)
|
||||
|
||||
assert dev_options["httponly"] is True
|
||||
assert dev_options["secure"] is False
|
||||
assert dev_options["samesite"] == "lax"
|
||||
assert prod_options["secure"] is True
|
||||
assert prod_options["samesite"] == "strict"
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_login_redirect_url_contains_required_oidc_params() -> None:
|
||||
settings = Settings()
|
||||
|
||||
url = build_login_redirect_url(
|
||||
settings=settings,
|
||||
redirect_uri="http://localhost:8000/auth/callback",
|
||||
state="state-123",
|
||||
nonce="nonce-123",
|
||||
)
|
||||
|
||||
assert "response_type=code" in url
|
||||
assert "client_id=headquarter-web" in url
|
||||
assert "scope=openid+profile+email" in url
|
||||
assert "state=state-123" in url
|
||||
assert "nonce=nonce-123" in url
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_mint_and_decode_internal_access_token_round_trip() -> None:
|
||||
settings = Settings()
|
||||
expires_at = datetime.now(UTC) + timedelta(minutes=15)
|
||||
|
||||
token = mint_access_token(
|
||||
settings=settings,
|
||||
subject="user-123",
|
||||
email="dev@headquarter.local",
|
||||
name="Dev User",
|
||||
expires_at=expires_at,
|
||||
)
|
||||
|
||||
claims = decode_access_token(settings=settings, token=token)
|
||||
|
||||
assert claims["sub"] == "user-123"
|
||||
assert claims["email"] == "dev@headquarter.local"
|
||||
assert claims["name"] == "Dev User"
|
||||
assert "exp" in claims
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_refresh_token_hash_is_deterministic_and_non_reversible() -> None:
|
||||
raw_token = "refresh-token-abc"
|
||||
|
||||
first_hash = hash_refresh_token(raw_token)
|
||||
second_hash = hash_refresh_token(raw_token)
|
||||
|
||||
assert first_hash == second_hash
|
||||
assert first_hash != raw_token
|
||||
assert len(first_hash) == 64
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_decode_access_token_rejects_invalid_signature() -> None:
|
||||
settings = Settings()
|
||||
other_settings = Settings(jwt_secret="different-secret")
|
||||
expires_at = datetime.now(UTC) + timedelta(minutes=15)
|
||||
|
||||
token = mint_access_token(
|
||||
settings=other_settings,
|
||||
subject="user-123",
|
||||
email="dev@headquarter.local",
|
||||
name="Dev User",
|
||||
expires_at=expires_at,
|
||||
)
|
||||
|
||||
with pytest.raises(Exception):
|
||||
decode_access_token(settings=settings, token=token)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
n@pytest.mark.integration
|
||||
|
||||
async def test_exchange_code_for_tokens_posts_expected_payload() -> None:
|
||||
settings = Settings()
|
||||
|
||||
def handler(request: httpx.Request) -> httpx.Response:
|
||||
assert request.url == httpx.URL(settings.resolved_authentik_token_url)
|
||||
payload = dict(httpx.QueryParams(request.content.decode("utf-8")))
|
||||
assert payload["grant_type"] == "authorization_code"
|
||||
assert payload["code"] == "auth-code"
|
||||
assert payload["redirect_uri"] == "http://localhost:8000/auth/callback"
|
||||
return httpx.Response(200, json={"access_token": "provider-token", "refresh_token": "provider-refresh"})
|
||||
|
||||
transport = httpx.MockTransport(handler)
|
||||
async with httpx.AsyncClient(transport=transport) as client:
|
||||
token_payload = await exchange_code_for_tokens(
|
||||
settings=settings,
|
||||
code="auth-code",
|
||||
redirect_uri="http://localhost:8000/auth/callback",
|
||||
client=client,
|
||||
)
|
||||
|
||||
assert token_payload["access_token"] == "provider-token"
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_verify_provider_access_token_with_jwks_oct_key() -> None:
|
||||
settings = Settings(authentik_audience="headquarter-web", authentik_issuer="https://authentik.local/")
|
||||
shared_secret = b"shared-secret-123"
|
||||
jwks = {
|
||||
"keys": [
|
||||
{
|
||||
"kty": "oct",
|
||||
"alg": "HS256",
|
||||
"k": base64.urlsafe_b64encode(shared_secret).decode("utf-8").rstrip("="),
|
||||
"kid": "kid-1",
|
||||
}
|
||||
]
|
||||
}
|
||||
|
||||
from jose import jwt # type: ignore[import-untyped]
|
||||
|
||||
token = jwt.encode(
|
||||
{
|
||||
"sub": "authentik-user",
|
||||
"iss": settings.authentik_issuer,
|
||||
"aud": settings.authentik_audience,
|
||||
"exp": int((datetime.now(UTC) + timedelta(minutes=5)).timestamp()),
|
||||
},
|
||||
shared_secret,
|
||||
algorithm="HS256",
|
||||
headers={"kid": "kid-1"},
|
||||
)
|
||||
|
||||
claims = verify_provider_access_token(settings=settings, token=token, jwks=jwks)
|
||||
|
||||
assert claims["sub"] == "authentik-user"
|
||||
|
||||
|
||||
TEST_DATABASE_URL = build_database_url(
|
||||
user="headquarter",
|
||||
password="headquarter",
|
||||
host="localhost",
|
||||
port=5432,
|
||||
database="headquarter",
|
||||
)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def db_session() -> AsyncSession:
|
||||
engine = create_async_engine(TEST_DATABASE_URL)
|
||||
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
||||
|
||||
async with engine.begin() as connection:
|
||||
await connection.run_sync(Base.metadata.create_all)
|
||||
|
||||
async with session_factory() as session:
|
||||
await session.execute(text("TRUNCATE TABLE refresh_tokens, users RESTART IDENTITY CASCADE"))
|
||||
await session.commit()
|
||||
yield session
|
||||
await session.rollback()
|
||||
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
n@pytest.mark.integration
|
||||
|
||||
async def test_refresh_store_create_rotate_and_revoke(db_session: AsyncSession) -> None:
|
||||
user = User(email="dev-auth@headquarter.local", name="Dev Auth", authentik_id="auth-dev", avatar_url=None)
|
||||
db_session.add(user)
|
||||
await db_session.commit()
|
||||
await db_session.refresh(user)
|
||||
|
||||
raw_refresh_token, stored_token = await create_refresh_token(
|
||||
session=db_session,
|
||||
user_id=user.id,
|
||||
expires_at=datetime.now(UTC) + timedelta(days=7),
|
||||
user_agent="pytest",
|
||||
ip_address="127.0.0.1",
|
||||
)
|
||||
assert raw_refresh_token
|
||||
assert stored_token.revoked_at is None
|
||||
|
||||
rotated_raw, rotated_stored = await rotate_refresh_token(
|
||||
session=db_session,
|
||||
raw_token=raw_refresh_token,
|
||||
user_agent="pytest-rotated",
|
||||
ip_address="127.0.0.2",
|
||||
)
|
||||
assert rotated_raw != raw_refresh_token
|
||||
assert rotated_stored.revoked_at is None
|
||||
assert stored_token.revoked_at is not None
|
||||
|
||||
revoked = await revoke_refresh_token(session=db_session, raw_token=rotated_raw)
|
||||
assert revoked is True
|
||||
@@ -0,0 +1,169 @@
|
||||
from collections.abc import AsyncIterator
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
|
||||
from src.config import build_database_url
|
||||
from src.models import Base
|
||||
from src.models.base import TimestampMixin, UUIDPrimaryKeyMixin
|
||||
from src.models.git_repository import GitRepository
|
||||
from src.models.project import Project
|
||||
from src.models.refresh_token import RefreshToken
|
||||
from src.models.ssh_key import SSHKey
|
||||
from src.models.user import User
|
||||
from src.models.user_config import UserConfig
|
||||
|
||||
|
||||
TEST_DATABASE_URL = build_database_url(
|
||||
user="headquarter",
|
||||
password="headquarter",
|
||||
host="localhost",
|
||||
port=5432,
|
||||
database="headquarter",
|
||||
)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def db_session() -> AsyncIterator[AsyncSession]:
|
||||
engine = create_async_engine(TEST_DATABASE_URL)
|
||||
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
||||
|
||||
async with session_factory() as session:
|
||||
table_rows = await session.execute(
|
||||
text(
|
||||
"SELECT tablename FROM pg_tables "
|
||||
"WHERE schemaname = 'public' "
|
||||
"AND tablename = ANY(:table_names)"
|
||||
),
|
||||
{
|
||||
"table_names": [
|
||||
"refresh_tokens",
|
||||
"user_configs",
|
||||
"git_repositories",
|
||||
"projects",
|
||||
"ssh_keys",
|
||||
"users",
|
||||
]
|
||||
},
|
||||
)
|
||||
existing_tables = [row[0] for row in table_rows]
|
||||
if existing_tables:
|
||||
await session.execute(text(f"TRUNCATE TABLE {', '.join(existing_tables)} RESTART IDENTITY CASCADE"))
|
||||
await session.commit()
|
||||
yield session
|
||||
await session.rollback()
|
||||
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_base_metadata_collects_declared_tables() -> None:
|
||||
assert isinstance(Base.metadata.tables, dict)
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_shared_mixins_define_expected_columns() -> None:
|
||||
assert "id" in UUIDPrimaryKeyMixin.__dict__
|
||||
assert "created_at" in TimestampMixin.__dict__
|
||||
assert "updated_at" in TimestampMixin.__dict__
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_expected_tables_are_registered() -> None:
|
||||
assert set(Base.metadata.tables) == {
|
||||
"refresh_tokens",
|
||||
"git_repositories",
|
||||
"projects",
|
||||
"ssh_keys",
|
||||
"user_configs",
|
||||
"users",
|
||||
}
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_user_table_has_required_columns() -> None:
|
||||
columns = User.__table__.columns
|
||||
|
||||
assert set(columns.keys()) == {
|
||||
"id",
|
||||
"email",
|
||||
"name",
|
||||
"authentik_id",
|
||||
"avatar_url",
|
||||
"created_at",
|
||||
"updated_at",
|
||||
}
|
||||
assert columns["email"].unique is True
|
||||
assert columns["authentik_id"].unique is True
|
||||
assert columns["avatar_url"].nullable is True
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_project_relationships_point_to_owner_and_default_ssh_key() -> None:
|
||||
owner_fk = next(iter(Project.__table__.c.owner_id.foreign_keys))
|
||||
ssh_fk = next(iter(Project.__table__.c.default_ssh_key_id.foreign_keys))
|
||||
|
||||
assert owner_fk.target_fullname == "users.id"
|
||||
assert ssh_fk.target_fullname == "ssh_keys.id"
|
||||
assert Project.owner.property.mapper.class_ is User
|
||||
assert Project.default_ssh_key.property.mapper.class_ is SSHKey
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_repository_and_user_config_relationships_are_registered() -> None:
|
||||
project_fk = next(iter(GitRepository.__table__.c.project_id.foreign_keys))
|
||||
owner_fk = next(iter(GitRepository.__table__.c.owner_id.foreign_keys))
|
||||
user_config_fk = next(iter(UserConfig.__table__.c.user_id.foreign_keys))
|
||||
|
||||
assert project_fk.target_fullname == "projects.id"
|
||||
assert owner_fk.target_fullname == "users.id"
|
||||
assert user_config_fk.target_fullname == "users.id"
|
||||
assert GitRepository.project.property.mapper.class_ is Project
|
||||
assert GitRepository.owner.property.mapper.class_ is User
|
||||
assert UserConfig.user.property.mapper.class_ is User
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_refresh_token_table_has_required_columns_and_relationships() -> None:
|
||||
columns = RefreshToken.__table__.columns
|
||||
user_fk = next(iter(RefreshToken.__table__.c.user_id.foreign_keys))
|
||||
|
||||
assert set(columns.keys()) == {
|
||||
"id",
|
||||
"user_id",
|
||||
"token_hash",
|
||||
"expires_at",
|
||||
"revoked_at",
|
||||
"user_agent",
|
||||
"ip_address",
|
||||
"created_at",
|
||||
}
|
||||
assert columns["token_hash"].unique is True
|
||||
assert columns["revoked_at"].nullable is True
|
||||
assert user_fk.target_fullname == "users.id"
|
||||
assert RefreshToken.user.property.mapper.class_ is User
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
n@pytest.mark.integration
|
||||
|
||||
async def test_async_session_can_insert_and_load_user(db_session: AsyncSession) -> None:
|
||||
user = User(email="dev@headquarter.local", name="Dev User", authentik_id="dev-user", avatar_url=None)
|
||||
|
||||
db_session.add(user)
|
||||
await db_session.commit()
|
||||
await db_session.refresh(user)
|
||||
|
||||
loaded_user = await db_session.get(User, user.id)
|
||||
|
||||
assert loaded_user is not None
|
||||
assert loaded_user.email == "dev@headquarter.local"
|
||||
@@ -0,0 +1,286 @@
|
||||
import uuid
|
||||
from datetime import UTC, datetime, timedelta
|
||||
import asyncio
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
import pytest
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
|
||||
|
||||
from src.auth.jwt_service import mint_access_token
|
||||
from src.config import Settings, build_database_url
|
||||
from src.models import Base
|
||||
from src.models.project import Project
|
||||
from src.models.user import User
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def configure_local_database(monkeypatch) -> None:
|
||||
local_url = build_database_url(
|
||||
user="headquarter",
|
||||
password="headquarter",
|
||||
host="localhost",
|
||||
port=5432,
|
||||
database="headquarter",
|
||||
)
|
||||
monkeypatch.setenv("DATABASE_URL", local_url)
|
||||
|
||||
|
||||
def _prepare_test_db() -> None:
|
||||
async def _run() -> None:
|
||||
engine = create_async_engine(
|
||||
build_database_url(
|
||||
user="headquarter",
|
||||
password="headquarter",
|
||||
host="localhost",
|
||||
port=5432,
|
||||
database="headquarter",
|
||||
)
|
||||
)
|
||||
async with engine.begin() as connection:
|
||||
await connection.run_sync(Base.metadata.create_all)
|
||||
await connection.execute(text("TRUNCATE TABLE git_repositories, ssh_keys, projects, users RESTART IDENTITY CASCADE"))
|
||||
await engine.dispose()
|
||||
|
||||
asyncio.run(_run())
|
||||
|
||||
|
||||
def _load_app():
|
||||
import importlib
|
||||
import src.database as database_module
|
||||
import src.api.auth as auth_module
|
||||
import src.api.projects as projects_module
|
||||
import src.main as main_module
|
||||
|
||||
# Dispose old engine connections before reload to prevent pool exhaustion
|
||||
if hasattr(database_module, 'engine'):
|
||||
import asyncio
|
||||
asyncio.run(database_module.engine.dispose())
|
||||
|
||||
importlib.reload(database_module)
|
||||
importlib.reload(auth_module)
|
||||
importlib.reload(projects_module)
|
||||
importlib.reload(main_module)
|
||||
return main_module.app
|
||||
|
||||
|
||||
def _mint_token(user_id: str) -> str:
|
||||
settings = Settings()
|
||||
return mint_access_token(
|
||||
settings=settings,
|
||||
subject=user_id,
|
||||
email="test@headquarter.local",
|
||||
name="Test User",
|
||||
expires_at=datetime.now(UTC) + timedelta(minutes=15),
|
||||
)
|
||||
|
||||
|
||||
def _insert_user(user_id: str, email: str = "test@headquarter.local") -> None:
|
||||
async def _run() -> None:
|
||||
engine = create_async_engine(
|
||||
build_database_url(
|
||||
user="headquarter",
|
||||
password="headquarter",
|
||||
host="localhost",
|
||||
port=5432,
|
||||
database="headquarter",
|
||||
)
|
||||
)
|
||||
async with engine.begin() as connection:
|
||||
await connection.run_sync(Base.metadata.create_all)
|
||||
|
||||
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
||||
async with session_factory() as session:
|
||||
user = User(
|
||||
id=uuid.UUID(user_id),
|
||||
email=email,
|
||||
name="Test User",
|
||||
authentik_id=f"authentik-{user_id}",
|
||||
avatar_url=None,
|
||||
)
|
||||
await session.merge(user)
|
||||
await session.commit()
|
||||
await engine.dispose()
|
||||
|
||||
asyncio.run(_run())
|
||||
|
||||
|
||||
def _insert_project(project_id: str, owner_id: str, name: str = "Test Project") -> None:
|
||||
async def _run() -> None:
|
||||
engine = create_async_engine(
|
||||
build_database_url(
|
||||
user="headquarter",
|
||||
password="headquarter",
|
||||
host="localhost",
|
||||
port=5432,
|
||||
database="headquarter",
|
||||
)
|
||||
)
|
||||
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
||||
async with session_factory() as session:
|
||||
project = Project(
|
||||
id=uuid.UUID(project_id),
|
||||
name=name,
|
||||
description="A test project",
|
||||
owner_id=uuid.UUID(owner_id),
|
||||
default_ssh_key_id=None,
|
||||
)
|
||||
await session.merge(project)
|
||||
await session.commit()
|
||||
await engine.dispose()
|
||||
|
||||
asyncio.run(_run())
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_create_project_requires_authentication() -> None:
|
||||
_prepare_test_db()
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
|
||||
response = client.post("/projects", json={"name": "New Project", "description": "Description"})
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_create_project_successfully() -> None:
|
||||
_prepare_test_db()
|
||||
user_id = "11111111-1111-1111-1111-111111111111"
|
||||
_insert_user(user_id)
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(user_id))
|
||||
|
||||
response = client.post("/projects", json={"name": "New Project", "description": "Description"})
|
||||
|
||||
assert response.status_code == 201
|
||||
data = response.json()
|
||||
assert data["name"] == "New Project"
|
||||
assert data["description"] == "Description"
|
||||
assert data["owner_id"] == user_id
|
||||
assert "id" in data
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_list_projects_returns_only_owned_projects() -> None:
|
||||
_prepare_test_db()
|
||||
user1_id = "11111111-1111-1111-1111-111111111111"
|
||||
user2_id = "22222222-2222-2222-2222-222222222222"
|
||||
_insert_user(user1_id)
|
||||
_insert_user(user2_id, "other@headquarter.local")
|
||||
_insert_project("aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa", user1_id, "User1 Project")
|
||||
_insert_project("bbbbbbbb-bbbb-bbbb-bbbb-bbbbbbbbbbbb", user2_id, "User2 Project")
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(user1_id))
|
||||
|
||||
response = client.get("/projects")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert len(data) == 1
|
||||
assert data[0]["name"] == "User1 Project"
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_update_project_requires_ownership() -> None:
|
||||
_prepare_test_db()
|
||||
owner_id = "11111111-1111-1111-1111-111111111111"
|
||||
other_id = "22222222-2222-2222-2222-222222222222"
|
||||
project_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
|
||||
_insert_user(owner_id)
|
||||
_insert_user(other_id, "other@headquarter.local")
|
||||
_insert_project(project_id, owner_id)
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(other_id))
|
||||
|
||||
response = client.patch(f"/projects/{project_id}", json={"name": "Hacked"})
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_update_project_successfully() -> None:
|
||||
_prepare_test_db()
|
||||
owner_id = "11111111-1111-1111-1111-111111111111"
|
||||
project_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
|
||||
_insert_user(owner_id)
|
||||
_insert_project(project_id, owner_id)
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(owner_id))
|
||||
|
||||
response = client.patch(f"/projects/{project_id}", json={"name": "Updated Name"})
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["name"] == "Updated Name"
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_delete_project_requires_ownership() -> None:
|
||||
_prepare_test_db()
|
||||
owner_id = "11111111-1111-1111-1111-111111111111"
|
||||
other_id = "22222222-2222-2222-2222-222222222222"
|
||||
project_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
|
||||
_insert_user(owner_id)
|
||||
_insert_user(other_id, "other@headquarter.local")
|
||||
_insert_project(project_id, owner_id)
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(other_id))
|
||||
|
||||
response = client.delete(f"/projects/{project_id}")
|
||||
|
||||
assert response.status_code == 403
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_delete_project_successfully() -> None:
|
||||
_prepare_test_db()
|
||||
owner_id = "11111111-1111-1111-1111-111111111111"
|
||||
project_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
|
||||
_insert_user(owner_id)
|
||||
_insert_project(project_id, owner_id)
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(owner_id))
|
||||
|
||||
response = client.delete(f"/projects/{project_id}")
|
||||
|
||||
assert response.status_code == 204
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_set_default_ssh_key_requires_ownership() -> None:
|
||||
_prepare_test_db()
|
||||
owner_id = "11111111-1111-1111-1111-111111111111"
|
||||
other_id = "22222222-2222-2222-2222-222222222222"
|
||||
project_id = "aaaaaaaa-aaaa-aaaa-aaaa-aaaaaaaaaaaa"
|
||||
_insert_user(owner_id)
|
||||
_insert_user(other_id, "other@headquarter.local")
|
||||
_insert_project(project_id, owner_id)
|
||||
|
||||
app = _load_app()
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _mint_token(other_id))
|
||||
|
||||
response = client.patch(f"/projects/{project_id}/default-ssh-key", json={"ssh_key_id": "cccccccc-cccc-cccc-cccc-cccccccccccc"})
|
||||
|
||||
assert response.status_code == 403
|
||||
@@ -0,0 +1,59 @@
|
||||
from collections.abc import AsyncIterator
|
||||
|
||||
import pytest
|
||||
import pytest_asyncio
|
||||
from sqlalchemy import select, text
|
||||
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
|
||||
|
||||
from src.config import build_database_url
|
||||
from src.models.user import User
|
||||
from src.scripts.seed import build_seed_user, seed_database
|
||||
|
||||
|
||||
TEST_DATABASE_URL = build_database_url(
|
||||
user="headquarter",
|
||||
password="headquarter",
|
||||
host="localhost",
|
||||
port=5432,
|
||||
database="headquarter",
|
||||
)
|
||||
|
||||
|
||||
@pytest_asyncio.fixture
|
||||
async def db_session() -> AsyncIterator[AsyncSession]:
|
||||
engine = create_async_engine(TEST_DATABASE_URL)
|
||||
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
||||
|
||||
async with session_factory() as session:
|
||||
await session.execute(text("TRUNCATE TABLE git_repositories, projects, users RESTART IDENTITY CASCADE"))
|
||||
await session.commit()
|
||||
yield session
|
||||
await session.execute(text("TRUNCATE TABLE git_repositories, projects, users RESTART IDENTITY CASCADE"))
|
||||
await session.commit()
|
||||
|
||||
await engine.dispose()
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_build_seed_user_returns_deterministic_payload() -> None:
|
||||
payload = build_seed_user()
|
||||
|
||||
assert payload == {
|
||||
"email": "dev@headquarter.local",
|
||||
"name": "Development User",
|
||||
"authentik_id": "dev-authentik-user",
|
||||
"avatar_url": None,
|
||||
}
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
n@pytest.mark.integration
|
||||
|
||||
async def test_seed_database_creates_development_user(db_session: AsyncSession) -> None:
|
||||
await seed_database(db_session)
|
||||
|
||||
seeded_user = await db_session.scalar(select(User).where(User.email == "dev@headquarter.local"))
|
||||
|
||||
assert seeded_user is not None
|
||||
assert seeded_user.authentik_id == "dev-authentik-user"
|
||||
@@ -0,0 +1,26 @@
|
||||
import pytest
|
||||
from httpx import AsyncClient
|
||||
|
||||
from src.main import app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def async_client():
|
||||
async with AsyncClient(app=app, base_url="http://test") as client:
|
||||
yield client
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
n@pytest.mark.integration
|
||||
|
||||
async def test_create_ssh_key_requires_authentication(async_client: AsyncClient) -> None:
|
||||
response = await async_client.post("/ssh-keys", json={"name": "test-key"})
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
n@pytest.mark.integration
|
||||
|
||||
async def test_list_ssh_keys_requires_authentication(async_client: AsyncClient) -> None:
|
||||
response = await async_client.get("/ssh-keys")
|
||||
assert response.status_code == 401
|
||||
@@ -0,0 +1,241 @@
|
||||
import uuid
|
||||
from datetime import UTC, datetime, timedelta
|
||||
import asyncio
|
||||
import io
|
||||
|
||||
from fastapi.testclient import TestClient
|
||||
import pytest
|
||||
from sqlalchemy import text
|
||||
from sqlalchemy.ext.asyncio import create_async_engine
|
||||
|
||||
from src.auth.jwt_service import mint_access_token
|
||||
from src.config import Settings, build_database_url
|
||||
from src.models import Base
|
||||
from src.models.user import User
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def configure_local_database(monkeypatch) -> None:
|
||||
local_url = build_database_url(
|
||||
user="headquarter",
|
||||
password="headquarter",
|
||||
host="localhost",
|
||||
port=5432,
|
||||
database="headquarter",
|
||||
)
|
||||
monkeypatch.setenv("DATABASE_URL", local_url)
|
||||
|
||||
|
||||
def _prepare_users_test_db() -> None:
|
||||
async def _run() -> None:
|
||||
engine = create_async_engine(
|
||||
build_database_url(
|
||||
user="headquarter",
|
||||
password="headquarter",
|
||||
host="localhost",
|
||||
port=5432,
|
||||
database="headquarter",
|
||||
)
|
||||
)
|
||||
async with engine.begin() as connection:
|
||||
await connection.run_sync(Base.metadata.create_all)
|
||||
await connection.execute(text("TRUNCATE TABLE refresh_tokens, users RESTART IDENTITY CASCADE"))
|
||||
await engine.dispose()
|
||||
|
||||
asyncio.run(_run())
|
||||
|
||||
|
||||
def _load_app():
|
||||
import importlib
|
||||
import src.database as database_module
|
||||
import src.api.users as users_module
|
||||
import src.main as main_module
|
||||
|
||||
importlib.reload(database_module)
|
||||
importlib.reload(users_module)
|
||||
importlib.reload(main_module)
|
||||
return main_module.app
|
||||
|
||||
|
||||
def _insert_test_user(user_id: str) -> None:
|
||||
async def _run() -> None:
|
||||
engine = create_async_engine(
|
||||
build_database_url(
|
||||
user="headquarter",
|
||||
password="headquarter",
|
||||
host="localhost",
|
||||
port=5432,
|
||||
database="headquarter",
|
||||
)
|
||||
)
|
||||
async with engine.begin() as connection:
|
||||
await connection.run_sync(Base.metadata.create_all)
|
||||
from sqlalchemy.ext.asyncio import async_sessionmaker
|
||||
|
||||
session_factory = async_sessionmaker(engine, expire_on_commit=False)
|
||||
async with session_factory() as session:
|
||||
user = User(
|
||||
id=uuid.UUID(user_id),
|
||||
email="test@headquarter.local",
|
||||
name="Test User",
|
||||
authentik_id="test-user",
|
||||
avatar_url=None,
|
||||
)
|
||||
await session.merge(user)
|
||||
await session.commit()
|
||||
await engine.dispose()
|
||||
|
||||
asyncio.run(_run())
|
||||
|
||||
|
||||
def _create_auth_cookie(user_id: str) -> str:
|
||||
settings = Settings()
|
||||
return mint_access_token(
|
||||
settings=settings,
|
||||
subject=user_id,
|
||||
email="test@headquarter.local",
|
||||
name="Test User",
|
||||
expires_at=datetime.now(UTC) + timedelta(minutes=15),
|
||||
)
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_get_profile_returns_401_without_cookie() -> None:
|
||||
_prepare_users_test_db()
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
response = client.get("/users/me")
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_get_profile_returns_user_data() -> None:
|
||||
_prepare_users_test_db()
|
||||
user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
|
||||
_insert_test_user(user_id)
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _create_auth_cookie(user_id))
|
||||
response = client.get("/users/me")
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["email"] == "test@headquarter.local"
|
||||
assert data["name"] == "Test User"
|
||||
assert data["avatar_url"] is None
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_update_profile_changes_name_and_email() -> None:
|
||||
_prepare_users_test_db()
|
||||
user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
|
||||
_insert_test_user(user_id)
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _create_auth_cookie(user_id))
|
||||
response = client.put("/users/me", json={"name": "Updated Name", "email": "updated@headquarter.local"})
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["name"] == "Updated Name"
|
||||
assert data["email"] == "updated@headquarter.local"
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_update_profile_rejects_empty_name() -> None:
|
||||
_prepare_users_test_db()
|
||||
user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
|
||||
_insert_test_user(user_id)
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _create_auth_cookie(user_id))
|
||||
response = client.put("/users/me", json={"name": " "})
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_update_profile_rejects_invalid_email() -> None:
|
||||
_prepare_users_test_db()
|
||||
user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
|
||||
_insert_test_user(user_id)
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _create_auth_cookie(user_id))
|
||||
response = client.put("/users/me", json={"email": "not-an-email"})
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_upload_avatar_updates_avatar_url() -> None:
|
||||
_prepare_users_test_db()
|
||||
user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
|
||||
_insert_test_user(user_id)
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _create_auth_cookie(user_id))
|
||||
|
||||
# Create a simple 1x1 PNG
|
||||
png_data = b"\x89PNG\r\n\x1a\n" + b"\x00" * 100
|
||||
response = client.post(
|
||||
"/users/me/avatar",
|
||||
files={"file": ("test.png", io.BytesIO(png_data), "image/png")},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["avatar_url"] is not None
|
||||
assert data["avatar_url"].startswith("/uploads/avatars/")
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_upload_avatar_rejects_invalid_file_type() -> None:
|
||||
_prepare_users_test_db()
|
||||
user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
|
||||
_insert_test_user(user_id)
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _create_auth_cookie(user_id))
|
||||
|
||||
response = client.post(
|
||||
"/users/me/avatar",
|
||||
files={"file": ("test.txt", io.BytesIO(b"not an image"), "text/plain")},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
|
||||
|
||||
n@pytest.mark.integration
|
||||
|
||||
def test_upload_avatar_rejects_oversized_file() -> None:
|
||||
_prepare_users_test_db()
|
||||
user_id = "7f4b7ad8-c4ce-4d1b-8c83-7ce0f4f66dfb"
|
||||
_insert_test_user(user_id)
|
||||
app = _load_app()
|
||||
|
||||
client = TestClient(app)
|
||||
client.cookies.set("access_token", _create_auth_cookie(user_id))
|
||||
|
||||
large_png = b"\x89PNG\r\n\x1a\n" + b"\x00" * (3 * 1024 * 1024) # 3MB
|
||||
response = client.post(
|
||||
"/users/me/avatar",
|
||||
files={"file": ("large.png", io.BytesIO(large_png), "image/png")},
|
||||
)
|
||||
|
||||
assert response.status_code == 400
|
||||
Reference in New Issue
Block a user