Create a subprocess for uploading the report as it goes

Author: ian-sift

python/lib/sift_client/_internal/low_level_wrappers/test_results.py

@@ -39,6 +39,7 @@ def compare_test_report_fields(simulated: TestReport, actual: TestReport) -> Non
     assert simulated.system_operator == actual.system_operator
     assert simulated.start_time == actual.start_time
     assert simulated.end_time == actual.end_time
+    assert simulated.metadata == actual.metadata
 
 
 def compare_test_step_fields(simulated: TestStep, actual: TestStep) -> None:
@@ -667,21 +668,13 @@ def test_replay_log_file_round_trip(self, sift_client, nostromo_run, tmp_path):
 
         # Report: updates should have been folded in before create
         compare_test_report_fields(replay_result.report, direct["report"])
-        assert replay_result.report.status == TestStatus.FAILED
 
         # Steps (matched by name)
         replayed_steps_by_name = {s.name: s for s in replay_result.steps}
         for direct_step in direct["steps"].values():
             replayed_step = replayed_steps_by_name[direct_step.name]
             compare_test_step_fields(replayed_step, direct_step)
 
-        assert replayed_steps_by_name["RT Step 2"].status == TestStatus.FAILED
-
-        # Nested step parent should point to the replayed step1
-        assert replayed_steps_by_name["RT Step 1.1"].parent_step_id == (
-            replayed_steps_by_name["RT Step 1"].id_
-        )
-
         # Measurements (matched by name)
         replayed_measurements_by_name = {m.name: m for m in replay_result.measurements}
         for direct_m in direct["measurements"].values():

python/lib/sift_client/_tests/resources/test_test_results.py

@@ -1,3 +1,4 @@
+import time
 from datetime import datetime, timezone
 
 import numpy as np
@@ -16,11 +17,46 @@
     assign_value_to_measurement,
     evaluate_measurement_bounds,
 )
-from sift_client.util.test_results.context_manager import NewStep
+from sift_client.util.test_results.context_manager import NewStep, ReportContext
 
 pytestmark = pytest.mark.integration
 
 
+class TestLogReplay:
+    """Test that the incremental replay subprocess creates real API objects."""
+
+    def test_replay_creates_report(self, sift_client):
+        unique_name = f"replay-test-{datetime.now(timezone.utc).isoformat()}"
+
+        with ReportContext(
+            sift_client,
+            name=unique_name,
+            test_case="test_replay_creates_report",
+            log_file=True,
+        ) as rc:
+            with rc.new_step(name="Step A") as step_a:
+                with step_a.substep(name="Step A.1"):
+                    pass
+            with rc.new_step(name="Step B"):
+                pass
+
+        # Wait to ensure the report creation has completed.
+        time.sleep(2)
+
+        reports = sift_client.test_results.list_(name=unique_name)
+        assert len(reports) >= 1, f"Expected report '{unique_name}' to be created by replay"
+        replay_report = reports[0]
+        assert replay_report.name == unique_name
+
+        steps = sift_client.test_results.list_steps(test_reports=[replay_report])
+        step_names = {s.name for s in steps}
+        assert "Step A" in step_names
+        assert "Step A.1" in step_names
+        assert "Step B" in step_names
+
+        sift_client.test_results.delete(test_report=replay_report)
+
+
 class TestContextManager:
     def test_link_run_to_report(self, report_context, nostromo_run):
         report_context.report.update({"run_id": nostromo_run.id_})

python/lib/sift_client/_tests/util/test_test_results_utils.py

@@ -1963,15 +1963,20 @@ class TestResultsAPI:
         """
         ...
 
-    def replay_log_file(self, log_file: str | Path) -> ReplayResult:
-        """Replay a log file by parsing each entry, simulating the results, then creating for real.
+    def replay_log_file(self, log_file: str | Path, *, incremental: bool = False) -> ReplayResult:
+        """Replay a log file, creating real API objects from the logged simulation data.
 
-        This method reads a log file created by the simulation logging, reconstructs
-        all the objects via simulation, and then creates them via the actual API.
-        IDs are mapped from simulated to real during the creation process.
+        Two modes are available:
+
+        * **batch** (default): Parse the entire log, reconstruct objects via
+          simulation, then create them all via the API in one pass.
+        * **incremental**: Walk the log line-by-line, issuing the real API call
+          for each entry. The ``LogTracking`` header is updated after every
+          successful call so a subsequent invocation picks up where it left off.
 
         Args:
             log_file: Path to the log file to replay.
+            incremental: If True, use incremental mode.
 
         Returns:
             A ReplayResult containing the created report, steps, and measurements.

python/lib/sift_client/resources/sync_stubs/__init__.pyi

@@ -619,20 +619,27 @@ async def delete_measurement(self, *, test_measurement: str | TestMeasurement) -
     async def replay_log_file(
         self,
         log_file: str | Path,
+        *,
+        incremental: bool = False,
     ) -> ReplayResult:
-        """Replay a log file by parsing each entry, simulating the results, then creating for real.
+        """Replay a log file, creating real API objects from the logged simulation data.
+
+        Two modes are available:
 
-        This method reads a log file created by the simulation logging, reconstructs
-        all the objects via simulation, and then creates them via the actual API.
-        IDs are mapped from simulated to real during the creation process.
+        * **batch** (default): Parse the entire log, reconstruct objects via
+          simulation, then create them all via the API in one pass.
+        * **incremental**: Walk the log line-by-line, issuing the real API call
+          for each entry. The ``LogTracking`` header is updated after every
+          successful call so a subsequent invocation picks up where it left off.
 
         Args:
             log_file: Path to the log file to replay.
+            incremental: If True, use incremental mode.
 
         Returns:
             A ReplayResult containing the created report, steps, and measurements.
         """
