python(feat): Test results logging now happens in a subprocess while test runs.

Author: ian-sift

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

@@ -697,24 +697,23 @@ def test_import_log_file_round_trip(self, sift_client, nostromo_run, tmp_path):
             compare_test_measurement_fields(replayed_m, direct_m)
 
     @pytest.mark.asyncio
-    async def test_malformed_log_line_raises(self, tmp_path):
-        """import_log_file raises ValueError on a line that doesn't match the expected format."""
+    async def test_malformed_log_line_skipped(self, tmp_path):
+        """Malformed lines are skipped; a file with no valid entries raises 'No CreateTestReport'."""
         log_file = tmp_path / "bad.jsonl"
-        log_file.write_text("this is not a valid log line\n")
+        log_file.write_text(
+            '[LogTracking] {"lastUploadedLine":0,"idMap":{}}\nthis is not a valid log line\n'
+        )
 
         client = TestResultsLowLevelClient(grpc_client=MagicMock())
-        with pytest.raises(ValueError, match="malformed log line"):
+        with pytest.raises(ValueError, match="Invalid log line: this is not a valid log lin"):
             await client.import_log_file(log_file)
 
     @pytest.mark.asyncio
-    async def test_malformed_line_after_valid_lines_raises(self, tmp_path):
-        """A malformed line after valid entries still raises."""
-        log_file = tmp_path / "mixed.jsonl"
-        log_file.write_text(
-            '[CreateTestReport] {"name":"r","testCase":"c","testSystemName":"s"}\n'
-            "totally broken line\n"
-        )
+    async def test_empty_log_file_raises(self, tmp_path):
+        """A log file with only a LogTracking header and no entries raises."""
+        log_file = tmp_path / "empty.jsonl"
+        log_file.write_text('[LogTracking] {"lastUploadedLine":0,"idMap":{}}\n')
 
         client = TestResultsLowLevelClient(grpc_client=MagicMock())
-        with pytest.raises(ValueError, match="malformed log line"):
+        with pytest.raises(ValueError, match="No CreateTestReport found"):
             await client.import_log_file(log_file)

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

@@ -1,43 +1,14 @@
-"""Override report_context to disable log file simulation for integration tests in this directory so that we can exercise the context manager when no log file is provided."""
-
-from __future__ import annotations
-
-from typing import TYPE_CHECKING, Generator
-
 import pytest
 
-from sift_client.util.test_results.pytest_util import _report_context_impl, _step_impl
-
-if TYPE_CHECKING:
-    from sift_client.client import SiftClient
-    from sift_client.util.test_results.context_manager import NewStep, ReportContext
-
-
-@pytest.fixture(scope="session", autouse=True)
-def report_context(
-    sift_client: SiftClient, client_has_connection: bool, request: pytest.FixtureRequest
-) -> Generator[ReportContext | None, None, None]:
-    if client_has_connection:
-        yield from _report_context_impl(sift_client, request, log_file=None)
-    else:
-        yield None
-
 
-@pytest.fixture(autouse=True)
-def step(
-    report_context: ReportContext, client_has_connection: bool, request: pytest.FixtureRequest
-) -> Generator[NewStep | None, None, None]:
-    if client_has_connection:
-        yield from _step_impl(report_context, request)
-    else:
-        yield None
+def pytest_addoption(parser: pytest.Parser) -> None:
+    existing_options = [opt.names() for opt in parser._anonymous.options]
+    # Flatten the list of lists into a single list of strings
+    flat_options = [item for sublist in existing_options for item in sublist]
+    if not any("--sift-test-results-log-file" in name for name in flat_options):
+        parser.addoption("--sift-test-results-log-file", action="store_true", default="false")
 
 
-@pytest.fixture(scope="module", autouse=True)
-def module_substep(
-    report_context: ReportContext, client_has_connection: bool, request: pytest.FixtureRequest
-) -> Generator[NewStep | None, None, None]:
-    if client_has_connection:
-        yield from _step_impl(report_context, request)
-    else:
-        yield None
+def pytest_configure(config: pytest.Config) -> None:
+    """Configure the pytest configuration to disable the Sift test results log file."""
+    config.option.sift_test_results_log_file = False

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

@@ -2001,15 +2001,16 @@ class TestResultsAPI:
         """
         ...
 
-    def import_log_file(self, log_file: str | Path) -> ReplayResult:
+    def import_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.
 
         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.
 
         Args:
-            log_file: Path to the log file to replay.
+            log_file: Path to the log file to import.
+            incremental: (internal tooling) If True, goes line by line and calls every event vs. reading the entire file at once and sending resultant test report.
 
         Returns:
             A ReplayResult containing the created report, steps, and measurements.

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

@@ -619,6 +619,7 @@ async def delete_measurement(self, *, test_measurement: str | TestMeasurement) -
     async def import_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.
 
@@ -627,12 +628,13 @@ async def import_log_file(
         IDs are mapped from simulated to real during the creation process.
 
         Args:
-            log_file: Path to the log file to replay.
+            log_file: Path to the log file to import.
+            incremental: (internal tooling) If True, goes line by line and calls every event vs. reading the entire file at once and sending resultant test report.
 
         Returns:
             A ReplayResult containing the created report, steps, and measurements.
         """
-        result = await self._low_level_client.import_log_file(log_file)
+        result = await self._low_level_client.import_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,10 +3,45 @@
 from __future__ import annotations
 
 import argparse
+import logging
 import os
+import select
+import sys
+import tempfile
+from typing import TYPE_CHECKING
 
 from sift_client import SiftClient, SiftConnectionConfig
 
