Hypothesis stateful tests for metrics/tracing/logging hooks (cuprum/adapters/metrics_adapter.py)

[coderabbitai]

Overview

Add Hypothesis stateful tests for the metrics, tracing, and logging hooks in cuprum/adapters/metrics_adapter.py (around line 206).

Justification

Event streams can arrive in varied phase/order/pid combinations, and current tests do not exercise these combinations systematically.

Benefit

• Proves counters, spans, and log records are created exactly when intended
• Proves active maps drain correctly

Cost Trade-Off

Medium. Requires synthetic event generators.

Verifiability Improvements

Extract event-to-operation reducers from side-effecting collector/span calls.

Recommended Order

Phase 3 — state-machine work after refactors isolate pure transitions.

---

Raised from PR #62 by @leynos.

[leynos]

@CodeRabbit plan

[coderabbitai]

Coding Plan

Summary

• Extract pure "reducer" functions from each adapter hook that map events to intended operations, enabling direct Hypothesis testing without mocking.
• Create a reusable @st.composite ExecEvent strategy respecting phase-field constraints and generating coherent PID lifecycles.
• Use RuleBasedStateMachine for TracingHook (the only stateful adapter) and simpler @given sequences for MetricsHook and the logging hook.
• Prove the key invariants: counters accumulate correctly, histogram observations match exit events, and _active_spans drains when all started PIDs exit.

Design Choices

Design Choice 1: Scope of "stateful tests" for stateless adapters

Options Considered:

1. Apply Hypothesis RuleBasedStateMachine only to TracingHook; use simpler @given property tests for MetricsHook and logging hook.
2. Force all three adapters into a RuleBasedStateMachine pattern regardless of internal state.

Chosen Option: 1

Rationale: Use the test pattern that matches each adapter's actual state model—RuleBasedStateMachine for TracingHook, @given sequences for MetricsHook and logging hook. The logging hook is a pure closure and MetricsHook holds no internal state beyond a collector reference. The ticket's mention of "active maps drain correctly" applies only to TracingHook.

💡 User Tips

Regenerate the plan with different choices with @coderabbitai <feedback>.

Implementation Steps

Phase 1: Extract Pure Reducers from Adapters

Refactor each adapter to separate the pure event-to-operation logic from side-effecting collector/span/logger calls. This enables direct testing of the mapping logic without mocking.

Task 1: Extract MetricsHook Reducer

Introduce a pure function that maps an ExecEvent to a sequence of intended metric operations, separating the "what" from the "do".

• Create a MetricOp dataclass (or NamedTuple) in cuprum/adapters/metrics_adapter.py representing a single metric operation: counter increment or histogram observation, with name, value, and labels.
• Extract a module-level pure function compute_metric_ops(event: ExecEvent) -> tuple[MetricOp, ...] that encodes the existing match/case dispatch logic.
• Refactor MetricsHook.__call__ to call compute_metric_ops and iterate over the results, invoking _collector.inc_counter or _collector.observe_histogram accordingly.
• Preserve the existing behaviour for all phases including the silent no-op on "plan".

Task 2: Extract TracingHook Reducer

Introduce a pure function that maps an ExecEvent plus current active-PID state to a sequence of intended span operations.

• Create a SpanOp union type (or tagged dataclass variants) in cuprum/adapters/tracing_adapter.py representing: start-span, set-attribute, add-event, set-status, end-span.
• Extract a pure function compute_span_ops(event: ExecEvent, active_pids: frozenset[int]) -> tuple[SpanOp, ...] that returns the operations required without performing them.
• Refactor TracingHook.__call__ to call compute_span_ops and apply the operations to _active_spans and the tracer.
• Ensure the _active_spans mutation semantics (insert on start, pop on exit) remain correct.

Task 3: Extract Logging Hook Reducer

Introduce a pure function that maps an ExecEvent to a log-record specification.

• Create a LogSpec dataclass in cuprum/adapters/logging_adapter.py representing the intended log record: level, message, and extra dict.
• Extract the existing _build_extra and _format_message into a single pure function compute_log_spec(event: ExecEvent, levels: LogLevels) -> LogSpec | None (returns None if the level is filtered).
• Refactor the hook closure inside structured_logging_hook to call compute_log_spec and emit via log.log only when non-None.

🤖 Prompt for AI agents

Refactor the three adapter modules in `cuprum/adapters/` to extract pure reducer
functions from each hook, separating event-to-operation logic from
side-effecting calls. This enables direct Hypothesis testing without mocking.