-        result = await self._low_level_client.replay_log_file(log_file)
+        result = await self._low_level_client.replay_log_file(log_file, incremental=incremental)
         result.report = self._apply_client_to_instance(result.report)
         result.steps = self._apply_client_to_instances(result.steps)
         result.measurements = self._apply_client_to_instances(result.measurements)

python/lib/sift_client/resources/test_results.py

@@ -3,17 +3,38 @@
 from __future__ import annotations
 
 import argparse
+import logging
 import os
+import select
+import sys
+import tempfile
 
 from sift_client import SiftClient, SiftConnectionConfig
 
+logger = logging.getLogger(__name__)
+
+
+def _print_result(result) -> None:
+    print(f"Report: {result.report.name} (id={result.report.id_})")
+    print(f"Steps:  {len(result.steps)}")
+    for step in result.steps:
+        print(f"  - {step.step_path} [{step.status}]")
+    print(f"Measurements: {len(result.measurements)}")
+    for m in result.measurements:
+        print(f"  - {m.name}: passed={m.passed}")
+
 
 def main() -> None:
     """Replay a test result simulation log file against the Sift API."""
     parser = argparse.ArgumentParser(
         description="Replay a test result simulation log file against the Sift API.",
     )
     parser.add_argument("log_file", help="Path to the .jsonl log file to replay.")
+    parser.add_argument(
+        "--incremental",
+        action="store_true",
+        help="Replay line-by-line, tracking progress so reruns pick up where they left off.",
+    )
     parser.add_argument("--grpc-url", default=os.getenv("SIFT_GRPC_URI", "localhost:50051"))
     parser.add_argument("--rest-url", default=os.getenv("SIFT_REST_URI", "localhost:8080"))
     parser.add_argument("--api-key", default=os.getenv("SIFT_API_KEY", ""))
@@ -30,15 +51,38 @@ def main() -> None:
         )
     )
 
-    result = client.test_results.replay_log_file(args.log_file)
+    try:
+        if args.incremental:
+            result = _incremental_loop(client, args.log_file)
+        else:
+            result = client.test_results.replay_log_file(args.log_file)
+            fp = os.path.abspath(args.log_file)
+            if fp.startswith(tempfile.gettempdir()):
+                os.remove(fp)
+        if result:
+            _print_result(result)
+    except Exception as e:
+        logger.error(e)
+        logger.error(
+            f"Error replaying log file: {args.log_file}.\n"
+            f"  Can replay with `replay-test-result-log {args.log_file}`."
+        )
+        raise
 
-    print(f"Report: {result.report.name} (id={result.report.id_})")
-    print(f"Steps:  {len(result.steps)}")
-    for step in result.steps:
-        print(f"  - {step.step_path} [{step.status}]")
-    print(f"Measurements: {len(result.measurements)}")
-    for m in result.measurements:
-        print(f"  - {m.name}: passed={m.passed}")
+
+def _incremental_loop(client: SiftClient, log_file: str):
+    """Replay incrementally in a loop until stdin is closed (EOF)."""
+    result = None
+    while True:
+        received_signal, _, _ = select.select([sys.stdin], [], [], 1.0)
+        result = client.test_results.replay_log_file(log_file, incremental=True)
+        if received_signal:
+            break
+    logger.info(f"Replay completed: {result}")
+    fp = os.path.abspath(log_file)
+    if fp.startswith(tempfile.gettempdir()):
+        os.remove(fp)
+    return result
 
 
 if __name__ == "__main__":

python/lib/sift_client/scripts/replay_test_result_log.py

