File size: 5,557 Bytes
948a05a | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 | """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!")
|