diff --git a/python/CHANGELOG.md b/python/CHANGELOG.md index 16fab1cc9..466f740ca 100644 --- a/python/CHANGELOG.md +++ b/python/CHANGELOG.md @@ -5,6 +5,24 @@ This project adheres to [Semantic Versioning](http://semver.org/). ## [Unreleased] +### What's New + +#### Interrupted test-result uploads resume + +`import-test-result-log ` now finishes an upload that was cut short instead of creating a second report. Every upload records what reached the server in a tracking sidecar beside the log (`.jsonl.tracking`), so a re-run reuses the report the earlier attempt created and sends only the entries that are missing. A log that is already fully uploaded is a no-op. + +```bash +# Resume, or start fresh if there is nothing to resume +import-test-result-log ./offline-runs/a1b2c3/a1b2c3.jsonl + +# Abandon the partial upload and create a new report +import-test-result-log --new-report ./offline-runs/a1b2c3/a1b2c3.jsonl +``` + +`--new-report` moves the existing sidecar to `.jsonl.tracking.bak` rather than overwriting it, so the abandoned report's ID stays recoverable. Resuming requires that report to still exist; if it was deleted, or the sidecar came from another environment, the command fails and names `--new-report` as the way forward. + +`client.test_results.import_log_file(...)` takes the matching `new_report` argument. The recovery hints printed by the pytest plugin drop the `--incremental` flag, which is no longer needed to avoid a duplicate report. + ### Bugfixes - Fix `assets.archive` raising `AttributeError` by reading the correct response field, `archived_run_ids`. diff --git a/python/docs/guides/pytest_plugin/index.md b/python/docs/guides/pytest_plugin/index.md index 26ed3652e..215dc9048 100644 --- a/python/docs/guides/pytest_plugin/index.md +++ b/python/docs/guides/pytest_plugin/index.md @@ -105,7 +105,7 @@ so a misconfigured job fails immediately instead of silently producing no report During the run, every create and update is appended to a JSONL log file. A background worker uploads new entries to Sift incrementally. If the connection drops mid-test, the test keeps running and the log keeps writing locally. -The remaining entries can be uploaded afterward by running import-test-result-log, which the plugin prints on exit. +The remaining entries can be uploaded afterward by running import-test-result-log, which the plugin prints on exit. That command resumes into the report the interrupted run created rather than starting a second one. See [Running Modes](running_modes.md) for the log-file and replay pipeline, overriding the connection check, and replaying a saved log. diff --git a/python/docs/guides/pytest_plugin/running_modes.md b/python/docs/guides/pytest_plugin/running_modes.md index 6b6ba0e22..2e63b0205 100644 --- a/python/docs/guides/pytest_plugin/running_modes.md +++ b/python/docs/guides/pytest_plugin/running_modes.md @@ -119,11 +119,10 @@ get noticed until somebody goes looking for the report, which is usually weeks later, which is usually too late. With the JSONL log on by default, create/update calls are written to a log file -in the run's output directory during the run, and an -`import-test-result-log --incremental` worker replays them against Sift in the -background. If the worker crashes mid-session (connection failure, API error) or -is still draining its backlog at session end, the failure is logged at session -end with an `import-test-result-log` command for manual recovery. Test outcomes +in the run's output directory during the run, and a background worker replays +them against Sift as they are written. If the worker crashes mid-session +(connection failure, API error) or is still draining its backlog at session +end, the failure is logged at session end with an `import-test-result-log` command for manual recovery. Test outcomes are unaffected and the local log file is preserved. Pass `--no-sift-log-file` to make every create/update synchronous against the API instead. @@ -150,8 +149,8 @@ The override is ignored under `--sift-offline` and `--sift-disabled`. Same fixtures, same `step.measure(...)` semantics as online. The difference is where the writes go: every create/update lands in a JSONL log file instead of hitting the Sift API. The session-start ping is skipped, missing `SIFT_*` env -vars are tolerated (placeholders are filled), and the replay worker -(`import-test-result-log --incremental`) does not get spawned at session end. +vars are tolerated (placeholders are filled), and the replay worker does not +get spawned at session end. ```bash pytest --sift-offline --sift-output-dir=./offline-runs @@ -165,8 +164,8 @@ import-test-result-log ./offline-runs/a1b2c3/a1b2c3.jsonl ``` That replay creates the report, steps, and measurements against Sift. See -[Replaying a saved log file](#replaying-a-saved-log-file) for cleanup and the -incremental flag. +[Replaying a saved log file](#replaying-a-saved-log-file) for cleanup and for +what happens when a replay is interrupted. `--no-sift-log-file` is rejected when offline is set, since the log is the only sink in offline mode and without it the results are gone. @@ -217,5 +216,30 @@ When the worker doesn't finish cleanly the plugin will print a hint mentioning import-test-result-log ``` -That replays the saved JSONL log as a single batch (no `--incremental`) and -deletes the file when it lives under the system temp dir. +That replays the saved JSONL log as a single batch and deletes the file when it +lives under the system temp dir. + +`import-test-result-log` runs in one of two modes: + +| Mode | How to get it | What it does | +| --- | --- | --- | +| New upload | the default, when nothing has been uploaded yet | Uploads the whole log as a new report | +| Resume | the default, when a tracking sidecar records an interrupted upload | Continues into the report that upload created, sending only what is missing | + +Run the same command again if a replay is interrupted. Each upload records what +it created in a tracking sidecar next to the log (`.jsonl.tracking`), so a +second run continues into the report the first one created and sends only what +is missing, rather than leaving a duplicate report behind. A log that is already +fully uploaded is a no-op. + +```bash +# Upload as a new report instead, abandoning the partial one +import-test-result-log --new-report +``` + +The existing sidecar is moved to `.jsonl.tracking.bak` rather than +overwritten, so the abandoned report's ID can still be looked up and cleaned up. + +Resuming needs the report the earlier attempt created to still exist. If it was +deleted, or the sidecar came from a different environment, the command fails and +names `--new-report` as the way forward. diff --git a/python/lib/sift_client/_internal/low_level_wrappers/_test_results_log.py b/python/lib/sift_client/_internal/low_level_wrappers/_test_results_log.py index 15ef0b115..aae8ca259 100644 --- a/python/lib/sift_client/_internal/low_level_wrappers/_test_results_log.py +++ b/python/lib/sift_client/_internal/low_level_wrappers/_test_results_log.py @@ -16,7 +16,9 @@ map. Written only by the replay subprocess via :meth:`LogTracking.save` using a temp-file + ``os.replace`` so a crash can't leave a half-written sidecar. Read once at replay start via :meth:`LogTracking.load`. Never touched by the - test process. + test process. Restarting an upload against a new report moves any existing + sidecar to ``foo.jsonl.tracking.bak`` (see :meth:`LogTracking.archive`) so the + abandoned report's ID stays recoverable. # Concurrency @@ -90,17 +92,26 @@ class LogTracking: file itself is append-only and stores only API-call data lines. * ``last_uploaded_line`` is the count of data lines that have been - successfully replayed against the server. Each data line corresponds to a - single API call, so line granularity matches the atomic unit of work: a - line is either fully replayed or must be retried in its entirety. Data - lines are strictly append-only, so this counter is stable across runs. + successfully replayed against the server, in log order. Each data line + corresponds to a single API call, so line granularity matches the atomic + unit of work: a line is either fully replayed or must be retried in its + entirety. Data lines are strictly append-only, so this counter is stable + across runs. A batch upload creates in collapsed order rather than log + order and so leaves it at zero; ``complete`` is what marks that upload + finished. * ``id_map`` maps simulated response IDs (created during the original test run) to the real IDs assigned by the server during replay. Subsequent - ``Update*`` entries consult this map to translate IDs. + ``Update*`` entries consult this map to translate IDs, and a resumed + replay uses it to skip the creates that already reached the server. + * ``complete`` marks the whole log as uploaded, so a stray re-run does + nothing instead of replaying a finished upload. Sidecars written before + this field existed default to False and are treated as resumable, which + costs a no-op walk at worst. """ last_uploaded_line: int = 0 id_map: dict[str, str] = field(default_factory=dict) + complete: bool = False client_version: str = field(default_factory=_client_version) @staticmethod @@ -109,6 +120,28 @@ def sidecar_path(log_path: str | Path) -> Path: p = Path(log_path) return p.with_name(p.name + ".tracking") + @staticmethod + def backup_path(log_path: str | Path) -> Path: + """Return where a sidecar is kept when an upload is restarted (``.tracking.bak``).""" + sidecar = LogTracking.sidecar_path(log_path) + return sidecar.with_name(sidecar.name + ".bak") + + @staticmethod + def archive(log_path: str | Path) -> Path | None: + """Move an existing sidecar aside to ``.tracking.bak``; return its new path. + + Used when an upload is deliberately restarted against a new report. The + old sidecar still names the report the abandoned upload created, so it + is preserved rather than overwritten: without it that report is only + findable by hand. Returns None when there was nothing to move. + """ + sidecar = LogTracking.sidecar_path(log_path) + if not sidecar.exists(): + return None + backup = LogTracking.backup_path(log_path) + os.replace(sidecar, backup) + return backup + @classmethod def load(cls, log_path: str | Path) -> LogTracking: """Read tracking state for ``log_path``; return a fresh instance if missing or corrupt. @@ -126,6 +159,7 @@ def load(cls, log_path: str | Path) -> LogTracking: return cls( last_uploaded_line=data.get("lastUploadedLine", 0), id_map=data.get("idMap", {}), + complete=data.get("complete", False), client_version=data.get("clientVersion", "unknown"), ) @@ -141,6 +175,7 @@ def save(self, log_path: str | Path) -> None: { "clientVersion": self.client_version, "lastUploadedLine": self.last_uploaded_line, + "complete": self.complete, "idMap": self.id_map, }, separators=(",", ":"), diff --git a/python/lib/sift_client/_internal/low_level_wrappers/test_results.py b/python/lib/sift_client/_internal/low_level_wrappers/test_results.py index 67fef412f..e0d33dec8 100644 --- a/python/lib/sift_client/_internal/low_level_wrappers/test_results.py +++ b/python/lib/sift_client/_internal/low_level_wrappers/test_results.py @@ -6,6 +6,7 @@ from typing import TYPE_CHECKING, Any, NamedTuple, TypeVar, cast from google.protobuf import json_format +from grpc import RpcError, StatusCode from sift.test_reports.v1.test_reports_pb2 import ( CreateTestMeasurementRequest, CreateTestMeasurementResponse, @@ -72,6 +73,12 @@ _EntityT = TypeVar("_EntityT", TestReport, TestStep, TestMeasurement) +# Create entries a resumed replay can skip outright once the sidecar id map +# names them. CreateTestMeasurements is absent on purpose: one line covers many +# measurements, of which an interrupted run may have created only some, so that +# handler filters per measurement instead. +_WHOLE_ENTRY_CREATES = frozenset({"CreateTestReport", "CreateTestStep", "CreateTestMeasurement"}) + class _EntryIds(NamedTuple): """The entity a replayed log entry acted on, for the audit trace. @@ -81,10 +88,14 @@ class _EntryIds(NamedTuple): entity; for an update both identify the target being mutated, so the ``replay.upload`` line names which step/report/measurement it touched instead of leaving the columns blank. + + ``skipped`` marks an entry a resume found already on the server, so the + audit trail distinguishes what this run sent from what it inherited. """ sim_id: str | None = None real_id: str | None = None + skipped: bool = False class TestResultsLowLevelClient(LowLevelClientBase, WithGrpcClient): @@ -923,35 +934,120 @@ async def import_log_file( self, log_file: str | Path, incremental: bool = False, + new_report: bool = False, ) -> ReplayResult: """Replay a log file, creating real API objects from the logged simulation data. - Two modes are available: - - * **batch** (default): Parse the entire log, reconstruct objects via - simulation, then create them all via the API in one pass. The - ``LogTracking`` header on line 0 is ignored. - * **incremental** (``incremental=True``): Walk the log line-by-line, - issuing the real API call for each entry as it is encountered. - ``LogTracking.last_uploaded_line`` is advanced only after the call - succeeds, so a failure during a line causes the entire line to be - retried on the next invocation; already-uploaded lines are skipped. + Three modes, of which the caller picks at most one: + + * **new upload** (the default when the tracking sidecar records nothing): + the whole log is collapsed to final state via simulation and created in + one pass, which is fewer API calls than replaying every logged update. + Each created entity is recorded in the sidecar as it goes, so an upload + interrupted partway can be finished later. + * **resume** (the default when the sidecar records unfinished work): the + report the earlier attempt created is reused, and replay walks the log + line by line, skipping the entries that already reached the server. A + sidecar marked complete means there is nothing to do. + * **incremental** (``incremental=True``): follow a log that is still + being written, uploading entries as they appear. This is the mode the + plugin's background worker ticks in during a test run, not a way to + finish an interrupted upload. Args: log_file: Path to the log file to replay. - incremental: If True, use incremental mode. + incremental: Select the worker's follow mode. The log is treated as + still growing, so reaching its end does not finish the upload. + new_report: Ignore any partial upload and create a new report. The + existing sidecar is moved aside rather than overwritten, so the + abandoned report's ID stays recoverable. Returns: A ReplayResult containing the created report, steps, and measurements. + + Raises: + FileNotFoundError: If the log file does not exist. + ValueError: If both ``incremental`` and ``new_report`` are set, or if + a resume cannot find the report the earlier attempt created, + which happens when it was deleted or was uploaded to a different + environment. """ log_path = Path(log_file) if not log_path.exists(): raise FileNotFoundError(f"Log file not found: {log_file}") if incremental: + if new_report: + raise ValueError( + "incremental and new_report are mutually exclusive: incremental replay " + "continues whatever the sidecar records, which is what new_report discards." + ) return await self._incremental_import_log_file(log_path) - return await self._batch_import_log_file(log_path) + if new_report: + backup = LogTracking.archive(log_path) + if backup is not None: + log_event(logger, logging.INFO, "replay.restart", backup=str(backup)) + return await self._batch_import_log_file(log_path) + + tracking = LogTracking.load(log_path) + if tracking.complete: + log_event(logger, logging.INFO, "replay.already_complete", log=str(log_path)) + return ReplayResult() + if not tracking.id_map: + return await self._batch_import_log_file(log_path) + + existing_report = await self._resume_report(log_path, tracking) + log_event( + logger, + logging.INFO, + "replay.resume", + log=str(log_path), + report=existing_report._id_or_error, + uploaded=len(tracking.id_map), + ) + result = await self._incremental_import_log_file(log_path, resuming=True) + # The report was created by the interrupted run, so replay never sees a + # create line for it; name it here so callers can still link the report. + result.report = result.report or existing_report + return result + + async def _resume_report(self, log_path: Path, tracking: LogTracking) -> TestReport: + """Fetch the report a partial upload created, so replay continues into it. + + Fetching up front turns a report that was deleted, or a sidecar carried + to a different environment, into one clear failure before anything is + uploaded, rather than a bare NOT_FOUND partway through the replay. + """ + raw_lines = await _read_log_lines(log_path) + logged_report_id = next( + ( + response_id + for request_type, response_id, _ in parse_log_data_lines(raw_lines) + if request_type == "CreateTestReport" + ), + None, + ) + real_report_id = tracking.id_map.get(logged_report_id) if logged_report_id else None + sidecar = LogTracking.sidecar_path(log_path) + if not real_report_id: + raise ValueError( + f"The upload recorded in {sidecar} has no test report to resume into. " + f"Re-run with --new-report to upload {log_path} as a new report." + ) + try: + return await self.get_test_report(real_report_id) + except RpcError as exc: + # RpcError itself carries no status; only the Call subclasses the + # channel actually raises do, and the async ones are not grpc.Call. + code = getattr(exc, "code", lambda: None)() + if code != StatusCode.NOT_FOUND: + raise + raise ValueError( + f"Test report {real_report_id}, from the interrupted upload recorded in " + f"{sidecar}, no longer exists. Re-run with --new-report to upload " + f"{log_path} as a new report." + ) from exc # ------------------------------------------------------------------ # Shared replay dispatch @@ -990,6 +1086,13 @@ async def _import_entry( handler = handlers.get(request_type) if handler is None: return _EntryIds() + if not simulate and request_type in _WHOLE_ENTRY_CREATES and response_id in id_map: + # A resumed replay walks the log from the first line, so every create + # an interrupted run completed is already in the sidecar id map. + # Re-issuing it would duplicate the entity, which is the outcome + # resuming exists to avoid. Skipping here rather than inside each + # handler keeps a new create type resume-safe by default. + return _EntryIds(response_id, id_map[response_id], skipped=True) return await handler(json_str, response_id, simulate=simulate, id_map=id_map, state=state) @staticmethod @@ -1083,15 +1186,31 @@ async def _replay_create_measurements( state.measurements_order.append(meas._id_or_error) created_ids.append(meas._id_or_error) else: - _, real_ids = await self.create_test_measurements(request=request) - for i, real_id in enumerate(real_ids): - if i < len(original_ids): - id_map[original_ids[i]] = real_id + # Batch replay creates measurements one at a time, so an interrupted + # run can leave part of a batch line already on the server. Re-sending + # the whole line would duplicate those, so send only what is missing. + pending: list[tuple[str | None, TestMeasurementProto]] = [] + for i, tm in enumerate(request.test_measurements): + logged_id = original_ids[i] if i < len(original_ids) else None + if logged_id and logged_id in id_map: + continue + pending.append((logged_id, tm)) + real_ids: list[str] = [] + if pending: + _, real_ids = await self.create_test_measurements( + request=CreateTestMeasurementsRequest( + test_measurements=[tm for _, tm in pending] + ) + ) + for (logged_id, _), real_id in zip(pending, real_ids): + if logged_id: + id_map[logged_id] = real_id created_ids.append(real_id) # Batch line covers many measurements; comma-join both sides so the # audit row still names every entity (fields are space-free, so commas - # keep it one token). - return _EntryIds(response_id, ",".join(created_ids) or None) + # keep it one token). A resume that found the whole line already on the + # server creates nothing, which is what marks the entry skipped. + return _EntryIds(response_id, ",".join(created_ids) or None, skipped=not created_ids) async def _replay_update_report( self, @@ -1171,8 +1290,16 @@ async def _replay_update_measurement( # ------------------------------------------------------------------ async def _batch_import_log_file(self, log_path: Path) -> ReplayResult: + """Collapse the whole log to final state, then create it in one pass. + + Each created entity is recorded in the tracking sidecar before the next + one is sent, so an interrupted batch upload can be finished by a resuming + replay instead of creating a second report. The sidecar is written fresh: + batch runs only when nothing is left over from an earlier attempt. + """ id_map: dict[str, str] = {} state = _ReplayState() + tracking = LogTracking() raw_lines = await _read_log_lines(log_path) for request_type, response_id, json_str in parse_log_data_lines(raw_lines): @@ -1188,10 +1315,29 @@ async def _batch_import_log_file(self, log_path: Path) -> ReplayResult: if state.report is None: raise ValueError("No CreateTestReport found in log file") + # Collapsed entities are keyed by the simulate pass's own IDs, while the + # sidecar is keyed by the logged response IDs a resume looks entities up + # by, so invert the map the pass just built. + logged_by_simulated = {simulated: logged for logged, simulated in id_map.items()} real_id_map: dict[str, str] = {} + def record_created(simulated_id: str, real_id: str) -> None: + """Note a real entity against both the in-run map and the sidecar. + + The sidecar is saved per entity so an upload interrupted at any point + is resumable. That is one small atomic rewrite per created entity; + the incremental path already pays the same cost per log line. + """ + real_id_map[simulated_id] = real_id + logged_id = logged_by_simulated.get(simulated_id) + if not logged_id: + return + tracking.id_map[logged_id] = real_id + tracking.save(log_path) + real_report = await self._create_report_from_simulated(state.report) real_report_id = real_report._id_or_error + record_created(state.report._id_or_error, real_report_id) real_steps: list[TestStep] = [] for sim_step_id in state.steps_order: @@ -1206,7 +1352,7 @@ async def _batch_import_log_file(self, log_path: Path) -> ReplayResult: ) real_step = await self.create_test_step(step_create) real_steps.append(real_step) - real_id_map[sim_step_id] = real_step._id_or_error + record_created(sim_step_id, real_step._id_or_error) real_measurements: list[TestMeasurement] = [] for sim_measurement_id in state.measurements_order: @@ -1219,6 +1365,13 @@ async def _batch_import_log_file(self, log_path: Path) -> ReplayResult: ) real_measurement = await self.create_test_measurement(measurement_create) real_measurements.append(real_measurement) + record_created(sim_measurement_id, real_measurement._id_or_error) + + # Everything in the log reached the server. The cursor stays at zero + # because batch created in collapsed order, not log order; the flag is + # what stops a later run from replaying a finished upload. + tracking.complete = True + tracking.save(log_path) return ReplayResult( report=real_report, @@ -1230,7 +1383,12 @@ async def _batch_import_log_file(self, log_path: Path) -> ReplayResult: # Incremental replay # ------------------------------------------------------------------ - async def _incremental_import_log_file(self, log_path: Path) -> ReplayResult: + async def _incremental_import_log_file( + self, + log_path: Path, + *, + resuming: bool = False, + ) -> ReplayResult: """Replay line-by-line, issuing real API calls and updating tracking. Resumes from ``LogTracking.last_uploaded_line`` (loaded from the @@ -1239,9 +1397,25 @@ async def _incremental_import_log_file(self, log_path: Path) -> ReplayResult: single atomic API call; if replay of a line fails, ``last_uploaded_line`` is not advanced so the whole line is retried next tick. + + A batch upload records what it created but keeps its cursor at zero, + since it creates in collapsed order rather than log order. Finishing one + therefore re-walks the log from the first line, and the create entries it + already sent are skipped by their sidecar id-map entry instead of the + cursor. + + ``resuming`` says the log is final and this call is finishing an earlier + upload: the create line for the report was consumed by that upload, so + its absence is not an error, and reaching the end marks the sidecar + complete. The live worker leaves it False, since for a log still being + written the end of the file is not the end of the run. """ tracking = LogTracking.load(log_path) - resuming = tracking.last_uploaded_line > 0 + # Two separate questions, and conflating them would let a worker tick + # mark a log still being written as complete: whether an earlier pass + # already consumed the report's create line, and whether this call is + # finishing a final log. + report_already_created = resuming or tracking.last_uploaded_line > 0 id_map = tracking.id_map state = _ReplayState() @@ -1274,10 +1448,11 @@ async def _incremental_import_log_file(self, log_path: Path) -> ReplayResult: tracking.last_uploaded_line += 1 tracking.save(log_path) - # One line per uploaded entity: what was sent, the sim->real id of + # One line per replayed entity: what was sent, the sim->real id of # the entity it acted on (the new entity for a create, the target - # for an update), and the sidecar cursor + id-map size after the - # save so a reader can follow exactly what reached the server. + # for an update), whether a resume found it already on the server, + # and the sidecar cursor + id-map size after the save so a reader can + # follow exactly what reached the server. log_event( logger, logging.DEBUG, @@ -1286,15 +1461,22 @@ async def _incremental_import_log_file(self, log_path: Path) -> ReplayResult: line=tracking.last_uploaded_line, sim_id=entry_ids.sim_id or "-", real_id=entry_ids.real_id or "-", + skipped="yes" if entry_ids.skipped else "no", idmap=len(id_map), ) # On a resume tick the CreateTestReport line was consumed on an earlier # tick, so state.report is expected to be None; the report already exists # on the server. Only a genuine first pass over an empty log is an error. - if state.report is None and not resuming: + if state.report is None and not report_already_created: raise ValueError("No CreateTestReport found in log file") + if resuming: + # The log is final and the walk reached its end, so the upload is + # done; a stray re-run now does nothing instead of replaying it. + tracking.complete = True + tracking.save(log_path) + return ReplayResult( report=state.report, steps=[state.steps_by_id[sid] for sid in state.steps_order], diff --git a/python/lib/sift_client/_internal/pytest_plugin/replay_worker.py b/python/lib/sift_client/_internal/pytest_plugin/replay_worker.py new file mode 100644 index 000000000..8dcaf1917 --- /dev/null +++ b/python/lib/sift_client/_internal/pytest_plugin/replay_worker.py @@ -0,0 +1,142 @@ +"""Background replay worker: follow a log that is still being written. + +Spawned by the report context (see ``util.test_results.context_manager``) for +the duration of a pytest session, never run by hand. It ticks the incremental +importer against a growing log and exits when its stdin closes, which is the +context manager's signal that the session is over. + +The user-facing ``import-test-result-log`` command is the other half: it uploads +or resumes a log that is final. Keeping the follow mode here rather than behind +a flag on that command means there is no way to invoke it by accident, and the +public command has one fewer mode to explain. +""" + +from __future__ import annotations + +import argparse +import logging +import os +import select +import shutil +import sys +import tempfile +from pathlib import Path +from typing import TYPE_CHECKING + +from sift_client import SiftClient, SiftConnectionConfig +from sift_client._internal.low_level_wrappers._test_results_log import LogTracking +from sift_client._internal.pytest_plugin.audit_log import log_event +from sift_client.util.test_results.context_manager import log_replay_instructions + +if TYPE_CHECKING: + from sift_client._internal.low_level_wrappers.test_results import ReplayResult + +logger = logging.getLogger(__name__) + + +def cleanup_temp_log(log_file: str) -> None: + """Remove temp artifacts after a successful upload when audit logging is off. + + Called only when audit logging is off: without an audit trail there's no + reason to retain the buffer, so default temp artifacts are reclaimed + immediately. An explicit ``--sift-output-dir`` (not under the temp dir) is + the user's to keep and is never touched. + + Session-dir layout (``/sift_test_results//``): the whole + directory is removed, cleaning up the JSONL, tracking sidecar, lock, and + any audit files in one shot. + + Legacy flat-temp layout (file directly in tmpdir): only the JSONL and its + tracking sidecars are removed individually. + + Shared with the public replay command, which reclaims the same artifacts + after a one-shot upload. + """ + fp = Path(log_file).absolute() + if not str(fp).startswith(tempfile.gettempdir()): + return + session_dir = fp.parent + if session_dir.parent == Path(tempfile.gettempdir()) / "sift_test_results": + shutil.rmtree(session_dir, ignore_errors=True) + log_event(logger, logging.DEBUG, "replay.cleanup", log=str(fp), dir=str(session_dir)) + return + fp.unlink(missing_ok=True) + LogTracking.sidecar_path(fp).unlink(missing_ok=True) + LogTracking.backup_path(fp).unlink(missing_ok=True) + log_event(logger, logging.DEBUG, "replay.cleanup", log=str(fp)) + + +def _incremental_import_loop( + client: SiftClient, log_file: str, *, keep_log: bool +) -> ReplayResult | None: + """Replay incrementally in a loop until stdin is closed (EOF). + + Per-entity upload detail and sidecar advances are logged inside the + incremental importer (``replay.upload`` / ``replay.error``); idle ticks + that upload nothing are silent on purpose. + + When ``keep_log`` is False (audit logging off) the temp log is deleted on a + clean finish; with audit logging on it's retained alongside the audit trail. + """ + 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 + log_event(logger, logging.INFO, "replay.complete", log=log_file) + if not keep_log: + cleanup_temp_log(log_file) + return result + + +def main() -> None: + """Follow a growing test result log, uploading entries as they are written.""" + parser = argparse.ArgumentParser( + description=( + "Internal replay worker for the Sift pytest plugin. Follows a log file " + "while the session that writes it is still running, and exits when its " + "stdin closes. Not meant to be run by hand: to upload or finish a log " + "that is final, use import-test-result-log." + ) + ) + parser.add_argument("log_file", help="Path to the .jsonl log file to follow.") + 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( + "--audit-log", default=None, help="Path to the replay worker's DEBUG audit log." + ) + args = parser.parse_args() + + if args.audit_log: + from sift_client._internal.pytest_plugin.audit_log import attach_file_handler + + attach_file_handler(Path(args.audit_log)) + + if not args.grpc_url or not args.rest_url or not args.api_key: + raise ValueError("SIFT_GRPC_URI, SIFT_REST_URI, and SIFT_API_KEY must be set") + + use_ssl = "localhost" not in args.grpc_url and "localhost" not in args.rest_url + + client = SiftClient( + connection_config=SiftConnectionConfig( + api_key=args.api_key, + grpc_url=args.grpc_url, + rest_url=args.rest_url, + use_ssl=use_ssl, + ) + ) + + # The worker is spawned with --audit-log only when audit logging is on, so + # its presence is the signal to retain the buffer after a clean upload. + try: + _incremental_import_loop(client, args.log_file, keep_log=bool(args.audit_log)) + except Exception as e: + log_event(logger, logging.ERROR, "replay.failed", error=repr(e)) + log_replay_instructions(args.log_file) + raise + + +if __name__ == "__main__": + main() diff --git a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_incremental_replay.py b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_incremental_replay.py index 8c1f66053..63bbc8533 100644 --- a/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_incremental_replay.py +++ b/python/lib/sift_client/_tests/_internal/low_level_wrappers/test_incremental_replay.py @@ -10,11 +10,14 @@ from __future__ import annotations +import json import logging +from contextlib import contextmanager from datetime import datetime, timezone from unittest.mock import AsyncMock, MagicMock import pytest +from grpc import RpcError, StatusCode from sift_client._internal.low_level_wrappers._test_results_log import LogTracking from sift_client._internal.low_level_wrappers.test_results import ( @@ -22,6 +25,9 @@ TestResultsLowLevelClient as ResultsLowLevelClient, ) from sift_client.sift_types.test_report import ( + TestMeasurement, + TestMeasurementCreate, + TestMeasurementType, TestReport, TestReportCreate, TestReportUpdate, @@ -65,6 +71,39 @@ def _make_step(id_: str) -> TestStep: ) +@contextmanager +def _captured_replay_logs(): + """Collect the replay module's log messages for the duration of the block. + + Captured on the module logger directly: the Sift plugin sets + propagate=False on the sift_client logger, so caplog's root handler would + not see these records. + """ + module_logger = logging.getLogger("sift_client._internal.low_level_wrappers.test_results") + messages: list[str] = [] + handler = logging.Handler() + handler.emit = lambda record: messages.append(record.getMessage()) # type: ignore[method-assign] + prior_level = module_logger.level + module_logger.addHandler(handler) + module_logger.setLevel(logging.DEBUG) + try: + yield messages + finally: + module_logger.removeHandler(handler) + module_logger.setLevel(prior_level) + + +def _make_measurement(id_: str) -> TestMeasurement: + return TestMeasurement( + id_=id_, + test_step_id="real-step", + name="m", + passed=True, + timestamp=T0, + measurement_type=TestMeasurementType.DOUBLE, + ) + + def _report_create() -> TestReportCreate: return TestReportCreate( status=TestStatus.IN_PROGRESS, @@ -182,20 +221,8 @@ async def test_replay_upload_log_names_update_target(tmp_path): client.create_test_step = AsyncMock(return_value=_make_step("real-step")) client.update_test_step = AsyncMock(return_value=_make_step("real-step")) - # Capture directly on the module logger: the Sift plugin sets propagate=False - # on the sift_client logger, so caplog's root handler wouldn't see the records. - module_logger = logging.getLogger("sift_client._internal.low_level_wrappers.test_results") - messages: list[str] = [] - handler = logging.Handler() - handler.emit = lambda record: messages.append(record.getMessage()) # type: ignore[method-assign] - prior_level = module_logger.level - module_logger.addHandler(handler) - module_logger.setLevel(logging.DEBUG) - try: + with _captured_replay_logs() as messages: await client.import_log_file(log_file, incremental=True) - finally: - module_logger.removeHandler(handler) - module_logger.setLevel(prior_level) upload_lines = [m for m in messages if m.startswith("replay.upload")] update_line = next(line for line in upload_lines if "type=UpdateTestStep" in line) @@ -204,6 +231,529 @@ async def test_replay_upload_log_names_update_target(tmp_path): assert "real_id=real-step" in update_line +# --------------------------------------------------------------------------- +# Resuming an interrupted upload +# --------------------------------------------------------------------------- + + +class _NotFoundError(RpcError): + """Stand-in for the server's response when a report has been deleted.""" + + def code(self) -> StatusCode: + return StatusCode.NOT_FOUND + + +class _PermissionDeniedError(RpcError): + """Stand-in for a failure that resuming into a new report would not fix.""" + + def code(self) -> StatusCode: + return StatusCode.PERMISSION_DENIED + + +async def _build_log(client, log_file, *, step_names=("s1", "s2")): + """Write a log offline: create a report, create each step, then close step one. + + Returns the simulated report and steps, whose IDs are the keys a later + resume looks up in the tracking sidecar. + """ + report = await client.create_test_report(test_report=_report_create(), log_file=log_file) + steps = [] + for index, name in enumerate(step_names, start=1): + steps.append( + await client.create_test_step( + test_step=TestStepCreate( + test_report_id=report.id_, + name=name, + step_type=TestStepType.ACTION, + step_path=str(index), + status=TestStatus.IN_PROGRESS, + start_time=T0, + end_time=T0, + ), + log_file=log_file, + ) + ) + step_update = StepUpdate(status=TestStatus.PASSED) + step_update.resource_id = steps[0].id_ + await client.update_test_step(update=step_update, log_file=log_file) + return report, steps + + +def _answer_real_creates(client, *, report_id, step_ids=(), measurement_ids=()): + """Answer the real create calls with canned IDs, leaving simulation alone. + + Batch replay drives its in-memory collapse through the same create methods + with ``simulate=True``, so a blanket mock would swallow those too. Returns + the list that records the name of each real create, in order; the report is + recorded as ``"report"`` since a report create carries no step name. + """ + created: list[str] = [] + + def answer(name, canned_ids, describe): + real = getattr(client, name) + remaining = iter(canned_ids) + + async def call(*args, **kwargs): + if kwargs.get("simulate") or kwargs.get("log_file"): + return await real(*args, **kwargs) + created.append(describe(*args)) + return next(remaining) + + setattr(client, name, call) + + answer("create_test_report", [_make_report(report_id)], lambda *_: "report") + answer("create_test_step", [_make_step(sid) for sid in step_ids], lambda create: create.name) + answer( + "create_test_measurement", + [_make_measurement(mid) for mid in measurement_ids], + lambda create: create.name, + ) + return created + + +@pytest.mark.asyncio +async def test_batch_upload_records_what_it_created(tmp_path): + """A batch upload leaves a sidecar naming every entity it created. + + Without it an interrupted batch upload is unrecoverable: nothing on disk + says which report and steps already reached the server. + """ + log_file = tmp_path / "batch.jsonl" + client = ResultsLowLevelClient(grpc_client=MagicMock()) + report, steps = await _build_log(client, log_file) + + created = _answer_real_creates( + client, report_id="real-report", step_ids=["real-step-1", "real-step-2"] + ) + + await client.import_log_file(log_file) + + assert created == ["report", "s1", "s2"] + tracking = LogTracking.load(log_file) + assert tracking.id_map == { + report.id_: "real-report", + steps[0].id_: "real-step-1", + steps[1].id_: "real-step-2", + } + # Everything reached the server, so a re-run has nothing to do. The cursor + # stays at zero: batch creates in collapsed order, not log order. + assert tracking.complete + assert tracking.last_uploaded_line == 0 + + +@pytest.mark.asyncio +async def test_interrupted_batch_upload_resumes_into_same_report(tmp_path): + """Re-running after a batch upload died finishes it instead of duplicating it.""" + log_file = tmp_path / "partial.jsonl" + client = ResultsLowLevelClient(grpc_client=MagicMock()) + report, steps = await _build_log(client, log_file) + + # The batch upload created the report and the first step, then died. Its + # cursor stays at zero: batch creates in collapsed order, not log order. + LogTracking( + last_uploaded_line=0, + id_map={report.id_: "real-report", steps[0].id_: "real-step-1"}, + ).save(log_file) + + client.get_test_report = AsyncMock(return_value=_make_report("real-report")) + client.create_test_report = AsyncMock(return_value=_make_report("duplicate-report")) + client.create_test_step = AsyncMock(return_value=_make_step("real-step-2")) + client.update_test_step = AsyncMock(return_value=_make_step("real-step-1")) + + result = await client.import_log_file(log_file) + + # The report and the first step were already sent, so neither is re-created. + client.create_test_report.assert_not_awaited() + client.create_test_step.assert_awaited_once() + assert client.create_test_step.await_args.kwargs["request"].test_step.name == "s2" + # The first step's closing update still has to be applied, or it stays open. + client.update_test_step.assert_awaited_once() + assert ( + client.update_test_step.await_args.kwargs["request"].test_step.test_step_id == "real-step-1" + ) + assert result.report is not None + assert result.report.id_ == "real-report" + assert LogTracking.load(log_file).id_map[steps[1].id_] == "real-step-2" + + +@pytest.mark.asyncio +async def test_worker_tick_never_marks_the_log_complete(tmp_path): + """A worker tick must not mark a log that is still being written as complete. + + The worker ticks against a growing log, so reaching the end of the file is + not the end of the run. Marking it complete would make a manual re-run after + the worker died silently upload nothing, dropping every entry logged after + the last tick. + """ + log_file = tmp_path / "worker_tick.jsonl" + client = ResultsLowLevelClient(grpc_client=MagicMock()) + report, steps = await _build_log(client, log_file, step_names=("s1",)) + + # An earlier tick uploaded the report; the cursor is past line one. + LogTracking(last_uploaded_line=1, id_map={report.id_: "real-report"}).save(log_file) + + client.create_test_step = AsyncMock(return_value=_make_step("real-step-1")) + client.update_test_step = AsyncMock(return_value=_make_step("real-step-1")) + + await client.import_log_file(log_file, incremental=True) + + tracking = LogTracking.load(log_file) + assert not tracking.complete + assert tracking.last_uploaded_line == 3 + + +@pytest.mark.asyncio +async def test_idle_worker_tick_is_silent_and_writes_nothing(tmp_path): + """A tick with nothing new to upload must not log or touch the sidecar. + + The worker ticks once a second for the whole session, so anything it does on + an idle tick is multiplied by the length of the run: audit-log noise that + buries the real entries, and sidecar rewrites that are pure churn. + """ + log_file = tmp_path / "idle_tick.jsonl" + client = ResultsLowLevelClient(grpc_client=MagicMock()) + report, _ = await _build_log(client, log_file, step_names=("s1",)) + + # The sidecar is caught up with every line currently in the log. + LogTracking(last_uploaded_line=3, id_map={report.id_: "real-report"}).save(log_file) + sidecar = LogTracking.sidecar_path(log_file) + before = sidecar.read_bytes(), sidecar.stat().st_mtime_ns + + with _captured_replay_logs() as messages: + await client.import_log_file(log_file, incremental=True) + + assert messages == [] + assert (sidecar.read_bytes(), sidecar.stat().st_mtime_ns) == before + + +@pytest.mark.asyncio +async def test_resume_marks_the_log_complete(tmp_path): + """Finishing a resumed upload marks it complete, so a re-run does nothing.""" + log_file = tmp_path / "resume_completes.jsonl" + client = ResultsLowLevelClient(grpc_client=MagicMock()) + report, steps = await _build_log(client, log_file, step_names=("s1",)) + + LogTracking(id_map={report.id_: "real-report"}).save(log_file) + + client.get_test_report = AsyncMock(return_value=_make_report("real-report")) + client.create_test_step = AsyncMock(return_value=_make_step("real-step-1")) + client.update_test_step = AsyncMock(return_value=_make_step("real-step-1")) + + await client.import_log_file(log_file) + + assert LogTracking.load(log_file).complete + + +@pytest.mark.asyncio +async def test_completed_upload_is_a_noop(tmp_path): + """A log whose sidecar is caught up is not replayed again.""" + log_file = tmp_path / "done.jsonl" + client = ResultsLowLevelClient(grpc_client=MagicMock()) + report, steps = await _build_log(client, log_file) + + LogTracking( + complete=True, + id_map={report.id_: "real-report", steps[0].id_: "real-step-1"}, + ).save(log_file) + + client.get_test_report = AsyncMock() + client.create_test_report = AsyncMock() + client.create_test_step = AsyncMock() + + result = await client.import_log_file(log_file) + + client.get_test_report.assert_not_awaited() + client.create_test_report.assert_not_awaited() + client.create_test_step.assert_not_awaited() + assert result.report is None + assert result.steps == [] + + +@pytest.mark.asyncio +async def test_new_report_abandons_the_partial_upload(tmp_path): + """``new_report`` starts over and keeps the old sidecar as a backup.""" + log_file = tmp_path / "restart.jsonl" + client = ResultsLowLevelClient(grpc_client=MagicMock()) + report, steps = await _build_log(client, log_file) + + LogTracking(id_map={report.id_: "abandoned-report"}).save(log_file) + + client.get_test_report = AsyncMock() + created = _answer_real_creates( + client, report_id="fresh-report", step_ids=["real-step-1", "real-step-2"] + ) + + result = await client.import_log_file(log_file, new_report=True) + + client.get_test_report.assert_not_awaited() + assert created == ["report", "s1", "s2"] + assert result.report is not None + assert result.report.id_ == "fresh-report" + # The abandoned report's ID survives, so it can still be found and cleaned up. + backup = LogTracking.backup_path(log_file) + assert json.loads(backup.read_text())["idMap"] == {report.id_: "abandoned-report"} + assert LogTracking.load(log_file).id_map[report.id_] == "fresh-report" + + +@pytest.mark.asyncio +async def test_resume_into_deleted_report_explains_the_override(tmp_path): + """A report that no longer exists fails before anything is uploaded.""" + log_file = tmp_path / "deleted.jsonl" + client = ResultsLowLevelClient(grpc_client=MagicMock()) + report, _ = await _build_log(client, log_file) + + LogTracking(id_map={report.id_: "gone-report"}).save(log_file) + + client.get_test_report = AsyncMock(side_effect=_NotFoundError()) + client.create_test_report = AsyncMock() + client.create_test_step = AsyncMock() + + with pytest.raises(ValueError, match="--new-report"): + await client.import_log_file(log_file) + + client.create_test_report.assert_not_awaited() + client.create_test_step.assert_not_awaited() + + +async def _build_measurement_log(client, log_file, *, batched): + """Write a log holding a report, one step, and three measurements. + + ``batched`` picks how the measurements are logged: one ``CreateTestMeasurements`` + line covering all three, or a separate ``CreateTestMeasurement`` line each. + """ + report = await client.create_test_report(test_report=_report_create(), log_file=log_file) + step = await client.create_test_step( + test_step=TestStepCreate( + test_report_id=report.id_, + name="s1", + step_type=TestStepType.ACTION, + step_path="1", + status=TestStatus.PASSED, + start_time=T0, + end_time=T0, + ), + log_file=log_file, + ) + creates = [ + TestMeasurementCreate( + name=f"m{index}", + test_step_id=step.id_, + passed=True, + timestamp=T0, + numeric_value=float(index), + ) + for index in (1, 2, 3) + ] + if batched: + _, measurement_ids = await client.create_test_measurements( + test_measurements=creates, log_file=log_file + ) + else: + measurement_ids = [ + (await client.create_test_measurement(test_measurement=create, log_file=log_file)).id_ + for create in creates + ] + return report, step, measurement_ids + + +@pytest.mark.asyncio +async def test_resume_sends_only_the_missing_part_of_a_batched_line(tmp_path): + """One log line can create many measurements, and only some may have made it. + + Batch replay creates measurements one at a time, so an interrupted run can + leave a batched line half done. Re-sending the whole line would duplicate + the measurements that already exist. + """ + log_file = tmp_path / "batched_measurements.jsonl" + client = ResultsLowLevelClient(grpc_client=MagicMock()) + report, step, measurement_ids = await _build_measurement_log(client, log_file, batched=True) + + # The interrupted run got through the report, the step, and the first + # measurement of the batched line. + LogTracking( + id_map={ + report.id_: "real-report", + step.id_: "real-step", + measurement_ids[0]: "real-meas-1", + }, + ).save(log_file) + + client.get_test_report = AsyncMock(return_value=_make_report("real-report")) + client.create_test_measurements = AsyncMock(return_value=(2, ["real-meas-2", "real-meas-3"])) + + await client.import_log_file(log_file) + + client.create_test_measurements.assert_awaited_once() + sent = client.create_test_measurements.await_args.kwargs["request"] + assert [m.name for m in sent.test_measurements] == ["m2", "m3"] + tracking = LogTracking.load(log_file) + assert tracking.id_map[measurement_ids[1]] == "real-meas-2" + assert tracking.id_map[measurement_ids[2]] == "real-meas-3" + # The one that already existed keeps the ID the interrupted run recorded. + assert tracking.id_map[measurement_ids[0]] == "real-meas-1" + + +@pytest.mark.asyncio +async def test_resume_skips_a_measurement_already_created(tmp_path): + """A measurement logged on its own line is skipped once it is in the id map.""" + log_file = tmp_path / "single_measurements.jsonl" + client = ResultsLowLevelClient(grpc_client=MagicMock()) + report, step, measurement_ids = await _build_measurement_log(client, log_file, batched=False) + + LogTracking( + id_map={ + report.id_: "real-report", + step.id_: "real-step", + measurement_ids[0]: "real-meas-1", + measurement_ids[1]: "real-meas-2", + }, + ).save(log_file) + + client.get_test_report = AsyncMock(return_value=_make_report("real-report")) + client.create_test_measurement = AsyncMock(return_value=_make_measurement("real-meas-3")) + + await client.import_log_file(log_file) + + client.create_test_measurement.assert_awaited_once() + sent = client.create_test_measurement.await_args.kwargs["request"] + assert sent.test_measurement.name == "m3" + assert sent.test_measurement.test_step_id == "real-step" + + +@pytest.mark.asyncio +async def test_batch_upload_records_measurements(tmp_path): + """Measurements created by a batch upload are recorded like steps are. + + Without this the tail of a large upload is the part a resume cannot skip. + """ + log_file = tmp_path / "batch_measurements.jsonl" + client = ResultsLowLevelClient(grpc_client=MagicMock()) + report, step, measurement_ids = await _build_measurement_log(client, log_file, batched=False) + + created = _answer_real_creates( + client, + report_id="real-report", + step_ids=["real-step"], + measurement_ids=["real-meas-1", "real-meas-2", "real-meas-3"], + ) + + await client.import_log_file(log_file) + + assert created == ["report", "s1", "m1", "m2", "m3"] + tracking = LogTracking.load(log_file) + assert [tracking.id_map[mid] for mid in measurement_ids] == [ + "real-meas-1", + "real-meas-2", + "real-meas-3", + ] + + +@pytest.mark.asyncio +async def test_batch_upload_tolerates_an_untagged_create(tmp_path): + """A create logged without a response ID is uploaded but cannot be recorded. + + Nothing in the client writes such a line today, and one carrying later + updates would fail replay outright since the updates could not be remapped. + A trailing untagged create still has to reach the server; it just cannot be + skipped by a later resume, which beats aborting the whole upload. + """ + log_file = tmp_path / "untagged.jsonl" + client = ResultsLowLevelClient(grpc_client=MagicMock()) + report = await client.create_test_report(test_report=_report_create(), log_file=log_file) + step = await client.create_test_step( + test_step=TestStepCreate( + test_report_id=report.id_, + name="s1", + step_type=TestStepType.ACTION, + step_path="1", + status=TestStatus.PASSED, + start_time=T0, + end_time=T0, + ), + log_file=log_file, + ) + log_file.write_text( + log_file.read_text().replace(f"[CreateTestStep:{step.id_}]", "[CreateTestStep]") + ) + + created = _answer_real_creates(client, report_id="real-report", step_ids=["real-step-1"]) + + await client.import_log_file(log_file) + + assert created == ["report", "s1"] + tracking = LogTracking.load(log_file) + assert tracking.id_map == {report.id_: "real-report"} + + +@pytest.mark.asyncio +async def test_resume_without_a_recorded_report_explains_the_override(tmp_path): + """A sidecar that records work but no report cannot say what to resume into.""" + log_file = tmp_path / "no_report.jsonl" + client = ResultsLowLevelClient(grpc_client=MagicMock()) + _, steps = await _build_log(client, log_file) + + LogTracking(id_map={steps[0].id_: "real-step-1"}).save(log_file) + + client.get_test_report = AsyncMock() + client.create_test_report = AsyncMock() + + with pytest.raises(ValueError, match="--new-report"): + await client.import_log_file(log_file) + + client.get_test_report.assert_not_awaited() + client.create_test_report.assert_not_awaited() + + +@pytest.mark.asyncio +async def test_resume_propagates_errors_other_than_a_missing_report(tmp_path): + """Only a missing report is turned into resume guidance. + + A permissions or connectivity failure is not something --new-report fixes, + so it surfaces as itself. + """ + log_file = tmp_path / "denied.jsonl" + client = ResultsLowLevelClient(grpc_client=MagicMock()) + report, _ = await _build_log(client, log_file) + + LogTracking(id_map={report.id_: "real-report"}).save(log_file) + + client.get_test_report = AsyncMock(side_effect=_PermissionDeniedError()) + + with pytest.raises(RpcError): + await client.import_log_file(log_file) + + +@pytest.mark.asyncio +async def test_incremental_and_new_report_are_rejected_together(tmp_path): + """The two flags contradict each other, so asking for both is an error. + + Incremental replay continues whatever the sidecar records, which is exactly + what new_report discards. Silently honouring one of them would upload into + the wrong report. + """ + log_file = tmp_path / "conflict.jsonl" + client = ResultsLowLevelClient(grpc_client=MagicMock()) + await _build_log(client, log_file, step_names=("s1",)) + + with pytest.raises(ValueError, match="mutually exclusive"): + await client.import_log_file(log_file, incremental=True, new_report=True) + + +@pytest.mark.asyncio +async def test_new_report_on_a_fresh_log_writes_no_backup(tmp_path): + """``new_report`` against a log that was never uploaded has nothing to move aside.""" + log_file = tmp_path / "fresh.jsonl" + client = ResultsLowLevelClient(grpc_client=MagicMock()) + await _build_log(client, log_file, step_names=("s1",)) + + created = _answer_real_creates(client, report_id="fresh-report", step_ids=["real-step-1"]) + + await client.import_log_file(log_file, new_report=True) + + assert created == ["report", "s1"] + assert not LogTracking.backup_path(log_file).exists() + + # --------------------------------------------------------------------------- # Session directory grouping # --------------------------------------------------------------------------- @@ -238,14 +788,14 @@ def test_make_session_dir_concurrent_calls_unique(tmp_path, monkeypatch): def test_cleanup_temp_log_removes_session_dir(tmp_path, monkeypatch): - """``_cleanup_temp_log`` removes the whole session dir when audit is off. + """``cleanup_temp_log`` removes the whole session dir when audit is off. Session dir layout: ``/sift_test_results//``. The JSONL, its tracking sidecar, and any audit files in the dir are all removed. """ import tempfile - from sift_client.scripts.import_test_result_log import _cleanup_temp_log + from sift_client._internal.pytest_plugin.replay_worker import cleanup_temp_log monkeypatch.setattr(tempfile, "gettempdir", lambda: str(tmp_path)) session_dir = tmp_path / "sift_test_results" / "abc123" @@ -256,21 +806,21 @@ def test_cleanup_temp_log_removes_session_dir(tmp_path, monkeypatch): for f in (log, tracking, audit): f.write_text("{}") - _cleanup_temp_log(str(log)) + cleanup_temp_log(str(log)) assert not session_dir.exists() def test_cleanup_temp_log_ignores_explicit_path(tmp_path, monkeypatch): - """``_cleanup_temp_log`` does not touch a log outside the temp dir.""" + """``cleanup_temp_log`` does not touch a log outside the temp dir.""" import tempfile - from sift_client.scripts.import_test_result_log import _cleanup_temp_log + from sift_client._internal.pytest_plugin.replay_worker import cleanup_temp_log monkeypatch.setattr(tempfile, "gettempdir", lambda: str(tmp_path)) explicit_log = tmp_path.parent / "my_project_log.jsonl" explicit_log.write_text("{}") - _cleanup_temp_log(str(explicit_log)) + cleanup_temp_log(str(explicit_log)) assert explicit_log.exists() explicit_log.unlink() @@ -279,7 +829,7 @@ def test_cleanup_temp_log_legacy_flat_layout(tmp_path, monkeypatch): """Legacy flat-temp layout: only the JSONL and its tracking sidecar are removed.""" import tempfile - from sift_client.scripts.import_test_result_log import _cleanup_temp_log + from sift_client._internal.pytest_plugin.replay_worker import cleanup_temp_log monkeypatch.setattr(tempfile, "gettempdir", lambda: str(tmp_path)) log = tmp_path / "tmp12345.jsonl" @@ -288,7 +838,7 @@ def test_cleanup_temp_log_legacy_flat_layout(tmp_path, monkeypatch): for f in (log, tracking, other): f.write_text("{}") - _cleanup_temp_log(str(log)) + cleanup_temp_log(str(log)) assert not log.exists() assert not tracking.exists() diff --git a/python/lib/sift_client/_tests/util/test_report_context.py b/python/lib/sift_client/_tests/util/test_report_context.py index ead4e5f73..7033f5eee 100644 --- a/python/lib/sift_client/_tests/util/test_report_context.py +++ b/python/lib/sift_client/_tests/util/test_report_context.py @@ -76,9 +76,11 @@ def test_worker_timeout_kills_and_warns() -> None: assert rc._import_proc.poll() is not None messages = "\n".join(str(w.message) for w in recorded) assert "did not exit in 0.2s" in messages - # Recovery must resume from the tracking cursor, not batch-replay (which would - # duplicate already-uploaded entries), so the hint carries --incremental. - assert "import-test-result-log --incremental" in messages + # Recovery must resume rather than re-upload (which would duplicate what the + # worker already sent). The plain command resumes off the tracking sidecar, + # so the hint must not send anyone to --incremental for it. + assert "import-test-result-log" in messages + assert "--incremental" not in messages def test_worker_nonzero_exit_warns_stderr_no_raise() -> None: @@ -98,7 +100,8 @@ def test_worker_nonzero_exit_warns_stderr_no_raise() -> None: messages = "\n".join(str(w.message) for w in recorded) assert "exited with code 2" in messages assert "rpc deadline exceeded" in messages - assert "import-test-result-log --incremental" in messages + assert "import-test-result-log" in messages + assert "--incremental" not in messages def test_replay_command_runs_module_through_current_interpreter() -> None: @@ -112,4 +115,8 @@ def test_replay_command_runs_module_through_current_interpreter() -> None: """ rc = ReportContext(_make_simulate_client(), name="test", log_file=True) cmd = rc._build_replay_command() - assert cmd[:3] == [sys.executable, "-m", "sift_client.scripts.import_test_result_log"] + assert cmd[:3] == [ + sys.executable, + "-m", + "sift_client._internal.pytest_plugin.replay_worker", + ] diff --git a/python/lib/sift_client/resources/sync_stubs/__init__.pyi b/python/lib/sift_client/resources/sync_stubs/__init__.pyi index fd6a3b6c2..e92fbd35d 100644 --- a/python/lib/sift_client/resources/sync_stubs/__init__.pyi +++ b/python/lib/sift_client/resources/sync_stubs/__init__.pyi @@ -3141,16 +3141,26 @@ class TestResultsAPI: """ ... - def import_log_file(self, log_file: str | Path, incremental: bool = False) -> ReplayResult: + def import_log_file( + self, log_file: str | Path, incremental: bool = False, new_report: 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. + There are three modes. By default the log is uploaded as a new report, + or, if the tracking sidecar beside it records an upload that was + interrupted partway, the report that upload created is reused and only + the missing entries are sent. The third mode belongs to the plugin's + background worker, which follows a log while it is still being written. + Args: log_file: Path to the log file to import. - incremental: (internal tooling) If True, goes line by line and calls API every event -- keeps track of last line sent so it can be called after some updates and be additive vs. replaying the entire log file each time(i.e. when False, reads the entire log file, building a test report in memory, then sends the calls for each step/measurement to the API). + incremental: (internal tooling) If True, goes line by line and calls the API for every event, tracking the last line sent so it can be called repeatedly against a log that is still growing and stay additive. This is the worker's follow mode during a test run, not the way to finish an interrupted upload. + new_report: If True, ignore any partial upload and create a new report. + Mutually exclusive with incremental. Returns: A ReplayResult containing the created report, steps, and measurements. diff --git a/python/lib/sift_client/resources/test_results.py b/python/lib/sift_client/resources/test_results.py index 10ef70920..987d7c92d 100644 --- a/python/lib/sift_client/resources/test_results.py +++ b/python/lib/sift_client/resources/test_results.py @@ -656,6 +656,7 @@ async def import_log_file( self, log_file: str | Path, incremental: bool = False, + new_report: bool = False, ) -> ReplayResult: """Replay a log file by parsing each entry, simulating the results, then creating for real. @@ -663,14 +664,24 @@ async def import_log_file( all the objects via simulation, and then creates them via the actual API. IDs are mapped from simulated to real during the creation process. + There are three modes. By default the log is uploaded as a new report, + or, if the tracking sidecar beside it records an upload that was + interrupted partway, the report that upload created is reused and only + the missing entries are sent. The third mode belongs to the plugin's + background worker, which follows a log while it is still being written. + Args: log_file: Path to the log file to import. - incremental: (internal tooling) If True, goes line by line and calls API every event -- keeps track of last line sent so it can be called after some updates and be additive vs. replaying the entire log file each time(i.e. when False, reads the entire log file, building a test report in memory, then sends the calls for each step/measurement to the API). + incremental: (internal tooling) If True, goes line by line and calls the API for every event, tracking the last line sent so it can be called repeatedly against a log that is still growing and stay additive. This is the worker's follow mode during a test run, not the way to finish an interrupted upload. + new_report: If True, ignore any partial upload and create a new report. + Mutually exclusive with incremental. Returns: A ReplayResult containing the created report, steps, and measurements. """ - result = await self._low_level_client.import_log_file(log_file, incremental=incremental) + result = await self._low_level_client.import_log_file( + log_file, incremental=incremental, new_report=new_report + ) if result.report is not None: result.report = self._apply_client_to_instance(result.report) result.steps = self._apply_client_to_instances(result.steps) diff --git a/python/lib/sift_client/scripts/import_test_result_log.py b/python/lib/sift_client/scripts/import_test_result_log.py index 370b69e2e..78876dfac 100644 --- a/python/lib/sift_client/scripts/import_test_result_log.py +++ b/python/lib/sift_client/scripts/import_test_result_log.py @@ -5,15 +5,13 @@ import argparse import logging import os -import select -import shutil -import sys -import tempfile from pathlib import Path from typing import TYPE_CHECKING from sift_client import SiftClient, SiftConnectionConfig +from sift_client._internal.low_level_wrappers._test_results_log import LogTracking from sift_client._internal.pytest_plugin.audit_log import log_event +from sift_client._internal.pytest_plugin.replay_worker import cleanup_temp_log from sift_client.util.test_results.context_manager import log_replay_instructions if TYPE_CHECKING: @@ -22,6 +20,29 @@ logger = logging.getLogger(__name__) +def _describe_upload(log_file: str, new_report: bool) -> None: + """Say up front whether this run continues an upload or starts one. + + The sidecar decides, so without this the same command can do several quite + different things with no way to tell which from the output. The branches + read the same recorded state the importer routes on, so the two cannot + drift into disagreeing about what is about to happen. + """ + if new_report: + print(f"Uploading {log_file} as a new report.") + return + tracking = LogTracking.load(log_file) + if tracking.complete: + print(f"{log_file} is already fully uploaded; nothing to do.") + elif tracking.id_map: + print( + f"Resuming the interrupted upload of {log_file} " + f"({len(tracking.id_map)} already uploaded)." + ) + else: + print(f"Uploading {log_file}.") + + def _print_result(result: ReplayResult) -> None: if result.report is not None: print(f"Report: {result.report.name} (id={result.report.id_})") @@ -33,69 +54,25 @@ def _print_result(result: ReplayResult) -> None: print(f" - {m.name}: passed={m.passed}") -def _cleanup_temp_log(log_file: str) -> None: - """Remove temp artifacts after a successful upload when audit logging is off. - - Called only when audit logging is off: without an audit trail there's no - reason to retain the buffer, so default temp artifacts are reclaimed - immediately. An explicit ``--sift-output-dir`` (not under the temp dir) is - the user's to keep and is never touched. - - Session-dir layout (``/sift_test_results//``): the whole - directory is removed, cleaning up the JSONL, tracking sidecar, lock, and - any audit files in one shot. - - Legacy flat-temp layout (file directly in tmpdir): only the JSONL and its - tracking sidecar are removed individually. - """ - fp = Path(log_file).absolute() - if not str(fp).startswith(tempfile.gettempdir()): - return - session_dir = fp.parent - if session_dir.parent == Path(tempfile.gettempdir()) / "sift_test_results": - shutil.rmtree(session_dir, ignore_errors=True) - log_event(logger, logging.DEBUG, "replay.cleanup", log=str(fp), dir=str(session_dir)) - return - fp.unlink(missing_ok=True) - fp.with_name(fp.name + ".tracking").unlink(missing_ok=True) - log_event(logger, logging.DEBUG, "replay.cleanup", log=str(fp)) - - -def _incremental_import_loop( - client: SiftClient, log_file: str, *, keep_log: bool -) -> ReplayResult | None: - """Replay incrementally in a loop until stdin is closed (EOF). - - Per-entity upload detail and sidecar advances are logged inside the - incremental importer (``replay.upload`` / ``replay.error``); idle ticks - that upload nothing are silent on purpose. - - When ``keep_log`` is False (audit logging off) the temp log is deleted on a - clean finish; with audit logging on it's retained alongside the audit trail. - """ - 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 - log_event(logger, logging.INFO, "replay.complete", log=log_file) - if not keep_log: - _cleanup_temp_log(log_file) - return result - - 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.", + epilog=( + "Runs in one of two modes. With no flags it uploads the log as a new " + "report, or resumes into the report an interrupted earlier run created, " + "whichever the tracking sidecar calls for. --new-report forces the first." + ), ) parser.add_argument("log_file", help="Path to the .jsonl log file to replay.") 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." + "--new-report", + action="store_true", + help="Ignore a partially uploaded report and upload the log as a new one. " + "By default an interrupted upload is resumed into the report it created.", ) parser.add_argument( "--audit-log", default=None, help="Path to the replay worker's DEBUG audit log." @@ -121,16 +98,13 @@ def main() -> None: ) ) - # The worker is spawned with --audit-log only when audit logging is on, so - # its presence is the signal to retain the buffer after a clean upload. - keep_log = bool(args.audit_log) + _describe_upload(args.log_file, args.new_report) try: - if args.incremental: - result = _incremental_import_loop(client, args.log_file, keep_log=keep_log) - else: - result = client.test_results.import_log_file(args.log_file) - if not keep_log: - _cleanup_temp_log(args.log_file) + result = client.test_results.import_log_file(args.log_file, new_report=args.new_report) + # An audit log means the run is being traced, so the buffer is retained + # alongside the trail rather than reclaimed. + if not args.audit_log: + cleanup_temp_log(args.log_file) except Exception as e: log_event(logger, logging.ERROR, "replay.failed", error=repr(e)) log_replay_instructions(args.log_file) diff --git a/python/lib/sift_client/util/test_results/context_manager.py b/python/lib/sift_client/util/test_results/context_manager.py index a3d59b105..ad0003fed 100644 --- a/python/lib/sift_client/util/test_results/context_manager.py +++ b/python/lib/sift_client/util/test_results/context_manager.py @@ -89,7 +89,8 @@ def log_replay_instructions(log_file: str | Path | None) -> None: return warnings.warn( f"Sift log file was not fully replayed: {log_file}. " - f"Re-run with `import-test-result-log --incremental {log_file}` to complete the upload.", + f"Re-run `import-test-result-log {log_file}` to complete the upload; it resumes " + f"into the report the interrupted run created rather than starting a new one.", SiftWarning, stacklevel=2, ) @@ -258,9 +259,9 @@ def __init__( on top of git metadata when ``include_git_metadata`` is True, so explicit keys win on collision. replay_log_file: When True (the default) and ``log_file`` is set, - spawn ``import-test-result-log --incremental`` to push log - entries to Sift in the background during the session. When - False, the log file is just a record and no worker is spawned. + spawn the background replay worker to push log entries to Sift + during the session. When False, the log file is just a record + and no worker is spawned. Replay happens later via ``import-test-result-log ``. Has no effect when ``log_file`` is None. audit_log: When set, the path of a DEBUG audit log. The replay worker @@ -319,23 +320,22 @@ def __init__( self.report = client.test_results.create(create, log_file=self.log_file) def _build_replay_command(self) -> list[str]: - """Build the argv for the import-test-result-log replay subprocess. + """Build the argv for the background replay worker subprocess. Factored out for testability: tests substitute commands that exit with controlled returncodes / stderr to exercise the ``__exit__`` - branches without depending on the real replay binary. + branches without depending on the real replay worker. """ cmd = [ - # Invoked through the running interpreter rather than by the bare - # ``import-test-result-log`` name, which resolves through PATH. The - # console script installs into the venv's bin/, so a bare-name spawn - # raises FileNotFoundError wherever that bin/ isn't on PATH: under + # Invoked through the running interpreter so the worker imports the + # same sift_client the session is using. Spawning by a bare command + # name would instead resolve through PATH, which raises + # FileNotFoundError wherever the venv's bin/ isn't on it: under # ``sudo`` (sudoers' secure_path replaces PATH even with ``-E``), or # running ``python -m pytest`` against a non-activated venv. sys.executable, "-m", - "sift_client.scripts.import_test_result_log", - "--incremental", + "sift_client._internal.pytest_plugin.replay_worker", str(self.log_file), "--grpc-url", self.client.grpc_client._config.uri,