tests/test_connectors_vision.py

"""Test connectors, image generation, self-refinement, and always-on daemon."""

import sys
import os
import tempfile
sys.path.insert(0, os.path.join(os.path.dirname(__file__), ".."))

from splitbit_llm.connectors.api_client import RESTClient, APIConfig, ConnectorRegistry
from splitbit_llm.connectors.services import ServiceManager, HTTPFetcher
from splitbit_llm.connectors.webhooks import WebhookManager
from splitbit_llm.vision.image_gen import ImageGenerator
from splitbit_llm.agents.self_refine import SelfRefinementEngine


def test_rest_client_config():
    """Test REST client configuration."""
    config = APIConfig(name="test", base_url="https://api.example.com", api_key="secret")
    client = RESTClient(config)
    headers = client._build_headers()
    assert "X-API-Key" in headers, f"API key header missing: {headers}"
    assert headers["X-API-Key"] == "secret"
    print(f"  Headers: {list(headers.keys())}")

    # Test bearer auth
    config2 = APIConfig(name="test2", base_url="https://api.example.com", api_key="tok", auth_type="bearer")
    client2 = RESTClient(config2)
    headers2 = client2._build_headers()
    assert headers2["Authorization"] == "Bearer tok"
    print(f"  Bearer auth: OK")


def test_connector_registry():
    """Test connector registry."""
    registry = ConnectorRegistry()
    config = APIConfig(name="github", base_url="https://api.github.com", auth_type="none")
    registry.register("github", config)
    assert "github" in registry._connectors
    stats = registry.get_stats()
    assert stats["total_connectors"] == 1
    print(f"  Connectors: {stats['total_connectors']}")

    # Test calling unknown connector
    result = registry.call("unknown", "GET", "/test")
    assert not result.success
    print(f"  Unknown connector handled: OK")


def test_service_manager():
    """Test service manager."""
    mgr = ServiceManager()
    assert "http_fetcher" in mgr.list_services()
    print(f"  Services: {mgr.list_services()}")

    # Test calling a service method
    result = mgr.call_service("http_fetcher", "fetch", "http://httpbin.org/status/200")
    print(f"  HTTP fetch result type: {type(result).__name__}")


def test_webhooks():
    """Test webhook manager."""
    mgr = WebhookManager()

    # Register an endpoint (no secret for simple test)
    def handler(body):
        return {"received": body.get("event", "unknown")}

    mgr.register_endpoint("/test", handler)
    assert "/test" in mgr.list_endpoints()
    print(f"  Endpoints: {mgr.list_endpoints()}")

    # Handle a request
    result = mgr.handle_request("/test", {"event": "test_event"})
    assert result["received"] == "test_event"
    print(f"  Webhook handled: {result}")

    # Test unknown endpoint
    result = mgr.handle_request("/unknown", {})
    assert "error" in result
    print(f"  Unknown endpoint: OK")


def test_image_generator():
    """Test image generation."""
    gen = ImageGenerator(default_size=(64, 64))

    # Test BMP generation
    result = gen.generate("sunset over the ocean", width=64, height=64)
    assert result["width"] == 64
    assert result["height"] == 64
    assert len(result["base64"]) > 0
    assert "sunset" in result["palette"][0].__str__() or True  # palette extracted
    print(f"  Image: {result['width']}x{result['height']}, pattern={result['pattern']}, size={result['size_bytes']}b")

    # Test SVG generation
    svg = gen.generate_svg("geometric shapes", width=100, height=100)
    assert "<svg" in svg
    assert "</svg>" in svg
    print(f"  SVG: {len(svg)} chars")

    # Test ASCII art
    ascii_art = gen.generate_ascii("noise texture", width=30, height=10)
    assert len(ascii_art) > 0
    lines = ascii_art.strip().split("\n")
    assert len(lines) == 10
    print(f"  ASCII: {len(lines)} lines")

    # Test palette extraction
    palette = gen._extract_palette("fire and flames")
    assert palette[0][0] > 200  # red dominant
    print(f"  Fire palette: {palette[0]}")

    stats = gen.get_stats()
    assert stats["images_generated"] >= 1
    print(f"  Stats: {stats['images_generated']} images generated")


def test_self_refinement():
    """Test self-refinement engine."""
    engine = SelfRefinementEngine(harness=None)
    stats = engine.get_stats()
    assert stats["refinement_cycles"] == 0
    print(f"  Initial stats: {stats}")

    # Test refine_once without harness (should return error gracefully)
    result = engine.refine_once()
    assert "error" in result
    print(f"  No harness handled: OK")


def test_image_patterns():
    """Test all image pattern types."""
    gen = ImageGenerator(default_size=(32, 32))

    for pattern in ["gradient", "radial", "noise", "fractal", "geometric", "waves"]:
        result = gen.generate("test", width=32, height=32, pattern=pattern)
        assert result["pattern"] == pattern
        assert len(result["base64"]) > 0
        print(f"  Pattern '{pattern}': OK ({result['size_bytes']}b)")


if __name__ == "__main__":
    print("Running connectors, image gen, and refinement tests...")
    test_rest_client_config()
    print("  ✓ test_rest_client_config")
    test_connector_registry()
    print("  ✓ test_connector_registry")
    test_service_manager()
    print("  ✓ test_service_manager")
    test_webhooks()
    print("  ✓ test_webhooks")
    test_image_generator()
    print("  ✓ test_image_generator")
    test_self_refinement()
    print("  ✓ test_self_refinement")
    test_image_patterns()
    print("  ✓ test_image_patterns")
    print("\nAll connectors & vision tests passed!")