import builtins import pytest from proofframe.config import ConfigurationError, Settings from proofframe.models import Campaign, CampaignManifest from proofframe.providers import GenblazeMediaProvider, create_media_provider from proofframe.storage import B2StorageBackend, create_storage_backend def test_b2_backend_requires_explicit_config(): settings = Settings(storage_backend="b2") with pytest.raises(ConfigurationError, match="B2 storage was requested"): create_storage_backend(settings) def test_genblaze_backend_requires_explicit_config(): settings = Settings(generation_backend="genblaze") with pytest.raises(ConfigurationError, match="Genblaze generation was requested"): create_media_provider(settings) def test_genblaze_provider_fails_closed_without_package(monkeypatch): real_import = builtins.__import__ def fake_import(name, globals=None, locals=None, fromlist=(), level=0): if name in {"genblaze_core", "genblaze_gmicloud"}: raise ModuleNotFoundError(name) return real_import(name, globals, locals, fromlist, level) provider = GenblazeMediaProvider( api_key="test-key", image_model="test-model", base_url="http://localhost:8800/v1", ) campaign = Campaign(title="Launch", brief="Generate one safe image.") monkeypatch.setattr(builtins, "__import__", fake_import) with pytest.raises(ConfigurationError, match="Genblaze generation requires"): provider.generate(campaign) def test_openai_genblaze_provider_fails_closed_without_package(monkeypatch): real_import = builtins.__import__ def fake_import(name, globals=None, locals=None, fromlist=(), level=0): if name == "genblaze_openai": raise ModuleNotFoundError(name) return real_import(name, globals, locals, fromlist, level) provider = GenblazeMediaProvider( api_key="test-key", image_model="gpt-image-1", genblaze_provider="openai", ) campaign = Campaign(title="Launch", brief="Generate one safe image.") monkeypatch.setattr(builtins, "__import__", fake_import) with pytest.raises(ConfigurationError, match="GENBLAZE_PROVIDER=openai requires"): provider.generate(campaign) def test_genblaze_settings_parse_official_defaults(): settings = Settings.from_env( { "PROOFFRAME_GENERATION_BACKEND": "genblaze", "PROOFFRAME_STORAGE_BACKEND": "b2", "B2_ENDPOINT_URL": "https://s3.us-west-004.backblazeb2.com", "B2_BUCKET": "proof-bucket", "B2_KEY_ID": "key-id", "B2_APPLICATION_KEY": "application-key", "GMI_API_KEY": "test-key", "GENBLAZE_IMAGE_MODEL": "seedream-5.0-lite", "GENBLAZE_TIMEOUT_SECONDS": "240", } ) assert settings.genblaze_base_url == "" assert settings.genblaze_aspect_ratio == "16:9" assert settings.genblaze_timeout_seconds == 240 assert settings.b2_region_for_backblaze() == "us-west-004" provider = create_media_provider(settings) assert isinstance(provider, GenblazeMediaProvider) assert provider.b2_sink_enabled is True assert provider.b2_region == "us-west-004" def test_genblaze_settings_parse_openai_provider(): settings = Settings.from_env( { "PROOFFRAME_GENERATION_BACKEND": "genblaze", "GENBLAZE_PROVIDER": "openai", "OPENAI_API_KEY": "test-key", "GENBLAZE_IMAGE_MODEL": "gpt-image-1", } ) assert settings.genblaze_provider == "openai" assert settings.genblaze_provider_key() == "test-key" provider = create_media_provider(settings) assert isinstance(provider, GenblazeMediaProvider) assert provider.genblaze_provider == "openai" assert provider.api_key == "test-key" def test_openai_provider_does_not_accept_non_openai_aliases_as_key(): settings = Settings.from_env( { "PROOFFRAME_GENERATION_BACKEND": "genblaze", "GENBLAZE_PROVIDER": "openai", "GENBLAZE_API_KEY": "wrong-provider-key", "GMI_API_KEY": "wrong-provider-key", "GENBLAZE_IMAGE_MODEL": "gpt-image-1", } ) assert settings.genblaze_provider_key() == "" with pytest.raises(ConfigurationError, match="OPENAI_API_KEY"): create_media_provider(settings) def test_local_genblaze_provider_requires_no_api_key_and_generates_asset(): settings = Settings.from_env( { "PROOFFRAME_GENERATION_BACKEND": "genblaze", "GENBLAZE_PROVIDER": "local", "GENBLAZE_IMAGE_MODEL": "local-svg-v1", } ) assert settings.genblaze_provider == "local" assert settings.genblaze_provider_key() == "" assert settings.genblaze_configured() is True provider = create_media_provider(settings) campaign = Campaign(title="Launch", brief="Generate one safe image.") media = provider.generate(campaign, count=1)[0] assert isinstance(provider, GenblazeMediaProvider) assert provider.genblaze_provider == "local" assert media.provider == "genblaze/local-image" assert media.model == "local-svg-v1" assert media.content_type == "image/svg+xml" assert media.generation_metadata["genblaze_manifest_verified"] == "True" def test_genblaze_b2_sink_builds_from_official_packages_without_network(): provider = GenblazeMediaProvider( api_key="test-key", image_model="seedream-5.0-lite", b2_sink_enabled=True, b2_bucket="proof-bucket", b2_key_id="key-id", b2_application_key="application-key", b2_region="us-west-004", ) sink, read_backend = provider._build_genblaze_b2_sink("cmp_test", preflight=False) assert sink is not None assert read_backend is not None provider._close_if_possible(read_backend) provider._close_if_possible(sink) def test_genblaze_fetch_asset_from_file_url(tmp_path): image_path = tmp_path / "asset.png" image_path.write_bytes(b"fake-png") data, content_type, filename = GenblazeMediaProvider._fetch_asset( image_path.as_uri(), "cmp_test", 2, ) assert data == b"fake-png" assert content_type == "image/png" assert filename == "cmp_test-genblaze-2.png" def test_genblaze_fetch_asset_uses_b2_backend_for_private_sink_url(): class FakeReadBackend: def __init__(self): self.keys = [] def key_from_url(self, url): assert url == "https://s3.us-west-004.backblazeb2.com/proof-bucket/genblaze/a.png" return "genblaze/a.png" def get(self, key): self.keys.append(key) return b"private-b2-image" backend = FakeReadBackend() data, content_type, filename = GenblazeMediaProvider._fetch_asset( "https://s3.us-west-004.backblazeb2.com/proof-bucket/genblaze/a.png", "cmp_test", 3, storage_backend=backend, ) assert data == b"private-b2-image" assert content_type == "image/png" assert filename == "cmp_test-genblaze-3.png" assert backend.keys == ["genblaze/a.png"] class FakeS3Client: def __init__(self): self.puts = [] def put_object(self, **kwargs): self.puts.append(kwargs) def test_b2_backend_puts_media_and_manifest_with_fake_client(): fake = FakeS3Client() backend = B2StorageBackend( endpoint_url="https://s3.us-west-004.backblazeb2.com", bucket="proof-bucket", key_id="key-id", application_key="application-key", public_base_url="https://cdn.example.test", client=fake, ) campaign = Campaign(title="Launch", brief="Generate one safe image.") media = create_media_provider(Settings()).generate(campaign, count=1)[0] stored_media = backend.put_media(campaign.id, media) assert stored_media.storage_backend == "b2" assert stored_media.storage_key.startswith(f"campaigns/{campaign.id}/media/") assert stored_media.public_url == f"https://cdn.example.test/{stored_media.storage_key}" manifest = CampaignManifest(campaign=campaign, assets=[]) stored_manifest = backend.put_manifest(campaign.id, manifest) assert stored_manifest.storage_key == ( f"campaigns/{campaign.id}/manifests/{campaign.id}-manifest.json" ) assert len(fake.puts) == 2 assert fake.puts[0]["Bucket"] == "proof-bucket" assert fake.puts[0]["Metadata"]["sha256"] == stored_media.sha256 def test_b2_backend_normalizes_endpoint_without_scheme(): backend = B2StorageBackend( endpoint_url="s3.us-west-004.backblazeb2.com", bucket="proof-bucket", key_id="key-id", application_key="application-key", client=FakeS3Client(), ) assert backend.endpoint_url == "https://s3.us-west-004.backblazeb2.com"