**`cuprum/adapters/metrics_adapter.py`**
- Define a `MetricOp` dataclass or `NamedTuple` with fields for operation type
(counter increment or histogram observation), name, value, and labels.
- Extract a module-level pure function `compute_metric_ops(event: ExecEvent) ->
tuple[MetricOp, ...]` containing the existing `match/case` dispatch logic.
- Refactor `MetricsHook.__call__` to call `compute_metric_ops`, then iterate
over the returned ops, invoking `_collector.inc_counter` or
`_collector.observe_histogram` as appropriate.
- Preserve the silent no-op for the `"plan"` phase.

**`cuprum/adapters/tracing_adapter.py`**
- Define a `SpanOp` union type or tagged dataclass variants representing:
start-span, set-attribute, add-event, set-status, and end-span operations.
- Extract a pure function `compute_span_ops(event: ExecEvent, active_pids:
frozenset[int]) -> tuple[SpanOp, ...]` that returns required operations without
performing them.
- Refactor `TracingHook.__call__` to call `compute_span_ops` and apply the
returned ops to `_active_spans` and the tracer.
- Maintain correct `_active_spans` mutation semantics: insert on start, pop on
exit.

**`cuprum/adapters/logging_adapter.py`**
- Define a `LogSpec` dataclass with fields for level, message, and extra dict.
- Merge the logic of `_build_extra` and `_format_message` into a single pure
function `compute_log_spec(event: ExecEvent, levels: LogLevels) -> LogSpec |
None`, returning `None` when the level is filtered.
- Refactor the `hook` closure inside `structured_logging_hook` to call
`compute_log_spec` and emit via `log.log` only when the result is non-`None`.

Phase 2: Hypothesis Strategies and Property Tests

Create reusable ExecEvent generation strategies and property-based tests exercising the reducers and full hooks.

Task 1: Create ExecEvent Hypothesis Strategy

Build a composite strategy producing valid ExecEvents with controllable phase distributions and field constraints.

• Create a new module cuprum/unittests/strategies/exec_event.py containing the strategy.
• Define strategies for each optional field (pid, line, exit_code, duration_s, byte_count, note) that respect phase constraints (e.g., pid is None only for "plan").
• Provide a @st.composite strategy exec_events() that generates single events, and exec_event_sequences(min_size, max_size) that generates coherent sequences where every started PID eventually exits.
• Ensure the tags field includes both present and absent "project" key to exercise the fallback.

Task 2: MetricsHook Property Tests

Add Hypothesis tests verifying the metrics reducer and hook invariants.

• Create test functions in cuprum/unittests/test_adapters.py (or a new test_adapters_hypothesis.py if preferred for clarity).
• Test compute_metric_ops invariants: "plan" → empty tuple; "stdin" with byte_count=None → empty tuple; "exit" with exit_code=0 → no failure counter; accumulation of byte counts is additive.
• Test full MetricsHook over random event sequences: final counter values equal the expected sum, histogram lengths match exit-event counts.
• Use @settings(max_examples=100, deadline=None, derandomize=True).

Task 3: TracingHook Stateful Test

Add a Hypothesis RuleBasedStateMachine test proving _active_spans drains correctly.

• Create a TracingHookStateMachine class in cuprum/unittests/test_adapters.py (or test_adapters_hypothesis.py) extending hypothesis.stateful.RuleBasedStateMachine.
• Maintain a model active_pids: set[int] alongside a real TracingHook + InMemoryTracer.
• Define rules: start_execution adds a PID to model and emits a "start" event; emit_output emits "stdout" or "stderr" for an active PID; finish_execution emits "exit" for an active PID and removes it from model.
• Add an invariant: the length of TracingHook._active_spans equals len(active_pids).
• Add a teardown check: after all rules, _active_spans is empty iff active_pids is empty.

Task 4: Logging Hook Property Tests

Add Hypothesis tests verifying the logging reducer produces correct specs.

• Test compute_log_spec invariants: each phase maps to the correct log level; extra dict contains exactly the expected cuprum_* keys based on event field presence.
• Test full hook over random event sequences with a capturing logging.Handler: record count equals event count (minus any level-filtered events), and each record's cuprum_phase matches the input event.
• Verify the "project" fallback to "unknown" when tags lacks the key.

🤖 Prompt for AI agents

Implement reusable `ExecEvent` Hypothesis strategies and property-based tests
for all three adapter hooks. This phase produces the new test infrastructure and
the tests themselves.

**`cuprum/unittests/strategies/exec_event.py`** (new file)
- Define per-field strategies for optional fields (`pid`, `line`, `exit_code`,
`duration_s`, `byte_count`, `note`) that respect phase constraints (e.g., `pid`
is `None` only for `"plan"`).
- Implement a `@st.composite` strategy `exec_events()` generating single valid
`ExecEvent` instances.
- Implement `exec_event_sequences(min_size, max_size)` generating coherent
sequences where every started PID eventually exits.
- Ensure `tags` exercises both presence and absence of the `"project"` key.

