stonks-oracle/tests/test_pbt_provider_routing.py

"""Property-based tests for provider routing and protocol resolution.

Feature: intelligence-pipeline-v3

Validates that the inference factory:
- Never silently falls back to Ollama for unknown protocols
- Correctly resolves "vllm" to "openai_chat"
- Fails closed with a typed error for any unrecognized protocol

Uses Hypothesis to verify these properties hold for arbitrary string inputs.

**Validates: Requirements 2.2, 2.6**
"""
from __future__ import annotations

from uuid import uuid4

from hypothesis import given, settings
from hypothesis import strategies as st

from services.shared.inference.clients.ollama_native import OllamaNativeClient
from services.shared.inference.clients.openai_compatible import OpenAICompatibleClient
from services.shared.inference.errors import InferenceError, InferenceErrorCategory
from services.shared.inference.factory import (
    KNOWN_PROTOCOLS,
    PROTOCOL_ALIASES,
    create_client,
    resolve_protocol,
)
from services.shared.inference.models import InferenceTarget, ProviderCapabilities

# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------


def _make_target(protocol: str) -> InferenceTarget:
    """Build a minimal InferenceTarget for testing protocol routing."""
    return InferenceTarget(
        endpoint_id=uuid4(),
        deployment_id=uuid4(),
        protocol=protocol,  # type: ignore[arg-type]
        base_url="http://localhost:11434",
        model="test-model",
        capabilities=ProviderCapabilities(
            chat_completions=True,
            json_schema=True,
        ),
    )


def _is_known_or_alias(s: str) -> bool:
    """Check if a string is a known protocol or alias after normalization."""
    normalized = s.strip().lower()
    return normalized in KNOWN_PROTOCOLS or normalized in PROTOCOL_ALIASES


# ---------------------------------------------------------------------------
# Property tests
# ---------------------------------------------------------------------------


class TestPropertyUnknownProtocolsFailClosed:
    """Property: Unknown protocols always raise InferenceError.

    For any string that is NOT in PROTOCOL_ALIASES keys AND not a known
    protocol, resolve_protocol() MUST raise InferenceError with
    CAPABILITY_UNAVAILABLE category.

    **Validates: Requirements 2.6**
    """

    @given(protocol=st.text(min_size=0, max_size=50))
    @settings(max_examples=100)
    def test_unknown_protocol_raises_error(self, protocol: str):
        """**Validates: Requirements 2.6**

        Any string not in known protocols or aliases must raise InferenceError.
        """
        if _is_known_or_alias(protocol):
            return  # skip known protocols — those should resolve fine

        try:
            resolve_protocol(protocol)
            raise AssertionError(
                f"resolve_protocol({protocol!r}) did not raise for unknown protocol"
            )
        except InferenceError as exc:
            assert exc.category == InferenceErrorCategory.CAPABILITY_UNAVAILABLE, (
                f"Expected CAPABILITY_UNAVAILABLE, got {exc.category} for {protocol!r}"
            )


class TestPropertyUnknownProtocolNeverProducesOllama:
    """Property: No unknown protocol input to create_client() produces an Ollama client.

    For any string that is NOT a known protocol or alias, create_client()
    must raise InferenceError — it must NEVER return an OllamaNativeClient
    as a silent fallback.

    **Validates: Requirements 2.6**
    """

    @given(protocol=st.text(min_size=0, max_size=50))
    @settings(max_examples=100)
    def test_unknown_protocol_never_returns_ollama_client(self, protocol: str):
        """**Validates: Requirements 2.6**

        create_client() with an unknown protocol must never return an
        OllamaNativeClient instance. It must raise InferenceError.
        """
        if _is_known_or_alias(protocol):
            return  # skip valid protocols

        target = _make_target(protocol)
        try:
            client = create_client(target)
            # If we got here, the factory returned a client for an unknown protocol
            assert not isinstance(client, OllamaNativeClient), (
                f"create_client() returned OllamaNativeClient for unknown protocol {protocol!r}"
            )
            assert not isinstance(client, OpenAICompatibleClient), (
                f"create_client() returned a client for unknown protocol {protocol!r} "
                f"instead of raising InferenceError"
            )
            raise AssertionError(
                f"create_client() returned {type(client)} for unknown protocol {protocol!r} "
                f"instead of raising InferenceError"
            )
        except InferenceError:
            pass  # Expected behavior — unknown protocol fails closed


class TestPropertyVLLMResolvesToOpenAIChat:
    """Property: "vllm" always resolves to "openai_chat".

    The backward-compatible alias must consistently map to the
    canonical openai_chat protocol regardless of whitespace or casing.

    **Validates: Requirements 2.2**
    """

    @given(
        padding_left=st.text(
            alphabet=st.sampled_from([" ", "\t"]),
            min_size=0,
            max_size=5,
        ),
        padding_right=st.text(
            alphabet=st.sampled_from([" ", "\t"]),
            min_size=0,
            max_size=5,
        ),
    )
    @settings(max_examples=100)
    def test_vllm_always_resolves_to_openai_chat(
        self, padding_left: str, padding_right: str
    ):
        """**Validates: Requirements 2.2**

        "vllm" with arbitrary surrounding whitespace always resolves to "openai_chat".
        """
        import warnings as _warnings

        protocol_input = f"{padding_left}vllm{padding_right}"
        with _warnings.catch_warnings():
            _warnings.simplefilter("ignore", DeprecationWarning)
            result = resolve_protocol(protocol_input)

        assert result == "openai_chat", (
            f"Expected 'openai_chat' for input {protocol_input!r}, got {result!r}"
        )

    @given(
        case_variant=st.sampled_from(["vllm", "VLLM", "Vllm", "vLLM", "VlLm"]),
    )
    @settings(max_examples=100)
    def test_vllm_case_insensitive(self, case_variant: str):
        """**Validates: Requirements 2.2**

        "vllm" in any case resolves to "openai_chat".
        """
        import warnings as _warnings

        with _warnings.catch_warnings():
            _warnings.simplefilter("ignore", DeprecationWarning)
            result = resolve_protocol(case_variant)

        assert result == "openai_chat", (
            f"Expected 'openai_chat' for input {case_variant!r}, got {result!r}"
        )