import os import hashlib import json import threading from io import BytesIO from types import SimpleNamespace import google.auth import pytest import torch from PIL import Image from custom_nodes.ComfyUI_Gemini_Expanded_API import gemini_nodes from custom_nodes.ComfyUI_Gemini_Expanded_API.response_schema import SSL_GeminiResponseSchema, default_state, default_entry @pytest.fixture def structured_client(monkeypatch): node = gemini_nodes.SSL_GeminiTextPrompt for cache in (node._cache, node._client_cache, node._context_cache, node._seed_map_cache): cache.clear() captured = SimpleNamespace(calls=[], creations=[], clients=[], responses=[], cache_error=None, on_client=None) class Models: def generate_content(self, **kwargs): captured.calls.append({**kwargs, "config": kwargs["config"].model_copy(deep=True)}) response = captured.responses.pop(0) if captured.responses else [('final', '{"answer":"ok"}')] if callable(response): response = response() if isinstance(response, Exception): raise response parts = [SimpleNamespace(text=text, thought=kind == "thought", inline_data=None) for kind, text in response] return SimpleNamespace(candidates=[SimpleNamespace(content=SimpleNamespace(parts=parts))]) class Caches: def create(self, **kwargs): captured.creations.append(kwargs) if captured.cache_error: raise captured.cache_error return SimpleNamespace(name=f"cachedContents/{len(captured.creations)}") class Client: def __init__(self, **kwargs): captured.clients.append(kwargs) if captured.on_client: captured.on_client() self.models, self.caches = Models(), Caches() monkeypatch.setattr(gemini_nodes.genai, "Client", Client) yield captured for cache in (node._cache, node._client_cache, node._context_cache, node._seed_map_cache): cache.clear() @pytest.mark.parametrize("model, settings", [ ("gemini-2.5-flash", {}), ("gemini-3.1-pro-preview", {"thinking_level": "high"}), (gemini_nodes.SSL_GeminiTextPrompt.GEMINI_3_7_FLASH, {"thinking_level": "medium"}), (gemini_nodes.SSL_GeminiTextPrompt.GEMINI_3_8_FLASH, {"thinking_level": "medium"}), (gemini_nodes.SSL_GeminiTextPrompt.GEMINI_4_ARGON, {"thinking_level": "medium"}), (gemini_nodes.SSL_GeminiTextPrompt.GEMINI_4_ARGON_CYBER_PERMISSIVE, {"thinking_level": "high"}), ]) @pytest.mark.parametrize("client_config", [{}, {"vertexai_express": True}, {"use_vertexai_env": True, "vertexai_project": "test", "vertexai_location": "global"}]) def test_builder_schema_reaches_generate_content(structured_client, monkeypatch, model, settings, client_config): target = gemini_nodes.SSL_GeminiTextPrompt.define_schema().inputs[-2] assert target.get_io_type() == SSL_GeminiResponseSchema.define_schema().outputs[0].get_io_type() == "GEMINI_RESPONSE_SCHEMA" monkeypatch.setenv("GOOGLE_GENAI_USE_ENTERPRISE", "true") state = default_state() state["entries"]["n1"] = default_entry("string") state["entries"]["n1"]["name"] = "answer" state["entries"]["n0"]["fields"] = ["n1"] state["next_id"] = 2 schema = SSL_GeminiResponseSchema.execute(json.dumps(state))[0] args = _execute_kwargs(_config(api_key="test", **client_config)) args.update(model=model, response_schema=schema, include_thoughts=True, **settings) structured_client.responses = [[("thought", "thinking"), ("final", '{"answer":'), ("final", ' "ok"}')]] output = gemini_nodes.SSL_GeminiTextPrompt.execute(**args) assert output[0] == output[3] == '{"answer": "ok"}' assert output[4] == "thinking" sent = structured_client.calls[0]["config"] assert sent.response_mime_type == "application/json" assert sent.response_json_schema == schema assert sent.response_schema is None @pytest.mark.parametrize("schema", [None, {}]) def test_absent_schema_is_distinct_from_empty_schema(structured_client, schema): structured_client.responses = [[("thought", "separate"), ("final", "{}")]] output = gemini_nodes.SSL_GeminiTextPrompt.execute(**_execute_kwargs(_config()), response_schema=schema) assert output[0] == "{}" assert output[3] == ("" if schema is None else "{}") assert output[4] == "separate" assert structured_client.calls[0]["config"].response_mime_type == (None if schema is None else "application/json") @pytest.mark.parametrize("text", [' {"x": 1}\n', '[1, 2]', '"hello"', '1', '1.2', 'true', 'null']) def test_structured_json_keeps_original_text(structured_client, text): structured_client.responses = [[("final", text)]] output = gemini_nodes.SSL_GeminiTextPrompt.execute(**_execute_kwargs(_config()), response_schema={}) assert output[0] == output[3] == text @pytest.mark.parametrize("text", ["", "{", '```json\n{}\n```', "NaN", "Infinity", "-Infinity", '{"x":NaN}']) def test_invalid_structured_answer_is_not_cached(structured_client, text): structured_client.responses = [[("thought", "must not leak"), ("final", text)]] output = gemini_nodes.SSL_GeminiTextPrompt.execute(**_execute_kwargs(_config()), response_schema={}, use_seed=True, seed=4) assert output[0].startswith("API call/processing error:") assert output[3] == output[4] == "" assert not gemini_nodes.SSL_GeminiTextPrompt._cache def test_schema_snapshot_and_result_cache(structured_client): schema = {"type": "object", "properties": {"answer": {"type": "string"}}} structured_client.on_client = lambda: schema["properties"].update(extra={"type": "number"}) structured_client.responses = [[("thought", "saved thought"), ("final", '{"answer":"ok"}')]] args = _execute_kwargs(_config()) args.update(use_seed=True, seed=7, response_schema=schema) first = gemini_nodes.SSL_GeminiTextPrompt.execute(**args) sent = structured_client.calls[0]["config"].response_json_schema assert "extra" not in sent["properties"] args["response_schema"] = sent second = gemini_nodes.SSL_GeminiTextPrompt.execute(**args) assert len(structured_client.calls) == 1 assert second[3] == first[3] and second[4] == "saved thought" assert len(next(iter(gemini_nodes.SSL_GeminiTextPrompt._cache.values()))) == 5 args["response_schema"] = schema gemini_nodes.SSL_GeminiTextPrompt.execute(**args) assert len(structured_client.calls) == 2 args["response_schema"] = None assert gemini_nodes.SSL_GeminiTextPrompt.execute(**args)[3] == "" assert len(structured_client.calls) == 3 @pytest.mark.parametrize("failure", [False, True]) def test_schema_changes_reuse_context_and_survive_cache_fallback(structured_client, failure): if failure: structured_client.cache_error = RuntimeError("cache unavailable") args = _execute_kwargs(_config(use_cache=True)) for schema in ({"type": "object"}, {"type": "object", "description": "changed"}): gemini_nodes.SSL_GeminiTextPrompt.execute(**args, response_schema=schema) assert structured_client.calls[-1]["config"].response_json_schema == schema assert structured_client.calls[-1]["config"].response_mime_type == "application/json" assert len(structured_client.creations) == (2 if failure else 1) if not failure: assert structured_client.calls[0]["config"].cached_content == structured_client.calls[1]["config"].cached_content def test_expired_context_recreation_keeps_schema(structured_client): structured_client.responses = [RuntimeError("cached content expired"), [("final", "{}")]] output = gemini_nodes.SSL_GeminiTextPrompt.execute(**_execute_kwargs(_config(use_cache=True)), response_schema={}) assert output[3] == "{}" assert len(structured_client.creations) == 2 assert all(call["config"].response_json_schema == {} for call in structured_client.calls) def test_regex_retry_uses_only_final_text_and_keeps_latest_thoughts(structured_client): structured_client.responses = [ [("thought", "old"), ("final", '{"answer":"retry"}')], [("thought", "retry in thought only"), ("final", '{"answer":"done"}')], ] args = _execute_kwargs(_config()) args.update(response_schema={}, use_seed=True, seed=5, retry_pattern="retry", max_retries=2) output = gemini_nodes.SSL_GeminiTextPrompt.execute(**args) assert len(structured_client.calls) == 2 assert output[3] == '{"answer":"done"}' assert output[4] == "retry in thought only" assert all(call["config"].response_json_schema == {} for call in structured_client.calls) key = next(iter(gemini_nodes.SSL_GeminiTextPrompt._seed_map_cache)) assert key[1][-1] == "{}" assert len(key[1]) == len(next(iter(gemini_nodes.SSL_GeminiTextPrompt._cache))) def test_failed_retry_and_outer_exception_are_not_cached(structured_client, monkeypatch): node = gemini_nodes.SSL_GeminiTextPrompt structured_client.responses = [[("thought", "old"), ("final", '"retry"')], RuntimeError("failed")] args = _execute_kwargs(_config()) args.update(response_schema={}, use_seed=True, seed=5, retry_pattern="retry", max_retries=1) output = node.execute(**args) assert output[3] == output[4] == "" assert not node._cache and not node._seed_map_cache def broken(**_kwargs): raise RuntimeError("config failed") monkeypatch.setattr(node, "_build_generate_content_config", broken) output = node.execute(**args) assert output[0].startswith("Unhandled error:") assert output[3] == output[4] == "" assert not node._cache @pytest.mark.parametrize("schema", [[], {"bad": float("nan")}, {"bad": object()}]) def test_bad_schema_fails_before_client_creation(structured_client, schema): with pytest.raises((ValueError, TypeError)): gemini_nodes.SSL_GeminiTextPrompt.execute(**_execute_kwargs(_config()), response_schema=schema) assert not structured_client.clients def test_schema_rejects_effective_image_generation_only(structured_client): args = _execute_kwargs(_config()) args.update(include_images=True, response_schema={}, model="gemini-3-pro-image") with pytest.raises(ValueError, match="image generation"): gemini_nodes.SSL_GeminiTextPrompt.execute(**args) assert not structured_client.clients args.update(model="gemini-2.5-flash", image_inputs={"image_1": torch.zeros(1, 8, 8, 3)}) assert gemini_nodes.SSL_GeminiTextPrompt.execute(**args)[3] == '{"answer":"ok"}' def test_schema_order_and_request_settings_isolate_retry_seed_keys(structured_client): node = gemini_nodes.SSL_GeminiTextPrompt args = _execute_kwargs(_config()) args.update(model="gemini-3.1-pro-preview", use_seed=True, seed=7, retry_pattern="never", max_retries=1, response_schema={"type": "object", "properties": {"a": {"type": "string"}, "b": {"type": "string"}}}, thinking_level="low") node.execute(**args) for change in [ {"response_schema": {"type": "object", "properties": {"b": {"type": "string"}, "a": {"type": "string"}}}}, {"thinking_level": "high"}, {"media_resolution": "high"}, {"max_retries": 2}, {"retry_pattern": "different"}, ]: node.execute(**{**args, **change}) assert len(node._seed_map_cache) == len(node._cache) == len(structured_client.calls) == 6 def test_success_then_retry_timeout_discards_outputs(structured_client): release = threading.Event() completed = threading.Event() def block(): try: release.wait(timeout=1) raise RuntimeError("released") finally: completed.set() structured_client.responses = [[("thought", "first"), ("final", '"retry"')], block] args = _execute_kwargs(_config()) args.update(response_schema={}, use_seed=True, seed=4, retry_pattern="retry", max_retries=1, timeout=0.01, timeout_fallback_text="fallback") try: output = gemini_nodes.SSL_GeminiTextPrompt.execute(**args) finally: release.set() completed.wait(timeout=1) assert output[0] == "fallback" assert output[3] == output[4] == "" assert not gemini_nodes.SSL_GeminiTextPrompt._cache assert not gemini_nodes.SSL_GeminiTextPrompt._seed_map_cache def test_cache_seed_explicitly_enables_seed_widget_control(): inputs = gemini_nodes.SSL_GeminiAPIKeyConfig.INPUT_TYPES() kind, options = inputs["optional"]["cache_seed"] assert kind == "INT" assert options["control_after_generate"] is True assert options["min"] == 0 assert options["max"] == 2147483647 @pytest.mark.parametrize("use_cache", [False, True]) def test_torch_seed_generation_preserves_request_and_fingerprint(structured_client, monkeypatch, use_cache): node = gemini_nodes.SSL_GeminiTextPrompt def forbidden(*args, **kwargs): raise AssertionError("Seed handling must not use global random generators") monkeypatch.setattr(gemini_nodes.random, "randint", forbidden) monkeypatch.setattr(gemini_nodes.random, "seed", forbidden) monkeypatch.setattr(gemini_nodes.np.random, "seed", forbidden) monkeypatch.setattr(torch, "manual_seed", forbidden) monkeypatch.setattr(torch.cuda, "manual_seed_all", forbidden) before = torch.get_rng_state().clone() generated = node._handle_seed(True, 0) assert 0 <= generated < 2147483647 assert node._handle_seed(False, 7) is None args = _execute_kwargs(_config(use_cache=use_cache, cache_seed=123)) args.update(use_seed=True, seed=456) first = node.execute(**args) second = node.execute(**args) assert first[2] == second[2] == (123 if use_cache else 456) assert len(structured_client.calls) == 1 assert structured_client.calls[0]["config"].seed == first[2] node._cache.clear() structured_client.calls.clear() structured_client.responses = [[("final", "retry")], [("final", "done")]] result = node.execute(**args, retry_pattern="retry", max_retries=1) assert len(structured_client.calls) == 2 assert 0 <= result[2] < 2147483647 assert structured_client.calls[-1]["config"].seed == result[2] assert torch.equal(before, torch.get_rng_state()) @pytest.mark.parametrize("value", [130610794845696, 450167232274663, 303772010844099, 854079365259032]) @pytest.mark.parametrize("use_cache", [False, True]) def test_high_precision_seed_is_retained_in_fingerprint_but_mapped_for_api(structured_client, use_cache, value): node = gemini_nodes.SSL_GeminiTextPrompt args = _execute_kwargs(_config(use_cache=use_cache, cache_seed=value)) args.update(use_seed=True, seed=value) first = node.execute(**args) wire_seed = structured_client.calls[-1]["config"].seed assert 0 <= wire_seed < 2**31 assert first[2] == wire_seed fingerprint = next(iter(node._cache)) assert fingerprint[14] == value if use_cache: assert fingerprint[17] == value node.execute(**args) assert len(structured_client.calls) == 1 node._cache.clear() assert node.execute(**args)[2] == wire_seed assert structured_client.calls[-1]["config"].seed == wire_seed def test_unseeded_image_branch_also_respects_wire_seed_range(structured_client): args = _execute_kwargs(_config()) args.update(model="gemini-3-pro-image", include_images=True, use_seed=False, seed=854079365259032) gemini_nodes.SSL_GeminiTextPrompt.execute(**args) assert 0 <= structured_client.calls[-1]["config"].seed < 2**31 def _execute_kwargs(config): return { "config": config, "prompt": "test prompt", "system_instruction": "test system instruction", "model": "gemini-2.0-flash", "temperature": 1.0, "top_p": 0.95, "top_k": 40, "max_output_tokens": 128, "include_images": False, "aspect_ratio": "None", "bypass_mode": "None", "thinking_budget": 0, } def _config(**overrides): config = { "api_key": "", "api_version": "v1", "use_vertexai_env": False, "vertexai_express": False, "vertexai_project": "", "vertexai_location": "", "google_application_credentials": "", } config.update(overrides) return config def _install_success_client(monkeypatch, captured): class FakeModels: @staticmethod def generate_content(**kwargs): part = SimpleNamespace(text="ok", inline_data=None) content = SimpleNamespace(parts=[part]) return SimpleNamespace(candidates=[SimpleNamespace(content=content)]) class FakeClient: def __init__(self, **kwargs): captured.update(kwargs) self.models = FakeModels() monkeypatch.setattr(gemini_nodes.genai, "Client", FakeClient) gemini_nodes.SSL_GeminiTextPrompt._client_cache.clear() class FakeVideo: def __init__(self, data): self.data = data self.save_calls = [] def save_to(self, buffer, format, codec): self.save_calls.append((format, codec)) buffer.write(self.data) def _video_config(data, fps=1): return {"video": FakeVideo(data), "fps": fps} def test_image_autogrow_is_last_input(): schema = gemini_nodes.SSL_GeminiTextPrompt.define_schema() assert schema.inputs[-3].id == "video" assert schema.inputs[-2].id == "response_schema" assert schema.inputs[-2].optional is True assert schema.inputs[-1].id == "image_inputs" assert "response_schema" in gemini_nodes.SSL_GeminiTextPrompt.INPUT_TYPES()["optional"] assert [output.id for output in schema.outputs] == ["text", "image", "final_actual_seed", "structured_output", "thoughts"] def test_video_config_exposes_native_video_and_integer_fps(): schema = gemini_nodes.SSL_GeminiVideoConfig.define_schema() fps = next(input_ for input_ in schema.inputs if input_.id == "fps") pad_at_start = next(input_ for input_ in schema.inputs if input_.id == "pad_at_start") duration_aware_padding = next(input_ for input_ in schema.inputs if input_.id == "duration_aware_padding") assert [input_.id for input_ in schema.inputs] == ["video", "fps", "pad_at_start", "duration_aware_padding"] assert fps.default == 1 assert fps.min == 1 assert fps.max == 24 assert pad_at_start.default is False assert duration_aware_padding.default is False native_video = object() output = gemini_nodes.SSL_GeminiVideoConfig.execute(native_video, 2, True, True) assert output[0] == { "video": native_video, "fps": 2, "pad_at_start": True, "duration_aware_padding": True, } def test_video_start_padding_matches_resolution_and_half_sampling_interval(monkeypatch): images = torch.ones((4, 3, 5, 3)) waveform = torch.ones((1, 2, 8000)) video = SimpleNamespace( get_components=lambda: gemini_nodes.Types.VideoComponents( images=images, audio={"waveform": waveform, "sample_rate": 8000}, frame_rate=4, ), get_bit_depth=lambda: 8, get_color_space=lambda: "sRGB", ) captured = {} class CapturingVideo: def __init__(self, components, bit_depth, color_space): captured.update(components=components, bit_depth=bit_depth, color_space=color_space) def save_to(self, buffer, format, codec): buffer.write(b"padded-video") monkeypatch.setattr(gemini_nodes, "VideoFromComponents", CapturingVideo) video_bytes, mime_type = gemini_nodes.SSL_GeminiTextPrompt._serialize_video(video, fps=2, pad_at_start=True) assert video_bytes == b"padded-video" assert mime_type == "video/mp4" assert captured["components"].frame_rate == 4 assert captured["components"].images.shape == (5, 3, 5, 3) assert torch.count_nonzero(captured["components"].images[0]) == 0 assert torch.equal(captured["components"].images[1], images[0]) assert captured["components"].audio["waveform"].shape[-1] == 10000 assert torch.count_nonzero(captured["components"].audio["waveform"][..., :2000]) == 0 def test_duration_aware_padding_preserves_native_fps_and_rounds_down_to_frames(monkeypatch): images = torch.ones((422, 1, 1, 3)) video = SimpleNamespace( get_components=lambda: gemini_nodes.Types.VideoComponents(images=images, frame_rate=40), get_duration=lambda: 10.559, get_bit_depth=lambda: 8, get_color_space=lambda: "sRGB", ) captured = {} class CapturingVideo: def __init__(self, components, bit_depth, color_space): captured["components"] = components def save_to(self, buffer, format, codec): buffer.write(b"duration-aware-video") monkeypatch.setattr(gemini_nodes, "VideoFromComponents", CapturingVideo) gemini_nodes.SSL_GeminiTextPrompt._serialize_video( video, fps=4, pad_at_start=True, duration_aware_padding=True, ) components = captured["components"] assert components.frame_rate == 40 assert components.images.shape[0] == 22 + images.shape[0] assert torch.count_nonzero(components.images[:22]) == 0 assert torch.equal(components.images[22], images[0]) def test_context_cache_reuses_full_input_after_local_fingerprint_miss(monkeypatch): captured = {"creates": [], "generations": []} class FakeCaches: @staticmethod def create(**kwargs): captured["creates"].append(kwargs) return SimpleNamespace(name="cachedContents/test", expire_time=None) @staticmethod def update(**kwargs): raise AssertionError("TTL should not update") class FakeModels: @staticmethod def generate_content(**kwargs): captured["generations"].append(kwargs) part = SimpleNamespace(text="ok", inline_data=None) content = SimpleNamespace(parts=[part]) return SimpleNamespace(candidates=[SimpleNamespace(content=content)]) class FakeClient: def __init__(self, **kwargs): self.caches = FakeCaches() self.models = FakeModels() monkeypatch.setattr(gemini_nodes.genai, "Client", FakeClient) gemini_nodes.SSL_GeminiTextPrompt._cache.clear() gemini_nodes.SSL_GeminiTextPrompt._client_cache.clear() gemini_nodes.SSL_GeminiTextPrompt._context_cache.clear() config = { "api_key": "test-key", "api_version": "v1", "use_vertexai_env": False, "vertexai_express": False, "vertexai_project": "", "vertexai_location": "", "google_application_credentials": "", "use_cache": True, "cache_ttl_minutes": 15, "cache_seed": 123, } kwargs = _execute_kwargs(config) kwargs.update( use_seed=True, seed=1, video=_video_config(b"normalized-video", fps=2), image_inputs={"image_1": torch.full((1, 2, 2, 3), 0.25)}, ) first = gemini_nodes.SSL_GeminiTextPrompt.execute(**kwargs) kwargs["seed"] = 2 second = gemini_nodes.SSL_GeminiTextPrompt.execute(**kwargs) third = gemini_nodes.SSL_GeminiTextPrompt.execute(**kwargs) assert first[0] == second[0] == third[0] == "ok" assert len(captured["creates"]) == 1 create_config = captured["creates"][0]["config"] assert create_config.system_instruction == "test system instruction" assert create_config.ttl == "900s" assert create_config.contents[0].parts[0].inline_data.mime_type == "image/png" assert create_config.contents[0].parts[1].inline_data.data == b"normalized-video" assert create_config.contents[0].parts[1].inline_data.mime_type == "video/mp4" assert create_config.contents[0].parts[1].video_metadata.fps == 2 assert create_config.contents[0].parts[-1].text == "test prompt" assert len(create_config.contents[0].parts) == 3 assert len(captured["generations"]) == 2 assert all(call["contents"] == " " for call in captured["generations"]) assert all(call["config"].cached_content == "cachedContents/test" for call in captured["generations"]) assert all(call["config"].system_instruction is None for call in captured["generations"]) assert all(call["config"].seed == 123 for call in captured["generations"]) def test_context_cache_creation_failure_uses_original_request(monkeypatch): captured = {} class FakeCaches: @staticmethod def create(**kwargs): raise RuntimeError("unsupported cache") class FakeModels: @staticmethod def generate_content(**kwargs): captured.update(kwargs) part = SimpleNamespace(text="ok", inline_data=None) content = SimpleNamespace(parts=[part]) return SimpleNamespace(candidates=[SimpleNamespace(content=content)]) class FakeClient: def __init__(self, **kwargs): self.caches = FakeCaches() self.models = FakeModels() monkeypatch.setattr(gemini_nodes.genai, "Client", FakeClient) gemini_nodes.SSL_GeminiTextPrompt._cache.clear() gemini_nodes.SSL_GeminiTextPrompt._client_cache.clear() gemini_nodes.SSL_GeminiTextPrompt._context_cache.clear() config = { "api_key": "test-key", "api_version": "v1", "use_vertexai_env": False, "vertexai_express": False, "vertexai_project": "", "vertexai_location": "", "google_application_credentials": "", "use_cache": True, "cache_ttl_minutes": 60, "cache_seed": 456, } kwargs = _execute_kwargs(config) kwargs.update( model="gemini-3-pro-image", include_images=True, use_seed=True, seed=1, ) output = gemini_nodes.SSL_GeminiTextPrompt.execute(**kwargs) assert output[0] == "ok" assert captured["contents"] == "test prompt" assert captured["config"].cached_content is None assert captured["config"].system_instruction[0].text == "test system instruction" assert captured["config"].seed == 456 kwargs.update(use_seed=False, seed=2) output = gemini_nodes.SSL_GeminiTextPrompt.execute(**kwargs) assert output[0] == "ok" assert captured["config"].seed is None kwargs["config"] = config | {"use_cache": False} output = gemini_nodes.SSL_GeminiTextPrompt.execute(**kwargs) assert output[0] == "ok" assert captured["config"].seed == 2 def test_missing_context_cache_is_recreated_once(monkeypatch): captured = {"creates": 0, "generations": 0} class FakeCaches: @staticmethod def create(**kwargs): captured["creates"] += 1 return SimpleNamespace( name=f"cachedContents/test-{captured['creates']}", expire_time=None, ) class FakeModels: @staticmethod def generate_content(**kwargs): captured["generations"] += 1 if captured["generations"] == 1: raise RuntimeError("cached content not found") part = SimpleNamespace(text="ok", inline_data=None) content = SimpleNamespace(parts=[part]) return SimpleNamespace(candidates=[SimpleNamespace(content=content)]) class FakeClient: def __init__(self, **kwargs): self.caches = FakeCaches() self.models = FakeModels() monkeypatch.setattr(gemini_nodes.genai, "Client", FakeClient) gemini_nodes.SSL_GeminiTextPrompt._cache.clear() gemini_nodes.SSL_GeminiTextPrompt._client_cache.clear() gemini_nodes.SSL_GeminiTextPrompt._context_cache.clear() config = { "api_key": "test-key", "api_version": "v1", "use_vertexai_env": False, "vertexai_express": False, "vertexai_project": "", "vertexai_location": "", "google_application_credentials": "", "use_cache": True, "cache_ttl_minutes": 60, "cache_seed": 456, } kwargs = _execute_kwargs(config) kwargs.update(use_seed=True, seed=1) output = gemini_nodes.SSL_GeminiTextPrompt.execute(**kwargs) assert output[0] == "ok" assert captured == {"creates": 2, "generations": 2} def test_context_cache_ttl_change_updates_existing_resource(): captured = {"creates": 0, "updates": []} class FakeCaches: @staticmethod def create(**kwargs): captured["creates"] += 1 return SimpleNamespace(name="cachedContents/test", expire_time=None) @staticmethod def update(**kwargs): captured["updates"].append(kwargs) return SimpleNamespace(name="cachedContents/test", expire_time=None) client = SimpleNamespace(caches=FakeCaches()) gemini_nodes.SSL_GeminiTextPrompt._context_cache.clear() first = gemini_nodes.SSL_GeminiTextPrompt._get_or_create_context_cache( client, ("context",), "gemini-2.0-flash", "prompt", "system", 15 ) second = gemini_nodes.SSL_GeminiTextPrompt._get_or_create_context_cache( client, ("context",), "gemini-2.0-flash", "prompt", "system", 30 ) assert first == second == "cachedContents/test" assert captured["creates"] == 1 assert len(captured["updates"]) == 1 assert captured["updates"][0]["config"].ttl == "1800s" def test_context_cache_identity_covers_every_cached_input(): build_key = gemini_nodes.SSL_GeminiTextPrompt._build_context_cache_key base = build_key(("standard", "key-a"), "model-a", "system-a", "prompt-a", "video-a", "video/mp4", 1, ["image-a"], "low") variants = { build_key(("standard", "key-b"), "model-a", "system-a", "prompt-a", "video-a", "video/mp4", 1, ["image-a"], "low"), build_key(("standard", "key-a"), "model-b", "system-a", "prompt-a", "video-a", "video/mp4", 1, ["image-a"], "low"), build_key(("standard", "key-a"), "model-a", "system-b", "prompt-a", "video-a", "video/mp4", 1, ["image-a"], "low"), build_key(("standard", "key-a"), "model-a", "system-a", "prompt-b", "video-a", "video/mp4", 1, ["image-a"], "low"), build_key(("standard", "key-a"), "model-a", "system-a", "prompt-a", "video-b", "video/mp4", 1, ["image-a"], "low"), build_key(("standard", "key-a"), "model-a", "system-a", "prompt-a", "video-a", "video/webm", 1, ["image-a"], "low"), build_key(("standard", "key-a"), "model-a", "system-a", "prompt-a", "video-a", "video/mp4", 2, ["image-a"], "low"), build_key(("standard", "key-a"), "model-a", "system-a", "prompt-a", "video-a", "video/mp4", 1, ["image-b"], "low"), build_key(("standard", "key-a"), "model-a", "system-a", "prompt-a", "video-a", "video/mp4", 1, ["image-a"], "high"), } assert base not in variants assert len(variants) == 9 def test_video_is_normalized_sent_first_and_locally_fingerprinted(monkeypatch): captured = {"generations": []} class FakeModels: @staticmethod def generate_content(**kwargs): captured["generations"].append(kwargs) part = SimpleNamespace(text="ok", inline_data=None) return SimpleNamespace(candidates=[SimpleNamespace(content=SimpleNamespace(parts=[part]))]) class FakeClient: def __init__(self, **kwargs): self.models = FakeModels() monkeypatch.setattr(gemini_nodes.genai, "Client", FakeClient) gemini_nodes.SSL_GeminiTextPrompt._cache.clear() gemini_nodes.SSL_GeminiTextPrompt._client_cache.clear() kwargs = _execute_kwargs(_config(api_key="test-key")) kwargs.update( model="gemini-3-flash-preview", media_resolution="low", use_seed=True, seed=7, video=_video_config(b"video-a", fps=2), ) first = gemini_nodes.SSL_GeminiTextPrompt.execute(**kwargs) kwargs["video"] = _video_config(b"video-a", fps=2) second = gemini_nodes.SSL_GeminiTextPrompt.execute(**kwargs) kwargs["video"] = _video_config(b"video-b", fps=2) third = gemini_nodes.SSL_GeminiTextPrompt.execute(**kwargs) assert first[0] == second[0] == third[0] == "ok" assert len(captured["generations"]) == 2 parts = captured["generations"][0]["contents"][0].parts assert parts[0].inline_data.data == b"video-a" assert parts[0].inline_data.mime_type == "video/mp4" assert parts[0].media_resolution.level == "MEDIA_RESOLUTION_LOW" assert parts[-1].text == "test prompt" assert kwargs["video"]["video"].save_calls == [ (gemini_nodes.Types.VideoContainer.MP4, gemini_nodes.Types.VideoCodec.H264) ] def test_video_at_inline_limit_returns_clear_error(monkeypatch): monkeypatch.setattr(gemini_nodes.SSL_GeminiTextPrompt, "VIDEO_INLINE_LIMIT_BYTES", 4) output = gemini_nodes.SSL_GeminiTextPrompt.execute( **_execute_kwargs(_config()), video=_video_config(b"1234"), ) assert output[0] == "Error processing input video: Gemini video input must be smaller than 100 MB after MP4/H.264 normalization." def test_vertex_video_over_cache_blob_limit_uses_full_request(monkeypatch): captured = {"cache_creates": 0, "generations": []} class FakeCaches: @staticmethod def create(**kwargs): captured["cache_creates"] += 1 raise AssertionError("Vertex cache creation must be skipped") class FakeModels: @staticmethod def generate_content(**kwargs): captured["generations"].append(kwargs) part = SimpleNamespace(text="ok", inline_data=None) return SimpleNamespace(candidates=[SimpleNamespace(content=SimpleNamespace(parts=[part]))]) class FakeClient: def __init__(self, **kwargs): self.caches = FakeCaches() self.models = FakeModels() monkeypatch.setattr(gemini_nodes.genai, "Client", FakeClient) monkeypatch.setattr(gemini_nodes.SSL_GeminiTextPrompt, "VERTEX_CACHE_INLINE_LIMIT_BYTES", 1) gemini_nodes.SSL_GeminiTextPrompt._cache.clear() gemini_nodes.SSL_GeminiTextPrompt._client_cache.clear() gemini_nodes.SSL_GeminiTextPrompt._context_cache.clear() config = _config(api_key="test-key", vertexai_express=True, use_cache=True, cache_seed=11) output = gemini_nodes.SSL_GeminiTextPrompt.execute( **_execute_kwargs(config), use_seed=True, seed=2, video=_video_config(b"12"), ) assert output[0] == "ok" assert captured["cache_creates"] == 0 assert len(captured["generations"]) == 1 assert captured["generations"][0]["contents"] != " " assert captured["generations"][0]["config"].seed == 11 def test_vertex_config_socket_ids_remain_stable(): schema = gemini_nodes.SSL_GeminiAPIKeyConfig.define_schema() assert [input_.id for input_ in schema.inputs] == [ "api_key", "api_version", "use_vertexai_env", "vertexai_express", "vertexai_project", "vertexai_location", "google_application_credentials", "use_cache", "cache_ttl_minutes", "cache_seed", ] assert all(input_.optional for input_ in schema.inputs[-3:]) def test_image_inputs_use_clean_autogrow_socket_ids(): schema = gemini_nodes.SSL_GeminiTextPrompt.define_schema() image_inputs = next(input_ for input_ in schema.inputs if input_.id == "image_inputs") assert image_inputs.template.min == 0 assert image_inputs.template.names[:3] == [ "image_1", "image_2", "image_3", ] assert image_inputs.template.names[-1] == "image_100" assert not {"input_image", "input_image_2"} & {input_.id for input_ in schema.inputs} def test_image_inputs_flatten_in_numeric_socket_and_batch_order(): first_batch = torch.stack( [ torch.full((2, 2, 3), 0.1), torch.full((2, 2, 3), 0.2), torch.full((2, 2, 3), 0.3), torch.full((2, 2, 3), 0.4), torch.full((2, 2, 3), 0.5), ] ) second_socket = torch.full((1, 2, 2, 3), 0.6) frames, batch_counts = gemini_nodes.SSL_GeminiTextPrompt._ordered_image_frames( { "image_2": second_socket, "image_1": first_batch, } ) assert [(socket, index) for socket, index, _ in frames] == [ ("image_1", 0), ("image_1", 1), ("image_1", 2), ("image_1", 3), ("image_1", 4), ("image_2", 0), ] assert [round(float(frame[0, 0, 0]), 1) for _, _, frame in frames] == [ 0.1, 0.2, 0.3, 0.4, 0.5, 0.6, ] assert batch_counts == [("image_1", 5), ("image_2", 1)] def test_image_inputs_reject_non_bhwc_tensor_with_socket_name(): with pytest.raises(ValueError, match=r"Gemini image_2 must be a BHWC image tensor"): gemini_nodes.SSL_GeminiTextPrompt._ordered_image_frames( {"image_2": torch.zeros((2, 2, 3))} ) def test_all_ordered_images_are_sent_to_gemini_2_and_3_before_prompt(monkeypatch): captured = {"call_count": 0} class FakeModels: @staticmethod def generate_content(**kwargs): captured["call_count"] += 1 captured.update(kwargs) part = SimpleNamespace(text="ok", inline_data=None) content = SimpleNamespace(parts=[part]) return SimpleNamespace(candidates=[SimpleNamespace(content=content)]) class FakeClient: def __init__(self, **kwargs): self.models = FakeModels() monkeypatch.setattr(gemini_nodes.genai, "Client", FakeClient) gemini_nodes.SSL_GeminiTextPrompt._client_cache.clear() config = { "api_key": "test-key", "api_version": "v1", "use_vertexai_env": False, "vertexai_express": False, "vertexai_project": "", "vertexai_location": "", "google_application_credentials": "", } kwargs = _execute_kwargs(config) kwargs["image_inputs"] = { "image_2": torch.full((1, 2, 2, 3), 0.6), "image_1": torch.stack( [ torch.full((2, 2, 3), 0.1), torch.full((2, 2, 3), 0.2), torch.full((2, 2, 3), 0.3), torch.full((2, 2, 3), 0.4), torch.full((2, 2, 3), 0.5), ] ), } def assert_ordered_parts(): assert captured["call_count"] == 1 assert len(captured["contents"]) == 1 parts = captured["contents"][0].parts assert len(parts) == 7 assert parts[-1].text == "test prompt" pixels = [ Image.open(BytesIO(part.inline_data.data)).getpixel((0, 0))[0] for part in parts[:-1] ] assert pixels == [25, 51, 76, 102, 127, 153] for model in ("gemini-2.0-flash", "gemini-3-flash-preview"): captured.clear() captured["call_count"] = 0 kwargs["model"] = model output = gemini_nodes.SSL_GeminiTextPrompt.execute(**kwargs) assert output[0] == "ok" assert_ordered_parts() monkeypatch.delenv("GOOGLE_GENAI_USE_ENTERPRISE", raising=False) monkeypatch.setenv("GOOGLE_GENAI_USE_VERTEXAI", "true") monkeypatch.setenv("GOOGLE_CLOUD_PROJECT", "test-project") monkeypatch.setenv("GOOGLE_CLOUD_LOCATION", "global") gemini_nodes.SSL_GeminiTextPrompt._client_cache.clear() kwargs["config"] = config | {"use_vertexai_env": True, "api_key": ""} kwargs["model"] = "gemini-2.5-flash" captured.clear() captured["call_count"] = 0 output = gemini_nodes.SSL_GeminiTextPrompt.execute(**kwargs) assert output[0] == "ok" assert_ordered_parts() def test_fingerprint_hashes_every_ordered_image(): common = { "config": {"api_key": "key", "api_version": "v1"}, "prompt": "prompt", "system_instruction": "system", "model": "gemini-2.0-flash", "temperature": 1.0, "top_p": 0.95, "top_k": 40, "max_output_tokens": 128, "include_images": False, "aspect_ratio": "None", "bypass_mode": "None", "thinking_budget": 0, "use_seed": True, "seed": 1, } first = torch.zeros((1, 2, 2, 3)) second = torch.ones((1, 2, 2, 3)) fingerprint_a, _ = gemini_nodes.SSL_GeminiTextPrompt._compute_fingerprint_and_check_cache( **common, image_inputs={"image_1": first, "image_2": second}, ) fingerprint_b, _ = gemini_nodes.SSL_GeminiTextPrompt._compute_fingerprint_and_check_cache( **common, image_inputs={"image_1": second, "image_2": first}, ) assert fingerprint_a != fingerprint_b def test_api_key_is_hashed_in_fingerprint(): api_key = "secret-api-key" fingerprint, cached = gemini_nodes.SSL_GeminiTextPrompt._compute_fingerprint_and_check_cache( config={"api_key": api_key, "api_version": "v1"}, prompt="prompt", system_instruction="system", model="gemini-2.0-flash", temperature=1.0, top_p=0.95, top_k=40, max_output_tokens=128, include_images=False, aspect_ratio="None", bypass_mode="None", thinking_budget=0, use_seed=True, seed=1, ) assert cached is None assert api_key not in fingerprint[0] assert hashlib.sha256(api_key.encode("utf-8")).hexdigest() in fingerprint[0] def test_false_vertex_environment_value_is_rejected(monkeypatch): monkeypatch.delenv("GOOGLE_GENAI_USE_ENTERPRISE", raising=False) monkeypatch.setenv("GOOGLE_GENAI_USE_VERTEXAI", "false") monkeypatch.setenv("GOOGLE_CLOUD_PROJECT", "test-project") monkeypatch.setenv("GOOGLE_CLOUD_LOCATION", "global") monkeypatch.setattr( gemini_nodes.genai, "Client", lambda **kwargs: (_ for _ in ()).throw(AssertionError("Client must not be created")), ) config = { "api_key": "", "api_version": "v1", "use_vertexai_env": True, "vertexai_express": False, "vertexai_project": "", "vertexai_location": "", "google_application_credentials": "", } output = gemini_nodes.SSL_GeminiTextPrompt.execute(**_execute_kwargs(config)) assert output[0] == ( "Invalid Enterprise/Vertex AI configuration: GOOGLE_GENAI_USE_ENTERPRISE or " "GOOGLE_GENAI_USE_VERTEXAI must be enabled when use_vertexai_env is true" ) def test_adc_path_is_loaded_and_passed_to_vertex_client(monkeypatch, tmp_path): credentials_path = tmp_path / "application_default_credentials.json" credentials_path.write_text("{}", encoding="utf-8") credentials = object() captured = {} monkeypatch.delenv("GOOGLE_GENAI_USE_ENTERPRISE", raising=False) monkeypatch.setenv("GOOGLE_GENAI_USE_VERTEXAI", "true") monkeypatch.delenv("GOOGLE_CLOUD_PROJECT", raising=False) monkeypatch.setenv("GOOGLE_CLOUD_LOCATION", "global") monkeypatch.setattr( google.auth, "load_credentials_from_file", lambda path, scopes: (credentials, "credentials-project"), ) class FakeModels: @staticmethod def generate_content(**kwargs): part = SimpleNamespace(text="ok", inline_data=None) content = SimpleNamespace(parts=[part]) return SimpleNamespace(candidates=[SimpleNamespace(content=content)]) class FakeClient: def __init__(self, **kwargs): captured.update(kwargs) self.models = FakeModels() monkeypatch.setattr(gemini_nodes.genai, "Client", FakeClient) gemini_nodes.SSL_GeminiTextPrompt._client_cache.clear() config = { "api_key": "", "api_version": "v1", "use_vertexai_env": True, "vertexai_express": False, "vertexai_project": "", "vertexai_location": "", "google_application_credentials": str(credentials_path), } output = gemini_nodes.SSL_GeminiTextPrompt.execute(**_execute_kwargs(config)) assert output[0] == "ok" assert captured["enterprise"] is True assert "vertexai" not in captured assert captured["credentials"] is credentials assert captured["project"] == "credentials-project" def test_enterprise_environment_takes_precedence_over_legacy_vertex(monkeypatch): captured = {} monkeypatch.setenv("GOOGLE_GENAI_USE_ENTERPRISE", "true") monkeypatch.setenv("GOOGLE_GENAI_USE_VERTEXAI", "false") monkeypatch.setenv("GOOGLE_CLOUD_PROJECT", "test-project") monkeypatch.setenv("GOOGLE_CLOUD_LOCATION", "global") _install_success_client(monkeypatch, captured) with pytest.warns(UserWarning, match="GOOGLE_GENAI_USE_ENTERPRISE takes precedence"): output = gemini_nodes.SSL_GeminiTextPrompt.execute( **_execute_kwargs(_config(use_vertexai_env=True)) ) assert output[0] == "ok" assert captured["enterprise"] is True assert captured["project"] == "test-project" assert captured["location"] == "global" def test_vertex_environment_mode_defaults_to_enterprise_when_selector_is_absent(monkeypatch): captured = {} monkeypatch.delenv("GOOGLE_GENAI_USE_ENTERPRISE", raising=False) monkeypatch.delenv("GOOGLE_GENAI_USE_VERTEXAI", raising=False) monkeypatch.setenv("GOOGLE_CLOUD_PROJECT", "test-project") monkeypatch.setenv("GOOGLE_CLOUD_LOCATION", "global") _install_success_client(monkeypatch, captured) output = gemini_nodes.SSL_GeminiTextPrompt.execute( **_execute_kwargs(_config(use_vertexai_env=True)) ) assert output[0] == "ok" assert captured["enterprise"] is True def test_malformed_enterprise_environment_value_is_rejected(monkeypatch): monkeypatch.setenv("GOOGLE_GENAI_USE_ENTERPRISE", "yes") monkeypatch.delenv("GOOGLE_GENAI_USE_VERTEXAI", raising=False) monkeypatch.setenv("GOOGLE_CLOUD_PROJECT", "test-project") monkeypatch.setenv("GOOGLE_CLOUD_LOCATION", "global") monkeypatch.setattr( gemini_nodes.genai, "Client", lambda **kwargs: (_ for _ in ()).throw(AssertionError("Client must not be created")), ) output = gemini_nodes.SSL_GeminiTextPrompt.execute( **_execute_kwargs(_config(use_vertexai_env=True)) ) assert output[0] == ( "Invalid Enterprise/Vertex AI configuration: " "GOOGLE_GENAI_USE_ENTERPRISE must be true, false, 1, or 0" ) def test_enterprise_express_requires_only_api_key(monkeypatch): captured = {} monkeypatch.delenv("GOOGLE_CLOUD_PROJECT", raising=False) monkeypatch.delenv("GOOGLE_CLOUD_LOCATION", raising=False) _install_success_client(monkeypatch, captured) output = gemini_nodes.SSL_GeminiTextPrompt.execute( **_execute_kwargs(_config(api_key="express-key", vertexai_express=True)) ) assert output[0] == "ok" assert captured["enterprise"] is True assert captured["api_key"] == "express-key" assert "project" not in captured assert "location" not in captured def test_proxy_uses_sdk_http_options_without_mutating_environment(monkeypatch): captured = {} monkeypatch.setenv("HTTP_PROXY", "http://existing-proxy:9000") _install_success_client(monkeypatch, captured) kwargs = _execute_kwargs(_config(api_key="test-key")) kwargs.update(use_proxy=True, proxy_host="proxy.example", proxy_port=8080) output = gemini_nodes.SSL_GeminiTextPrompt.execute(**kwargs) assert output[0] == "ok" assert captured["http_options"].client_args == { "proxy": "http://proxy.example:8080" } assert os.environ["HTTP_PROXY"] == "http://existing-proxy:9000" def test_confirmed_current_models_are_visible_without_deprecated_image_aliases(): schema = gemini_nodes.SSL_GeminiTextPrompt.define_schema() model_input = next(input_ for input_ in schema.inputs if input_.id == "model") input_ids = [input_.id for input_ in schema.inputs] aspect_ratio = next(input_ for input_ in schema.inputs if input_.id == "aspect_ratio") image_size = next(input_ for input_ in schema.inputs if input_.id == "image_size") for model in ( "gemini-4-argon", "gemini-4-argon-cyber-permissive", "gemini-3.7-flash", "gemini-3.8-flash", "gemini-3.5-flash", "gemini-3.1-flash-image", "gemini-3.1-flash-lite-image", "gemini-3-pro-image", ): assert model in model_input.options assert "gemini-2.5-flash-image-preview" not in model_input.options assert "nano-banana-pro-preview" not in model_input.options assert input_ids[input_ids.index("include_images") + 1] == "aspect_ratio" assert aspect_ratio.options == ["None", "1:1", "9:16", "16:9", "3:4", "4:3", "3:2", "2:3", "5:4", "4:5", "21:9"] assert aspect_ratio.default == "None" assert input_ids[input_ids.index("timeout_fallback_text") + 1] == "image_size" assert image_size.options == ["None", "512", "1K", "2K", "4K"] assert image_size.default == "None" def test_image_config_omits_none_aspect_ratio(): config = gemini_nodes.SSL_GeminiTextPrompt._build_generate_content_config( model="gemini-3.1-flash-image", temperature=1.0, top_p=0.95, top_k=40, max_output_tokens=8192, seed=7, include_images=True, response_modalities=["TEXT", "IMAGE"], aspect_ratio="None", padded_system_instruction="system", thinking_level="None", thinking_budget=0, include_thoughts=False, media_resolution="unspecified", ) assert config.response_modalities == ["TEXT", "IMAGE"] assert config.image_config is None def test_image_config_preserves_concrete_aspect_ratio(): config = gemini_nodes.SSL_GeminiTextPrompt._build_generate_content_config( model="gemini-3-pro-image", temperature=1.0, top_p=0.95, top_k=40, max_output_tokens=8192, seed=7, include_images=True, response_modalities=["IMAGE", "TEXT"], aspect_ratio="16:9", padded_system_instruction="system", thinking_level="None", thinking_budget=0, include_thoughts=False, media_resolution="unspecified", ) assert config.image_config.aspect_ratio == "16:9" def test_image_config_preserves_requested_image_size(): config = gemini_nodes.SSL_GeminiTextPrompt._build_generate_content_config( model="gemini-3.1-flash-image", temperature=1.0, top_p=0.95, top_k=40, max_output_tokens=8192, seed=7, include_images=True, response_modalities=["IMAGE", "TEXT"], aspect_ratio="None", padded_system_instruction="system", thinking_level="None", thinking_budget=0, include_thoughts=False, media_resolution="unspecified", image_size="2K", ) assert config.image_config.aspect_ratio is None assert config.image_config.image_size == "2K" @pytest.mark.parametrize( ("model", "image_size", "allowed"), [ ("gemini-3.1-flash-lite-image", "2K", "1K"), ("gemini-3-pro-image", "512", "1K, 2K, 4K"), ], ) def test_image_config_rejects_unsupported_image_size(model, image_size, allowed): with pytest.raises(ValueError, match=rf"^{model} image_size must be one of: {allowed}\.$"): gemini_nodes.SSL_GeminiTextPrompt._build_generate_content_config( model=model, temperature=1.0, top_p=0.95, top_k=40, max_output_tokens=8192, seed=7, include_images=True, response_modalities=["IMAGE", "TEXT"], aspect_ratio="None", padded_system_instruction="system", thinking_level="None", thinking_budget=0, include_thoughts=False, media_resolution="unspecified", image_size=image_size, ) def test_enterprise_image_config_allows_all_people(): config = gemini_nodes.SSL_GeminiTextPrompt._build_generate_content_config( model="gemini-3.1-flash-image", temperature=1.0, top_p=0.95, top_k=40, max_output_tokens=8192, seed=7, include_images=True, response_modalities=["IMAGE", "TEXT"], aspect_ratio="None", padded_system_instruction="system", thinking_level="None", thinking_budget=0, include_thoughts=False, media_resolution="unspecified", allow_all_people=True, ) assert config.image_config.aspect_ratio is None assert config.image_config.person_generation == "ALLOW_ALL" @pytest.mark.parametrize( ("selected_level", "expected_level"), [("None", "medium"), ("low", "low"), ("medium", "medium"), ("high", "high")], ) def test_gemini_3_7_uses_supported_thinking_levels(selected_level, expected_level): config = gemini_nodes.SSL_GeminiTextPrompt._build_generate_content_config( model="gemini-3.7-flash", temperature=0.4, top_p=0.8, top_k=20, max_output_tokens=1024, seed=7, include_images=False, response_modalities=["TEXT"], aspect_ratio="None", padded_system_instruction="system", thinking_level=selected_level, thinking_budget=4096, include_thoughts=True, media_resolution="unspecified", ) assert config.thinking_config.thinking_level.value == expected_level.upper() assert config.thinking_config.thinking_budget is None assert config.temperature is None assert config.top_p is None assert config.top_k is None def test_gemini_3_8_uses_gemini_3_thinking_config(): config = gemini_nodes.SSL_GeminiTextPrompt._build_generate_content_config( model="gemini-3.8-flash", temperature=0.4, top_p=0.8, top_k=20, max_output_tokens=1024, seed=7, include_images=False, response_modalities=["TEXT"], aspect_ratio="None", padded_system_instruction="system", thinking_level="None", thinking_budget=4096, include_thoughts=True, media_resolution="unspecified", ) assert config.thinking_config.thinking_level.value == "MEDIUM" assert config.thinking_config.thinking_budget is None assert config.temperature is None assert config.top_p is None assert config.top_k is None def test_gemini_3_8_rejects_minimal_thinking_level(): with pytest.raises(ValueError, match="gemini-3.8-flash thinking_level"): gemini_nodes.SSL_GeminiTextPrompt._resolve_gemini_3_7_thinking_level( "minimal", gemini_nodes.SSL_GeminiTextPrompt.GEMINI_3_8_FLASH ) def test_gemini_4_argon_uses_gemini_3_thinking_config(): config = gemini_nodes.SSL_GeminiTextPrompt._build_generate_content_config( model="gemini-4-argon", temperature=0.4, top_p=0.8, top_k=20, max_output_tokens=1024, seed=7, include_images=False, response_modalities=["TEXT"], aspect_ratio="None", padded_system_instruction="system", thinking_level="None", thinking_budget=4096, include_thoughts=True, media_resolution="unspecified", ) assert config.thinking_config.thinking_level.value == "MEDIUM" assert config.thinking_config.thinking_budget is None assert config.temperature is None assert config.top_p is None assert config.top_k is None # Cyber permissive variant defaults to HIGH thinking level config_cyber = gemini_nodes.SSL_GeminiTextPrompt._build_generate_content_config( model="gemini-4-argon-cyber-permissive", temperature=0.4, top_p=0.8, top_k=20, max_output_tokens=1024, seed=7, include_images=False, response_modalities=["TEXT"], aspect_ratio="None", padded_system_instruction="system", thinking_level="None", thinking_budget=4096, include_thoughts=True, media_resolution="unspecified", ) assert config_cyber.thinking_config.thinking_level.value == "HIGH" assert config_cyber.thinking_config.thinking_budget is None assert config_cyber.temperature is None assert config_cyber.top_p is None assert config_cyber.top_k is None def test_gemini_4_argon_rejects_minimal_thinking_level(): with pytest.raises(ValueError, match="gemini-4-argon thinking_level"): gemini_nodes.SSL_GeminiTextPrompt._resolve_gemini_3_7_thinking_level( "minimal", gemini_nodes.SSL_GeminiTextPrompt.GEMINI_4_ARGON ) with pytest.raises(ValueError, match="gemini-4-argon-cyber-permissive thinking_level"): gemini_nodes.SSL_GeminiTextPrompt._resolve_gemini_3_7_thinking_level( "minimal", gemini_nodes.SSL_GeminiTextPrompt.GEMINI_4_ARGON_CYBER_PERMISSIVE ) def test_gemini_4_argon_fingerprint_omits_sampling_params(): common = { "config": {"api_key": "key", "api_version": "v1"}, "prompt": "prompt", "system_instruction": "system", "model": "gemini-4-argon", "max_output_tokens": 1024, "include_images": False, "aspect_ratio": "None", "bypass_mode": "None", "use_seed": False, "seed": 0, "include_thoughts": True, } fp1, _ = gemini_nodes.SSL_GeminiTextPrompt._compute_fingerprint_and_check_cache( **common, temperature=0.1, top_p=0.5, top_k=10, thinking_budget=0, thinking_level="medium", ) fp2, _ = gemini_nodes.SSL_GeminiTextPrompt._compute_fingerprint_and_check_cache( **common, temperature=0.9, top_p=0.99, top_k=90, thinking_budget=1000, thinking_level="medium", ) assert fp1 == fp2 def test_gemini_4_argon_skips_explicit_context_cache(structured_client): structured_client.responses = [[("final", "argon response")]] args = _execute_kwargs(_config(use_cache=True, cache_seed=42)) args["model"] = "gemini-4-argon" output = gemini_nodes.SSL_GeminiTextPrompt.execute(**args) assert output[0] == "argon response" assert len(structured_client.calls) == 1 # Explicit cache was not set on the request config assert structured_client.calls[0]["config"].cached_content is None def test_gemini_3_7_fingerprint_matches_effective_request_controls(): common = { "config": {"api_key": "key", "api_version": "v1"}, "prompt": "prompt", "system_instruction": "system", "model": "gemini-3.7-flash", "max_output_tokens": 1024, "include_images": False, "aspect_ratio": "None", "bypass_mode": "None", "use_seed": False, "seed": 0, "include_thoughts": True, } default_fingerprint, _ = gemini_nodes.SSL_GeminiTextPrompt._compute_fingerprint_and_check_cache( **common, temperature=0.4, top_p=0.8, top_k=20, thinking_budget=0, thinking_level="None", ) explicit_medium_fingerprint, _ = gemini_nodes.SSL_GeminiTextPrompt._compute_fingerprint_and_check_cache( **common, temperature=1.0, top_p=0.95, top_k=40, thinking_budget=8192, thinking_level="medium", ) high_fingerprint, _ = gemini_nodes.SSL_GeminiTextPrompt._compute_fingerprint_and_check_cache( **common, temperature=1.0, top_p=0.95, top_k=40, thinking_budget=8192, thinking_level="high", ) minimal_fingerprint, _ = gemini_nodes.SSL_GeminiTextPrompt._compute_fingerprint_and_check_cache( **common, temperature=1.0, top_p=0.95, top_k=40, thinking_budget=8192, thinking_level="minimal", ) low_fingerprint, _ = gemini_nodes.SSL_GeminiTextPrompt._compute_fingerprint_and_check_cache( **common, temperature=0.2, top_p=0.7, top_k=10, thinking_budget=0, thinking_level="low", ) assert default_fingerprint == explicit_medium_fingerprint assert high_fingerprint != default_fingerprint assert minimal_fingerprint == low_fingerprint def test_gemini_3_7_coerces_minimal_to_low_with_warning(capsys): resolved = gemini_nodes.SSL_GeminiTextPrompt._resolve_gemini_3_7_thinking_level( "minimal" ) assert resolved == "low" assert "does not support minimal thinking; using low instead" in capsys.readouterr().out def test_gemini_4_placeholder_matches_3_6_without_ui_exposure(): placeholder = gemini_nodes.SSL_GeminiTextPrompt.GEMINI_4_FLASH_PREVIEW reference = "gemini-3.6-flash" for capability_list in ( gemini_nodes.SSL_GeminiTextPrompt.THINKING_MODELS, gemini_nodes.SSL_GeminiTextPrompt.GEN3_THINKING_MODELS, gemini_nodes.SSL_GeminiTextPrompt.IMAGE_MODELS, gemini_nodes.SSL_GeminiTextPrompt.MEDIA_RES_MODELS, ): assert (placeholder in capability_list) == (reference in capability_list) schema = gemini_nodes.SSL_GeminiTextPrompt.define_schema() model_input = next(input_ for input_ in schema.inputs if input_.id == "model") assert placeholder not in model_input.options def test_timeout_returns_custom_fallback_without_caching(monkeypatch): release_request = threading.Event() class BlockingModels: @staticmethod def generate_content(**kwargs): release_request.wait(timeout=1) raise RuntimeError("request released") class FakeClient: def __init__(self, **kwargs): self.models = BlockingModels() monkeypatch.setattr(gemini_nodes.genai, "Client", FakeClient) gemini_nodes.SSL_GeminiTextPrompt._client_cache.clear() gemini_nodes.SSL_GeminiTextPrompt._cache.clear() config = { "api_key": "test-key", "api_version": "v1", "use_vertexai_env": False, "vertexai_express": False, "vertexai_project": "", "vertexai_location": "", "google_application_credentials": "", } kwargs = _execute_kwargs(config) kwargs.update( use_seed=True, seed=123, timeout=0.01, timeout_fallback_text="configured fallback", ) try: output = gemini_nodes.SSL_GeminiTextPrompt.execute(**kwargs) finally: release_request.set() assert output[0] == "configured fallback" assert output[3] == output[4] == "" assert gemini_nodes.SSL_GeminiTextPrompt._cache == {}