alex
22474cdba5
Backend (ruff): - Fix 106 errors: move imports to top of file (E402) - Remove unused imports (F401) - Add missing imports for undefined names (F821) - Remove unused variables (F841) - Fix test_models.py broken RefreshToken test - Fix test_projects_api.py missing TestClient import Frontend (eslint): - Remove unused imports/variables across 10 files - Fix explicit any types in client.ts and sessions.ts - Clean up empty block statements in terminal.tsx Quality gates: ruff (pass), eslint (pass), tsc --noEmit (pass), pytest (98 passed, 4 pre-existing failures)
267 lines
8.3 KiB
Python
267 lines
8.3 KiB
Python
import uuid
|
|
from datetime import UTC, datetime, timedelta
|
|
import asyncio
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from sqlalchemy import text
|
|
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker
|
|
|
|
from src.auth.session import create_session_cookie
|
|
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
|
|
|
|
|
|
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 create_session_cookie(
|
|
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())
|
|
|
|
|
|
@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
|
|
|
|
|
|
@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
|
|
|
|
|
|
@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"
|
|
|
|
|
|
@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
|
|
|
|
|
|
@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"
|
|
|
|
|
|
@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
|
|
|
|
|
|
@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
|
|
|
|
|
|
@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
|