shinkan-jinkendo/backend/tests/test_access_layer_integration.py

"""
Cross-Tenant-Integrationstests (PostgreSQL).

Voraussetzungen:
  - Migrierte Datenbank (wie lokaler Docker-Postgres).
  - ACCESS_LAYER_INTEGRATION=1 oder true
  - DB_HOST, DB_PORT, DB_NAME, DB_USER, DB_PASSWORD passend setzen
    (lokal z. B. DB_HOST=localhost DB_PORT=5434 bei docker-compose Port-Mapping).

Aufruf aus backend/:
  set ACCESS_LAYER_INTEGRATION=1
  pytest tests/test_access_layer_integration.py -v -m integration

main wird mit SKIP_DB_MIGRATE=1 importiert, damit beim Testlauf keine Migrationen erneut laufen.
"""
from __future__ import annotations

import os
import uuid
from dataclasses import dataclass

import pytest
from fastapi.testclient import TestClient


def _integration_enabled() -> bool:
    return os.getenv("ACCESS_LAYER_INTEGRATION", "").strip().lower() in ("1", "true", "yes")


pytestmark = [
    pytest.mark.integration,
    pytest.mark.skipif(
        not _integration_enabled(),
        reason="ACCESS_LAYER_INTEGRATION=1 setzen und PostgreSQL (DB_*) konfigurieren",
    ),
]


def _db_ping() -> bool:
    try:
        from db import get_db, get_cursor

        with get_db() as conn:
            cur = get_cursor(conn)
            cur.execute("SELECT 1 AS ok")
            row = cur.fetchone()
            return row is not None and row.get("ok") == 1
    except Exception:
        return False


@pytest.fixture(scope="module")
def integration_app():
    """FastAPI-App ohne erneuten Migrationslauf beim Import."""
    os.environ.setdefault("SKIP_DB_MIGRATE", "1")
    from main import app

    return app


@pytest.fixture(scope="module")
def db_ready():
    if not _db_ping():
        pytest.skip("PostgreSQL nicht erreichbar (DB_HOST/DB_PORT/… prüfen)")


@dataclass
class CrossTenantFixture:
    suffix: str
    club_a_id: int
    club_b_id: int
    user_a_id: int
    user_b_id: int
    token_a: str
    token_b: str
    exercise_official_id: int
    exercise_club_b_id: int


@pytest.fixture
def cross_tenant(integration_app, db_ready) -> CrossTenantFixture:
    from auth import hash_pin, make_token
    from db import get_db, get_cursor

    suffix = uuid.uuid4().hex[:10]
    email_a = f"access_layer_it_{suffix}_a@test.local"
    email_b = f"access_layer_it_{suffix}_b@test.local"
    name_a = f"Club A User {suffix}"
    name_b = f"Club B User {suffix}"
    club_a_name = f"access_layer_it_club_a_{suffix}"
    club_b_name = f"access_layer_it_club_b_{suffix}"
    token_a = f"it-token-a-{suffix}"
    token_b = f"it-token-b-{suffix}"

    pin_hash = hash_pin("integration-test-pin")

    with get_db() as conn:
        cur = get_cursor(conn)

        cur.execute(
            "INSERT INTO clubs (name, abbreviation, status) VALUES (%s, %s, %s) RETURNING id",
            (club_a_name, "ITA", "active"),
        )
        club_a_id = int(cur.fetchone()["id"])

        cur.execute(
            "INSERT INTO clubs (name, abbreviation, status) VALUES (%s, %s, %s) RETURNING id",
            (club_b_name, "ITB", "active"),
        )
        club_b_id = int(cur.fetchone()["id"])

        cur.execute(
            """
            INSERT INTO profiles (email, pin_hash, name, role, active_club_id)
            VALUES (%s, %s, %s, %s, %s)
            RETURNING id
            """,
            (email_a, pin_hash, name_a, "trainer", club_a_id),
        )
        user_a_id = int(cur.fetchone()["id"])

        cur.execute(
            """
            INSERT INTO profiles (email, pin_hash, name, role, active_club_id)
            VALUES (%s, %s, %s, %s, %s)
            RETURNING id
            """,
            (email_b, pin_hash, name_b, "trainer", club_b_id),
        )
        user_b_id = int(cur.fetchone()["id"])

        cur.execute(
            "INSERT INTO club_members (profile_id, club_id, status) VALUES (%s, %s, %s)",
            (user_a_id, club_a_id, "active"),
        )
        cur.execute(
            "INSERT INTO club_members (profile_id, club_id, status) VALUES (%s, %s, %s)",
            (user_b_id, club_b_id, "active"),
        )

        cur.execute(
            """
            INSERT INTO sessions (profile_id, token, expires_at)
            VALUES (%s, %s, NOW() + INTERVAL '2 days')
            """,
            (user_a_id, token_a),
        )
        cur.execute(
            """
            INSERT INTO sessions (profile_id, token, expires_at)
            VALUES (%s, %s, NOW() + INTERVAL '2 days')
            """,
            (user_b_id, token_b),
        )

        title_official = f"access_layer_it_official_{suffix}"
        title_club_b = f"access_layer_it_club_b_only_{suffix}"

        cur.execute(
            """
            INSERT INTO exercises
              (title, summary, goal, execution, visibility, status, created_by, club_id)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
            RETURNING id
            """,
            (
                title_official,
                None,
                "Goal",
                "Execution",
                "official",
                "approved",
                user_b_id,
                None,
            ),
        )
        exercise_official_id = int(cur.fetchone()["id"])

        cur.execute(
            """
            INSERT INTO exercises
              (title, summary, goal, execution, visibility, status, created_by, club_id)
            VALUES (%s, %s, %s, %s, %s, %s, %s, %s)
            RETURNING id
            """,
            (
                title_club_b,
                None,
                "Goal",
                "Execution",
                "club",
                "approved",
                user_b_id,
                club_b_id,
            ),
        )
        exercise_club_b_id = int(cur.fetchone()["id"])

        conn.commit()

    fx = CrossTenantFixture(
        suffix=suffix,
        club_a_id=club_a_id,
        club_b_id=club_b_id,
        user_a_id=user_a_id,
        user_b_id=user_b_id,
        token_a=token_a,
        token_b=token_b,
        exercise_official_id=exercise_official_id,
        exercise_club_b_id=exercise_club_b_id,
    )

    try:
        yield fx
    finally:
        with get_db() as conn:
            cur = get_cursor(conn)
            cur.execute(
                "DELETE FROM exercises WHERE id IN (%s, %s)",
                (exercise_official_id, exercise_club_b_id),
            )
            cur.execute("DELETE FROM sessions WHERE token IN (%s, %s)", (token_a, token_b))
            cur.execute(
                "DELETE FROM club_members WHERE profile_id IN (%s, %s)",
                (user_a_id, user_b_id),
            )
            cur.execute("DELETE FROM profiles WHERE id IN (%s, %s)", (user_a_id, user_b_id))
            cur.execute("DELETE FROM clubs WHERE id IN (%s, %s)", (club_a_id, club_b_id))
            conn.commit()


