Pipeline/backend/tests/test_forgejo_metrics_api.py

# ruff: noqa: INP001
"""Integration tests for Forgejo metrics scoping."""

from __future__ import annotations

from datetime import datetime
from uuid import uuid4

import pytest
from fastapi import APIRouter, FastAPI
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncEngine, async_sessionmaker, create_async_engine
from sqlmodel import SQLModel
from sqlmodel.ext.asyncio.session import AsyncSession

from app import models as _models
from app.api.deps import require_org_member
from app.api.forgejo_metrics import router as forgejo_metrics_router
from app.db.session import get_session
from app.models.board_repository_links import BoardRepositoryLink
from app.models.boards import Board
from app.models.forgejo_connections import ForgejoConnection
from app.models.forgejo_issues import ForgejoIssue
from app.models.forgejo_repositories import ForgejoRepository
from app.models.organization_members import OrganizationMember
from app.models.organizations import Organization
from app.services.organizations import OrganizationContext


async def _make_engine() -> AsyncEngine:
    engine = create_async_engine("sqlite+aiosqlite:///:memory:")
    async with engine.begin() as conn:
        await conn.run_sync(SQLModel.metadata.create_all)
    return engine


def _build_test_app(
    session_maker: async_sessionmaker[AsyncSession],
    ctx: OrganizationContext,
) -> FastAPI:
    app = FastAPI()
    api_v1 = APIRouter(prefix="/api/v1")
    api_v1.include_router(forgejo_metrics_router)
    app.include_router(api_v1)

    async def _override_get_session() -> AsyncSession:
        async with session_maker() as session:
            yield session

    async def _override_require_org_member() -> OrganizationContext:
        return ctx

    app.dependency_overrides[get_session] = _override_get_session
    app.dependency_overrides[require_org_member] = _override_require_org_member
    return app


def _issue(
    *,
    organization_id,
    repository_id,
    issue_number: int,
) -> ForgejoIssue:
    return ForgejoIssue(
        id=uuid4(),
        organization_id=organization_id,
        repository_id=repository_id,
        forgejo_issue_number=issue_number,
        title=f"Issue {issue_number}",
        body_preview="Cached issue body",
        state="open",
        is_pull_request=False,
        labels=[],
        assignees=[],
        author="kaspa",
        html_url=f"https://forgejo.example.local/openclaw/pipeline/issues/{issue_number}",
        forgejo_created_at=datetime(2026, 5, 19, 12, 0, 0),
        forgejo_updated_at=datetime(2026, 5, 19, 12, 5, 0),
    )


@pytest.mark.asyncio
async def test_forgejo_metrics_are_scoped_to_active_organization() -> None:
    engine = await _make_engine()
    session_maker = async_sessionmaker(engine, class_=AsyncSession, expire_on_commit=False)
    org_a = Organization(id=uuid4(), name="Org A")
    org_b = Organization(id=uuid4(), name="Org B")
    member = OrganizationMember(
        id=uuid4(),
        organization_id=org_a.id,
        user_id=uuid4(),
        role="owner",
    )
    app = _build_test_app(
        session_maker,
        OrganizationContext(organization=org_a, member=member),
    )

    conn_a = ForgejoConnection(
        id=uuid4(),
        organization_id=org_a.id,
        name="A Forgejo",
        base_url="https://forgejo-a.example.local",
    )
    conn_b = ForgejoConnection(
        id=uuid4(),
        organization_id=org_b.id,
        name="B Forgejo",
        base_url="https://forgejo-b.example.local",
    )
    repo_a = ForgejoRepository(
        id=uuid4(),
        organization_id=org_a.id,
        connection_id=conn_a.id,
        owner="openclaw",
        repo="pipeline",
        display_name="Pipeline",
    )
    repo_b = ForgejoRepository(
        id=uuid4(),
        organization_id=org_b.id,
        connection_id=conn_b.id,
        owner="other",
        repo="private",
        display_name="Private",
    )
    board_b = Board(
        id=uuid4(),
        organization_id=org_b.id,
        name="Other Board",
        slug="other-board",
    )
    link_b = BoardRepositoryLink(
        id=uuid4(),
        organization_id=org_b.id,
        board_id=board_b.id,
        repository_id=repo_b.id,
    )

    try:
        async with session_maker() as session:
            session.add(org_a)
            session.add(org_b)
            session.add(conn_a)
            session.add(conn_b)
            session.add(repo_a)
            session.add(repo_b)
            session.add(board_b)
            session.add(link_b)
            session.add(
                _issue(
                    organization_id=org_a.id,
                    repository_id=repo_a.id,
                    issue_number=1,
                )
            )
            session.add(
                _issue(
                    organization_id=org_b.id,
                    repository_id=repo_b.id,
                    issue_number=2,
                )
            )
            await session.commit()

        async with AsyncClient(
            transport=ASGITransport(app=app),
            base_url="http://testserver",
        ) as client:
            own_org = await client.get(
                f"/api/v1/forgejo/metrics?organization_id={org_a.id}"
            )
            other_org = await client.get(
                f"/api/v1/forgejo/metrics?organization_id={org_b.id}"
            )
            other_repo = await client.get(
                f"/api/v1/forgejo/metrics?repository_id={repo_b.id}"
            )
            other_board = await client.get(
                f"/api/v1/forgejo/metrics?board_id={board_b.id}"
            )

        assert own_org.status_code == 200
        assert own_org.json()["open_issues"] == 1
        assert other_org.status_code == 403
        assert other_repo.status_code == 200
        assert other_repo.json()["open_issues"] == 0
        assert other_board.status_code == 200
        assert other_board.json()["open_issues"] == 0
    finally:
        await engine.dispose()