svrnty-vision/tests/test_integration_e2e.py

"""End-to-end integration tests — hit live Spark hosts via svrnty-vision.

Run with:   pytest -m integration -v
Skip by default in CI / offline environments.

Hosts required:
  VLM   — svrnty-steev (Strix Halo) · 100.88.167.87:11434 · qwen3-vl:32b on Ollama
  FLUX  — gx10-f38f · 100.90.100.10:8188 · ComfyUI + flux2_dev_fp8mixed
"""

from __future__ import annotations

import base64
import io
import os
from decimal import Decimal

import httpx
import pytest
from PIL import Image

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

BASE_URL = os.environ.get("SVRNTY_VISION_URL", "http://localhost:8092")
VLM_HOST = "100.88.167.87"
FLUX_HOST = "100.90.100.10"

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _make_png_b64(color: tuple[int, int, int] = (220, 80, 60), size: int = 128) -> str:
    img = Image.new("RGB", (size, size), color=color)
    buf = io.BytesIO()
    img.save(buf, format="PNG")
    return base64.b64encode(buf.getvalue()).decode("ascii")


def _host_reachable(host: str, port: int, timeout: float = 2.0) -> bool:
    import socket
    try:
        with socket.create_connection((host, port), timeout=timeout):
            return True
    except OSError:
        return False


# ---------------------------------------------------------------------------
# Skip conditions
# ---------------------------------------------------------------------------

vlm_available = pytest.mark.skipif(
    not _host_reachable(VLM_HOST, 11434),
    reason=f"VLM host {VLM_HOST}:11434 (svrnty-steev Ollama) not reachable",
)

flux_available = pytest.mark.skipif(
    not _host_reachable(FLUX_HOST, 8188),
    reason=f"FLUX host {FLUX_HOST}:8188 (gx10 ComfyUI) not reachable",
)

gateway_available = pytest.mark.skipif(
    not _host_reachable("127.0.0.1", 8092),
    reason="svrnty-vision gateway not running on localhost:8092",
)

# ---------------------------------------------------------------------------
# Gateway health
# ---------------------------------------------------------------------------


@pytest.mark.integration
@gateway_available
def test_gateway_healthz() -> None:
    resp = httpx.get(f"{BASE_URL}/healthz", timeout=5)
    assert resp.status_code == 200
    body = resp.json()
    assert body["status"] == "ok"
    assert "version" in body


# ---------------------------------------------------------------------------
# VLM — Qwen3-VL 32B on svrnty-steev
# ---------------------------------------------------------------------------