def test_list_hides_foreign_club_exercise_same_active_header(
    integration_app, cross_tenant: CrossTenantFixture
):
    """Nutzer nur Verein A sieht keine Vereins-Übung von Verein B (aktiver Header = A)."""
    client = TestClient(integration_app)
    r = client.get(
        "/api/exercises",
        params={"search": f"access_layer_it_club_b_only_{cross_tenant.suffix}", "limit": 50},
        headers={
            "X-Auth-Token": cross_tenant.token_a,
            "X-Active-Club-Id": str(cross_tenant.club_a_id),
        },
    )
    assert r.status_code == 200, r.text
    ids = [row["id"] for row in r.json()]
    assert cross_tenant.exercise_club_b_id not in ids


def test_list_shows_official_exercise_for_foreign_user(
    integration_app, cross_tenant: CrossTenantFixture
):
    """Offizielle Übung ist für alle authentifizierten Nutzer sichtbar."""
    client = TestClient(integration_app)
    r = client.get(
        "/api/exercises",
        params={"search": f"access_layer_it_official_{cross_tenant.suffix}", "limit": 50},
        headers={
            "X-Auth-Token": cross_tenant.token_a,
            "X-Active-Club-Id": str(cross_tenant.club_a_id),
        },
    )
    assert r.status_code == 200, r.text
    ids = [row["id"] for row in r.json()]
    assert cross_tenant.exercise_official_id in ids


def test_detail_forbidden_foreign_club_exercise(
    integration_app, cross_tenant: CrossTenantFixture
):
    """Detail: keine Mitgliedschaft im Übungs-Verein → 403."""
    client = TestClient(integration_app)
    r = client.get(
        f"/api/exercises/{cross_tenant.exercise_club_b_id}",
        headers={
            "X-Auth-Token": cross_tenant.token_a,
            "X-Active-Club-Id": str(cross_tenant.club_a_id),
        },
    )
    assert r.status_code == 403


def test_member_club_b_can_read_own_club_exercise(
    integration_app, cross_tenant: CrossTenantFixture
):
    client = TestClient(integration_app)
    r = client.get(
        f"/api/exercises/{cross_tenant.exercise_club_b_id}",
        headers={
            "X-Auth-Token": cross_tenant.token_b,
            "X-Active-Club-Id": str(cross_tenant.club_b_id),
        },
    )
    assert r.status_code == 200
    body = r.json()
    assert body["id"] == cross_tenant.exercise_club_b_id
    assert body["visibility"] == "club"