61 lines
2.6 KiB
Python
61 lines
2.6 KiB
Python
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import httpx
|
||
|
|
|
||
|
|
from atlas_portal import provisioning_tasks as tasks
|
||
|
|
|
||
|
|
|
||
|
|
class DummyResult:
|
||
|
|
def __init__(self, rows=None) -> None:
|
||
|
|
self.rows = rows or []
|
||
|
|
|
||
|
|
def fetchall(self):
|
||
|
|
return self.rows
|
||
|
|
|
||
|
|
|
||
|
|
class DummyConn:
|
||
|
|
def __init__(self, rows=None) -> None:
|
||
|
|
self.rows = rows or []
|
||
|
|
self.executed: list[tuple[str, object | None]] = []
|
||
|
|
|
||
|
|
def execute(self, query: str, params: object | None = None) -> DummyResult:
|
||
|
|
self.executed.append((query, params))
|
||
|
|
return DummyResult(self.rows)
|
||
|
|
|
||
|
|
|
||
|
|
def test_task_row_helpers_are_idempotent_and_status_based() -> None:
|
||
|
|
conn = DummyConn(rows=[{"task": "keycloak_user", "status": "ok"}, {"task": 7, "status": "ignored"}])
|
||
|
|
|
||
|
|
tasks.upsert_task(conn, "code", "keycloak_user", "ok", "created")
|
||
|
|
tasks.ensure_task_rows(conn, "code", ["keycloak_user", "mailu_sync"])
|
||
|
|
tasks.ensure_task_rows(conn, "code", [])
|
||
|
|
|
||
|
|
assert conn.executed[0][1] == ("code", "keycloak_user", "ok", "created")
|
||
|
|
assert conn.executed[1][1] == ("code", ["keycloak_user", "mailu_sync"])
|
||
|
|
assert tasks.task_statuses(conn, "code") == {"keycloak_user": "ok"}
|
||
|
|
assert tasks.all_tasks_ok(conn, "code", ["keycloak_user"]) is True
|
||
|
|
assert tasks.all_tasks_ok(conn, "code", ["keycloak_user", "mailu_sync"]) is False
|
||
|
|
|
||
|
|
|
||
|
|
def test_safe_error_detail_prefers_actionable_messages() -> None:
|
||
|
|
assert tasks.safe_error_detail(RuntimeError(" explicit failure "), "fallback") == "explicit failure"
|
||
|
|
assert tasks.safe_error_detail(RuntimeError(" "), "fallback") == "fallback"
|
||
|
|
assert tasks.safe_error_detail(httpx.TimeoutException("slow"), "fallback") == "timeout"
|
||
|
|
assert tasks.safe_error_detail(ValueError("bad"), "fallback") == "fallback"
|
||
|
|
|
||
|
|
|
||
|
|
def test_safe_error_detail_formats_http_status_payloads() -> None:
|
||
|
|
request = httpx.Request("GET", "https://example.invalid")
|
||
|
|
response = httpx.Response(409, request=request, json={"errorMessage": " duplicate user "})
|
||
|
|
exc = httpx.HTTPStatusError("conflict", request=request, response=response)
|
||
|
|
|
||
|
|
assert tasks.safe_error_detail(exc, "fallback") == "http 409: duplicate user"
|
||
|
|
|
||
|
|
text_response = httpx.Response(502, request=request, content=b" upstream offline ")
|
||
|
|
text_exc = httpx.HTTPStatusError("bad gateway", request=request, response=text_response)
|
||
|
|
assert tasks.safe_error_detail(text_exc, "fallback") == "http 502: upstream offline"
|
||
|
|
|
||
|
|
string_response = httpx.Response(400, request=request, json=" plain text ")
|
||
|
|
string_exc = httpx.HTTPStatusError("bad request", request=request, response=string_response)
|
||
|
|
assert tasks.safe_error_detail(string_exc, "fallback") == "http 400: plain text"
|