Python: Fix Foundry hosting test teardown across pytest retries

python/packages/foundry_hosting/tests/conftest.py

@@ -2,12 +2,17 @@
 
 """Fixtures for Foundry Hosting tests."""
 
+from collections.abc import Iterator
 from pathlib import Path
+from tempfile import TemporaryDirectory
 
 import pytest
 
 
 @pytest.fixture(autouse=True)
-def isolate_local_agentserver_state_root(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None:
-    """Give each test an independent local AgentServer state root."""
-    monkeypatch.setenv("AGENTSERVER_STATE_ROOT", str(tmp_path / "agentserver-state"))
+def isolate_local_agentserver_state_root(monkeypatch: pytest.MonkeyPatch) -> Iterator[None]:
+    """Give each test attempt an independent local AgentServer state root."""
+    # pytest-retry bypasses the report hooks needed by tmp_path's teardown bookkeeping.
+    with TemporaryDirectory(prefix="agentserver-test-") as directory:
+        monkeypatch.setenv("AGENTSERVER_STATE_ROOT", str(Path(directory) / "agentserver-state"))
+        yield

python/packages/foundry_hosting/tests/foundry_hosting/test_conftest.py

@@ -0,0 +1,110 @@
+# Copyright (c) Microsoft. All rights reserved.
+
+from __future__ import annotations
+
+import os
+import subprocess
+import sys
+import textwrap
+from pathlib import Path
+from xml.etree import ElementTree
+
+import pytest
+
+
+@pytest.mark.parametrize("original_root", [None, "existing-state-root"])
+@pytest.mark.parametrize("failures", [1, 3], ids=["passes-on-retry", "exhausts-retries"])
+def test_state_root_is_isolated_and_cleaned_per_attempt(
+    tmp_path: Path, original_root: str | None, failures: int
+) -> None:
+    fixture_source = (Path(__file__).parents[1] / "conftest.py").read_text(encoding="utf-8")
+    (tmp_path / "conftest.py").write_text(
+        fixture_source
+        + textwrap.dedent(
+            f"""
+
+            import os
+
+            @pytest.hookimpl(tryfirst=True)
+            def pytest_runtest_setup():
+                assert os.environ.get("AGENTSERVER_STATE_ROOT") == {original_root!r}
+
+            def pytest_sessionfinish():
+                assert os.environ.get("AGENTSERVER_STATE_ROOT") == {original_root!r}
+            """
+        ),
+        encoding="utf-8",
+    )
+    (tmp_path / "pytest.ini").write_text("[pytest]\n", encoding="utf-8")
+    (tmp_path / "test_retry.py").write_text(
+        textwrap.dedent(
+            f"""
+            import os
+            from pathlib import Path
+
+            import pytest
+
+            roots = []
+
+            @pytest.mark.flaky
+            def test_retry():
+                root = Path(os.environ["AGENTSERVER_STATE_ROOT"])
+                with Path("roots.txt").open("a", encoding="utf-8") as log:
+                    log.write(str(root) + "\\n")
+                assert root.name == "agentserver-state"
+                assert root.parent.is_dir()
+                assert not root.exists()
+                assert root not in roots
+                assert all(not previous.parent.exists() for previous in roots)
+                roots.append(root)
+                root.mkdir()
+                (root / "state.txt").write_text("attempt state", encoding="utf-8")
+                assert len(roots) > {failures}, "Synthetic attempt failure"
+            """
+        ),
+        encoding="utf-8",
+    )
+    # Isolate only the child harness, which needs pytest-retry but no application plugins.
+    env: dict[str, str] = {
+        **os.environ,
+        "PYTEST_DISABLE_PLUGIN_AUTOLOAD": "1",
+        "PYTEST_ADDOPTS": "",
+        "PYTEST_PLUGINS": "",
+    }
+    if original_root is None:
+        env.pop("AGENTSERVER_STATE_ROOT", None)
+    else:
+        env["AGENTSERVER_STATE_ROOT"] = original_root
+    result = subprocess.run(
+        [
+            sys.executable,
+            "-I",
+            "-m",
+            "pytest",
+            "-p",
+            "pytest_retry.retry_plugin",
+            "--import-mode=importlib",
+            "--retries=2",
+            "--retry-delay=0",
+            "--junitxml=results.xml",
+            "-q",
+        ],
+        cwd=tmp_path,
+        env=env,
+        capture_output=True,
+        text=True,
+        timeout=30,
+        check=False,
+    )
+    output = result.stdout + result.stderr
+    assert result.returncode == (0 if failures == 1 else 1), output
+    suite = ElementTree.parse(tmp_path / "results.xml").getroot().find("testsuite")
+    assert suite is not None, output
+    assert suite.attrib["tests"] == "1", output
+    assert suite.attrib["errors"] == "0", output
+    assert suite.attrib["failures"] == ("0" if failures == 1 else "1"), output
+    if failures == 3:
+        assert "Synthetic attempt failure" in output
+    roots = (tmp_path / "roots.txt").read_text(encoding="utf-8").splitlines()
+    assert len(roots) == len(set(roots)) == min(failures + 1, 3)
+    assert all(not Path(root).parent.exists() for root in roots)

python/packages/foundry_hosting/tests/test_state_store.py

@@ -1,8 +1,6 @@
 # Copyright (c) Microsoft. All rights reserved.
-import os
 from collections.abc import Callable
 from dataclasses import dataclass
-from pathlib import Path
 from types import SimpleNamespace
 from typing import Any
 from unittest.mock import AsyncMock, MagicMock, patch
@@ -74,13 +72,6 @@ def _platform_context(call_id: str = "call-1", user_id: str = "user-1") -> Found
     return FoundryAgentRequestContext(call_id=call_id, user_id=user_id)
 
 
-def test_local_agentserver_state_root_is_test_scoped(tmp_path: Path) -> None:
-    """Independent tests must not share the local fallback state directory."""
-    state_root = Path(os.environ["AGENTSERVER_STATE_ROOT"])
-
-    assert state_root.is_relative_to(tmp_path)
-
-
 def test_storage_providers_use_public_abstraction() -> None:
     assert issubclass(CheckpointStoreProvider, ContextScopedStoreProvider)
     assert not issubclass(CheckpointStoreProvider, StoreProvider)