test: retry transient external-service failures in snippet runners

The generative/hybrid search snippet tests and the Claude llms.txt fetch test
intermittently red the suite on transient blips (gRPC UNAVAILABLE, deadline
exceeded, read timeouts, 429/5xx) against the shared test cluster + LLM APIs.
Add a narrow transient-error retry (utils.retry_on_transient) around the Python,
TypeScript, and C# snippet runners, and retry the Claude fetch up to 3x. Real
failures (assertion/logic/syntax errors) still fail immediately — only messages
matching a small transient-marker set are retried.

Author: g-despot

tests/test_csharp.py

@@ -2,6 +2,8 @@
 import pytest
 import os
 
+import utils
+
 
 CSHARP_CSPROJ = "_includes/code/csharp/WeaviateProject.Tests.csproj"
 
@@ -14,9 +16,19 @@ def run_csharp_test(test_class, empty_weaviates):
     ]
     env = dict(os.environ)
 
+    def _run():
+        result = subprocess.run(command, env=env, capture_output=True, text=True)
+        if result.stdout.strip():
+            print(result.stdout)
+        if result.returncode != 0:
+            raise Exception(
+                f"C# {test_class} failed (exit {result.returncode})\n"
+                f"--- STDERR ---\n{result.stderr}\n--- STDOUT ---\n{result.stdout}"
+            )
+
     try:
-        subprocess.check_call(command, env=env)
-    except subprocess.CalledProcessError as error:
+        utils.retry_on_transient(_run, label=test_class)
+    except Exception as error:
         pytest.fail(f"C# {test_class} failed with error: {error}")
 
 

tests/test_docs_indexability.py

@@ -453,43 +453,53 @@ def test_claude_can_fetch_llms_txt():
     # The llms.txt file starts with "# Weaviate Documentation" and contains
     # section headings like "## agents", "## cloud", "## weaviate".
     # Ask Claude to quote specific content to prove it fetched the real file.
-    response = client.messages.create(
-        model="claude-haiku-4-5-20251001",
-        max_tokens=2048,
-        tools=[{
-            "type": "web_fetch_20250910",
-            "name": "web_fetch",
-            "max_uses": 1,
-            "allowed_domains": ["docs.weaviate.io", "weaviate.io"],
-        }],
-        messages=[{
-            "role": "user",
-            "content": (
-                f"Fetch {url} and tell me: "
-                "1) What is the first heading line of the file (copy it verbatim)? "
-                "2) List ALL the top-level section headings (lines starting with '## '). "
-                "3) Does it mention code examples in multiple languages? Which ones?"
-            ),
-        }],
-    )
+    # LLM responses are non-deterministic, so retry a few times: pass as soon
+    # as one attempt satisfies ALL conditions; only fail if every attempt does.
+    required_sections = ["agents", "cloud", "weaviate"]
+    last_text = ""
+    for attempt in range(3):
+        response = client.messages.create(
+            model="claude-haiku-4-5-20251001",
+            max_tokens=2048,
+            tools=[{
+                "type": "web_fetch_20250910",
+                "name": "web_fetch",
+                "max_uses": 1,
+                "allowed_domains": ["docs.weaviate.io", "weaviate.io"],
+            }],
+            messages=[{
+                "role": "user",
+                "content": (
+                    f"Fetch {url} and tell me: "
+                    "1) What is the first heading line of the file (copy it verbatim)? "
+                    "2) List ALL the top-level section headings (lines starting with '## '). "
+                    "3) Does it mention code examples in multiple languages? Which ones?"
+                ),
+            }],
+        )
+        last_text = _extract_text_from_response(response)
+        tl = last_text.lower()
+        if "weaviate" in tl and all(s in tl for s in required_sections) and "python" in tl:
+            return
+        time.sleep(5)
 
-    text = _extract_text_from_response(response)
-    text_lower = text.lower()
+    # All attempts fell short — surface the last response via the existing assertions
+    tl = last_text.lower()
 
     # Must identify Weaviate