+if TYPE_CHECKING:
+    from sift_client._internal.low_level_wrappers.test_results import ReplayResult
+
+logger = logging.getLogger(__name__)
+
+
+def _print_result(result: ReplayResult) -> 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 _incremental_import_loop(client: SiftClient, log_file: str) -> ReplayResult | None:
+    """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.import_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
+
 
 def main() -> None:
     """Replay a test result simulation log file against the Sift API."""
@@ -17,6 +52,9 @@ def main() -> None:
     parser.add_argument("--grpc-url", default=os.getenv("SIFT_GRPC_URI"))
     parser.add_argument("--rest-url", default=os.getenv("SIFT_REST_URI"))
     parser.add_argument("--api-key", default=os.getenv("SIFT_API_KEY"))
+    parser.add_argument(
+        "--incremental", action="store_true", help="Import the log file incrementally."
+    )
     args = parser.parse_args()
 
     if not args.grpc_url or not args.rest_url or not args.api_key:
@@ -33,15 +71,26 @@ def main() -> None:
         )
     )
 
-    result = client.test_results.import_log_file(args.log_file)
+    try:
+        if args.incremental:
+            result = _incremental_import_loop(client, args.log_file)
+        else:
+            result = client.test_results.import_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}")
+    if result:
+        _print_result(result)
 
 
 if __name__ == "__main__":

python/lib/sift_client/scripts/import_test_result_log.py

@@ -58,7 +58,12 @@ 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.
+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 can disable logging by passing ``--sift-test-results-log-file=false``.
+
+#### Configuration
+
+- Git metadata: You can configure the test results by passing the `--sift-test-results-git-metadata` flag to pytest in your commandline, conftest.py file, or as an addopt in your pyproject.toml file (https://docs.pytest.org/en/stable/reference/customize.html#configuration).
+- Log file: You can configure the log file by passing the `--sift-test-results-log-file` flag to pytest in your commandline, conftest.py file, or as an addopt in your pyproject.toml file.
 
 ###### Example at top of your test file or in your conftest.py file:
 
@@ -101,10 +106,10 @@ def test_example(report_context, step):
     client_has_connection,
     module_substep,
     module_substep_check_connection,
+    pytest_addoption,
     pytest_runtest_makereport,
     report_context,
     report_context_check_connection,
-    report_context_no_logging,
     step,
     step_check_connection,
 )
@@ -115,10 +120,10 @@ def test_example(report_context, step):
     "client_has_connection",
     "module_substep",
     "module_substep_check_connection",
+    "pytest_addoption",
     "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
@@ -37,6 +38,29 @@
 logger = logging.getLogger(__name__)
 
 
+def _git_metadata() -> dict[str, str] | None:
+    """Return git branch and commit hash, or None if not in a git repo."""
+    try:
+        branch = subprocess.check_output(
+            ["git", "rev-parse", "--abbrev-ref", "HEAD"],
+            stderr=subprocess.DEVNULL,
+            text=True,
+        ).strip()
+        commit = subprocess.check_output(
+            ["git", "rev-parse", "--short", "HEAD"],
+            stderr=subprocess.DEVNULL,
+            text=True,
+        ).strip()
+        repo = subprocess.check_output(
+            ["git", "remote", "get-url", "origin"],
+            stderr=subprocess.DEVNULL,
+            text=True,
+        ).strip()
+        return {"git_repo": repo, "git_branch": branch, "git_commit": commit}
+    except Exception:
+        return None
+
+
 class ReportContext(AbstractContextManager):
     """Context manager for a new TestReport. See usage example in __init__.py."""
 
@@ -48,6 +72,7 @@ class ReportContext(AbstractContextManager):
     step_number_at_depth: dict[int, int]
     open_step_results: dict[str, bool]
     any_failures: bool
+    _import_proc: subprocess.Popen | None = None
 
     def __init__(
         self,
@@ -57,6 +82,7 @@ def __init__(
         system_operator: str | None = None,
         test_case: str | None = None,
         log_file: str | Path | bool | None = None,
+        include_git_metadata: bool = False,
     ):
         """Initialize a new report context.
 
@@ -68,6 +94,7 @@ def __init__(
             test_case: The name of the test case. Will default to the basename of the file containing the test if not provided.
             log_file: If True, create a temp log file. If a path, use that path.
                 All create/update operations will be logged to this file.
+            include_git_metadata: If True, include git metadata in the report.
         """
         self.client = client
         self.step_is_open = False
@@ -97,10 +124,31 @@ def __init__(
             end_time=datetime.now(timezone.utc),
             status=TestStatus.IN_PROGRESS,
             system_operator=system_operator,
+            metadata=_git_metadata() if include_git_metadata else None,  # type: ignore
         )
         self.report = client.test_results.create(create, log_file=self.log_file)
 
+    def _open_import_proc(self):
+        """Open a subprocess to import the log file."""
+        # 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._import_proc = subprocess.Popen(
+                ["import-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)
+
     def __enter__(self):
+        if self.log_file:
+            self._open_import_proc()
         return self
 
     def __exit__(self, exc_type, exc_value, traceback):
@@ -112,19 +160,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._import_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.import_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 `import-test-result-log {self.log_file}`."
-                )
+                self._import_proc.communicate(timeout=10)
+            except subprocess.TimeoutExpired:
+                logger.error("Import process did not exit in 10s, killing it")
+                self._import_proc.kill()
+                self._import_proc.wait()
                 raise
 
         return True