From 50a8e61ab73756555de130181c81a85031a6bf36 Mon Sep 17 00:00:00 2001 From: Marlon Costa Date: Sat, 26 Sep 2026 10:07:53 -0300 Subject: [PATCH 1/2] fix(cli): let the remaining CLI_SAFE_EXCEPTIONS catches fail loud Remove every remaining catch-and-normalize over CLI_SAFE_EXCEPTIONS (ValueError, TypeError, KeyError), the same violation class #202 removed from prompts, and delete the constant now that nothing uses it. - u.Cli.cmd_validate_settings: no catch. The structure probe raises none of those types; any exception it raises now escapes with its cause. - u.Cli.tables_resolve_config: only the model's ValidationError is the declared failure outcome; the result carries it (exception=exc). Anything else escapes. - u.Cli.tables_render: renderer exceptions escape instead of becoming r.fail("Table formatting ..."). - cli.apply_to_config: only a CliParamsConfig ValidationError is a declared failure and the result carries it; params_apply failures stay r.fail from their owners; every other exception escapes. - u.Cli.process_mapping: removed with t.Cli.MappingProcessor. Its skip mode swallowed exceptions into a debug log and its collect mode aggregated them; no member of the workspace calls it. - Removed the constants only the removed catches used: CLI_SAFE_EXCEPTIONS, ERR_SETTINGS_VALIDATION_FAILED, OUTPUT_TABLE_FORMATTING_OPERATION. Tests that only asserted process_mapping's normalization were removed. The tables and params tests now assert the failure carries the ValidationError. Co-Authored-By: Claude Opus 5.5 (1M context) --- src/flext_cli/_constants/base.py | 5 -- src/flext_cli/_constants/errors.py | 3 -- src/flext_cli/_constants/output.py | 1 - src/flext_cli/_typings/domain.py | 1 - src/flext_cli/_utilities/cmd.py | 15 +++--- src/flext_cli/_utilities/tables.py | 39 ++++++++-------- src/flext_cli/_utilities/validation.py | 33 ------------- src/flext_cli/services/cli_params.py | 11 +++-- tests/unit/test_cli_params.py | 10 ++++ .../test_cmd_runtime_validation_branch_cov.py | 46 ------------------- tests/unit/test_tables_cov.py | 1 + tests/unit/test_utilities_cov.py | 36 --------------- 12 files changed, 44 insertions(+), 157 deletions(-) diff --git a/src/flext_cli/_constants/base.py b/src/flext_cli/_constants/base.py index d1e971b84..ab9730df8 100644 --- a/src/flext_cli/_constants/base.py +++ b/src/flext_cli/_constants/base.py @@ -26,11 +26,6 @@ class FlextCliConstantsBase: OP_EXECUTE_APPLICATION: ClassVar[str] = "execute CLI application" ERR_EXIT_WITH_CODE: ClassVar[str] = "CLI exited with code {exit_code}" - CLI_SAFE_EXCEPTIONS: ClassVar[t.VariadicTuple[type[Exception]]] = ( - ValueError, - TypeError, - KeyError, - ) # Pipeline retries fail loud by default; the bound prevents accidental # exponential fan-out when a caller declares a retry policy. PIPELINE_DEFAULT_RETRY: ClassVar[int] = 0 diff --git a/src/flext_cli/_constants/errors.py b/src/flext_cli/_constants/errors.py index b488330c2..e6853cefc 100644 --- a/src/flext_cli/_constants/errors.py +++ b/src/flext_cli/_constants/errors.py @@ -60,9 +60,6 @@ class FlextCliConstantsErrors: ERR_INVALID_CREDENTIALS: ClassVar[str] = ( "Invalid credentials: missing token or username/password" ) - ERR_SETTINGS_VALIDATION_FAILED: ClassVar[str] = ( - "Settings validation failed: {error}" - ) ERR_AUTH_SAVE_FAILED: ClassVar[str] = "Failed to save token: {error}" ERR_AUTH_LOAD_FAILED: ClassVar[str] = "Failed to load token: {error}" ERR_AUTH_FILE_NOT_FOUND: ClassVar[str] = "Token file does not exist" diff --git a/src/flext_cli/_constants/output.py b/src/flext_cli/_constants/output.py index 9c8c32559..ce6300aaf 100644 --- a/src/flext_cli/_constants/output.py +++ b/src/flext_cli/_constants/output.py @@ -70,7 +70,6 @@ class FlextCliConstantsOutput: ) OUTPUT_TABLE_NORMALIZATION_FAILED: ClassVar[str] = "Table normalization failed" OUTPUT_TABLE_ROW_INVALID: ClassVar[str] = "Table row invalid after validation" - OUTPUT_TABLE_FORMATTING_OPERATION: ClassVar[str] = "Table formatting" TABLE_FORMATS: ClassVar[t.StrMapping] = MappingProxyType({ ce.TabularFormat.PLAIN: "Minimal formatting, no borders", diff --git a/src/flext_cli/_typings/domain.py b/src/flext_cli/_typings/domain.py index 3fb080823..b5c037d92 100644 --- a/src/flext_cli/_typings/domain.py +++ b/src/flext_cli/_typings/domain.py @@ -69,7 +69,6 @@ class FlextCliTypesDomain: type SuccessMessageFormatter[TResult: ResultValue = ResultValue] = Callable[ [TResult], str ] - type MappingProcessor[T, U] = Callable[[str, T], U] type TomlMappingSource = ( t.JsonPayload | t.JsonMapping | t.ScalarMapping | Item | TOMLDocument ) diff --git a/src/flext_cli/_utilities/cmd.py b/src/flext_cli/_utilities/cmd.py index 9d349335e..83f0c8750 100644 --- a/src/flext_cli/_utilities/cmd.py +++ b/src/flext_cli/_utilities/cmd.py @@ -49,15 +49,12 @@ def cmd_show_settings(logger: p.Logger) -> p.Result[bool]: @staticmethod def cmd_validate_settings(logger: p.Logger) -> p.Result[bool]: """Validate canonical settings structure and log normalized results.""" - try: - results = us.validate_settings_structure() - if results: - logger.info( - c.Cli.LOG_MSG_SETTINGS_VALIDATION_RESULTS.format(results=results) - ) - return r[bool].ok(True) - except c.Cli.CLI_SAFE_EXCEPTIONS as exc: - return r[bool].fail(c.Cli.ERR_SETTINGS_VALIDATION_FAILED.format(error=exc)) + results = us.validate_settings_structure() + if results: + logger.info( + c.Cli.LOG_MSG_SETTINGS_VALIDATION_RESULTS.format(results=results) + ) + return r[bool].ok(True) __all__: t.MutableSequenceOf[str] = ["FlextCliUtilitiesCmd"] diff --git a/src/flext_cli/_utilities/tables.py b/src/flext_cli/_utilities/tables.py index d3203d6eb..5924bed55 100644 --- a/src/flext_cli/_utilities/tables.py +++ b/src/flext_cli/_utilities/tables.py @@ -39,22 +39,26 @@ def tables_resolve_config( settings: m.Cli.TableConfig | None = None, **settings_kwargs: t.Cli.TableConfigValue, ) -> p.Result[m.Cli.TableConfig]: - """Resolve table config via canonical Pydantic model contract.""" + """Resolve table config via canonical Pydantic model contract. + + An override the model rejects is the declared failure outcome and the + result carries the ``ValidationError``; every other exception escapes. + """ + if settings is not None and not settings_kwargs: + return r[m.Cli.TableConfig].ok(settings) + base_data = ( + settings.model_dump(exclude_computed_fields=True) + if settings is not None + else {} + ) + settings_data = {**base_data, **settings_kwargs} try: - if settings is not None and not settings_kwargs: - return r[m.Cli.TableConfig].ok(settings) - base_data = ( - settings.model_dump(exclude_computed_fields=True) - if settings is not None - else {} - ) - settings_data = {**base_data, **settings_kwargs} resolved = m.Cli.TableConfig.model_validate(settings_data) - return r[m.Cli.TableConfig].ok(resolved) - except c.Cli.CLI_SAFE_EXCEPTIONS as exc: + except c.ValidationError as exc: return r[m.Cli.TableConfig].fail( - c.Cli.OUTPUT_TABLE_CONFIG_INVALID_FMT.format(error=exc) + c.Cli.OUTPUT_TABLE_CONFIG_INVALID_FMT.format(error=exc), exception=exc ) + return r[m.Cli.TableConfig].ok(resolved) @staticmethod def tables_normalize_data( @@ -140,13 +144,10 @@ def tables_render( if colalign is not None and column_count > 0 and len(colalign) > column_count: colalign = colalign[:column_count] - try: - rendered_table = FlextCliUtilitiesTablesRenderer.render( - rows, headers, colalign=colalign or (), settings=settings - ) - return r[str].ok(rendered_table) - except c.Cli.CLI_SAFE_EXCEPTIONS as exc: - return r[str].fail_op(c.Cli.OUTPUT_TABLE_FORMATTING_OPERATION, exc) + rendered_table = FlextCliUtilitiesTablesRenderer.render( + rows, headers, colalign=colalign or (), settings=settings + ) + return r[str].ok(rendered_table) __all__: t.MutableSequenceOf[str] = ["FlextCliUtilitiesTables"] diff --git a/src/flext_cli/_utilities/validation.py b/src/flext_cli/_utilities/validation.py index c75f46243..5be779652 100644 --- a/src/flext_cli/_utilities/validation.py +++ b/src/flext_cli/_utilities/validation.py @@ -2,11 +2,7 @@ from __future__ import annotations -from collections.abc import MutableMapping -from typing import ClassVar - from flext_cli import c, m, p, r, t -from flext_core import u PydanticValidationError = m.ValidationError @@ -14,35 +10,6 @@ class FlextCliUtilitiesValidation: """Validation methods exposed directly on ``u.Cli``.""" - _module_logger: ClassVar[p.Logger] = u.fetch_logger(__name__) - - @staticmethod - def process_mapping[T, U]( - items: t.MappingKV[str, T], - processor: t.Cli.MappingProcessor[T, U], - on_error: str = "fail", - ) -> p.Result[t.MappingKV[str, U]]: - """Process a mapping of items with canonical error handling.""" - errors: t.MutableSequenceOf[str] = [] - values: MutableMapping[str, U] = {} - for key, value in items.items(): - try: - values[key] = processor(key, value) - except c.Cli.CLI_SAFE_EXCEPTIONS as exc: - if on_error == "fail": - return r[t.MappingKV[str, U]].fail(f"Error processing {key}: {exc}") - if on_error == "collect": - errors.append(f"{key}: {exc}") - else: - FlextCliUtilitiesValidation._module_logger.debug( - f"process_mapping skip key {key}: {exc}", exc_info=False - ) - return ( - r[t.MappingKV[str, U]].fail("; ".join(errors)) - if errors - else r[t.MappingKV[str, U]].ok(values) - ) - @staticmethod def validate_not_empty( val: t.Cli.CliValue | None, *, name: str = "field" diff --git a/src/flext_cli/services/cli_params.py b/src/flext_cli/services/cli_params.py index 936971041..ca8ef8648 100644 --- a/src/flext_cli/services/cli_params.py +++ b/src/flext_cli/services/cli_params.py @@ -27,15 +27,18 @@ def apply_to_config( """Apply CLI parameter values to FlextSettings using Pydantic validation. Business Rule: Applies CLI parameter values with Pydantic validation. - Trace mode requires debug mode to be enabled. + Trace mode requires debug mode to be enabled. Parameter values the + ``CliParamsConfig`` model rejects are the declared failure outcome and + the result carries the ``ValidationError``; every other exception + escapes. """ try: params_to_use = u.Cli.params_resolve(params, kwargs) - return u.Cli.params_apply(settings, params_to_use) - except c.Cli.CLI_SAFE_EXCEPTIONS as exc: + except c.ValidationError as exc: return r[p.Cli.Settings].fail( - c.Cli.CLI_PARAM_ERR_APPLY_FAILED_FMT.format(error=exc) + c.Cli.CLI_PARAM_ERR_APPLY_FAILED_FMT.format(error=exc), exception=exc ) + return u.Cli.params_apply(settings, params_to_use) @classmethod def create_option(cls, field_name: str) -> m.Cli.OptionSpec: diff --git a/tests/unit/test_cli_params.py b/tests/unit/test_cli_params.py index aa4aad4ae..5afc591a1 100644 --- a/tests/unit/test_cli_params.py +++ b/tests/unit/test_cli_params.py @@ -94,6 +94,16 @@ def test_apply_to_config_trace_without_debug_fails( tm.fail(result) tm.that((result.error or "").lower(), has="trace mode requires debug mode") + def test_apply_to_config_unknown_parameter_fails_carrying_validation_error( + self, settings: p.Cli.Settings + ) -> None: + """A parameter the params model rejects fails and keeps its cause.""" + result = cli.apply_to_config(settings, not_a_param=True) + + tm.fail(result) + tm.that((result.error or ""), has="not_a_param") + tm.that(result.exception, is_=c.ValidationError) + @pytest.mark.parametrize( ("field_name", "expected_fragments"), [ diff --git a/tests/unit/test_cmd_runtime_validation_branch_cov.py b/tests/unit/test_cmd_runtime_validation_branch_cov.py index 1a6ed1f62..01b90ace9 100644 --- a/tests/unit/test_cmd_runtime_validation_branch_cov.py +++ b/tests/unit/test_cmd_runtime_validation_branch_cov.py @@ -8,21 +8,11 @@ from __future__ import annotations -from collections.abc import Callable - import pytest from flext_tests import tm from tests import c, m, t, u -type MappingProcessor = Callable[[str, int], int] - - -def _raise_bad(_key: str, _value: int) -> int: - """Processor that always raises to drive the error-handling branches.""" - msg = "bad-item" - raise ValueError(msg) - class TestsFlextCliCmdRuntimeValidationBranchCov: """Public-contract behavior of cmd, runtime, and validation helpers.""" @@ -81,42 +71,6 @@ def test_process_env_removing_unknown_key_is_a_noop(self) -> None: tm.that(pruned, eq=baseline) # ---------------------------------------------------------- validation - def test_process_mapping_maps_every_value_on_success(self) -> None: - """Verify that process mapping maps every value on success.""" - result = u.Cli.process_mapping({"a": 1, "b": 2}, lambda _key, value: value * 10) - - tm.ok(result) - tm.that(result.unwrap(), eq={"a": 10, "b": 20}) - - def test_process_mapping_fail_mode_stops_at_first_error(self) -> None: - """Verify that process mapping fail mode stops at first error.""" - result = u.Cli.process_mapping({"a": 1}, _raise_bad, on_error="fail") - - tm.fail(result) - tm.that((result.error or ""), has="a") - tm.that((result.error or ""), has="bad-item") - - def test_process_mapping_collect_mode_aggregates_errors(self) -> None: - """Verify that process mapping collect mode aggregates errors.""" - result = u.Cli.process_mapping({"a": 1}, _raise_bad, on_error="collect") - - tm.fail(result) - tm.that((result.error or ""), has="a: bad-item") - - def test_process_mapping_skip_mode_yields_only_successful_items(self) -> None: - """Verify that process mapping skip mode yields only successful items.""" - - def raise_on_a(key: str, value: int) -> int: - if key == "a": - msg = "boom" - raise ValueError(msg) - return value - - result = u.Cli.process_mapping({"a": 1, "b": 2}, raise_on_a, on_error="skip") - - tm.ok(result) - tm.that(result.unwrap(), eq={"b": 2}) - @pytest.mark.parametrize("output_format", tuple(c.Cli.OUTPUT_FORMATS)) def test_validate_format_accepts_every_supported_format( self, output_format: str diff --git a/tests/unit/test_tables_cov.py b/tests/unit/test_tables_cov.py index 2084ca61a..977273e10 100644 --- a/tests/unit/test_tables_cov.py +++ b/tests/unit/test_tables_cov.py @@ -92,6 +92,7 @@ def test_resolve_config_invalid_kwarg_fails_with_config_message(self) -> None: result = u.Cli.tables_resolve_config(not_a_field="oops") tm.fail(result) tm.that((result.error or ""), has="Invalid table configuration") + tm.that(result.exception, is_=c.ValidationError) # ── tables_normalize_data ───────────────────────────────────────── diff --git a/tests/unit/test_utilities_cov.py b/tests/unit/test_utilities_cov.py index 411349c81..9a1b503ce 100644 --- a/tests/unit/test_utilities_cov.py +++ b/tests/unit/test_utilities_cov.py @@ -2,7 +2,6 @@ Exercises only the observable contract of the utility helpers: - ``u.process`` — item processing with predicate filtering and error policy. -- ``u.Cli.process_mapping`` — keyed processing with fail/collect/skip policy. - ``u.Cli.validate_not_empty`` — emptiness validation returning ``r[bool]``. - ``u.Cli.project_names_from_values`` / ``project_numbers_from_values`` — CLI selector normalization. @@ -24,11 +23,6 @@ def _raise_on_zero(value: int) -> int: return 10 // value -def _raise_on_zero_kv(_key: str, value: int) -> int: - """Delegate key/value arguments to :func:`_raise_on_zero`.""" - return _raise_on_zero(value) - - class TestsFlextCliUtilitiesCov: """Behavioral contract for CLI utility helpers.""" @@ -56,36 +50,6 @@ def test_process_predicate_excludes_items_before_processing(self) -> None: tm.ok(result) tm.that(list(result.unwrap()), eq=[10, 2]) - def test_process_mapping_returns_mapped_values_on_success(self) -> None: - """Verify that process mapping returns mapped values on success.""" - result = u.Cli.process_mapping({"a": 2, "b": 5}, _raise_on_zero_kv) - tm.ok(result) - tm.that(result.unwrap(), eq={"a": 5, "b": 2}) - - def test_process_mapping_fail_policy_reports_offending_key(self) -> None: - """Verify that process mapping fail policy reports offending key.""" - result = u.Cli.process_mapping( - {"ok": 2, "bad": 0}, _raise_on_zero_kv, on_error="fail" - ) - tm.fail(result) - tm.that(result.error or "", has="bad") - - def test_process_mapping_collect_policy_reports_offending_key(self) -> None: - """Verify that process mapping collect policy reports offending key.""" - result = u.Cli.process_mapping( - {"ok": 2, "bad": 0}, _raise_on_zero_kv, on_error="collect" - ) - tm.fail(result) - tm.that(result.error or "", has="bad") - - def test_process_mapping_skip_policy_keeps_only_successes(self) -> None: - """Verify that process mapping skip policy keeps only successes.""" - result = u.Cli.process_mapping( - {"ok": 2, "bad": 0}, _raise_on_zero_kv, on_error="skip" - ) - tm.ok(result) - tm.that(result.unwrap(), eq={"ok": 5}) - @pytest.mark.parametrize("value", [None, "", " "]) def test_validate_not_empty_fails_for_empty_inputs(self, value: str | None) -> None: """Verify that validate not empty fails for empty inputs.""" From 828053658ed9876ec3002d6cf7aad7d762390236 Mon Sep 17 00:00:00 2001 From: Marlon Costa Date: Sat, 26 Sep 2026 10:08:06 -0300 Subject: [PATCH 2/2] test(cli): assert process completion and containment without host timing Both runtime-process tests asserted wall-clock thresholds and failed under host load (3.09s against a 2s limit, and the 10s pytest timeout, at load ~9.8). The thresholds measured the host, not the behaviour. - test_completion_before_wake_clear_ends_monitoring (was ..._returns_promptly): the trace holds the monitor at its first wake.clear() until the root waiter has set process_done and wake (unbounded Event.wait, no 2s cap). A monitor that lost that notification would sleep to its deadline and report timed_out, so `not outcome.timed_out` is the proof. The elapsed <2s assertion is gone and the child only prints (no sleep), started with -I -S. - test_normal_root_exit_leaves_no_descendant: the descendant announces its pid and process group over a pipe and blocks on an Event forever; the root does a blocking readline, records it, and exits. No file polling, no heartbeat, no sleep(0.15), no elapsed <5s check. The proof is that the descendant pid and its process group no longer exist when run_to_file returns. Interpreters start with -I -S so the scenario does not pay site-packages startup. Co-Authored-By: Claude Opus 5.5 (1M context) --- tests/unit/test_runtime_process_completion.py | 19 +++++------ .../unit/test_runtime_process_descendants.py | 34 +++++++++---------- 2 files changed, 25 insertions(+), 28 deletions(-) diff --git a/tests/unit/test_runtime_process_completion.py b/tests/unit/test_runtime_process_completion.py index a2292fe75..50310970c 100644 --- a/tests/unit/test_runtime_process_completion.py +++ b/tests/unit/test_runtime_process_completion.py @@ -5,7 +5,6 @@ import linecache import sys import threading -import time from pathlib import Path from types import FrameType from typing import TYPE_CHECKING @@ -21,11 +20,14 @@ class TestsRuntimeProcessCompletion: """A completed child must not wait for its execution deadline.""" - def test_completion_before_wake_clear_returns_promptly( - self, tmp_path: Path - ) -> None: + def test_completion_before_wake_clear_ends_monitoring(self, tmp_path: Path) -> None: """Schedule a real waiter at the exact lost-notification boundary. + The trace holds the monitor at its first ``wake.clear()`` until the + root waiter has set both ``process_done`` and ``wake``, so completion + is always observed before the wake is cleared. A monitor that lost + that notification would sleep until its deadline and report + ``timed_out``; the outcome, not the host clock, proves it did not. Tracing only controls thread scheduling; process creation, events, waiting, output and cleanup all use the unmodified public runtime. """ @@ -43,26 +45,23 @@ def trace(frame: FrameType, event: str, _argument: object) -> TraceFunction: wake = frame.f_locals["wake"] assert isinstance(done, threading.Event) assert isinstance(wake, threading.Event) - assert done.wait(2) - assert wake.wait(2) + done.wait() + wake.wait() scheduled.append(True) return trace previous = sys.gettrace() - started = time.monotonic() try: sys.settrace(trace) result = u.Cli().run_to_file( - [sys.executable, "-c", "import time;time.sleep(.1);print('complete')"], + [sys.executable, "-I", "-S", "-c", "print('complete')"], tmp_path / "completion.log", timeout=4, ) finally: sys.settrace(previous) - elapsed = time.monotonic() - started outcome = tm.ok(result) assert scheduled == [True] assert outcome.raw_return_code == 0 assert not outcome.timed_out - assert elapsed < 2 assert (tmp_path / "completion.log").read_text() == "complete\n" diff --git a/tests/unit/test_runtime_process_descendants.py b/tests/unit/test_runtime_process_descendants.py index 457e53fbd..942e9291a 100644 --- a/tests/unit/test_runtime_process_descendants.py +++ b/tests/unit/test_runtime_process_descendants.py @@ -41,28 +41,30 @@ class TestsFlextCliRuntimeProcessDescendants: """Prove root completion is not mistaken for boundary completion.""" def test_normal_root_exit_leaves_no_descendant(self, tmp_path: Path) -> None: + """A descendant alive when the root exits is gone when the run returns. + + The descendant announces its identity over a pipe and then blocks + forever; the root reads that line (a blocking read, no polling), + records it, and exits. Only containment can end the descendant, so + its absence from the process table after ``run_to_file`` returns is + the observable proof. + """ output_file = tmp_path / "normal-exit.log" - heartbeat = tmp_path / "normal-heartbeat" process_info = tmp_path / "normal-process-info" child = ( - "import os,pathlib,sys,time;" - "path=pathlib.Path(sys.argv[1]);" - "group=getattr(os,'getpgrp',lambda:0)();" - "pathlib.Path(sys.argv[2]).write_text(f'{os.getpid()} {group}');" - "\nwhile True:\n path.write_text(str(time.monotonic()));time.sleep(.02)" + "import os,threading;" + "print(os.getpid(),getattr(os,'getpgrp',lambda:0)(),flush=True);" + "threading.Event().wait()" ) parent = ( - "import pathlib,subprocess,sys,time;" - f"subprocess.Popen([sys.executable,'-c',{child!r}," - "sys.argv[1],sys.argv[2]]);" - "heartbeat=pathlib.Path(sys.argv[1]);" - "info=pathlib.Path(sys.argv[2]);" - "\nwhile not heartbeat.exists() or not info.exists():\n time.sleep(.01)" + "import pathlib,subprocess,sys;" + f"child=subprocess.Popen([sys.executable,'-I','-S','-c',{child!r}]," + "stdout=subprocess.PIPE);" + "pathlib.Path(sys.argv[1]).write_bytes(child.stdout.readline())" ) - started = time.monotonic() result = u.Cli().run_to_file( - [sys.executable, "-c", parent, str(heartbeat), str(process_info)], + [sys.executable, "-I", "-S", "-c", parent, str(process_info)], output_file, ) @@ -71,14 +73,10 @@ def test_normal_root_exit_leaves_no_descendant(self, tmp_path: Path) -> None: child_pid, process_group = ( int(value) for value in process_info.read_text().split() ) - stopped_value = heartbeat.stat().st_mtime_ns - time.sleep(0.15) - tm.that(heartbeat.stat().st_mtime_ns, eq=stopped_value) tm.that(_process_exists(child_pid), eq=False) if os.name != "nt": with pytest.raises(ProcessLookupError): os.killpg(process_group, 0) - tm.that(time.monotonic() - started, lt=5.0) def test_windows_job_reports_zero_active_processes(self, tmp_path: Path) -> None: _ObservedWindowsCli.active_counts.clear()