repo_description: SRE Agent - CNCF Sandbox Project
repo_topics: aiops chatbot chatops devops devops-tools incident incident-management incident-response jira kubernetes llm llm-agent llm-framework llms monitoring observability prometheus site-reliability-engineering slack sre
code: import json
  import logging
  import os
  import re
  import urllib.parse
  import urllib.request
  from unittest.mock import MagicMock, patch
  
  import pytest
  import responses as responses_
  
  from holmes.core.tracing import get_active_branch_name, readable_timestamp
  from tests.llm.conftest import show_llm_summary_report
  from tests.llm.utils.braintrust import get_braintrust_url
  
  
  def pytest_addoption(parser):
      """Add custom pytest command line options"""
      parser.addoption(
          "--skip-setup",
          action="store_true",
          default=False,
          help="Skip running before_test commands for test cases (useful for iterative test development)",
      )
      parser.addoption(
          "--skip-cleanup",
          action="store_true",
          default=False,
          help="Skip running after_test commands for test cases (useful for debugging test failures)",
      )
      parser.addoption(
          "--only-setup",
          action="store_true",
          default=False,
          help="Only run before_test setup commands, skip the actual test execution",
      )
      parser.addoption(
          "--only-cleanup",
          action="store_true",
          default=False,
          help="Only run after_test cleanup commands, skip the actual test execution",
      )
      parser.addoption(
          "--retry-on-throttle",
          action="store_true",
          default=True,
          help="Retry tests when encountering API throttling/overload errors (default: True)",
      )
      parser.addoption(
          "--no-retry-on-throttle",
          action="store_false",
          dest="retry_on_throttle",
          help="Disable retrying tests on API throttling/overload errors",
      )
      parser.addoption(
          "--include-intermediate",
          action="store_true",
          default=True,
          help="Include intermediate LLM outputs when evaluating correctness (default: True)",
      )
      parser.addoption(
          "--no-include-intermediate",
          action="store_false",
          dest="include_intermediate",
          help="Exclude intermediate LLM outputs from evaluation (only use final answer)",
      )
      parser.addoption(
          "--strict-setup-mode",
          type=str,
          default="false",
          choices=["true", "false"],
          help="Fail entire pytest run if any test setup fails (true/false, default: false) - useful for benchmarks to ensure all tests run properly",
      )
      parser.addoption(
          "--strict-setup-exceptions",
          type=str,
          default="",
          help="Comma-separated list of test IDs that are allowed to have setup failures even with --strict-setup-mode",
      )
      parser.addoption(
          "--additional-system-prompt-url",
          action="store",
          default=None,
          help="URL returning the additional system prompt text or JSON (expects 'additional_system_prompt' field)",
      )
  
  
  def pytest_configure(config):
      """Configure pytest settings"""
      # Disable SSL verification if SSL_VERIFY env var is set to false/0
      # This is useful for running tests in environments with TLS interception proxies
      ssl_verify_env = os.environ.get("SSL_VERIFY", "true").lower()
      if ssl_verify_env in ("false", "0", "no"):
          try:
              import litellm
  
              litellm.ssl_verify = False
          except ImportError:
              pass
  
          # Also patch OpenAI client to use verify=False for httpx
          try:
              import httpx
              import openai
  
              _original_openai_init = openai.OpenAI.__init__
  
              def _patched_openai_init(self, *args, **kwargs):
                  if "http_client" not in kwargs:
                      kwargs["http_client"] = httpx.Client(verify=False)
                  return _original_openai_init(self, *args, **kwargs)
  
              openai.OpenAI.__init__ = _patched_openai_init
          except ImportError:
              pass
  
      # Auto-derive CONFLUENCE_SA_BASE_URL from CONFLUENCE_BASE_URL if not already set.
      # SA tokens require the Atlassian Cloud API gateway (api.atlassian.com/ex/confluence/{cloudId}/...)
      # rather than the direct instance URL. We discover the cloud ID via the public tenant_info endpoint.
      confluence_base = os.environ.get("CONFLUENCE_BASE_URL")
      if confluence_base and not os.environ.get("CONFLUENCE_SA_BASE_URL"):
          parsed = urllib.parse.urlparse(confluence_base)
          if parsed.scheme not in ("http", "https"):
              logging.warning(f"CONFLUENCE_BASE_URL has unsupported scheme '{parsed.scheme}', skipping SA URL derivation")
          else:
              try:
                  tenant_url = f"{confluence_base.rstrip('/')}/_edge/tenant_info"
                  with urllib.request.urlopen(tenant_url, timeout=10) as resp:
                      cloud_id = json.loads(resp.read())["cloudId"]
                  os.environ["CONFLUENCE_SA_BASE_URL"] = f"https://api.atlassian.com/ex/confluence/{cloud_id}"
                  os.environ["CONFLUENCE_CLOUD_ID"] = cloud_id
                  logging.info(f"Auto-derived CONFLUENCE_SA_BASE_URL and CONFLUENCE_CLOUD_ID from cloud ID {cloud_id}")
              except Exception as e:
                  logging.warning(f"Could not auto-derive CONFLUENCE_SA_BASE_URL: {e}")
  
      # Configure worker-specific log files for xdist compatibility
      # worker_id = getattr(config, "workerinput", {}).get("workerid", "master")
      # if worker_id != "master":
      #    # Set worker-specific log file to avoid conflicts
      #    config.option.log_file = f"tests-{worker_id}.log"
  
      # Determine worker id
      # Also see: https://pytest-xdist.readthedocs.io/en/latest/how-to.html#creating-one-log-file-for-each-worker
      # worker_id = os.environ.get("PYTEST_XDIST_WORKER", default="gw0")
  
      # # Create logs folder
      # logs_folder = os.environ.get("LOGS_FOLDER", default="logs_folder")
      # os.makedirs(logs_folder, exist_ok=True)
  
      # # Create file handler to output logs into corresponding worker file
      # file_handler = logging.FileHandler(f"{logs_folder}/logs_worker_{worker_id}.log", mode="w")
      # file_handler.setFormatter(
      #     logging.Formatter(
      #         fmt="{asctime} {levelname}:{name}:{lineno}:{message}",
      #         style="{",
      #     )
      # )
      # # Create stream handler to output logs on console
      # # This is a workaround for a known limitation:
      # # https://pytest-xdist.readthedocs.io/en/latest/known-limitations.html
      # console_handler = logging.StreamHandler(sys.stderr)  # pytest only prints error logs
      # console_handler.setFormatter(
      #     logging.Formatter(
      #         # Include worker id in log messages, \r is needed to separate lines in console
      #         fmt="\r{asctime} " + worker_id + ":{levelname}:{name}:{lineno}:{message}",
      #         style="{",
      #     )
      # )
      # # Configure logging
      # logging.basicConfig(level=logging.INFO, force=True, handlers=[console_handler, file_handler])
  
      # Suppress noisy LiteLLM logs during testing
      logging.getLogger("LiteLLM").setLevel(logging.ERROR)
      # Also suppress the verbose logger used by LiteLLM
      logging.getLogger("LiteLLM.verbose_logger").setLevel(logging.ERROR)
      # Suppress litellm sub-loggers
      logging.getLogger("litellm").setLevel(logging.ERROR)
      logging.getLogger("litellm.cost_calculator").setLevel(logging.ERROR)
      logging.getLogger("litellm.litellm_core_utils").setLevel(logging.ERROR)
      logging.getLogger("litellm.litellm_core_utils.litellm_logging").setLevel(
          logging.ERROR
      )
      # Suppress httpx HTTP request logs
      logging.getLogger("httpx").setLevel(logging.WARNING)
      logging.getLogger("httpcore").setLevel(logging.WARNING)
  
      if not os.getenv("EXPERIMENT_ID"):
          try:
              username = os.getlogin()
          except OSError:
              # os.getlogin() fails in environments without a terminal (e.g., GitHub Actions)
              username = os.getenv("USER", "ci")
          git_branch = get_active_branch_name()
  
          # Include pytest parameters in experiment ID
          params = []
          if config.option.keyword:
              params.append(f"k={config.option.keyword}")
          if config.option.markexpr:
              params.append(f"m={config.option.markexpr}")
  
          params_str = f"-{'-'.join(params)}" if params else ""
          os.environ["EXPERIMENT_ID"] = (
              f"{username}-{git_branch}{params_str}-{readable_timestamp()}"
          )
  
  
  def pytest_report_header(config):
      braintrust_api_key = os.environ.get("BRAINTRUST_API_KEY")
      if not braintrust_api_key:
          return ""
  
      experiment_url = get_braintrust_url()
      clickable_url = f"\033]8;;{experiment_url}\033\\{experiment_url}\033]8;;\033\\"
      return f"Eval results: (link valid once setup completes): {clickable_url}"
  
  
  # due to pytest quirks, we need to define this in the main conftest.py - when defined in the llm conftest.py it
  # is SOMETIMES picked up and sometimes not, depending on how the test was invokedr
  pytest_terminal_summary = show_llm_summary_report
  
  
  @pytest.fixture(autouse=True)
  def patch_supabase(monkeypatch):
      monkeypatch.setattr("holmes.core.supabase_dal.ROBUSTA_ACCOUNT_ID", "test-cluster")
      monkeypatch.setattr(
          "holmes.core.supabase_dal.STORE_URL", "https://fakesupabaseref.supabase.co"
      )
      monkeypatch.setattr(
          "holmes.core.supabase_dal.STORE_API_KEY",
          "[REDACTED_JWT]",
      )
      monkeypatch.setattr("holmes.core.supabase_dal.STORE_EMAIL", "mock_store_user")
      monkeypatch.setattr(
          "holmes.core.supabase_dal.STORE_PASSWORD", "mock_store_password"
      )
  
  
  @pytest.fixture(autouse=True, scope="session")
  def storage_dal_mock():
      with patch("holmes.config.SupabaseDal") as MockSupabaseDal:
          mock_supabase_dal_instance = MagicMock()
          MockSupabaseDal.return_value = mock_supabase_dal_instance
          mock_supabase_dal_instance.sign_in.return_value = "mock_supabase_user_id"
          mock_supabase_dal_instance.get_ai_credentials.return_value = (
              "mock_account_id",
              "mock_session_token",
          )
          yield mock_supabase_dal_instance
  
  
  @pytest.fixture(autouse=True)
  def responses():
      with responses_.RequestsMock() as rsps:
          rsps.add_passthru("https://www.braintrust.dev")
          rsps.add_passthru("https://api.braintrust.dev")  # Allow Braintrust API calls
          rsps.add_passthru(
              "https://api.newrelic.com/graphql"
          )  # Allow New Relic API calls
          rsps.add_passthru(
              "https://api.eu.newrelic.com/graphql"
          )  # Allow New Relic API calls
          rsps.add_passthru("http://localhost")
  
          # Allow all Datadog API calls to pass through (all regions and endpoints)
          rsps.add_passthru("https://api.datadoghq.com")
          rsps.add_passthru("https://api.datadoghq.eu")
          rsps.add_passthru("https://api.ddog-gov.com")
          rsps.add_passthru("https://api.us3.datadoghq.com")
          rsps.add_passthru("https://api.us5.datadoghq.com")
          rsps.add_passthru("https://api.ap1.datadoghq.com")
          rsps.add_passthru("https://app.datadoghq.com")
          rsps.add_passthru("https://app.datadoghq.eu")
          # Allow all Coralogix API calls (query and ingestion endpoints, all regions)
          rsps.add_passthru(re.compile(r"https://.*\.coralogix\.com"))
          rsps.add_passthru(re.compile(r"https://.*\.coralogix\.us"))
          rsps.add_passthru(re.compile(r"https://.*\.coralogix\.in"))
  
          # Allow Elasticsearch/OpenSearch Cloud API calls (various hosting regions)
          rsps.add_passthru(re.compile(r"https://.*\.cloud\.es\.io"))  # Elastic Cloud
          rsps.add_passthru(re.compile(r"https://.*\.elastic-cloud\.com"))  # Azure-hosted
          rsps.add_passthru(
              re.compile(r"https://.*\.es\.amazonaws\.com")
          )  # AWS OpenSearch
  
          # Allow Confluence/Atlassian Cloud API calls
          rsps.add_passthru(re.compile(r"https://.*\.atlassian\.net"))
          rsps.add_passthru("https://api.atlassian.com")  # Atlassian Cloud API gateway
  
          # Allow
          rsps.add_passthru("https://google.com")
          rsps.add_passthru("https://burgergooglenetworkspam.co.uk")
          rsps.add_passthru("https://www.google.com")
          rsps.add_passthru("https://www.burgergooglenetworkspam.co.uk")
          yield rsps
  