Coverage for app/routers/admin_apps.py: 91.11%
45 statements
« prev ^ index » next coverage.py v7.14.1, created at 2026-07-25 15:51 +0000
1from uuid import UUID
3from fastapi import APIRouter, Depends, HTTPException, status
4from sqlalchemy.orm import Session
6from app.crud import app_version_approvals as crud_approvals
7from app.crud import apps as crud_apps
8from app.database import get_db
9from app.models import User
10from app.routers.apps import _serialize_app, load_variable_definitions
11from app.schemas import (
12 AppResponse,
13 AppVersionApprovalDecision,
14 AppVersionApprovalResponse,
15 AppVersionApprovalWithApp,
16)
17from app.utils.permissions import require_admin
19router = APIRouter()
22# ----------------------------------------------------------------
23# PENDING REVIEW QUEUE
24# ----------------------------------------------------------------
25@router.get(
26 "/apps/versions/pending",
27 response_model=list[AppVersionApprovalWithApp],
28 tags=["Admin"],
29)
30def list_pending_versions(
31 db: Session = Depends(get_db),
32 _: User = Depends(require_admin),
33):
34 """Return all version submissions awaiting admin review, oldest first."""
35 return crud_approvals.get_pending_approvals(db)
38# ----------------------------------------------------------------
39# APPROVE VERSION
40# ----------------------------------------------------------------
41@router.post(
42 "/apps/{app_id}/versions/{version_tag}/approve",
43 response_model=AppVersionApprovalResponse,
44 tags=["Admin"],
45)
46def approve_version(
47 app_id: UUID,
48 version_tag: str,
49 db: Session = Depends(get_db),
50 current_user: User = Depends(require_admin),
51):
52 """Approve a PENDING version — makes it deployable by all users.
54 Validates the ``@openstack`` markers in this version's Terraform/Packer
55 variable files before flipping the status, using the same helper as
56 ``GET /apps/{id}/variables``.
57 """
58 app = _require_app(db, app_id)
60 # Block approval if any variable carries a marker error. If the git repo
61 # is unreachable (400/500), skip validation rather than hard-blocking.
62 try:
63 variables = load_variable_definitions(app, version_tag)
64 marker_errors = [
65 v.get("markerError") for v in variables if v.get("markerError")
66 ]
67 if marker_errors:
68 raise HTTPException(
69 status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
70 detail={
71 "message": (
72 "Version kann nicht approved werden — fehlerhafte "
73 "@openstack-Marker in den Variablen-Dateien"
74 ),
75 "marker_errors": marker_errors,
76 },
77 )
78 except HTTPException as exc:
79 if exc.status_code == status.HTTP_422_UNPROCESSABLE_ENTITY:
80 raise
81 # 400 (no git_link) or 500 (git unreachable) — skip validation
83 return crud_approvals.approve(db, app_id, version_tag, current_user.userId)
86# ----------------------------------------------------------------
87# REJECT VERSION
88# ----------------------------------------------------------------
89@router.post(
90 "/apps/{app_id}/versions/{version_tag}/reject",
91 response_model=AppVersionApprovalResponse,
92 tags=["Admin"],
93)
94def reject_version(
95 app_id: UUID,
96 version_tag: str,
97 body: AppVersionApprovalDecision,
98 db: Session = Depends(get_db),
99 current_user: User = Depends(require_admin),
100):
101 """Reject a PENDING version with a mandatory reason."""
102 _require_app(db, app_id)
103 return crud_approvals.reject(
104 db, app_id, version_tag, current_user.userId, body.rejection_reason
105 )
108# ----------------------------------------------------------------
109# REVOKE APPROVED VERSION
110# ----------------------------------------------------------------
111@router.post(
112 "/apps/{app_id}/versions/{version_tag}/revoke",
113 response_model=AppVersionApprovalResponse,
114 tags=["Admin"],
115)
116def revoke_version(
117 app_id: UUID,
118 version_tag: str,
119 body: AppVersionApprovalDecision,
120 db: Session = Depends(get_db),
121 current_user: User = Depends(require_admin),
122):
123 """Revoke a previously APPROVED version with a mandatory reason (sets status to REJECTED)."""
124 _require_app(db, app_id)
125 return crud_approvals.revoke(db, app_id, version_tag, current_user.userId, body.rejection_reason)
128# ----------------------------------------------------------------
129# EMERGENCY DEACTIVATION
130# ----------------------------------------------------------------
131@router.put(
132 "/apps/{app_id}",
133 response_model=AppResponse,
134 tags=["Admin"],
135)
136def deactivate_app(
137 app_id: UUID,
138 db: Session = Depends(get_db),
139 _: User = Depends(require_admin),
140):
141 """Emergency deactivation: set app to private so it disappears from
142 the store immediately. Does not delete the app or its deployments."""
143 from app.schemas import AppUpdate
145 _require_app(db, app_id)
146 updated = crud_apps.update_app(db, app_id, AppUpdate(is_private=True))
147 return _serialize_app(updated)
150# ----------------------------------------------------------------
151# HELPER
152# ----------------------------------------------------------------
153def _require_app(db: Session, app_id: UUID):
154 app = crud_apps.get_app(db, app_id, include_deleted=True)
155 if not app:
156 raise HTTPException(
157 status_code=status.HTTP_404_NOT_FOUND,
158 detail="App not found",
159 )
160 return app