**MetricsHook property tests** (in `cuprum/unittests/test_adapters.py` or
`test_adapters_hypothesis.py`)
- Test `compute_metric_ops` invariants:
  - `"plan"` phase → empty tuple
  - `"stdin"` with `byte_count=None` → empty tuple
  - `"exit"` with `exit_code=0` → no failure counter op
  - Accumulation of byte counts is additive
- Test full `MetricsHook` over random event sequences: assert final counter
values equal expected sums and histogram lengths match exit-event counts.
- Decorate all tests with `@settings(max_examples=100, deadline=None,
derandomize=True)`.

**TracingHook stateful test** (same file as above)
- Define `TracingHookStateMachine` extending
`hypothesis.stateful.RuleBasedStateMachine`.
- Maintain a model `active_pids: set[int]` alongside a real `TracingHook` +
`InMemoryTracer`.
- Define three rules: `start_execution` (adds PID to model, emits `"start"`
event), `emit_output` (emits `"stdout"` or `"stderr"` for an active PID),
`finish_execution` (emits `"exit"` for an active PID, removes from model).
- Add an invariant asserting `len(TracingHook._active_spans) ==
len(active_pids)`.
- Add a teardown check asserting `_active_spans` is empty iff `active_pids` is
empty.

**Logging hook property tests** (same file as above)
- Test `compute_log_spec` invariants: each phase maps to the correct log level;
`extra` dict contains exactly the expected `cuprum_*` keys for the event's
fields.
- Test the full hook over random event sequences using a capturing
`logging.Handler`: assert record count equals event count minus level-filtered
events, and each record's `cuprum_phase` matches the input event.
- Assert the `"project"` fallback to `"unknown"` when `tags` lacks the key.

Phase 3: Documentation and Cleanup

Ensure the new reducers and strategies are discoverable and the test suite remains maintainable.

Task 1: Update Adapter Docstrings

Document the new pure reducer functions and their role in testability.

• Add docstrings to compute_metric_ops, compute_span_ops, and compute_log_spec explaining their purpose and relationship to the hook classes.
• Update module docstrings in each adapter file to mention the reducer/applier separation.

Task 2: Export Strategies for Downstream Use

Make the ExecEvent strategy available for other test modules.

• Add cuprum/unittests/strategies/__init__.py exporting exec_events and exec_event_sequences.
• Add a brief README or docstring explaining usage for future property-based tests elsewhere in the suite.

🤖 Prompt for AI agents

Add documentation to the new reducer functions and expose the strategy module
for downstream use. This phase ensures everything introduced in Phases 1 and 2
is discoverable and maintainable.

**Adapter docstrings** (`cuprum/adapters/metrics_adapter.py`,
`tracing_adapter.py`, `logging_adapter.py`)
- Write docstrings for `compute_metric_ops`, `compute_span_ops`, and
`compute_log_spec` that explain their purpose and their relationship to the
corresponding hook classes.
- Update each adapter file's module-level docstring to mention the
reducer/applier separation pattern.

**Strategy package** (`cuprum/unittests/strategies/__init__.py`)
- Create this file and export `exec_events` and `exec_event_sequences` from
`exec_event.py`.
- Include a docstring or brief inline README explaining how to use the
strategies in future property-based tests elsewhere in the suite.

Research

The cuprum/adapters/ directory contains three telemetry adapters: MetricsHook (stateless, delegates to a MetricsCollector protocol), TracingHook (stateful, maintains _active_spans: dict[int, Span] keyed by PID), and structured_logging_hook (stateless closure). Existing tests in cuprum/unittests/test_adapters.py cover happy-path scenarios but do not exercise varied phase/order/PID combinations. The codebase already uses Hypothesis in test_stream_property_based.py and test_env_context_properties.py with patterns including @st.composite strategies, @given, @settings(max_examples=..., deadline=None, derandomize=True).

---

🚀 Next Steps

🤖 All AI agent prompts combined

Task: 1

Refactor the three adapter modules in `cuprum/adapters/` to extract pure reducer
functions from each hook, separating event-to-operation logic from
side-effecting calls. This enables direct Hypothesis testing without mocking.

**`cuprum/adapters/metrics_adapter.py`**
- Define a `MetricOp` dataclass or `NamedTuple` with fields for operation type
(counter increment or histogram observation), name, value, and labels.
- Extract a module-level pure function `compute_metric_ops(event: ExecEvent) ->
tuple[MetricOp, ...]` containing the existing `match/case` dispatch logic.
- Refactor `MetricsHook.__call__` to call `compute_metric_ops`, then iterate
over the returned ops, invoking `_collector.inc_counter` or
`_collector.observe_histogram` as appropriate.
- Preserve the silent no-op for the `"plan"` phase.