-    assert "weaviate" in text_lower, (
-        f"Claude couldn't identify Weaviate in llms.txt. Response: {text[:500]}"
+    assert "weaviate" in tl, (
+        f"Claude couldn't identify Weaviate in llms.txt. Response: {last_text[:500]}"
     )
 
     # Must find the key top-level sections from llms.txt
-    for section in ["agents", "cloud", "weaviate"]:
-        assert section in text_lower, (
-            f"Claude didn't find '{section}' section in llms.txt. Response: {text[:1000]}"
+    for section in required_sections:
+        assert section in tl, (
+            f"Claude didn't find '{section}' section in llms.txt. Response: {last_text[:1000]}"
         )
 
     # Must identify multi-language code examples
-    assert "python" in text_lower, (
-        f"Claude didn't find Python mentioned in llms.txt. Response: {text[:500]}"
+    assert "python" in tl, (
+        f"Claude didn't find Python mentioned in llms.txt. Response: {last_text[:500]}"
     )
 
 

tests/test_python.py

@@ -4,16 +4,19 @@
 
 
 def run_py_script(script_loc, custom_replace_pairs=None):
-    if custom_replace_pairs:
-        temp_proc_script_loc = utils.load_and_prep_temp_file(
-            script_loc, lang="py", custom_replace_pairs=custom_replace_pairs
-        )
-        utils.execute_py_script_as_module(
-            temp_proc_script_loc.read_text(), Path(script_loc).stem
-        )
-    else:
-        proc_script = utils.load_and_prep_script(script_loc)
-        utils.execute_py_script_as_module(proc_script, Path(script_loc).stem)
+    def _exec():
+        if custom_replace_pairs:
+            temp_proc_script_loc = utils.load_and_prep_temp_file(
+                script_loc, lang="py", custom_replace_pairs=custom_replace_pairs
+            )
+            utils.execute_py_script_as_module(
+                temp_proc_script_loc.read_text(), Path(script_loc).stem
+            )
+        else:
+            proc_script = utils.load_and_prep_script(script_loc)
+            utils.execute_py_script_as_module(proc_script, Path(script_loc).stem)
+
+    utils.retry_on_transient(_exec, label=str(script_loc))
 
 
 def run_pyv3_script(script_loc):

tests/test_typescript.py

@@ -9,7 +9,9 @@ def run_ts_script(script_loc, custom_replace_pairs=None):
     command = ["npx", "tsx", temp_proc_script_loc]
 
     try:
-        utils.run_script(command, script_loc)
+        utils.retry_on_transient(
+            lambda: utils.run_script(command, script_loc), label=str(script_loc)
+        )
     except Exception as e:
         pytest.fail(str(e))
 

tests/utils.py

@@ -2,12 +2,50 @@
 import re
 import subprocess
 import tempfile
+import time
 import runpy
 from pathlib import Path
 from dotenv import load_dotenv
 
 load_dotenv()
 
+
+# Substrings (case-insensitive) that indicate a transient external-service
+# failure worth retrying (gRPC blips, timeouts, rate limits, 5xx) rather than a
+# real snippet/assertion bug. Kept deliberately narrow.
+TRANSIENT_MARKERS = (
+    "unavailable", "deadline exceeded", "deadline_exceeded",
+    "timed out", "timeout", "read operation timed out",
+    "connection reset", "econnreset", "connection aborted",
+    "502", "503", "504", "429", "too many requests",
+    "rate limit", "overloaded", "temporarily unavailable",
+)
+
+
+def is_transient_error(err) -> bool:
+    """True if the error message looks like a transient external-service blip."""
+    msg = str(err).lower()
+    return any(marker in msg for marker in TRANSIENT_MARKERS)
+
+
+def retry_on_transient(fn, *, retries: int = 2, delay: int = 8, label: str = ""):
+    """Call fn(); on a TRANSIENT error, wait and retry up to `retries` times.
+    Non-transient errors (assertion/logic/syntax) re-raise immediately."""
+    for attempt in range(retries + 1):
+        try:
+            return fn()
+        except Exception as err:
+            if attempt < retries and is_transient_error(err):
+                print(
+                    f"[retry] transient failure on {label} "
+                    f"(attempt {attempt + 1}/{retries + 1}), retrying in {delay}s: "
+                    f"{str(err)[:300]}"
+                )
+                time.sleep(delay)
+                continue
+            raise
+
+
 def load_script(script_path: str) -> str:
     with open(script_path, "r") as f:
         code_block = f.read()