Spaces:
Running
Running
| import asyncio | |
| import base64 | |
| import sys | |
| from pathlib import Path | |
| import pytest | |
| from PIL import Image | |
| sys.path.insert(0, str(Path(__file__).resolve().parents[1])) | |
| from api import vision | |
| class FakeResponse: | |
| def __init__(self, status_code=200, payload=None, content=b"png-bytes", text=""): | |
| self.status_code = status_code | |
| self._payload = payload | |
| self.content = content | |
| self.text = text | |
| def json(self): | |
| if isinstance(self._payload, Exception): | |
| raise self._payload | |
| return self._payload | |
| class FakeClient: | |
| calls = [] | |
| responses = [] | |
| def __init__(self, *args, **kwargs): | |
| self.calls = [] | |
| async def __aenter__(self): | |
| FakeClient.active = self | |
| return self | |
| async def __aexit__(self, *args): | |
| return False | |
| async def post(self, url, **kwargs): | |
| self.calls.append((url, kwargs)) | |
| FakeClient.calls.append((url, kwargs)) | |
| return FakeClient.responses.pop(0) | |
| def test_analyze_uses_hf_vqa_without_openai(monkeypatch): | |
| FakeClient.calls = [] | |
| FakeClient.responses = [FakeResponse(payload=[{"answer": "un gatto"}])] | |
| monkeypatch.setattr(vision.httpx, "AsyncClient", FakeClient) | |
| monkeypatch.delenv("OPENAI_API_KEY", raising=False) | |
| monkeypatch.delenv("OPENAI_BASE_URL", raising=False) | |
| monkeypatch.delenv("GROQ_API_KEY", raising=False) | |
| monkeypatch.delenv("GEMINI_API_KEY", raising=False) | |
| result = asyncio.run(vision.analyze_image( | |
| vision.AnalyzeImageRequest(base64_image=base64.b64encode(b"image").decode(), question="Cosa vedi?") | |
| )) | |
| assert result == {"ok": True, "description": "un gatto", "provider": "blip-vqa"} | |
| assert len(FakeClient.calls) == 1 | |
| assert vision._HF_VQA_MODEL in FakeClient.calls[0][0] | |
| assert all("openai.com" not in call[0] for call in FakeClient.calls) | |
| def test_generate_uses_huggingface(monkeypatch): | |
| class FakeInferenceClient: | |
| def __init__(self, **kwargs): | |
| self.kwargs = kwargs | |
| def text_to_image(self, **kwargs): | |
| assert kwargs["model"] == "stabilityai/stable-diffusion-3-medium-diffusers" | |
| return Image.new("RGB", (1, 1), (0, 120, 255)) | |
| monkeypatch.setattr(vision, "InferenceClient", FakeInferenceClient) | |
| result = asyncio.run(vision.generate_image(vision.GenerateImageRequest(prompt="un paesaggio"))) | |
| assert result["ok"] is True | |
| assert result["mime"] == "image/png" | |
| assert len(base64.b64decode(result["image_b64"])) > 0 | |
| def test_edit_uses_inference_client_hf(monkeypatch): | |
| class FakeInferenceClient: | |
| def __init__(self, **kwargs): | |
| self.kwargs = kwargs | |
| def image_to_image(self, **kwargs): | |
| assert kwargs["model"] == "black-forest-labs/FLUX.1-Kontext-dev" | |
| return Image.new("RGB", (1, 1), (0, 120, 255)) | |
| monkeypatch.setattr(vision, "InferenceClient", FakeInferenceClient) | |
| result = asyncio.run(vision.edit_image( | |
| vision.EditImageRequest(prompt="rendi il cielo blu", base64_image="aW1hZ2U=") | |
| )) | |
| assert result["ok"] is True | |
| assert result["model"] == "FLUX.1-Kontext-dev" | |
| assert result["mime"] == "image/png" | |
| assert len(base64.b64decode(result["image_b64"])) > 0 | |