Stirling-PDF/testing/cucumber/features/environment.py

import os
import subprocess

import requests

_BASE_URL = "http://localhost:8080"
_CONTAINER_NAME = os.environ.get("TEST_CONTAINER_NAME", "")
_REPORT_DIR = os.environ.get("TEST_REPORT_DIR", "")

# Tags that indicate a scenario requires JWT Bearer auth to be functional.
# These scenarios are skipped when the server has JWT disabled (V2=false).
# @login and @register scenarios work in both modes.
# The "jwt" tag itself is included so that feature-level @jwt tagging is sufficient
# to mark an entire feature as JWT-dependent.
_JWT_DEPENDENT_TAGS = frozenset({
    # jwt_auth.feature scenario tags
    "me", "refresh", "logout", "role", "token", "mfa", "apikey",
    # proprietary/enterprise feature tags (all scenarios in these features need JWT)
    "jwt", "user_mgmt", "admin_settings", "audit", "signature", "team",
})


def _check_jwt_available():
    """Probe the server to determine whether JWT Bearer auth is functional.

    Logs in as admin, then attempts to use the returned token on /me.
    Returns True only when the full JWT round-trip succeeds (V2 enabled).
    """
    try:
        login = requests.post(
            f"{_BASE_URL}/api/v1/auth/login",
            json={"username": "admin", "password": "stirling"},
            timeout=10,
        )
        if login.status_code != 200:
            return False
        token = login.json().get("session", {}).get("access_token")
        if not token:
            return False
        me = requests.get(
            f"{_BASE_URL}/api/v1/auth/me",
            headers={"Authorization": f"Bearer {token}"},
            timeout=10,
        )
        return me.status_code == 200
    except Exception:
        return False


def _get_docker_log_line_count():
    if not _CONTAINER_NAME:
        return 0
    try:
        result = subprocess.run(
            ["docker", "logs", _CONTAINER_NAME],
            capture_output=True, text=True, timeout=10,
        )
        return len(result.stdout.splitlines()) + len(result.stderr.splitlines())
    except Exception:
        return 0


def _capture_docker_logs_window(start_line, scenario_name):
    if not _CONTAINER_NAME or not _REPORT_DIR:
        return
    try:
        result = subprocess.run(
            ["docker", "logs", _CONTAINER_NAME],
            capture_output=True, text=True, timeout=10,
        )
        all_lines = (result.stdout + result.stderr).splitlines()
        window = all_lines[start_line:]
        if not window:
            return

        safe_name = "".join(c if c.isalnum() or c in "-_" else "_" for c in scenario_name)
        log_path = os.path.join(_REPORT_DIR, f"scenario_{safe_name}.log")
        with open(log_path, "w") as f:
            f.write(f"=== Docker logs during: {scenario_name} ===\n")
            f.write(f"Container: {_CONTAINER_NAME}\n")
            f.write(f"Lines {start_line + 1}-{len(all_lines)} ({len(window)} lines)\n")
            f.write("---\n")
            f.write("\n".join(window[-200:]) + "\n")
    except Exception:
        pass


def before_all(context):
    context.endpoint = None
    context.request_data = None
    context.files = {}
    context.response = None
    context.jwt_available = _check_jwt_available()
    if not context.jwt_available:
        print(
            "\n[JWT] JWT Bearer authentication is not available in this environment "
            "(server likely running with V2=false). "
            "Scenarios tagged with JWT-dependent tags will be skipped."
        )


def before_scenario(context, scenario):
    """Reset all per-scenario state before each scenario runs."""
    # Skip scenarios that require JWT Bearer auth when it is not functional.
    scenario_tags = set(scenario.effective_tags)
    if _JWT_DEPENDENT_TAGS & scenario_tags and not context.jwt_available:
        scenario.skip(
            "JWT Bearer authentication not available in this environment (V2 disabled). "
            "Run against a server with V2=true to execute these scenarios."
        )
        return

    context.files = {}
    context.multi_files = []
    context.json_parts = {}
    context.request_data = None
    # JWT auth state
    context.jwt_token = None
    context.original_jwt_token = None
    # OR-status helper used by auth step definitions
    context._status_ok = False
    # Stored value used by enterprise step definitions for dynamic URL composition
    context.stored_value = None

    context._docker_log_line_count = _get_docker_log_line_count()


def after_scenario(context, scenario):
    if scenario.status == "failed":
        start_line = getattr(context, "_docker_log_line_count", 0)
        _capture_docker_logs_window(start_line, scenario.name)

    if hasattr(context, "files"):
        for file in context.files.values():
            try:
                file.close()
            except Exception:
                pass

    # Close any multi-file handles
    for _key, file in getattr(context, "multi_files", []):
        try:
            file.close()
        except Exception:
            pass

    if os.path.exists("response_file"):
        os.remove("response_file")
    # Guard against context.file_name being None (e.g. reset from a previous scenario)
    if hasattr(context, "file_name") and context.file_name and os.path.exists(context.file_name):
        os.remove(context.file_name)

    # Remove any temporary files generated during the scenario
    for temp_file in os.listdir("."):
        if temp_file.startswith("genericNonCustomisableName") or temp_file.startswith(
            "temp_image_"
        ):
            try:
                os.remove(temp_file)
            except Exception:
                pass

    # Reset all per-scenario state so stale handles don't bleed into the next scenario
    context.files = {}
    context.multi_files = []
    context.json_parts = {}
    context.request_data = None
    # JWT auth state
    context.jwt_token = None
    context.original_jwt_token = None
    context._status_ok = False