@pytest.mark.integration
@gateway_available
@vlm_available
def test_vlm_analyze_raw_mode_returns_text() -> None:
    """Raw mode: VLM describes the image freely — no score parsing."""
    resp = httpx.post(
        f"{BASE_URL}/vlm/analyze",
        json={
            "image_base64": _make_png_b64((220, 80, 60)),
            "brand_context": "Describe what you see in this image.",
            "rubric_mode": "raw",
        },
        timeout=60,
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["rubric_mode"] == "raw"
    assert body["brand_fit_score"] is None
    assert isinstance(body["raw_scores_json"], str)
    assert len(body["raw_scores_json"]) > 0
    assert "qwen" in body["model_id"].lower()


@pytest.mark.integration
@gateway_available
@vlm_available
def test_vlm_analyze_polished_returns_scores() -> None:
    """Polished mode: VLM returns brand_fit + visual_polish 0–5 scores."""
    resp = httpx.post(
        f"{BASE_URL}/vlm/analyze",
        json={
            "image_base64": _make_png_b64((50, 120, 200)),
            "brand_context": "Modern tech brand — clean, minimal, confident.",
            "rubric_mode": "polished",
        },
        timeout=120,
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["rubric_mode"] == "polished"

    brand_fit = Decimal(str(body["brand_fit_score"]))
    visual_polish = Decimal(str(body["visual_polish_score"]))
    assert Decimal("0") <= brand_fit <= Decimal("5"), f"brand_fit out of range: {brand_fit}"
    assert Decimal("0") <= visual_polish <= Decimal("5"), f"visual_polish out of range: {visual_polish}"
    assert isinstance(body["justification"], str)
    assert len(body["justification"]) > 0


@pytest.mark.integration
@gateway_available
@vlm_available
def test_vlm_analyze_ugc_mode() -> None:
    """UGC mode: same structure as polished, different rubric framing."""
    resp = httpx.post(
        f"{BASE_URL}/vlm/analyze",
        json={
            "image_base64": _make_png_b64((80, 180, 80)),
            "brand_context": "Fresh food delivery — organic, home-style.",
            "rubric_mode": "ugc",
        },
        timeout=120,
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()
    assert body["rubric_mode"] == "ugc"
    assert body["brand_fit_score"] is not None


# ---------------------------------------------------------------------------
# FLUX — ComfyUI on gx10-f38f
# ---------------------------------------------------------------------------


@pytest.mark.integration
@gateway_available
@flux_available
def test_flux_render_returns_valid_png() -> None:
    """Minimal FLUX render — 4 steps for speed, verifies PNG round-trip."""
    resp = httpx.post(
        f"{BASE_URL}/flux/render",
        json={
            "prompt": "a plain white circle on black background",
            "width": 512,
            "height": 512,
            "steps": 4,
            "guidance": 2.5,
        },
        timeout=300,
    )
    assert resp.status_code == 200, resp.text
    body = resp.json()

    assert "image_base64" in body
    assert body["content_type"] == "image/png"
    assert body["provider"] == "local"
    assert isinstance(body["duration_ms"], int)
    assert body["duration_ms"] > 0

    raw = base64.b64decode(body["image_base64"])
    img = Image.open(io.BytesIO(raw))
    assert img.width == 512
    assert img.height == 512


@pytest.mark.integration
@gateway_available
@flux_available
def test_flux_render_seeds_produce_different_images() -> None:
    """Two renders with different prompts → different images (non-trivial output)."""
    def render(prompt: str) -> bytes:
        resp = httpx.post(
            f"{BASE_URL}/flux/render",
            json={"prompt": prompt, "width": 512, "height": 512, "steps": 4},
            timeout=300,
        )
        assert resp.status_code == 200
        return base64.b64decode(resp.json()["image_base64"])

    img_a = render("solid red background, nothing else")
    img_b = render("solid blue background, nothing else")
    assert img_a != img_b, "Two different prompts produced identical output — likely cached/deduped"


# ---------------------------------------------------------------------------
# Palette — in-process (Pillow)
# ---------------------------------------------------------------------------


@pytest.mark.integration
@gateway_available
def test_palette_extract_live() -> None:
    """Palette extraction is in-process — always passes when gateway is up."""
    resp = httpx.post(
        f"{BASE_URL}/palette/extract",
        json={"image_base64": _make_png_b64((200, 50, 50)), "color_count": 4},
        timeout=10,
    )
    assert resp.status_code == 200
    body = resp.json()
    r, g, b = body["dominant"]
    assert r > 150, "dominant color should be red-dominant"
    assert body["color_count"] <= 4


# ---------------------------------------------------------------------------
# Rembg — in-process
# ---------------------------------------------------------------------------


@pytest.mark.integration
@gateway_available
def test_rembg_cutout_live() -> None:
    """Background removal — always passes when gateway is up (model downloads on first call)."""
    resp = httpx.post(
        f"{BASE_URL}/rembg/cutout",
        json={"image_base64": _make_png_b64()},
        timeout=120,  # first call downloads u2net ONNX model
    )
    assert resp.status_code == 200
    body = resp.json()
    raw = base64.b64decode(body["image_base64"])
    img = Image.open(io.BytesIO(raw))
    assert img.mode == "RGBA"


# ---------------------------------------------------------------------------
# Error surface — gateway must return correct HTTP codes
# ---------------------------------------------------------------------------


@pytest.mark.integration
@gateway_available
def test_vlm_analyze_missing_image_returns_400() -> None:
    resp = httpx.post(
        f"{BASE_URL}/vlm/analyze",
        json={"brand_context": "test", "rubric_mode": "raw"},
        timeout=10,
    )
    assert resp.status_code == 400


@pytest.mark.integration
@gateway_available
def test_flux_render_missing_prompt_returns_400() -> None:
    resp = httpx.post(f"{BASE_URL}/flux/render", json={"width": 512, "height": 512}, timeout=10)
    assert resp.status_code == 400