payfrit-qa/shared/lib/test_helpers.py

"""
test_helpers.py — Shared Python test utilities for Payfrit QA
Usage: from shared.lib.test_helpers import TestRunner
"""

import json
import sys
import time
import urllib.request
import urllib.error
import ssl
from dataclasses import dataclass, field
from typing import Optional


# Skip SSL verification for dev/self-signed certs
SSL_CTX = ssl.create_default_context()
SSL_CTX.check_hostname = False
SSL_CTX.verify_mode = ssl.CERT_NONE


@dataclass
class TestResult:
    name: str
    status: str  # pass, fail, skip
    detail: str = ""
    response_ms: float = 0


class TestRunner:
    """Lightweight test runner with console + JSON output."""

    def __init__(self, suite_name: str, json_mode: bool = False):
        self.suite_name = suite_name
        self.json_mode = json_mode
        self.results: list[TestResult] = []

    def pass_(self, name: str, detail: str = ""):
        r = TestResult(name, "pass", detail)
        self.results.append(r)
        if not self.json_mode:
            print(f"  ✅ {name}{f' — {detail}' if detail else ''}")

    def fail(self, name: str, detail: str = ""):
        r = TestResult(name, "fail", detail)
        self.results.append(r)
        if not self.json_mode:
            print(f"  ❌ {name}{f' — {detail}' if detail else ''}")

    def skip(self, name: str, reason: str = ""):
        r = TestResult(name, "skip", reason)
        self.results.append(r)
        if not self.json_mode:
            print(f"  ⏭️  {name}{f' — {reason}' if reason else ''}")

    def section(self, title: str):
        if not self.json_mode:
            print(f"\n━━━ {title} ━━━")

    def http_get(self, url: str, timeout: int = 10) -> dict:
        """Returns dict with keys: status, body, json, time_ms, error"""
        start = time.time()
        try:
            req = urllib.request.Request(url, method="GET")
            with urllib.request.urlopen(req, timeout=timeout, context=SSL_CTX) as resp:
                body = resp.read().decode("utf-8", errors="replace")
                elapsed = (time.time() - start) * 1000
                result = {"status": resp.status, "body": body, "time_ms": elapsed, "error": None}
                try:
                    result["json"] = json.loads(body)
                except json.JSONDecodeError:
                    result["json"] = None
                return result
        except urllib.error.HTTPError as e:
            elapsed = (time.time() - start) * 1000
            body = e.read().decode("utf-8", errors="replace") if e.fp else ""
            result = {"status": e.code, "body": body, "time_ms": elapsed, "error": str(e)}
            try:
                result["json"] = json.loads(body)
            except (json.JSONDecodeError, Exception):
                result["json"] = None
            return result
        except Exception as e:
            elapsed = (time.time() - start) * 1000
            return {"status": 0, "body": "", "json": None, "time_ms": elapsed, "error": str(e)}

    def http_post(self, url: str, data: dict = None, timeout: int = 10) -> dict:
        """POST JSON, returns same dict as http_get."""
        payload = json.dumps(data or {}).encode("utf-8")
        start = time.time()
        try:
            req = urllib.request.Request(url, data=payload, method="POST",
                                         headers={"Content-Type": "application/json"})
            with urllib.request.urlopen(req, timeout=timeout, context=SSL_CTX) as resp:
                body = resp.read().decode("utf-8", errors="replace")
                elapsed = (time.time() - start) * 1000
                result = {"status": resp.status, "body": body, "time_ms": elapsed, "error": None}
                try:
                    result["json"] = json.loads(body)
                except json.JSONDecodeError:
                    result["json"] = None
                return result
        except urllib.error.HTTPError as e:
            elapsed = (time.time() - start) * 1000
            body = e.read().decode("utf-8", errors="replace") if e.fp else ""
            result = {"status": e.code, "body": body, "time_ms": elapsed, "error": str(e)}
            try:
                result["json"] = json.loads(body)
            except (json.JSONDecodeError, Exception):
                result["json"] = None
            return result
        except Exception as e:
            elapsed = (time.time() - start) * 1000
            return {"status": 0, "body": "", "json": None, "time_ms": elapsed, "error": str(e)}

    def summary(self) -> int:
        """Print summary, return exit code (0=all pass, 1=any fail)."""
        passes = sum(1 for r in self.results if r.status == "pass")
        fails = sum(1 for r in self.results if r.status == "fail")
        skips = sum(1 for r in self.results if r.status == "skip")
        total = len(self.results)

        if self.json_mode:
            out = {
                "suite": self.suite_name,
                "total": total, "pass": passes, "fail": fails, "skip": skips,
                "results": [{"name": r.name, "status": r.status, "detail": r.detail} for r in self.results]
            }
            print(json.dumps(out, indent=2))
        else:
            print(f"\n{'━' * 42}")
            print(f"  TOTAL: {total}  |  PASS: {passes}  |  FAIL: {fails}  |  SKIP: {skips}")
            print(f"{'━' * 42}")

        return 0 if fails == 0 else 1


# Environment config
ENVS = {
    "payfrit": {
        "dev": "https://dev.payfrit.com/api",
        "prod": "https://biz.payfrit.com/api",
    },
    "grubflip": {
        "dev": "https://dev.grubflip.com/api",
        "prod": "https://api.grubflip.com",
    },
}


def get_base_url(project: str, env: str) -> str:
    return ENVS.get(project, {}).get(env, "")