fixed sonarqube scan issues

tools/python/common.py

@@ -9,6 +9,7 @@ import getpass
 import json
 import os
 import sys
+import traceback
 import uuid
 from dataclasses import dataclass
 from pathlib import Path
@@ -20,23 +21,30 @@ from urllib.request import Request, urlopen
 DEFAULT_BASE_URL = "https://wegwichtel.de"
 API_USER_ENV = "API_USER"
 API_PASSWORD_ENV = "API_PASSWORD"
+APPLICATION_JSON = "application/json"
+MAX_AUTH_ATTEMPTS = 3
+REQUEST_TIMEOUT_SECONDS = 120
 
 
 class ApiError(RuntimeError):
+    """Fehlerantwort der REST-API einschließlich HTTP-Status und Nutzdaten."""
+
     def __init__(self, status: int, message: str, payload: Any = None):
         super().__init__(f"HTTP {status}: {message}")
         self.status = status
         self.payload = payload
 
 
-@dataclass
+@dataclass(frozen=True)
 class Credentials:
+    """Zugangsdaten für eine nach HTTP 401 angeforderte Basic-Authentifizierung."""
+
     username: str
     password: str
 
 
 def credentials_from_environment() -> Credentials | None:
-    """Liest CI-Credentials, ohne einen Request vor dem ersten HTTP 401 zu authentifizieren."""
+    """Liest CI-Credentials, ohne den ersten Request vorab zu authentifizieren."""
     username = os.getenv(API_USER_ENV, "").strip()
     password = os.getenv(API_PASSWORD_ENV, "").strip()
     if username and password:
@@ -44,12 +52,35 @@ def credentials_from_environment() -> Credentials | None:
     return None
 
 
+def _json_bytes(value: dict[str, Any]) -> bytes:
+    return json.dumps(value, ensure_ascii=False).encode("utf-8")
+
+
+def _request_payload(
+    json_body: dict[str, Any] | None,
+    fields: dict[str, Any] | None,
+    files: dict[str, Path] | None,
+) -> tuple[bytes | None, dict[str, str]]:
+    headers = {"Accept": APPLICATION_JSON}
+    if files:
+        data, content_type = encode_multipart(fields or {}, files)
+        headers["Content-Type"] = content_type
+        return data, headers
+    if json_body is not None:
+        headers["Content-Type"] = APPLICATION_JSON
+        return _json_bytes(json_body), headers
+    if fields:
+        headers["Content-Type"] = APPLICATION_JSON
+        return _json_bytes(fields), headers
+    return None, headers
+
+
 class ApiClient:
-    """Kleine HTTP-Schicht mit dynamischer Basic-Auth-Nachfrage bei 401."""
+    """Kleine HTTP-Schicht mit dynamischer Basic-Auth-Nachfrage bei HTTP 401."""
 
     def __init__(self, base_url: str, username: str | None = None, password: str | None = None):
         self.base_url = base_url.rstrip("/") + "/"
-        self.credentials = Credentials(username, password) if username and password else None
+        self.credentials = Credentials(username.strip(), password.strip()) if username and password else None
 
     def prompt_credentials(self) -> None:
         environment_credentials = credentials_from_environment()
@@ -57,21 +88,60 @@ class ApiClient:
             self.credentials = environment_credentials
             return
 
-        default_user = self.credentials.username if self.credentials else os.getenv(API_USER_ENV, "")
-        username = input(f"Benutzername{f' [{default_user}]' if default_user else ''}: ").strip() or default_user
+        default_user = self.credentials.username if self.credentials else os.getenv(API_USER_ENV, "").strip()
+        suffix = f" [{default_user}]" if default_user else ""
+        username = input(f"Benutzername{suffix}: ").strip() or default_user
         if not username:
             raise ApiError(401, "Für den geschützten API-Zugriff ist ein Benutzername erforderlich.")
-        password = getpass.getpass("Passwort: ")
+
+        password = getpass.getpass("Passwort: ").strip()
+        if not password:
+            raise ApiError(401, "Für den geschützten API-Zugriff ist ein Passwort erforderlich.")
         self.credentials = Credentials(username, password)
 
     def _authorization_header(self) -> str | None:
         if not self.credentials:
             return None
-        token = base64.b64encode(
-            f"{self.credentials.username}:{self.credentials.password}".encode("utf-8")
-        ).decode("ascii")
+        value = f"{self.credentials.username}:{self.credentials.password}".encode("utf-8")
+        token = base64.b64encode(value).decode("ascii")
         return f"Basic {token}"
 