**`cuprum/adapters/tracing_adapter.py`**
- Define a `SpanOp` union type or tagged dataclass variants representing:
start-span, set-attribute, add-event, set-status, and end-span operations.
- Extract a pure function `compute_span_ops(event: ExecEvent, active_pids:
frozenset[int]) -> tuple[SpanOp, ...]` that returns required operations without
performing them.
- Refactor `TracingHook.__call__` to call `compute_span_ops` and apply the
returned ops to `_active_spans` and the tracer.
- Maintain correct `_active_spans` mutation semantics: insert on start, pop on
exit.

**`cuprum/adapters/logging_adapter.py`**
- Define a `LogSpec` dataclass with fields for level, message, and extra dict.
- Merge the logic of `_build_extra` and `_format_message` into a single pure
function `compute_log_spec(event: ExecEvent, levels: LogLevels) -> LogSpec |
None`, returning `None` when the level is filtered.
- Refactor the `hook` closure inside `structured_logging_hook` to call
`compute_log_spec` and emit via `log.log` only when the result is non-`None`.
===============================================================================

Task: 2

Implement reusable `ExecEvent` Hypothesis strategies and property-based tests
for all three adapter hooks. This phase produces the new test infrastructure and
the tests themselves.

**`cuprum/unittests/strategies/exec_event.py`** (new file)
- Define per-field strategies for optional fields (`pid`, `line`, `exit_code`,
`duration_s`, `byte_count`, `note`) that respect phase constraints (e.g., `pid`
is `None` only for `"plan"`).
- Implement a `@st.composite` strategy `exec_events()` generating single valid
`ExecEvent` instances.
- Implement `exec_event_sequences(min_size, max_size)` generating coherent
sequences where every started PID eventually exits.
- Ensure `tags` exercises both presence and absence of the `"project"` key.

**MetricsHook property tests** (in `cuprum/unittests/test_adapters.py` or
`test_adapters_hypothesis.py`)
- Test `compute_metric_ops` invariants:
  - `"plan"` phase → empty tuple
  - `"stdin"` with `byte_count=None` → empty tuple
  - `"exit"` with `exit_code=0` → no failure counter op
  - Accumulation of byte counts is additive
- Test full `MetricsHook` over random event sequences: assert final counter
values equal expected sums and histogram lengths match exit-event counts.
- Decorate all tests with `@settings(max_examples=100, deadline=None,
derandomize=True)`.

**TracingHook stateful test** (same file as above)
- Define `TracingHookStateMachine` extending
`hypothesis.stateful.RuleBasedStateMachine`.
- Maintain a model `active_pids: set[int]` alongside a real `TracingHook` +
`InMemoryTracer`.
- Define three rules: `start_execution` (adds PID to model, emits `"start"`
event), `emit_output` (emits `"stdout"` or `"stderr"` for an active PID),
`finish_execution` (emits `"exit"` for an active PID, removes from model).
- Add an invariant asserting `len(TracingHook._active_spans) ==
len(active_pids)`.
- Add a teardown check asserting `_active_spans` is empty iff `active_pids` is
empty.

**Logging hook property tests** (same file as above)
- Test `compute_log_spec` invariants: each phase maps to the correct log level;
`extra` dict contains exactly the expected `cuprum_*` keys for the event's
fields.
- Test the full hook over random event sequences using a capturing
`logging.Handler`: assert record count equals event count minus level-filtered
events, and each record's `cuprum_phase` matches the input event.
- Assert the `"project"` fallback to `"unknown"` when `tags` lacks the key.
===============================================================================

Task: 3

Add documentation to the new reducer functions and expose the strategy module
for downstream use. This phase ensures everything introduced in Phases 1 and 2
is discoverable and maintainable.

**Adapter docstrings** (`cuprum/adapters/metrics_adapter.py`,
`tracing_adapter.py`, `logging_adapter.py`)
- Write docstrings for `compute_metric_ops`, `compute_span_ops`, and
`compute_log_spec` that explain their purpose and their relationship to the
corresponding hook classes.
- Update each adapter file's module-level docstring to mention the
reducer/applier separation pattern.

**Strategy package** (`cuprum/unittests/strategies/__init__.py`)
- Create this file and export `exec_events` and `exec_event_sequences` from
`exec_event.py`.
- Include a docstring or brief inline README explaining how to use the
strategies in future property-based tests elsewhere in the suite.

💡 Iterate on the plan with: `@coderabbitai` <feedback>

Example Feedback
- `@coderabbitai` You can skip phase 3. Add a simple unit test case for phase 2.
- `@coderabbitai` For design choice 1 go ahead with option 3 and replan.

---

💬 Have feedback or questions? Drop into our discord!