"""Tests for inference registry seed migration helpers. Task 18: Migrate existing provider records. Validates: - Initial endpoints have correct protocol and URL - Initial deployments reference valid endpoints - Agent conversion maps ollama correctly - Agent conversion maps vllm correctly - Unknown providers raise error (don't silently convert) - Conflicting defaults are identified """ from __future__ import annotations import re from pathlib import Path from uuid import UUID import pytest from services.shared.inference.seed_migration import ( OLLAMA_DEPLOYMENT_ID, OLLAMA_ENDPOINT_ID, VLLM_DEPLOYMENT_ID, VLLM_ENDPOINT_ID, UnknownProviderError, convert_agent_providers, get_initial_deployments, get_initial_endpoints, identify_conflicting_defaults, ) # ─── SQL migration file checks ──────────────────────────────────────────────── MIGRATION_PATH = ( Path(__file__).resolve().parent.parent / "infra" / "migrations" / "042_seed_inference_registry.sql" ) class TestMigrationFileExists: def test_file_exists(self): assert MIGRATION_PATH.exists(), f"Migration file not found: {MIGRATION_PATH}" def test_file_not_empty(self): content = MIGRATION_PATH.read_text() assert len(content.strip()) > 100 def test_uses_on_conflict_do_nothing(self): """Migration is idempotent via ON CONFLICT DO NOTHING.""" content = MIGRATION_PATH.read_text() # Remove comments before counting sql = re.sub(r"--[^\n]*", "", content) inserts = re.findall(r"INSERT INTO", sql) on_conflicts = re.findall(r"ON CONFLICT", sql) assert len(inserts) == len(on_conflicts), ( f"Expected {len(inserts)} ON CONFLICT clauses, got {len(on_conflicts)}" ) # ─── Initial endpoints ───────────────────────────────────────────────────────── class TestInitialEndpoints: """Test that initial endpoints have correct protocol and URL.""" def test_returns_two_endpoints(self): endpoints = get_initial_endpoints() assert len(endpoints) == 2 def test_ollama_endpoint_protocol(self): endpoints = get_initial_endpoints() ollama = next(e for e in endpoints if e["name"] == "stonks-ollama") assert ollama["protocol"] == "ollama_native" def test_ollama_endpoint_url(self): endpoints = get_initial_endpoints() ollama = next(e for e in endpoints if e["name"] == "stonks-ollama") assert ollama["base_url"] == "http://ollama.ollama-service.svc.cluster.local:11434" def test_ollama_endpoint_id_is_uuid(self): endpoints = get_initial_endpoints() ollama = next(e for e in endpoints if e["name"] == "stonks-ollama") assert isinstance(ollama["id"], UUID) def test_ollama_health_path(self): endpoints = get_initial_endpoints() ollama = next(e for e in endpoints if e["name"] == "stonks-ollama") assert ollama["health_path"] == "/api/tags" def test_vllm_endpoint_protocol(self): endpoints = get_initial_endpoints() vllm = next(e for e in endpoints if e["name"] == "stonks-vllm") assert vllm["protocol"] == "openai_chat" def test_vllm_endpoint_url(self): endpoints = get_initial_endpoints() vllm = next(e for e in endpoints if e["name"] == "stonks-vllm") assert vllm["base_url"] == "http://kube-vllm.stonks-oracle.svc.cluster.local:8000" def test_vllm_endpoint_id_is_uuid(self): endpoints = get_initial_endpoints() vllm = next(e for e in endpoints if e["name"] == "stonks-vllm") assert isinstance(vllm["id"], UUID) def test_vllm_health_path(self): endpoints = get_initial_endpoints() vllm = next(e for e in endpoints if e["name"] == "stonks-vllm") assert vllm["health_path"] == "/health" def test_endpoints_have_unique_ids(self): endpoints = get_initial_endpoints() ids = [e["id"] for e in endpoints] assert len(set(ids)) == len(ids) def test_endpoints_are_enabled(self): endpoints = get_initial_endpoints() for ep in endpoints: assert ep["enabled"] is True # ─── Initial deployments ─────────────────────────────────────────────────────── class TestInitialDeployments: """Test that initial deployments reference valid endpoints.""" def test_returns_two_deployments(self): deployments = get_initial_deployments() assert len(deployments) == 2 def test_deployments_reference_valid_endpoint_ids(self): """Every deployment references an endpoint from the seed set.""" endpoints = get_initial_endpoints() endpoint_ids = {e["id"] for e in endpoints} deployments = get_initial_deployments() for dep in deployments: assert dep["endpoint_id"] in endpoint_ids, ( f"Deployment {dep['served_model_name']} references unknown endpoint {dep['endpoint_id']}" ) def test_ollama_deployment_model_name(self): deployments = get_initial_deployments() ollama_dep = next(d for d in deployments if d["endpoint_id"] == OLLAMA_ENDPOINT_ID) assert ollama_dep["served_model_name"] == "qwen3.5:9b" def test_vllm_deployment_model_name(self): deployments = get_initial_deployments() vllm_dep = next(d for d in deployments if d["endpoint_id"] == VLLM_ENDPOINT_ID) assert vllm_dep["served_model_name"] == "AxionML/Qwen3.5-9B-NVFP4" def test_vllm_deployment_context_window(self): deployments = get_initial_deployments() vllm_dep = next(d for d in deployments if d["endpoint_id"] == VLLM_ENDPOINT_ID) assert vllm_dep["context_window"] == 8192 def test_vllm_deployment_max_output_tokens(self): deployments = get_initial_deployments() vllm_dep = next(d for d in deployments if d["endpoint_id"] == VLLM_ENDPOINT_ID) assert vllm_dep["max_output_tokens"] == 2048 def test_vllm_deployment_quantization(self): deployments = get_initial_deployments() vllm_dep = next(d for d in deployments if d["endpoint_id"] == VLLM_ENDPOINT_ID) assert vllm_dep["quantization"] == "NVFP4" def test_vllm_deployment_has_json_schema_capability(self): deployments = get_initial_deployments() vllm_dep = next(d for d in deployments if d["endpoint_id"] == VLLM_ENDPOINT_ID) assert vllm_dep["capabilities"]["json_schema"] is True def test_ollama_deployment_lacks_json_schema(self): deployments = get_initial_deployments() ollama_dep = next(d for d in deployments if d["endpoint_id"] == OLLAMA_ENDPOINT_ID) assert ollama_dep["capabilities"]["json_schema"] is False def test_deployments_have_unique_ids(self): deployments = get_initial_deployments() ids = [d["id"] for d in deployments] assert len(set(ids)) == len(ids) def test_deployments_are_enabled(self): deployments = get_initial_deployments() for dep in deployments: assert dep["enabled"] is True # ─── Agent conversion: ollama ────────────────────────────────────────────────── class TestConvertAgentOllama: """Test agent conversion maps ollama provider correctly.""" def test_ollama_agent_produces_binding(self): agents = [ {"id": "00000000-0000-4000-8000-aaaaaaaaaaaa", "model_provider": "ollama", "slug": "document-extractor"} ] bindings = convert_agent_providers(agents) assert len(bindings) == 1 def test_ollama_binding_endpoint_id(self): agents = [ {"id": "00000000-0000-4000-8000-aaaaaaaaaaaa", "model_provider": "ollama", "slug": "document-extractor"} ] bindings = convert_agent_providers(agents) assert bindings[0]["endpoint_id"] == OLLAMA_ENDPOINT_ID def test_ollama_binding_deployment_id(self): agents = [ {"id": "00000000-0000-4000-8000-aaaaaaaaaaaa", "model_provider": "ollama", "slug": "document-extractor"} ] bindings = convert_agent_providers(agents) assert bindings[0]["model_deployment_id"] == str(OLLAMA_DEPLOYMENT_ID) def test_ollama_binding_stage_extraction(self): agents = [ {"id": "00000000-0000-4000-8000-aaaaaaaaaaaa", "model_provider": "ollama", "slug": "document-extractor"} ] bindings = convert_agent_providers(agents) assert bindings[0]["stage"] == "extraction" def test_ollama_event_classifier_stage(self): agents = [ {"id": "00000000-0000-4000-8000-bbbbbbbbbbbb", "model_provider": "ollama", "slug": "event-classifier"} ] bindings = convert_agent_providers(agents) assert bindings[0]["stage"] == "classification" def test_ollama_thesis_rewriter_stage(self): agents = [ {"id": "00000000-0000-4000-8000-cccccccccccc", "model_provider": "ollama", "slug": "thesis-rewriter"} ] bindings = convert_agent_providers(agents) assert bindings[0]["stage"] == "thesis_rewrite" def test_ollama_binding_is_active(self): agents = [ {"id": "00000000-0000-4000-8000-aaaaaaaaaaaa", "model_provider": "ollama", "slug": "document-extractor"} ] bindings = convert_agent_providers(agents) assert bindings[0]["is_active"] is True # ─── Agent conversion: vllm ─────────────────────────────────────────────────── class TestConvertAgentVllm: """Test agent conversion maps vllm provider correctly.""" def test_vllm_agent_produces_binding(self): agents = [ {"id": "00000000-0000-4000-8000-dddddddddddd", "model_provider": "vllm", "slug": "document-extractor"} ] bindings = convert_agent_providers(agents) assert len(bindings) == 1 def test_vllm_binding_endpoint_id(self): agents = [ {"id": "00000000-0000-4000-8000-dddddddddddd", "model_provider": "vllm", "slug": "document-extractor"} ] bindings = convert_agent_providers(agents) assert bindings[0]["endpoint_id"] == VLLM_ENDPOINT_ID def test_vllm_binding_deployment_id(self): agents = [ {"id": "00000000-0000-4000-8000-dddddddddddd", "model_provider": "vllm", "slug": "document-extractor"} ] bindings = convert_agent_providers(agents) assert bindings[0]["model_deployment_id"] == str(VLLM_DEPLOYMENT_ID) def test_vllm_binding_stage(self): agents = [ {"id": "00000000-0000-4000-8000-dddddddddddd", "model_provider": "vllm", "slug": "document-extractor"} ] bindings = convert_agent_providers(agents) assert bindings[0]["stage"] == "extraction" def test_vllm_binding_is_active(self): agents = [ {"id": "00000000-0000-4000-8000-dddddddddddd", "model_provider": "vllm", "slug": "document-extractor"} ] bindings = convert_agent_providers(agents) assert bindings[0]["is_active"] is True def test_multiple_vllm_agents(self): agents = [ {"id": "00000000-0000-4000-8000-111111111111", "model_provider": "vllm", "slug": "document-extractor"}, {"id": "00000000-0000-4000-8000-222222222222", "model_provider": "vllm", "slug": "event-classifier"}, {"id": "00000000-0000-4000-8000-333333333333", "model_provider": "vllm", "slug": "thesis-rewriter"}, ] bindings = convert_agent_providers(agents) assert len(bindings) == 3 stages = [b["stage"] for b in bindings] assert "extraction" in stages assert "classification" in stages assert "thesis_rewrite" in stages # ─── Unknown providers raise error ──────────────────────────────────────────── class TestUnknownProviderError: """Test that unknown providers raise error (don't silently convert).""" def test_unknown_provider_raises(self): agents = [ {"id": "00000000-0000-4000-8000-eeeeeeeeeeee", "model_provider": "openai", "slug": "document-extractor"} ] with pytest.raises(UnknownProviderError) as exc_info: convert_agent_providers(agents) assert "openai" in str(exc_info.value) def test_unknown_provider_includes_agent_id(self): agents = [ {"id": "00000000-0000-4000-8000-eeeeeeeeeeee", "model_provider": "anthropic", "slug": "test"} ] with pytest.raises(UnknownProviderError) as exc_info: convert_agent_providers(agents) assert "00000000-0000-4000-8000-eeeeeeeeeeee" in str(exc_info.value) def test_empty_provider_skipped(self): """Agents with no provider set are skipped, not errored.""" agents = [ {"id": "00000000-0000-4000-8000-eeeeeeeeeeee", "model_provider": "", "slug": "document-extractor"} ] bindings = convert_agent_providers(agents) assert len(bindings) == 0 def test_none_provider_skipped(self): """Agents with None provider are skipped, not errored.""" agents = [ {"id": "00000000-0000-4000-8000-eeeeeeeeeeee", "model_provider": None, "slug": "document-extractor"} ] bindings = convert_agent_providers(agents) assert len(bindings) == 0 def test_mixed_valid_and_invalid_raises_on_invalid(self): """If any agent has an unknown provider, conversion fails immediately.""" agents = [ {"id": "00000000-0000-4000-8000-111111111111", "model_provider": "vllm", "slug": "document-extractor"}, {"id": "00000000-0000-4000-8000-222222222222", "model_provider": "unknown_thing", "slug": "test"}, ] with pytest.raises(UnknownProviderError) as exc_info: convert_agent_providers(agents) assert "unknown_thing" in str(exc_info.value) # ─── Conflicting defaults identification ────────────────────────────────────── class TestConflictingDefaults: """Test that conflicting defaults are identified.""" def test_returns_non_empty_list(self): conflicts = identify_conflicting_defaults() assert len(conflicts) > 0 def test_identifies_config_py(self): conflicts = identify_conflicting_defaults() config_conflicts = [c for c in conflicts if "config.py" in c] assert len(config_conflicts) >= 1 def test_identifies_migrations(self): conflicts = identify_conflicting_defaults() migration_conflicts = [c for c in conflicts if "migrations" in c] assert len(migration_conflicts) >= 1 def test_identifies_helm_values(self): conflicts = identify_conflicting_defaults() helm_conflicts = [c for c in conflicts if "helm" in c] assert len(helm_conflicts) >= 1 def test_identifies_kube_vllm_deployment(self): conflicts = identify_conflicting_defaults() kube_conflicts = [c for c in conflicts if "kube-vllm" in c] assert len(kube_conflicts) >= 1 def test_all_entries_are_strings(self): conflicts = identify_conflicting_defaults() for c in conflicts: assert isinstance(c, str) assert len(c) > 10 # Meaningful content # ─── Well-known ID consistency ───────────────────────────────────────────────── class TestWellKnownIds: """Ensure Python constants match SQL migration UUIDs.""" def test_ollama_endpoint_id_matches_sql(self): content = MIGRATION_PATH.read_text() assert str(OLLAMA_ENDPOINT_ID) in content def test_vllm_endpoint_id_matches_sql(self): content = MIGRATION_PATH.read_text() assert str(VLLM_ENDPOINT_ID) in content def test_ollama_deployment_id_matches_sql(self): content = MIGRATION_PATH.read_text() assert str(OLLAMA_DEPLOYMENT_ID) in content def test_vllm_deployment_id_matches_sql(self): content = MIGRATION_PATH.read_text() assert str(VLLM_DEPLOYMENT_ID) in content