d81a8f41b4
- Implemented a comprehensive end-to-end testing script for the Sag module's HTTP API, covering various functionalities including case creation, updates, and file uploads. - Introduced a safe HTML sanitizer utility to ensure safe rendering of HTML content in the BMC Hub UI. - Added database migrations for new features including WAN connection marking for wall outlets, permanent audit trails for supplier invoices, and dedicated permissions for the Sag module. - Created a migration center for manual subscription and invoice migrations with relevant tables and indexes. - Added tests for migration center functionalities, ensuring stability and correctness of the new features.
419 lines
18 KiB
Python
419 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
"""Reversibel end-to-end test af Sag-modulets HTTP API.
|
|
|
|
Eksempel:
|
|
BMC_TEST_PASSWORD='...' python app/modules/sag/scripts/sag_module_e2e.py \
|
|
--username admin --customer-id 1
|
|
|
|
Alternativt kan et eksisterende JWT angives via BMC_TEST_TOKEN eller --token.
|
|
Testen opretter to tydeligt navngivne testsager og sletter dem igen i finally.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import getpass
|
|
import json
|
|
import os
|
|
import sys
|
|
import time
|
|
import uuid
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
from typing import Any, Callable
|
|
from urllib.error import HTTPError, URLError
|
|
from urllib.parse import urlencode
|
|
from urllib.request import Request, urlopen
|
|
|
|
|
|
@dataclass
|
|
class Result:
|
|
name: str
|
|
status: str
|
|
detail: str = ""
|
|
duration_ms: int = 0
|
|
|
|
|
|
class ApiError(RuntimeError):
|
|
def __init__(self, method: str, path: str, status: int, body: Any):
|
|
super().__init__(f"{method} {path} -> HTTP {status}: {body}")
|
|
self.status = status
|
|
self.body = body
|
|
|
|
|
|
class SagE2E:
|
|
def __init__(self, args: argparse.Namespace):
|
|
self.args = args
|
|
self.base = args.base_url.rstrip("/")
|
|
self.token = args.token or os.getenv("BMC_TEST_TOKEN")
|
|
self.results: list[Result] = []
|
|
self.created_cases: list[int] = []
|
|
self.cleanup: list[tuple[str, Callable[[], Any]]] = []
|
|
self.run_id = f"{datetime.now():%Y%m%d-%H%M%S}-{uuid.uuid4().hex[:6]}"
|
|
self.marker = f"SAG-E2E-{self.run_id}"
|
|
|
|
def request(
|
|
self,
|
|
method: str,
|
|
path: str,
|
|
data: Any = None,
|
|
*,
|
|
expected: tuple[int, ...] = (200,),
|
|
headers: dict[str, str] | None = None,
|
|
raw: bool = False,
|
|
) -> Any:
|
|
request_headers = {"Accept": "application/json"}
|
|
if self.token:
|
|
request_headers["Authorization"] = f"Bearer {self.token}"
|
|
if headers:
|
|
request_headers.update(headers)
|
|
body = data
|
|
if data is not None and not isinstance(data, bytes):
|
|
body = json.dumps(data).encode()
|
|
request_headers["Content-Type"] = "application/json"
|
|
req = Request(self.base + path, data=body, headers=request_headers, method=method)
|
|
try:
|
|
with urlopen(req, timeout=self.args.timeout) as response:
|
|
payload = response.read()
|
|
if response.status not in expected:
|
|
raise ApiError(method, path, response.status, payload.decode(errors="replace"))
|
|
if raw:
|
|
return payload
|
|
if not payload:
|
|
return None
|
|
content_type = response.headers.get("Content-Type", "")
|
|
return json.loads(payload) if "json" in content_type else payload.decode(errors="replace")
|
|
except HTTPError as exc:
|
|
payload = exc.read().decode(errors="replace")
|
|
try:
|
|
payload = json.loads(payload)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
if exc.code in expected:
|
|
return payload
|
|
raise ApiError(method, path, exc.code, payload) from exc
|
|
except URLError as exc:
|
|
raise RuntimeError(f"Kan ikke forbinde til {self.base}: {exc.reason}") from exc
|
|
|
|
def check(self, name: str, operation: Callable[[], Any]) -> Any:
|
|
started = time.monotonic()
|
|
try:
|
|
value = operation()
|
|
elapsed = int((time.monotonic() - started) * 1000)
|
|
self.results.append(Result(name, "PASS", duration_ms=elapsed))
|
|
print(f"PASS {name} ({elapsed} ms)")
|
|
return value
|
|
except Exception as exc:
|
|
elapsed = int((time.monotonic() - started) * 1000)
|
|
detail = str(exc).replace("\n", " ")[:800]
|
|
self.results.append(Result(name, "FAIL", detail, elapsed))
|
|
print(f"FAIL {name}: {detail}")
|
|
return None
|
|
|
|
def skip(self, name: str, reason: str) -> None:
|
|
self.results.append(Result(name, "SKIP", reason))
|
|
print(f"SKIP {name}: {reason}")
|
|
|
|
def authenticate(self) -> dict[str, Any]:
|
|
if not self.token:
|
|
if not self.args.username:
|
|
raise RuntimeError("Angiv --token/BMC_TEST_TOKEN eller --username.")
|
|
password = self.args.password or os.getenv("BMC_TEST_PASSWORD")
|
|
if not password and sys.stdin.isatty():
|
|
password = getpass.getpass("Adgangskode: ")
|
|
if not password:
|
|
raise RuntimeError("Angiv --password eller BMC_TEST_PASSWORD.")
|
|
payload = {"username": self.args.username, "password": password}
|
|
if self.args.otp:
|
|
payload["otp_code"] = self.args.otp
|
|
response = self.request("POST", "/api/v1/auth/login", payload)
|
|
self.token = response.get("access_token")
|
|
if not self.token:
|
|
raise RuntimeError("Login returnerede intet access_token.")
|
|
me = self.request("GET", "/api/v1/auth/me")
|
|
print(f"Bruger: {me.get('full_name') or me.get('username') or me.get('user_id')}")
|
|
return me
|
|
|
|
def resolve_customer_id(self) -> int:
|
|
if self.args.customer_id:
|
|
return self.args.customer_id
|
|
customers = self.request("GET", "/api/v1/customers?limit=1&is_active=true")
|
|
rows = customers.get("customers", []) if isinstance(customers, dict) else customers
|
|
if not rows:
|
|
raise RuntimeError("Ingen aktiv kunde fundet. Angiv --customer-id.")
|
|
return int(rows[0]["id"])
|
|
|
|
@staticmethod
|
|
def require(value: Any, message: str) -> Any:
|
|
if not value:
|
|
raise AssertionError(message)
|
|
return value
|
|
|
|
def create_case(self, customer_id: int, suffix: str) -> dict[str, Any]:
|
|
case = self.request(
|
|
"POST",
|
|
"/api/v1/sag",
|
|
{
|
|
"titel": f"[{self.marker}] {suffix}",
|
|
"beskrivelse": "Automatisk testsag. Må slettes.",
|
|
"customer_id": customer_id,
|
|
"type": "ticket",
|
|
"status": "åben",
|
|
},
|
|
expected=(200, 201),
|
|
)
|
|
case_id = int(case["id"])
|
|
self.created_cases.append(case_id)
|
|
return case
|
|
|
|
def multipart_file(
|
|
self, field: str, filename: str, content: bytes, media_type: str
|
|
) -> tuple[bytes, str]:
|
|
boundary = f"----SagE2E{uuid.uuid4().hex}"
|
|
body = (
|
|
f"--{boundary}\r\n"
|
|
f'Content-Disposition: form-data; name="{field}"; filename="{filename}"\r\n'
|
|
f"Content-Type: {media_type}\r\n\r\n"
|
|
).encode() + content + f"\r\n--{boundary}--\r\n".encode()
|
|
return body, f"multipart/form-data; boundary={boundary}"
|
|
|
|
def run(self) -> None:
|
|
self.check("Serverens health endpoint", lambda: self.request("GET", "/health"))
|
|
if self.check("Login og aktuel bruger", self.authenticate) is None:
|
|
return
|
|
customer_id = self.check("Find testkunde", self.resolve_customer_id)
|
|
if customer_id is None:
|
|
return
|
|
|
|
parent = self.check("Opret hovedsag", lambda: self.create_case(customer_id, "Hovedsag"))
|
|
child = self.check("Opret undersag", lambda: self.create_case(customer_id, "Undersag"))
|
|
if not parent or not child:
|
|
return
|
|
parent_id, child_id = int(parent["id"]), int(child["id"])
|
|
|
|
self.check("Hent sag", lambda: self.require(
|
|
self.request("GET", f"/api/v1/sag/{parent_id}").get("id") == parent_id,
|
|
"Forkert sag returneret",
|
|
))
|
|
self.check("Sag vises i sagslisten", lambda: self.require(
|
|
any(int(row["id"]) == parent_id for row in self.request(
|
|
"GET", f"/api/v1/sag?search={self.marker}"
|
|
)),
|
|
"Testsagen blev ikke fundet i listen",
|
|
))
|
|
self.check("Redigér sag", lambda: self.require(
|
|
self.request("PATCH", f"/api/v1/sag/{parent_id}", {
|
|
"titel": f"[{self.marker}] Hovedsag redigeret",
|
|
"status": "under behandling",
|
|
"deadline": (datetime.now(timezone.utc) + timedelta(days=14)).isoformat(),
|
|
}).get("id") == parent_id,
|
|
"Sag blev ikke opdateret",
|
|
))
|
|
self.check("Gem beskrivelse", lambda: self.request(
|
|
"PATCH", f"/api/v1/sag/{parent_id}/beskrivelse",
|
|
{"beskrivelse": "<p>Sikker <strong>HTML</strong></p><script>nope()</script>"},
|
|
))
|
|
self.check("Beskrivelseshistorik", lambda: self.require(
|
|
self.request("GET", f"/api/v1/sag/{parent_id}/beskrivelse/history"),
|
|
"Historikken er tom",
|
|
))
|
|
self.check("Hent modulindstillinger", lambda: self.request(
|
|
"GET", f"/api/v1/sag/{parent_id}/modules"
|
|
))
|
|
|
|
comment = self.check("Opret kommentar", lambda: self.request(
|
|
"POST", f"/api/v1/sag/{parent_id}/kommentarer",
|
|
{"indhold": f"Kommentar {self.marker}", "er_intern": True},
|
|
expected=(200, 201),
|
|
))
|
|
self.check("Læs kommentarer", lambda: self.require(
|
|
any(self.marker in str(row.get("indhold", "")) for row in self.request(
|
|
"GET", f"/api/v1/sag/{parent_id}/kommentarer"
|
|
)),
|
|
"Kommentaren blev ikke fundet",
|
|
))
|
|
self.check("Læs tidslinje", lambda: self.request(
|
|
"GET", f"/api/v1/sag/{parent_id}/timeline"
|
|
))
|
|
|
|
todo = self.check("Opret todo", lambda: self.request(
|
|
"POST", f"/api/v1/sag/{child_id}/todo-steps",
|
|
{"title": f"Næste handling {self.marker}", "description": "E2E", "due_date": None},
|
|
expected=(200, 201),
|
|
))
|
|
if todo:
|
|
todo_id = int(todo["id"])
|
|
self.check("Markér todo som næste", lambda: self.require(
|
|
self.request("PATCH", f"/api/v1/sag/todo-steps/{todo_id}", {"is_next": True}).get("is_next"),
|
|
"Todo er ikke markeret som næste",
|
|
))
|
|
self.check("Afslut todo", lambda: self.require(
|
|
self.request("PATCH", f"/api/v1/sag/todo-steps/{todo_id}", {"is_done": True}).get("is_done"),
|
|
"Todo er ikke afsluttet",
|
|
))
|
|
self.check("Slet todo", lambda: self.request(
|
|
"DELETE", f"/api/v1/sag/todo-steps/{todo_id}"
|
|
))
|
|
|
|
relation = self.check("Opret logisk undersag-relation", lambda: self.request(
|
|
"POST", f"/api/v1/sag/{parent_id}/relationer",
|
|
{"målsag_id": child_id, "relationstype": "undersag"},
|
|
expected=(200, 201),
|
|
))
|
|
if relation:
|
|
relation_id = int(relation["id"])
|
|
self.check("Læs relationer", lambda: self.require(
|
|
any(int(row["id"]) == relation_id for row in self.request(
|
|
"GET", f"/api/v1/sag/{parent_id}/relationer"
|
|
)),
|
|
"Relationen blev ikke fundet",
|
|
))
|
|
self.check("Afvis dubletrelation", lambda: self.request(
|
|
"POST", f"/api/v1/sag/{parent_id}/relationer",
|
|
{"målsag_id": child_id, "relationstype": "undersag"},
|
|
expected=(409,),
|
|
))
|
|
self.check("Slet relation", lambda: self.request(
|
|
"DELETE", f"/api/v1/sag/{parent_id}/relationer/{relation_id}"
|
|
))
|
|
|
|
tag = self.check("Opret tag", lambda: self.request(
|
|
"POST", f"/api/v1/sag/{parent_id}/tags", {"tag_navn": self.marker},
|
|
expected=(200, 201),
|
|
))
|
|
if tag:
|
|
self.check("Læs tags", lambda: self.require(
|
|
any(row.get("tag_navn") == self.marker for row in self.request(
|
|
"GET", f"/api/v1/sag/{parent_id}/tags"
|
|
)), "Tag blev ikke fundet",
|
|
))
|
|
self.check("Slet tag", lambda: self.request(
|
|
"DELETE", f"/api/v1/sag/{parent_id}/tags/{tag['id']}"
|
|
))
|
|
|
|
buzz = self.check("Opret buzzword", lambda: self.request(
|
|
"POST", f"/api/v1/sag/{parent_id}/buzzwords",
|
|
{"buzzword": self.marker.lower()}, expected=(200, 201),
|
|
))
|
|
if buzz:
|
|
self.check("Global søgning finder buzzword", lambda: self.require(
|
|
any(int(row.get("id", -1)) == parent_id for row in self.request(
|
|
"GET", "/api/v1/search/sag?" + urlencode({"q": self.marker.lower()})
|
|
)), "Global søgning fandt ikke sagen via buzzword",
|
|
))
|
|
self.check("Slet buzzword-link", lambda: self.request(
|
|
"DELETE", f"/api/v1/sag/{parent_id}/buzzwords/{buzz['buzzword_id']}"
|
|
))
|
|
|
|
item = self.check("Opret salgs-/ordrelinje", lambda: self.request(
|
|
"POST", f"/api/v1/sag/{parent_id}/sale-items",
|
|
{"type": "sale", "description": self.marker, "quantity": 1,
|
|
"unit_price": 125, "amount": 125, "currency": "DKK", "status": "draft"},
|
|
expected=(200, 201),
|
|
))
|
|
if item:
|
|
item_id = int(item["id"])
|
|
self.check("Redigér salgs-/ordrelinje", lambda: self.require(
|
|
self.request("PATCH", f"/api/v1/sag/{parent_id}/sale-items/{item_id}",
|
|
{"status": "confirmed", "amount": 150}).get("status") == "confirmed",
|
|
"Linjen blev ikke bekræftet",
|
|
))
|
|
self.check("Slet salgs-/ordrelinje", lambda: self.request(
|
|
"DELETE", f"/api/v1/sag/{parent_id}/sale-items/{item_id}"
|
|
))
|
|
|
|
# Minimal gyldig 1x1 PNG; PNG er blandt Sag-modulets tilladte filtyper.
|
|
file_content = bytes.fromhex(
|
|
"89504e470d0a1a0a0000000d49484452000000010000000108060000001f15c489"
|
|
"0000000d49444154789c6360000000020001e221bc330000000049454e44ae426082"
|
|
)
|
|
body, content_type = self.multipart_file(
|
|
"files", f"{self.marker}.png", file_content, "image/png"
|
|
)
|
|
uploaded = self.check("Upload fil", lambda: self.request(
|
|
"POST", f"/api/v1/sag/{parent_id}/files", body,
|
|
headers={"Content-Type": content_type}, expected=(200, 201),
|
|
))
|
|
if uploaded:
|
|
file_id = int(uploaded[0]["id"])
|
|
self.check("Download fil og kontrollér indhold", lambda: self.require(
|
|
self.request("GET", f"/api/v1/sag/{parent_id}/files/{file_id}", raw=True) == file_content,
|
|
"Downloadet fil har forkert indhold",
|
|
))
|
|
self.check("Slet fil", lambda: self.request(
|
|
"DELETE", f"/api/v1/sag/{parent_id}/files/{file_id}"
|
|
))
|
|
|
|
for name, reason in (
|
|
("Mailafsendelse og mail-link", "kræver en rigtig mailkonto/mail-ID"),
|
|
("AnyDesk-forbindelse", "kræver installeret klient og gyldigt AnyDesk-ID"),
|
|
("Direkte labelprint", "kræver fysisk printer"),
|
|
("Arbejdsseddel og underskrift", "kræver aktiv work-order/token"),
|
|
("Faktura-/økonomisynkronisering", "må ikke skrive i eksternt økonomisystem fra E2E"),
|
|
):
|
|
self.skip(name, reason)
|
|
|
|
def cleanup_all(self) -> None:
|
|
if self.args.keep_data:
|
|
print("Beholder testdata (--keep-data).")
|
|
return
|
|
for case_id in reversed(self.created_cases):
|
|
try:
|
|
self.request("DELETE", f"/api/v1/sag/{case_id}", expected=(200, 204, 404))
|
|
print(f"CLEAN Sag #{case_id}")
|
|
except Exception as exc:
|
|
self.results.append(Result(f"Oprydning af sag #{case_id}", "FAIL", str(exc)))
|
|
print(f"FAIL Oprydning af sag #{case_id}: {exc}")
|
|
|
|
def write_report(self) -> None:
|
|
counts = {status: sum(r.status == status for r in self.results)
|
|
for status in ("PASS", "FAIL", "SKIP")}
|
|
report = {
|
|
"run_id": self.run_id,
|
|
"base_url": self.base,
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
"summary": counts,
|
|
"results": [asdict(result) for result in self.results],
|
|
}
|
|
if self.args.json_report:
|
|
path = Path(self.args.json_report)
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(report, ensure_ascii=False, indent=2) + "\n")
|
|
print(f"Rapport: {path.resolve()}")
|
|
print(f"\nResultat: {counts['PASS']} PASS, {counts['FAIL']} FAIL, {counts['SKIP']} SKIP")
|
|
|
|
|
|
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
|
|
parser = argparse.ArgumentParser(description="Reversibel E2E-test af BMC Hub Sag-modulet")
|
|
parser.add_argument("--base-url", default=os.getenv("BMC_BASE_URL", "http://127.0.0.1:8001"))
|
|
parser.add_argument("--token", help="JWT; alternativt BMC_TEST_TOKEN")
|
|
parser.add_argument("--username", help="Loginbrugernavn")
|
|
parser.add_argument("--password", help="Frarådes i shellhistorik; brug BMC_TEST_PASSWORD")
|
|
parser.add_argument("--otp", help="2FA-kode hvis påkrævet")
|
|
parser.add_argument("--customer-id", type=int, help="Kunde til midlertidige testsager")
|
|
parser.add_argument("--timeout", type=float, default=20.0)
|
|
parser.add_argument("--keep-data", action="store_true", help="Behold testsager ved fejlsøgning")
|
|
parser.add_argument("--json-report", default="test-results/sag-e2e.json")
|
|
return parser.parse_args(argv)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
suite = SagE2E(parse_args(argv))
|
|
print(f"Sag E2E {suite.run_id} mod {suite.base}")
|
|
try:
|
|
suite.run()
|
|
except KeyboardInterrupt:
|
|
print("\nAfbrudt.")
|
|
except Exception as exc:
|
|
suite.results.append(Result("Testkørsel", "FAIL", str(exc)))
|
|
print(f"FAIL Testkørsel: {exc}")
|
|
finally:
|
|
suite.cleanup_all()
|
|
suite.write_report()
|
|
return 1 if any(result.status == "FAIL" for result in suite.results) else 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|