482ac3377d
- Add CI workflow running Ruff lint/format checks - Run full E2E cycle (Docker Compose + Playwright + tests) - Refactor code formatting to satisfy Ruff (line breaks, readability) - Use sys.executable in tests for interpreter-agnostic execution https://chatgpt.com/share/694a7f81-d96c-800f-88cb-7b25b4cdfe1a
126 lines
4.0 KiB
Python
126 lines
4.0 KiB
Python
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
import urllib.error
|
|
|
|
from .errors import MatomoNotReadyError, TokenCreationError
|
|
from .http import HttpClient
|
|
|
|
|
|
def _md5(text: str) -> str:
|
|
return hashlib.md5(text.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def _try_json(body: str) -> object:
|
|
try:
|
|
return json.loads(body)
|
|
except json.JSONDecodeError as exc:
|
|
raise TokenCreationError(f"Invalid JSON from Matomo API: {body[:400]}") from exc
|
|
|
|
|
|
def _dbg(msg: str, enabled: bool) -> None:
|
|
if enabled:
|
|
# Keep stdout clean (tests expect only token on stdout).
|
|
print(msg, file=sys.stderr)
|
|
|
|
|
|
class MatomoApi:
|
|
def __init__(self, *, client: HttpClient, debug: bool = False):
|
|
self.client = client
|
|
self.debug = debug
|
|
|
|
def assert_ready(self, timeout: int = 10) -> None:
|
|
"""
|
|
Minimal readiness check: Matomo UI should be reachable and look like Matomo.
|
|
"""
|
|
try:
|
|
status, body = self.client.get("/", {})
|
|
except Exception as exc: # pragma: no cover
|
|
raise MatomoNotReadyError(f"Matomo not reachable: {exc}") from exc
|
|
|
|
_dbg(f"[ready] GET / -> HTTP {status}", self.debug)
|
|
|
|
html = (body or "").lower()
|
|
if "matomo" not in html and "piwik" not in html:
|
|
raise MatomoNotReadyError("Matomo UI not detected at base URL")
|
|
|
|
def login_via_logme(self, admin_user: str, admin_password: str) -> None:
|
|
"""
|
|
Create an authenticated Matomo session (cookie jar) using Login controller.
|
|
Matomo accepts md5 hashed password in `password` parameter for action=logme.
|
|
"""
|
|
md5_password = _md5(admin_password)
|
|
try:
|
|
status, body = self.client.get(
|
|
"/index.php",
|
|
{
|
|
"module": "Login",
|
|
"action": "logme",
|
|
"login": admin_user,
|
|
"password": md5_password,
|
|
},
|
|
)
|
|
_dbg(f"[auth] logme HTTP {status} body[:120]={body[:120]!r}", self.debug)
|
|
except urllib.error.HTTPError as exc:
|
|
# Even 4xx/5xx can still set cookies; continue and let the API call validate.
|
|
try:
|
|
err_body = exc.read().decode("utf-8", errors="replace")
|
|
except Exception:
|
|
err_body = ""
|
|
_dbg(
|
|
f"[auth] logme HTTPError {exc.code} body[:120]={err_body[:120]!r}",
|
|
self.debug,
|
|
)
|
|
|
|
def create_app_specific_token(
|
|
self,
|
|
*,
|
|
admin_user: str,
|
|
admin_password: str,
|
|
description: str,
|
|
) -> str:
|
|
"""
|
|
Create an app-specific token using an authenticated session (cookies),
|
|
not UsersManager.getTokenAuth (not available in Matomo 5.3.x images).
|
|
"""
|
|
env_token = os.environ.get("MATOMO_BOOTSTRAP_TOKEN_AUTH")
|
|
if env_token:
|
|
_dbg(
|
|
"[auth] Using MATOMO_BOOTSTRAP_TOKEN_AUTH from environment.", self.debug
|
|
)
|
|
return env_token
|
|
|
|
self.login_via_logme(admin_user, admin_password)
|
|
|
|
status, body = self.client.post(
|
|
"/index.php",
|
|
{
|
|
"module": "API",
|
|
"method": "UsersManager.createAppSpecificTokenAuth",
|
|
"userLogin": admin_user,
|
|
"passwordConfirmation": admin_password,
|
|
"description": description,
|
|
"format": "json",
|
|
},
|
|
)
|
|
|
|
_dbg(
|
|
f"[auth] createAppSpecificTokenAuth HTTP {status} body[:200]={body[:200]!r}",
|
|
self.debug,
|
|
)
|
|
|
|
if status != 200:
|
|
raise TokenCreationError(
|
|
f"HTTP {status} during token creation: {body[:400]}"
|
|
)
|
|
|
|
data = _try_json(body)
|
|
token = data.get("value") if isinstance(data, dict) else None
|
|
if not token:
|
|
raise TokenCreationError(f"Unexpected response from token creation: {data}")
|
|
|
|
return str(token)
|