+    def _request_headers(self, headers: dict[str, str]) -> dict[str, str]:
+        request_headers = dict(headers)
+        authorization = self._authorization_header()
+        if authorization:
+            request_headers["Authorization"] = authorization
+        return request_headers
+
+    def _perform_request(
+        self,
+        method: str,
+        url: str,
+        data: bytes | None,
+        headers: dict[str, str],
+        expected: tuple[int, ...],
+    ) -> Any:
+        request = Request(
+            url,
+            data=data,
+            headers=self._request_headers(headers),
+            method=method.upper(),
+        )
+        with urlopen(request, timeout=REQUEST_TIMEOUT_SECONDS) as response:
+            raw = response.read()
+            if response.status not in expected:
+                raise ApiError(response.status, decode_error(raw), parse_json(raw))
+            return parse_response(raw, response.headers.get("Content-Type", ""))
+
+    def _handle_http_error(self, error: HTTPError, attempt: int) -> bool:
+        if error.code != 401 or attempt >= MAX_AUTH_ATTEMPTS:
+            return False
+        error.read()
+        error.close()
+        print("Die API verlangt Zugangsdaten (HTTP 401).", file=sys.stderr)
+        self.prompt_credentials()
+        return True
+
     def request(
         self,
         method: str,
@@ -82,42 +152,18 @@ class ApiClient:
         files: dict[str, Path] | None = None,
         expected: tuple[int, ...] = (200, 201),
     ) -> Any:
-        data: bytes | None = None
-        headers = {"Accept": "application/json"}
-
-        if files:
-            data, content_type = encode_multipart(fields or {}, files)
-            headers["Content-Type"] = content_type
-        elif json_body is not None:
-            data = json.dumps(json_body, ensure_ascii=False).encode("utf-8")
-            headers["Content-Type"] = "application/json"
-        elif fields:
-            data = json.dumps(fields, ensure_ascii=False).encode("utf-8")
-            headers["Content-Type"] = "application/json"
-
+        data, headers = _request_payload(json_body, fields, files)
         url = urljoin(self.base_url, path.lstrip("/"))
-        attempts = 0
-        while True:
-            request_headers = dict(headers)
-            authorization = self._authorization_header()
-            if authorization:
-                request_headers["Authorization"] = authorization
 
-            request = Request(url, data=data, headers=request_headers, method=method.upper())
+        attempt = 0
+        while True:
             try:
-                with urlopen(request, timeout=120) as response:
-                    raw = response.read()
-                    status = response.status
-                    if status not in expected:
-                        raise ApiError(status, decode_error(raw), parse_json(raw))
-                    return parse_response(raw, response.headers.get("Content-Type", ""))
+                return self._perform_request(method, url, data, headers, expected)
             except HTTPError as error:
-                raw = error.read()
-                if error.code == 401 and attempts < 3:
-                    print("Die API verlangt Zugangsdaten (HTTP 401).", file=sys.stderr)
-                    self.prompt_credentials()
-                    attempts += 1
+                if self._handle_http_error(error, attempt):
+                    attempt += 1
                     continue
+                raw = error.read()
                 raise ApiError(error.code, decode_error(raw), parse_json(raw)) from error
             except URLError as error:
                 raise RuntimeError(f"API nicht erreichbar: {error.reason}") from error
@@ -247,9 +293,18 @@ def print_result(result: Any) -> None:
     print(json.dumps(result, ensure_ascii=False, indent=2) if not isinstance(result, str) else result)
 
 
+def _print_failure(error: BaseException) -> None:
+    print(f"Fehler: {type(error).__name__}: {error}", file=sys.stderr)
+    traceback.print_exception(type(error), error, error.__traceback__, file=sys.stderr)
+
+
 def run(action: Callable[[], Any]) -> None:
+    """Führt eine Werkzeugaktion aus und gibt Fehler samt vollständigem Stacktrace aus."""
     try:
         print_result(action())
-    except (ApiError, RuntimeError, FileNotFoundError, ValueError) as error:
-        print(f"Fehler: {error}", file=sys.stderr)
+    except KeyboardInterrupt as error:
+        _print_failure(error)
+        raise SystemExit(130) from error
+    except Exception as error:
+        _print_failure(error)
         raise SystemExit(1) from error