@@ -9,6 +9,7 @@
 import asyncio
 import atexit
 import logging
+import os
 import threading
 from typing import Any
 from urllib.parse import urlparse
@@ -98,6 +99,8 @@ def __init__(self, config: GrpcConfig):
         Args:
             config: The gRPC client configuration.
         """
+        os.environ.setdefault("GRPC_ENABLE_FORK_SUPPORT", "0")
+
         self._config = config
         # map each asyncio loop to its async channel and stub dict
         self._channels_async: dict[asyncio.AbstractEventLoop, Any] = {}

python/lib/sift_client/transport/grpc_transport.py

@@ -58,6 +58,9 @@ def main(self):
   - If you want each module(file) to be marked as a step w/ each test as a substep, import the `module_substep` fixture as well.
 - The `report_context` fixture requires a fixture `sift_client` returning an `SiftClient` instance to be passed in.
 
+
+Note: FedRAMP users: report_context will log test results to a temp file to avoid API calls during test execution. If this is a shared environment, you should import the `report_context_no_logging` fixture instead.
+
 ###### Example at top of your test file or in your conftest.py file:
 
 ```python
@@ -102,6 +105,7 @@ def test_example(report_context, step):
     pytest_runtest_makereport,
     report_context,
     report_context_check_connection,
+    report_context_no_logging,
     step,
     step_check_connection,
 )
@@ -115,6 +119,7 @@ def test_example(report_context, step):
     "pytest_runtest_makereport",
     "report_context",
     "report_context_check_connection",
+    "report_context_no_logging",
     "step",
     "step_check_connection",
 ]

python/lib/sift_client/util/test_results/__init__.py

@@ -4,6 +4,7 @@
 import logging
 import os
 import socket
+import subprocess
 import tempfile
 import traceback
 from contextlib import AbstractContextManager
@@ -100,7 +101,28 @@ def __init__(
         )
         self.report = client.test_results.create(create, log_file=self.log_file)
 
+    def _open_replay_proc(self):
+        if self.log_file is not None:
+            # To avoid GRPC forking errors, temporarily redirect stderr at the fd level before forking, so the child inherits /dev/null on fd 2 when the atfork handler fires.
+            saved_stderr = os.dup(2)
+            devnull_fd = os.open(os.devnull, os.O_WRONLY)
+            os.dup2(devnull_fd, 2)
+            os.close(devnull_fd)
+            try:
+                self._replay_proc = subprocess.Popen(
+                    ["replay-test-result-log", "--incremental", str(self.log_file)],
+                    stdin=subprocess.PIPE,
+                    stdout=subprocess.DEVNULL,
+                    stderr=subprocess.DEVNULL,
+                )
+            finally:
+                os.dup2(saved_stderr, 2)
+                os.close(saved_stderr)
+        else:
+            self._replay_proc = None
+
     def __enter__(self):
+        self._open_replay_proc()
         return self
 
     def __exit__(self, exc_type, exc_value, traceback):
@@ -112,20 +134,14 @@ def __exit__(self, exc_type, exc_value, traceback):
         else:
             update["status"] = TestStatus.PASSED
         self.report.update(update, log_file=self.log_file)
-        if self.log_file:
+
+        if self._replay_proc is not None:
             try:
-                # Try replaying the log file and clean up the file if it's a temporary file.
-                self.client.test_results.replay_log_file(self.log_file)
-                fp = os.path.abspath(self.log_file)
-                tmp_dir = tempfile.gettempdir()
-                if fp.startswith(tmp_dir):
-                    os.remove(fp)
-            except Exception as e:
-                logger.error(e)
-                logger.error(
-                    f"Error replaying log file: {self.log_file}.\n  Can replay with `replay-test-result-log {self.log_file}`."
-                )
-                raise e
+                self._replay_proc.communicate(timeout=10)
+            except subprocess.TimeoutExpired:
+                logger.warning("Replay process did not exit in 10s, killing it")
+                self._replay_proc.kill()
+                self._replay_proc.wait()
 
         return True
 

python/lib/sift_client/util/test_results/context_manager.py

@@ -63,6 +63,14 @@ def report_context(
     yield from _report_context_impl(sift_client, request)
 
 
+@pytest.fixture(scope="session", autouse=True)
+def report_context_no_logging(
+    sift_client: SiftClient, request: pytest.FixtureRequest
+) -> Generator[ReportContext | None, None, None]:
+    """Create a report context for the session without log file."""
+    yield from _report_context_impl(sift_client, request, log_file=None)
+
+
 def _step_impl(
     report_context: ReportContext, request: pytest.FixtureRequest
 ) -> Generator[NewStep | None, None, None]: