Fusion
996ea73bbf
- Add POST /tool-types/validate endpoint for pre-creation validation - Add ToolConfigUpdate model with optional fields for PUT endpoint - Fix tool_configs POST to return 201 status code - Fix tool_configs list endpoint to return list instead of dict - Fix tool_configs defaults endpoint to return 'suggested_configs' - Fix tool_types create endpoint to include category and interfaces - Add model_validator to enforce dockerfile/compose template requirements - Update tests to match API response format
121 lines
3.6 KiB
Python
121 lines
3.6 KiB
Python
import pytest
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
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.ssh_key import SSHKey
|
|
from src.models.user import User
|
|
from src.models.user_config import UserConfig
|
|
|
|
|
|
@pytest.mark.integration
|
|
def test_base_metadata_collects_declared_tables() -> None:
|
|
assert isinstance(Base.metadata.tables, dict)
|
|
|
|
|
|
@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__
|
|
|
|
|
|
@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",
|
|
}
|
|
|
|
|
|
@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
|
|
|
|
|
|
@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
|
|
|
|
|
|
@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
|
|
|
|
|
|
@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
|
|
@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"
|