test classes

app/flask-postgres/app/app.py

@@ -15,14 +15,15 @@ from flask import (
 )
 from werkzeug.security import check_password_hash, generate_password_hash
 
-from config import Config
+from .config import Config
-from db import get_connection, fetchone_dict, fetchall_dict
+from .db import get_connection, fetchone_dict, fetchall_dict
-from auth import login_required
+from .auth import login_required
-from permissions import (
+from .permissions import is_video_allowed_for_level
+from .logging_config import setup_logging
+from .security import (
     admin_required,
     get_current_user,
     get_current_user_mandant_level,
-    is_video_allowed_for_level,
 )
 
 app = Flask(__name__)
@@ -31,20 +32,21 @@ app.secret_key = app.config["SECRET_KEY"]
 
 
 LOG_DIR = app.config["LOG_DIR"]
-os.makedirs(LOG_DIR, exist_ok=True)
+if not app.config.get("TESTING"):
+    os.makedirs(LOG_DIR, exist_ok=True)
 
-file_handler = RotatingFileHandler(
+# file_handler = RotatingFileHandler(
-    os.path.join(LOG_DIR, "flask-app.log"),
+#     os.path.join(LOG_DIR, "flask-app.log"),
-    maxBytes=5 * 1024 * 1024,
+#     maxBytes=5 * 1024 * 1024,
-    backupCount=5
+#     backupCount=5
-)
+# )
-file_handler.setLevel(logging.INFO)
+# file_handler.setLevel(logging.INFO)
-file_handler.setFormatter(
+# file_handler.setFormatter(
-    logging.Formatter("%(asctime)s %(levelname)s %(message)s")
+#     logging.Formatter("%(asctime)s %(levelname)s %(message)s")
-)
+# )
 
-app.logger.setLevel(logging.INFO)
+# app.logger.setLevel(logging.INFO)
-app.logger.addHandler(file_handler)
+# app.logger.addHandler(file_handler)
 
 
 

app/flask-postgres/app/config.py

@@ -10,4 +10,4 @@ class Config:
     DB_PASSWORD = os.getenv("DB_PASSWORD", "CertPWD")
     DB_PORT = os.getenv("DB_PORT", "5432")
 
-    LOG_DIR = os.getenv("LOG_DIR", "/logs")
+    LOG_DIR = os.getenv("LOG_DIR", "./logs")

app/flask-postgres/app/logging_config.py

@@ -0,0 +1,18 @@
+import os
+import logging
+from logging.handlers import RotatingFileHandler
+
+
+def setup_logging(app):
+    log_dir = app.config["LOG_DIR"]
+
+    os.makedirs(log_dir, exist_ok=True)
+
+    handler = RotatingFileHandler(
+        os.path.join(log_dir, "app.log"),
+        maxBytes=1_000_000,
+        backupCount=5
+    )
+
+    handler.setLevel(logging.INFO)
+    app.logger.addHandler(handler)

app/flask-postgres/app/permissions.py

@@ -1,35 +1,4 @@
 import os
-from functools import wraps
-
-from flask import session, redirect, url_for, request, abort
-
-from db import get_connection
-
-
-def get_current_user_mandant_level():
-    user_id = session.get("user_id")
-    if not user_id:
-        return None
-
-    conn = get_connection()
-    cur = conn.cursor()
-
-    cur.execute("""
-        SELECT m.level
-        FROM app_user u
-        JOIN mandant m ON m.id = u.mandant_id
-        WHERE u.id = %s
-    """, (user_id,))
-
-    row = cur.fetchone()
-
-    cur.close()
-    conn.close()
-
-    if row is None:
-        return None
-
-    return row[0]
 
 
 def is_video_allowed_for_level(filename: str, mandant_level: int | None) -> bool:
@@ -46,53 +15,4 @@ def is_video_allowed_for_level(filename: str, mandant_level: int | None) -> bool
     if mandant_level == 3:
         return first_char == "A"
 
-    return False
+    return False
-
-
-def user_is_admin():
-    user_id = session.get("user_id")
-    if not user_id:
-        return False
-
-    conn = get_connection()
-    cur = conn.cursor()
-
-    cur.execute("""
-        SELECT 1
-        FROM app_user u
-        JOIN user_group ug ON ug.user_id = u.id
-        JOIN app_group g ON g.id = ug.group_id
-        WHERE u.id = %s
-          AND ug.mandant_id = 1
-          AND g.mandant_id = 1
-          AND g.group_name = 'Administratoren'
-        LIMIT 1
-    """, (user_id,))
-
-    result = cur.fetchone()
-
-    cur.close()
-    conn.close()
-
-    return result is not None
-
-
-def get_current_user():
-    return {
-        "user_id": session.get("user_id"),
-        "user_name": session.get("user_name"),
-        "user_email": session.get("user_email"),
-        "is_logged_in": bool(session.get("user_id")),
-        "is_admin": user_is_admin() if session.get("user_id") else False,
-    }
-
-
-def admin_required(view_func):
-    @wraps(view_func)
-    def wrapper(*args, **kwargs):
-        if not session.get("user_id"):
-            return redirect(url_for("login", next=request.path))
-        if not user_is_admin():
-            abort(403)
-        return view_func(*args, **kwargs)
-    return wrapper

app/flask-postgres/app/requirements.txt

@@ -1,4 +1,5 @@
 Flask==3.0.2
 gunicorn==22.0.0
 psycopg2-binary==2.9.9
-Werkzeug==3.0.1
+Werkzeug==3.0.1
+pytest==8.3.2

app/flask-postgres/app/security.py

@@ -0,0 +1,79 @@
+from functools import wraps
+from flask import session, redirect, url_for, request, abort
+
+from .db import get_connection
+
+
+def get_current_user_mandant_level():
+    user_id = session.get("user_id")
+    if not user_id:
+        return None
+
+    conn = get_connection()
+    cur = conn.cursor()
+
+    cur.execute("""
+        SELECT m.level
+        FROM app_user u
+        JOIN mandant m ON m.id = u.mandant_id
+        WHERE u.id = %s
+    """, (user_id,))
+
+    row = cur.fetchone()
+
+    cur.close()
+    conn.close()
+
+    if row is None:
+        return None
+
+    return row[0]
+
+
+def user_is_admin():
+    user_id = session.get("user_id")
+    if not user_id:
+        return False
+
+    conn = get_connection()
+    cur = conn.cursor()
+
+    cur.execute("""
+        SELECT 1
+        FROM app_user u
+        JOIN user_group ug ON ug.user_id = u.id
+        JOIN app_group g ON g.id = ug.group_id
+        WHERE u.id = %s
+          AND ug.mandant_id = 1
+          AND g.mandant_id = 1
+          AND g.group_name = 'Administratoren'
+        LIMIT 1
+    """, (user_id,))
+
+    result = cur.fetchone()
+
+    cur.close()
+    conn.close()
+
+    return result is not None
+
+
+def get_current_user():
+    return {
+        "user_id": session.get("user_id"),
+        "user_name": session.get("user_name"),
+        "user_email": session.get("user_email"),
+        "is_logged_in": bool(session.get("user_id")),
+        "is_admin": user_is_admin() if session.get("user_id") else False,
+    }
+
+
+def admin_required(view_func):
+    @wraps(view_func)
+    def wrapper(*args, **kwargs):
+        if not session.get("user_id"):
+            return redirect(url_for("login", next=request.path))
+        if not user_is_admin():
+            abort(403)
+        return view_func(*args, **kwargs)
+    return wrapper

app/flask-postgres/tests/conftest.py

@@ -0,0 +1,5 @@
+import os
+import sys
+
+BASE_DIR = os.path.dirname(os.path.dirname(__file__))
+sys.path.insert(0, BASE_DIR)

app/flask-postgres/tests/test_app_smoke.py

@@ -0,0 +1,27 @@
+import pytest
+import os
+
+os.environ["DB_HOST"] = "192.168.0.10"
+os.environ["DB_NAME"] = "CertDB"
+os.environ["DB_USER"] = "CertUser"
+os.environ["DB_PASSWORD"] = "CertPWD"
+os.environ["DB_PORT"] = "55432"
+os.environ["LOG_DIR"] = "./logs"
+
+from app.app import app
+
+
+@pytest.fixture
+def client():
+    app.config["TESTING"] = True
+    with app.test_client() as client:
+        yield client
+
+
+def test_app_exists():
+    assert app is not None
+
+
+def test_protected_videos_requires_login(client):
+    response = client.get("/videos/A1.mp4", follow_redirects=False)
+    assert response.status_code in (302, 401, 403)

app/flask-postgres/tests/test_permissions.py

@@ -0,0 +1,34 @@
+from app.permissions import is_video_allowed_for_level
+
+
+def test_level_0_sees_everything():
+    assert is_video_allowed_for_level("A1.mp4", 0) is True
+    assert is_video_allowed_for_level("B2.mp4", 0) is True
+    assert is_video_allowed_for_level("C3.mp4", 0) is True
+
+
+def test_level_1_sees_everything():
+    assert is_video_allowed_for_level("A1.mp4", 1) is True
+    assert is_video_allowed_for_level("B2.mp4", 1) is True
+    assert is_video_allowed_for_level("C3.mp4", 1) is True
+
+
+def test_level_2_sees_only_a_and_b():
+    assert is_video_allowed_for_level("A1.mp4", 2) is True
+    assert is_video_allowed_for_level("B2.mp4", 2) is True
+    assert is_video_allowed_for_level("C3.mp4", 2) is False
+
+
+def test_level_3_sees_only_a():
+    assert is_video_allowed_for_level("A1.mp4", 3) is True
+    assert is_video_allowed_for_level("B2.mp4", 3) is False
+    assert is_video_allowed_for_level("C3.mp4", 3) is False
+
+
+def test_none_level_sees_nothing():
+    assert is_video_allowed_for_level("A1.mp4", None) is False
+
+
+def test_lowercase_filename_is_handled():
+    assert is_video_allowed_for_level("a1.mp4", 3) is True
+    assert is_video_allowed_for_level("b1.mp4", 3) is False