Merge pull request 'Mandantenfähigkeit V1.1' (#11) from develop into main
All checks were successful
Deploy Production / deploy (push) Successful in 37s
Test Suite / pytest-backend (push) Successful in 11s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 6s
Test Suite / playwright-tests (push) Successful in 22s
All checks were successful
Deploy Production / deploy (push) Successful in 37s
Test Suite / pytest-backend (push) Successful in 11s
Test Suite / lint-backend (push) Successful in 0s
Test Suite / build-frontend (push) Successful in 6s
Test Suite / playwright-tests (push) Successful in 22s
Reviewed-on: #11
This commit is contained in:
commit
0c1fbab0ef
|
|
@ -3,11 +3,44 @@ name: Test Suite
|
|||
on:
|
||||
push:
|
||||
branches: [main, develop]
|
||||
pull_request:
|
||||
branches: [main, develop]
|
||||
workflow_run:
|
||||
workflows: ["Deploy Development", "Deploy Production"]
|
||||
types: [completed]
|
||||
|
||||
jobs:
|
||||
# Wie Mitai-Jinkendo: pytest im laufenden backend-Container (Python aus Image, gleiche DB wie Deploy).
|
||||
pytest-backend:
|
||||
if: ${{ github.event_name != 'workflow_run' || github.event.workflow_run.conclusion == 'success' }}
|
||||
runs-on: ubuntu-latest
|
||||
steps:
|
||||
- name: Backend pytest im deployten Container
|
||||
run: |
|
||||
EVENT_NAME="${{ github.event_name }}"
|
||||
REF_NAME="${{ github.ref_name }}"
|
||||
RUN_WORKFLOW="${{ github.event.workflow_run.name }}"
|
||||
APP_DIR="/home/lars/docker/shinkan"
|
||||
COMPOSE_FILE="docker-compose.yml"
|
||||
|
||||
if [ "$EVENT_NAME" = "workflow_run" ]; then
|
||||
if [ "$RUN_WORKFLOW" = "Deploy Development" ]; then
|
||||
APP_DIR="/home/lars/docker/shinkan-dev"
|
||||
COMPOSE_FILE="docker-compose.dev-env.yml"
|
||||
fi
|
||||
elif [ "$REF_NAME" = "develop" ]; then
|
||||
APP_DIR="/home/lars/docker/shinkan-dev"
|
||||
COMPOSE_FILE="docker-compose.dev-env.yml"
|
||||
fi
|
||||
|
||||
cd "$APP_DIR"
|
||||
docker compose -f "$COMPOSE_FILE" exec -T backend sh -lc "
|
||||
pip install -r /app/requirements-dev.txt &&
|
||||
cd /app &&
|
||||
ACCESS_LAYER_STRICT=1 python scripts/check_access_layer_hints.py &&
|
||||
ACCESS_LAYER_INTEGRATION=1 SKIP_DB_MIGRATE=1 python -m pytest tests -m 'not slow' -ra -vv --tb=short
|
||||
"
|
||||
|
||||
lint-backend:
|
||||
if: ${{ github.event_name != 'workflow_run' || github.event.workflow_run.conclusion == 'success' }}
|
||||
runs-on: ubuntu-latest
|
||||
|
|
@ -172,7 +205,8 @@ jobs:
|
|||
|
||||
- name: Upload test screenshots
|
||||
if: failure()
|
||||
uses: actions/upload-artifact@v4
|
||||
# v4 ist auf Gitea (GHES-kompatibles Actions-Backend) oft nicht verfügbar
|
||||
uses: actions/upload-artifact@v3
|
||||
with:
|
||||
name: playwright-screenshots
|
||||
path: screenshots/
|
||||
|
|
|
|||
|
|
@ -29,8 +29,10 @@ class PasswordResetConfirm(BaseModel):
|
|||
new_password: str
|
||||
|
||||
class ProfileCreate(BaseModel):
|
||||
name: str
|
||||
email: Optional[str] = None
|
||||
"""Nur für POST /api/profiles (Plattform-Admin): neues Nutzerprofil ohne Self-Registration."""
|
||||
|
||||
name: str = Field(min_length=2, max_length=200)
|
||||
email: EmailStr
|
||||
|
||||
class ProfileUpdate(BaseModel):
|
||||
name: Optional[str] = None
|
||||
|
|
|
|||
|
|
@ -1,3 +1,10 @@
|
|||
[pytest]
|
||||
testpaths = tests
|
||||
pythonpath = .
|
||||
python_files = test_*.py
|
||||
python_functions = test_*
|
||||
addopts = -q --tb=short
|
||||
markers =
|
||||
smoke: Schnelle Kern-Regression.
|
||||
integration: PostgreSQL-Mandanten-Integration (ACCESS_LAYER_INTEGRATION=1).
|
||||
slow: Lange/schwere Tests; in CI wie Mitai-Jinkendo ausgeschlossen (Auswahl: not slow).
|
||||
|
|
|
|||
|
|
@ -3,14 +3,14 @@ Profile Management Endpoints for Mitai Jinkendo
|
|||
|
||||
Handles profile CRUD operations for both admin and current user.
|
||||
"""
|
||||
import uuid
|
||||
import secrets
|
||||
from typing import Optional
|
||||
from datetime import datetime
|
||||
|
||||
from fastapi import APIRouter, HTTPException, Header, Depends
|
||||
|
||||
from db import get_db, get_cursor, r2d
|
||||
from auth import require_auth
|
||||
from auth import require_auth, hash_pin
|
||||
from club_tenancy import assert_club_member, memberships_with_roles, is_platform_admin
|
||||
from tenant_context import resolve_tenant_context, TenantContext, get_tenant_context
|
||||
from models import ProfileCreate, ProfileUpdate
|
||||
|
|
@ -87,17 +87,51 @@ def list_profiles(session=Depends(require_auth)):
|
|||
|
||||
@router.post("/profiles")
|
||||
def create_profile(p: ProfileCreate, session=Depends(require_auth)):
|
||||
"""Create new profile (admin)."""
|
||||
pid = str(uuid.uuid4())
|
||||
"""
|
||||
Neues Profil anlegen — nur Plattform-Admin.
|
||||
|
||||
Nutzt gleiches Kernschema wie Registrierung (SERIAL id); PIN wird zufällig gesetzt — Nutzer
|
||||
setzt das Passwort über „Passwort vergessen“ oder ein späteres Admin-Tool.
|
||||
"""
|
||||
role = session.get("role") or ""
|
||||
if not is_platform_admin(role):
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail="Nur Plattform-Administratoren dürfen Profile anlegen",
|
||||
)
|
||||
|
||||
email_norm = str(p.email).strip().lower()
|
||||
name_s = (p.name or "").strip()
|
||||
|
||||
pin_hash = hash_pin(secrets.token_urlsafe(24))
|
||||
|
||||
with get_db() as conn:
|
||||
cur = get_cursor(conn)
|
||||
cur.execute("""INSERT INTO profiles (id,name,avatar_color,sex,dob,height,goal_weight,goal_bf_pct,created,updated)
|
||||
VALUES (%s,%s,%s,%s,%s,%s,%s,%s,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)""",
|
||||
(pid,p.name,p.avatar_color,p.sex,p.dob,p.height,p.goal_weight,p.goal_bf_pct))
|
||||
with get_db() as conn:
|
||||
cur = get_cursor(conn)
|
||||
cur.execute("SELECT * FROM profiles WHERE id=%s", (pid,))
|
||||
return r2d(cur.fetchone())
|
||||
cur.execute(
|
||||
"SELECT id FROM profiles WHERE email IS NOT NULL AND lower(trim(email)) = %s",
|
||||
(email_norm,),
|
||||
)
|
||||
if cur.fetchone():
|
||||
raise HTTPException(status_code=409, detail="E-Mail bereits registriert")
|
||||
|
||||
cur.execute(
|
||||
"""
|
||||
INSERT INTO profiles (
|
||||
name, email, pin_hash, auth_type, role, tier,
|
||||
email_verified, created_at, updated_at
|
||||
)
|
||||
VALUES (%s, %s, %s, 'email', 'user', 'free', TRUE,
|
||||
CURRENT_TIMESTAMP, CURRENT_TIMESTAMP)
|
||||
RETURNING id
|
||||
""",
|
||||
(name_s, email_norm, pin_hash),
|
||||
)
|
||||
row = cur.fetchone()
|
||||
new_id = row["id"] if row else None
|
||||
if new_id is None:
|
||||
raise HTTPException(status_code=500, detail="Profil konnte nicht angelegt werden")
|
||||
|
||||
return profile_document(str(new_id))
|
||||
|
||||
|
||||
def profile_document(pid: str) -> dict:
|
||||
|
|
@ -114,8 +148,12 @@ def profile_document(pid: str) -> dict:
|
|||
|
||||
|
||||
@router.get("/profiles/{pid}")
|
||||
def get_profile(pid: str, _session=Depends(require_auth)):
|
||||
"""Get profile by ID."""
|
||||
def get_profile(pid: str, session=Depends(require_auth)):
|
||||
"""Profil nach ID — nur eigenes Profil oder Plattform-Admin (wie PUT)."""
|
||||
viewer = session["profile_id"]
|
||||
role = session.get("role") or ""
|
||||
if str(viewer) != str(pid) and not is_platform_admin(role):
|
||||
raise HTTPException(status_code=403, detail="Keine Berechtigung für dieses Profil")
|
||||
return profile_document(pid)
|
||||
def _run_profile_update(pid: str, p: ProfileUpdate, tenant: TenantContext) -> dict:
|
||||
"""Gemeinsame PUT-Logik für /profiles/{id} und Legacy /profile."""
|
||||
|
|
@ -249,7 +287,13 @@ def update_profile(pid: str, p: ProfileUpdate, tenant: TenantContext = Depends(g
|
|||
|
||||
@router.delete("/profiles/{pid}")
|
||||
def delete_profile(pid: str, session=Depends(require_auth)):
|
||||
"""Delete profile (admin)."""
|
||||
"""Profil löschen — nur Plattform-Admin; letztes Profil wird geschützt."""
|
||||
role = session.get("role") or ""
|
||||
if not is_platform_admin(role):
|
||||
raise HTTPException(
|
||||
status_code=403,
|
||||
detail="Nur Plattform-Administratoren dürfen Profile löschen",
|
||||
)
|
||||
with get_db() as conn:
|
||||
cur = get_cursor(conn)
|
||||
cur.execute("SELECT COUNT(*) as count FROM profiles")
|
||||
|
|
|
|||
323
backend/tests/test_access_layer_integration.py
Normal file
323
backend/tests/test_access_layer_integration.py
Normal file
|
|
@ -0,0 +1,323 @@
|
|||
"""
|
||||
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"
|
||||
|
||||
|
||||
def test_get_profile_forbidden_other_user_integration(
|
||||
integration_app, cross_tenant: CrossTenantFixture
|
||||
):
|
||||
"""GET /profiles/{id}: Fremdes Profil → 403 (PostgreSQL-Integration)."""
|
||||
client = TestClient(integration_app)
|
||||
r = client.get(
|
||||
f"/api/profiles/{cross_tenant.user_b_id}",
|
||||
headers={"X-Auth-Token": cross_tenant.token_a},
|
||||
)
|
||||
assert r.status_code == 403
|
||||
|
||||
|
||||
def test_get_profile_ok_own_integration(
|
||||
integration_app, cross_tenant: CrossTenantFixture
|
||||
):
|
||||
client = TestClient(integration_app)
|
||||
r = client.get(
|
||||
f"/api/profiles/{cross_tenant.user_a_id}",
|
||||
headers={"X-Auth-Token": cross_tenant.token_a},
|
||||
)
|
||||
assert r.status_code == 200
|
||||
body = r.json()
|
||||
assert body["id"] == cross_tenant.user_a_id
|
||||
assert "pin_hash" not in body
|
||||
193
backend/tests/test_profiles_read_access.py
Normal file
193
backend/tests/test_profiles_read_access.py
Normal file
|
|
@ -0,0 +1,193 @@
|
|||
"""
|
||||
Zugriffsrechte Profile: POST/GET/DELETE /api/profiles/{pid}.
|
||||
|
||||
TestClient + Dependency-Override für Sessions; GET mockt profile_document, DELETE/POST mocken DB,
|
||||
damit keine echte Datenbank nötig ist.
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
from fastapi.testclient import TestClient
|
||||
|
||||
os.environ.setdefault("SKIP_DB_MIGRATE", "1")
|
||||
|
||||
from auth import require_auth
|
||||
from main import app
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def client() -> TestClient:
|
||||
return TestClient(app)
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _clear_auth_override():
|
||||
yield
|
||||
app.dependency_overrides.pop(require_auth, None)
|
||||
|
||||
|
||||
def test_get_profile_forbidden_other_user(client: TestClient) -> None:
|
||||
def auth_viewer() -> dict:
|
||||
return {"profile_id": 42, "role": "trainer"}
|
||||
|
||||
app.dependency_overrides[require_auth] = auth_viewer
|
||||
with patch("routers.profiles.profile_document") as pd_mock:
|
||||
r = client.get("/api/profiles/99", headers={"X-Auth-Token": "dummy"})
|
||||
assert r.status_code == 403
|
||||
assert pd_mock.call_count == 0
|
||||
|
||||
|
||||
def test_get_profile_ok_same_numeric_id(client: TestClient) -> None:
|
||||
payload = {"id": 42, "name": "Self"}
|
||||
|
||||
def auth_self() -> dict:
|
||||
return {"profile_id": 42, "role": "trainer"}
|
||||
|
||||
app.dependency_overrides[require_auth] = auth_self
|
||||
with patch("routers.profiles.profile_document", return_value=payload) as pd_mock:
|
||||
r = client.get("/api/profiles/42", headers={"X-Auth-Token": "dummy"})
|
||||
assert r.status_code == 200
|
||||
assert r.json() == payload
|
||||
pd_mock.assert_called_once_with("42")
|
||||
|
||||
|
||||
def test_get_profile_ok_same_string_session_id(client: TestClient) -> None:
|
||||
"""Session liefert profile_id als String (Backward-Compatibility)."""
|
||||
payload = {"id": 7, "name": "S"}
|
||||
|
||||
def auth_self_str() -> dict:
|
||||
return {"profile_id": "7", "role": "user"}
|
||||
|
||||
app.dependency_overrides[require_auth] = auth_self_str
|
||||
with patch("routers.profiles.profile_document", return_value=payload):
|
||||
r = client.get("/api/profiles/7", headers={"X-Auth-Token": "dummy"})
|
||||
assert r.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.parametrize("admin_role", ["admin", "superadmin"])
|
||||
def test_get_profile_ok_platform_admin_views_other(client: TestClient, admin_role: str) -> None:
|
||||
payload = {"id": 99, "name": "Other"}
|
||||
|
||||
def auth_admin() -> dict:
|
||||
return {"profile_id": 1, "role": admin_role}
|
||||
|
||||
app.dependency_overrides[require_auth] = auth_admin
|
||||
with patch("routers.profiles.profile_document", return_value=payload) as pd_mock:
|
||||
r = client.get("/api/profiles/99", headers={"X-Auth-Token": "dummy"})
|
||||
assert r.status_code == 200
|
||||
assert r.json() == payload
|
||||
pd_mock.assert_called_once_with("99")
|
||||
|
||||
|
||||
def test_delete_profile_forbidden_non_admin(client: TestClient) -> None:
|
||||
def auth_trainer() -> dict:
|
||||
return {"profile_id": 1, "role": "trainer"}
|
||||
|
||||
app.dependency_overrides[require_auth] = auth_trainer
|
||||
with patch("routers.profiles.get_db") as mock_get_db:
|
||||
r = client.delete("/api/profiles/1", headers={"X-Auth-Token": "dummy"})
|
||||
assert r.status_code == 403
|
||||
mock_get_db.assert_not_called()
|
||||
|
||||
|
||||
def test_delete_profile_admin_last_profile_protected(client: TestClient) -> None:
|
||||
mock_conn = MagicMock()
|
||||
mock_cm = MagicMock()
|
||||
mock_cm.__enter__.return_value = mock_conn
|
||||
mock_cm.__exit__.return_value = False
|
||||
mock_cur = MagicMock()
|
||||
mock_cur.fetchone.return_value = {"count": 1}
|
||||
|
||||
app.dependency_overrides[require_auth] = lambda: {"profile_id": 99, "role": "admin"}
|
||||
with patch("routers.profiles.get_db", return_value=mock_cm), patch(
|
||||
"routers.profiles.get_cursor", return_value=mock_cur
|
||||
):
|
||||
r = client.delete("/api/profiles/5", headers={"X-Auth-Token": "dummy"})
|
||||
assert r.status_code == 400
|
||||
|
||||
|
||||
def test_delete_profile_admin_ok_when_multiple_profiles(client: TestClient) -> None:
|
||||
mock_conn = MagicMock()
|
||||
mock_cm = MagicMock()
|
||||
mock_cm.__enter__.return_value = mock_conn
|
||||
mock_cm.__exit__.return_value = False
|
||||
mock_cur = MagicMock()
|
||||
mock_cur.fetchone.return_value = {"count": 4}
|
||||
|
||||
app.dependency_overrides[require_auth] = lambda: {"profile_id": 99, "role": "superadmin"}
|
||||
with patch("routers.profiles.get_db", return_value=mock_cm), patch(
|
||||
"routers.profiles.get_cursor", return_value=mock_cur
|
||||
):
|
||||
r = client.delete("/api/profiles/12", headers={"X-Auth-Token": "dummy"})
|
||||
assert r.status_code == 200
|
||||
assert r.json() == {"ok": True}
|
||||
|
||||
calls = mock_cur.execute.call_args_list
|
||||
assert any(
|
||||
args and "DELETE FROM profiles" in str(args[0][0]) for args in calls
|
||||
)
|
||||
|
||||
|
||||
def test_create_profile_forbidden_non_admin(client: TestClient) -> None:
|
||||
app.dependency_overrides[require_auth] = lambda: {"profile_id": 1, "role": "trainer"}
|
||||
with patch("routers.profiles.get_db") as mock_get_db:
|
||||
r = client.post(
|
||||
"/api/profiles",
|
||||
json={"name": "Neu", "email": "neu@example.com"},
|
||||
headers={"X-Auth-Token": "dummy"},
|
||||
)
|
||||
assert r.status_code == 403
|
||||
mock_get_db.assert_not_called()
|
||||
|
||||
|
||||
def test_create_profile_admin_duplicate_email(client: TestClient) -> None:
|
||||
mock_cm = MagicMock()
|
||||
mock_conn = MagicMock()
|
||||
mock_cm.__enter__.return_value = mock_conn
|
||||
mock_cm.__exit__.return_value = False
|
||||
mock_cur = MagicMock()
|
||||
mock_cur.fetchone.return_value = {"id": 99}
|
||||
|
||||
app.dependency_overrides[require_auth] = lambda: {"profile_id": 1, "role": "admin"}
|
||||
with patch("routers.profiles.get_db", return_value=mock_cm), patch(
|
||||
"routers.profiles.get_cursor", return_value=mock_cur
|
||||
), patch("routers.profiles.hash_pin", return_value="hashed"):
|
||||
r = client.post(
|
||||
"/api/profiles",
|
||||
json={"name": "Dup", "email": "dup@example.com"},
|
||||
headers={"X-Auth-Token": "dummy"},
|
||||
)
|
||||
assert r.status_code == 409
|
||||
|
||||
|
||||
def test_create_profile_admin_success(client: TestClient) -> None:
|
||||
mock_cm = MagicMock()
|
||||
mock_conn = MagicMock()
|
||||
mock_cm.__enter__.return_value = mock_conn
|
||||
mock_cm.__exit__.return_value = False
|
||||
mock_cur = MagicMock()
|
||||
mock_cur.fetchone.side_effect = [None, {"id": 501}]
|
||||
|
||||
doc = {"id": 501, "name": "Nu User", "email": "nu@example.com"}
|
||||
|
||||
app.dependency_overrides[require_auth] = lambda: {"profile_id": 1, "role": "superadmin"}
|
||||
with patch("routers.profiles.get_db", return_value=mock_cm), patch(
|
||||
"routers.profiles.get_cursor", return_value=mock_cur
|
||||
), patch("routers.profiles.profile_document", return_value=doc), patch(
|
||||
"routers.profiles.hash_pin", return_value="hashed"
|
||||
):
|
||||
r = client.post(
|
||||
"/api/profiles",
|
||||
json={"name": "Nu User", "email": "nu@example.com"},
|
||||
headers={"X-Auth-Token": "dummy"},
|
||||
)
|
||||
assert r.status_code == 200
|
||||
assert r.json() == doc
|
||||
|
||||
insert_calls = [
|
||||
c.args[0] for c in mock_cur.execute.call_args_list if c.args and "INSERT INTO profiles" in str(c.args[0])
|
||||
]
|
||||
assert len(insert_calls) >= 1
|
||||
|
|
@ -1,13 +1,13 @@
|
|||
# Shinkan Jinkendo Version Information
|
||||
|
||||
APP_VERSION = "0.8.29"
|
||||
APP_VERSION = "0.8.36"
|
||||
BUILD_DATE = "2026-05-05"
|
||||
DB_SCHEMA_VERSION = "20260505041"
|
||||
|
||||
MODULE_VERSIONS = {
|
||||
"auth": "1.2.1", # Login-Rate-Limit 30/minute pro IP (vorher 5/minute)
|
||||
"profiles": "1.4.1", # PUT /profiles*, Legacy /profile: Depends(get_tenant_context); profile_document für internes Laden
|
||||
"tenant_context": "1.0.3", # pytest: backend/tests/test_access_layer.py (Visibility-SQL, Header)
|
||||
"profiles": "1.6.0", # POST /profiles nur Plattform-Admin; Insert SERIAL + E-Mail wie Auth; Tests
|
||||
"tenant_context": "1.0.4", # pytest: Unit test_access_layer.py + optional Integration test_access_layer_integration (PostgreSQL)
|
||||
"clubs": "0.4.1", # Alle geschützten Endpoints Depends(get_tenant_context); profile_id/role aus TenantContext
|
||||
"club_memberships": "1.0.1", # Depends(get_tenant_context)
|
||||
"club_join_requests": "1.0.1", # Depends(get_tenant_context)
|
||||
|
|
@ -27,6 +27,60 @@ MODULE_VERSIONS = {
|
|||
}
|
||||
|
||||
CHANGELOG = [
|
||||
{
|
||||
"version": "0.8.36",
|
||||
"date": "2026-05-05",
|
||||
"changes": [
|
||||
"POST /api/profiles: nur Plattform-Admin; Anlage schema-konform (SERIAL, E-Mail, temporärer PIN-Hash); ProfileCreate mit Pflichtfeldern name + email",
|
||||
"pytest: POST-Szenarien in test_profiles_read_access.py",
|
||||
],
|
||||
},
|
||||
{
|
||||
"version": "0.8.35",
|
||||
"date": "2026-05-05",
|
||||
"changes": [
|
||||
"DELETE /api/profiles/{pid}: nur Plattform-Admin (403 sonst); Unit-Tests mit gemockter DB",
|
||||
],
|
||||
},
|
||||
{
|
||||
"version": "0.8.34",
|
||||
"date": "2026-05-05",
|
||||
"changes": [
|
||||
"GET /api/profiles/{pid}: Zugriff nur eigenes Profil oder Plattform-Admin (403 sonst); Unit-Tests ohne DB; zwei Integrationstests mit PostgreSQL",
|
||||
"Integration: zwei zusätzliche Profil-Lese-Tests in test_access_layer_integration",
|
||||
],
|
||||
},
|
||||
{
|
||||
"version": "0.8.33",
|
||||
"date": "2026-05-05",
|
||||
"changes": [
|
||||
"CI pytest-backend wie Mitai-Jinkendo: docker compose exec backend (deployte Stacks shinkan / shinkan-dev); keine Runner-Python/Postgres-Service",
|
||||
"pytest.ini: Marker smoke/slow, addopts/-q wie Schwesterprojekt; CI: -m \"not slow\", ACCESS_LAYER_STRICT + Integration im Container",
|
||||
],
|
||||
},
|
||||
{
|
||||
"version": "0.8.32",
|
||||
"date": "2026-05-05",
|
||||
"changes": [
|
||||
"CI pytest-backend: kein setup-python — venv aus System-python3 (arm64/Debian Self-Host, Raspi)",
|
||||
],
|
||||
},
|
||||
{
|
||||
"version": "0.8.31",
|
||||
"date": "2026-05-05",
|
||||
"changes": [
|
||||
"CI (Gitea): Job pytest-backend — Checkout, Postgres-Service, run_migrations, pytest (+ ACCESS_LAYER_INTEGRATION), ACCESS_LAYER_STRICT auf Router-Hinweis-Script",
|
||||
"Workflow test.yml: pull_request auf main/develop zusätzlich zu push",
|
||||
],
|
||||
},
|
||||
{
|
||||
"version": "0.8.30",
|
||||
"date": "2026-05-05",
|
||||
"changes": [
|
||||
"pytest: optionale PostgreSQL-Integration tests/test_access_layer_integration.py (Cross-Verein Übungen Liste/Detail; ACCESS_LAYER_INTEGRATION=1)",
|
||||
"pytest.ini: Marker „integration“",
|
||||
],
|
||||
},
|
||||
{
|
||||
"version": "0.8.29",
|
||||
"date": "2026-05-05",
|
||||
|
|
|
|||
|
|
@ -1,6 +1,6 @@
|
|||
// Shinkan Jinkendo Frontend Version
|
||||
|
||||
export const APP_VERSION = "0.8.29"
|
||||
export const APP_VERSION = "0.8.36"
|
||||
export const BUILD_DATE = "2026-05-05"
|
||||
|
||||
export const PAGE_VERSIONS = {
|
||||
|
|
|
|||
|
|
@ -76,8 +76,12 @@ test('4. Navigation zu Vereine', async ({ page }) => {
|
|||
await page.locator('.bottom-nav a[href="/clubs"]').click();
|
||||
await page.waitForLoadState('networkidle');
|
||||
|
||||
// Prüfe ob Vereine-Seite geladen
|
||||
await expect(page.locator('h1, h2, .page-title')).toContainText(/vereine|clubs/i, { timeout: 5000 });
|
||||
// ClubsPage: <h1>Vereinsverwaltung</h1> + Tab <h2>Vereine</h2> → ein kombinierter
|
||||
// Selektor löst 2 Treffer aus (Playwright strict mode). URL + primäre Überschrift reichen.
|
||||
await expect(page).toHaveURL(/\/clubs\/?$/, { timeout: 5000 });
|
||||
await expect(page.getByRole('heading', { level: 1, name: /Vereinsverwaltung/i })).toBeVisible({
|
||||
timeout: 5000,
|
||||
});
|
||||
|
||||
await page.screenshot({ path: 'screenshots/04-vereine.png' });
|
||||
console.log('✓ Vereine-Seite erreichbar');
|
||||
|
|
@ -115,15 +119,20 @@ test('7. Session-Persistenz nach Reload', async ({ page }) => {
|
|||
await login(page);
|
||||
|
||||
await expect(page.locator('.spinner')).toHaveCount(0, { timeout: 10000 });
|
||||
await expect(
|
||||
page.locator('.app-main').getByRole('heading', { level: 1, name: 'Dashboard' }),
|
||||
).toBeVisible({ timeout: 10000 });
|
||||
|
||||
await page.reload({ waitUntil: 'domcontentloaded' });
|
||||
await page.waitForLoadState('networkidle');
|
||||
|
||||
// Auth lädt erst nach Spinner – nicht auf /login stranden (stabiler als Button „Login“-Tab auf Login-Screen)
|
||||
await expect(page.locator('.spinner')).toHaveCount(0, { timeout: 20000 });
|
||||
await expect(page).not.toHaveURL('**/login', { timeout: 20000 });
|
||||
await expect(page.locator('h1').filter({ hasText: /^Dashboard$/ })).toBeVisible({
|
||||
timeout: 10000,
|
||||
await expect(page).not.toHaveURL(/\/login(?:\/|$|\?|#)/, { timeout: 20000 });
|
||||
await expect(
|
||||
page.locator('.app-main').getByRole('heading', { level: 1, name: 'Dashboard' }),
|
||||
).toBeVisible({
|
||||
timeout: 20000,
|
||||
});
|
||||
|
||||
await page.screenshot({ path: 'screenshots/07-nach-reload.png' });
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user