tracer-cloud--opensre/tests/tools/test_availability.py

"""Tests for tool availability helpers."""

from __future__ import annotations

from integrations.bitbucket.availability import bitbucket_available_or_backend
from integrations.cloudwatch.availability import cloudwatch_is_available
from integrations.datadog.availability import datadog_available_or_backend
from integrations.eks.availability import eks_available_or_backend


class TestEksAvailableOrBackend:
    def test_eks_missing(self) -> None:
        sources: dict[str, dict] = {}
        assert eks_available_or_backend(sources) is False

    def test_eks_empty(self) -> None:
        sources = {"eks": {}}
        assert eks_available_or_backend(sources) is False

    def test_eks_verified(self) -> None:
        sources = {"eks": {"connection_verified": True}}
        assert eks_available_or_backend(sources) is True

    def test_eks_backend(self) -> None:
        sources = {"eks": {"_backend": object()}}
        assert eks_available_or_backend(sources) is True

    def test_eks_not_available(self) -> None:
        sources = {"eks": {"connection_verified": False}}
        assert eks_available_or_backend(sources) is False

    def test_eks_backend_none(self) -> None:
        sources = {"eks": {"_backend": None}}
        assert eks_available_or_backend(sources) is False

    def test_eks_backend_overrides_failed_verification(self) -> None:
        sources = {"eks": {"connection_verified": False, "_backend": object()}}
        assert eks_available_or_backend(sources) is True

    def test_eks_check_ignores_bench_backend_in_dedicated_slot(self) -> None:
        """The bench adapter puts its replay backend at ``_bench_backend``,
        NOT ``_backend``. Production availability checks only look at
        ``_backend`` and ``connection_verified`` — they must stay completely
        unaware of bench backends. Regression-pin for the slot-separation
        refactor (no more ``is_cloudopsbench_backend`` marker checks)."""
        # Bench-style backend in its own slot — _backend remains None.
        sources = {"eks": {"_bench_backend": object()}}
        assert eks_available_or_backend(sources) is False

    def test_eks_check_uses_only_backend_and_verified_no_provider_specific_logic(
        self,
    ) -> None:
        """Belt-and-suspenders: the function's full decision MUST be a clean
        ``bool(connection_verified or _backend)``. No special-case branches,
        no marker attribute lookups, no provider-aware logic. Adding
        provider-specific branches here would re-couple production tool
        availability to bench backend types."""
        # Connection verified alone → True
        assert eks_available_or_backend({"eks": {"connection_verified": True}}) is True
        # Backend alone → True
        assert eks_available_or_backend({"eks": {"_backend": object()}}) is True
        # Both → True
        assert (
            eks_available_or_backend({"eks": {"connection_verified": True, "_backend": object()}})
            is True
        )
        # Neither → False
        assert eks_available_or_backend({"eks": {}}) is False
        # Extra unrelated slots (like _bench_backend) must not affect the result
        assert (
            eks_available_or_backend({"eks": {"_bench_backend": object(), "anything": True}})
            is False
        )


class TestDatadogAvailableOrBackend:
    def test_datadog_missing(self) -> None:
        sources: dict[str, dict] = {}
        assert datadog_available_or_backend(sources) is False

    def test_datadog_empty(self) -> None:
        sources = {"datadog": {}}
        assert datadog_available_or_backend(sources) is False

    def test_datadog_verified(self) -> None:
        sources = {"datadog": {"connection_verified": True}}
        assert datadog_available_or_backend(sources) is True

    def test_datadog_backend(self) -> None:
        sources = {"datadog": {"_backend": object()}}
        assert datadog_available_or_backend(sources) is True

    def test_datadog_not_available(self) -> None:
        sources = {"datadog": {"connection_verified": False}}
        assert datadog_available_or_backend(sources) is False

    def test_datadog_backend_none(self) -> None:
        sources = {"datadog": {"_backend": None}}
        assert datadog_available_or_backend(sources) is False

    def test_datadog_backend_overrides_failed_verification(self) -> None:
        sources = {"datadog": {"connection_verified": False, "_backend": object()}}
        assert datadog_available_or_backend(sources) is True


class TestBitbucketAvailableOrBackend:
    def test_bitbucket_missing(self) -> None:
        sources: dict[str, dict] = {}
        assert bitbucket_available_or_backend(sources) is False

    def test_bitbucket_empty(self) -> None:
        sources = {"bitbucket": {}}
        assert bitbucket_available_or_backend(sources) is False

    def test_bitbucket_verified_with_credentials(self) -> None:
        sources = {
            "bitbucket": {
                "connection_verified": True,
                "workspace": "acme",
                "username": "bb-user",
                "app_password": "bb-pass",
            }
        }
        assert bitbucket_available_or_backend(sources) is True

    def test_bitbucket_missing_credentials(self) -> None:
        sources = {"bitbucket": {"connection_verified": True, "workspace": "acme"}}
        assert bitbucket_available_or_backend(sources) is False

    def test_bitbucket_requires_connection_verification(self) -> None:
        sources = {
            "bitbucket": {
                "workspace": "acme",
                "username": "bb-user",
                "app_password": "bb-pass",
            }
        }
        assert bitbucket_available_or_backend(sources) is False

    def test_bitbucket_backend(self) -> None:
        sources = {"bitbucket": {"_backend": object()}}
        assert bitbucket_available_or_backend(sources) is True

    def test_bitbucket_backend_overrides_failed_verification(self) -> None:
        sources = {"bitbucket": {"connection_verified": False, "_backend": object()}}
        assert bitbucket_available_or_backend(sources) is True


class TestCloudwatchIsAvailable:
    def test_cloudwatch_missing(self) -> None:
        sources: dict[str, dict] = {}
        assert cloudwatch_is_available(sources) is False

    def test_cloudwatch_present_empty(self) -> None:
        sources = {"cloudwatch": {}}
        assert cloudwatch_is_available(sources) is False

    def test_cloudwatch_with_data(self) -> None:
        sources = {"cloudwatch": {"log_group": "test"}}
        assert cloudwatch_is_available(sources) is True