use dataclasses for observations in-memory

Author: tybug

hypothesis-python/RELEASE.rst

@@ -0,0 +1,3 @@
+RELEASE_TYPE: patch
+
+The observations passed to |TESTCASE_CALLBACKS| are now dataclasses, rather than dictionaries. The content written to ``.hypothesis/observed`` under ``HYPOTHESIS_EXPERIMENTAL_OBSERVABILITY`` remains the same.

hypothesis-python/docs/conf.py

@@ -286,6 +286,7 @@ def setup(app):
 .. |PrimitiveProvider.draw_bytes| replace:: :func:`~hypothesis.internal.conjecture.providers.PrimitiveProvider.draw_bytes`
 
 .. |AVAILABLE_PROVIDERS| replace:: :data:`~hypothesis.internal.conjecture.providers.AVAILABLE_PROVIDERS`
+.. |TESTCASE_CALLBACKS| replace:: :data:`~hypothesis.internal.observability.TESTCASE_CALLBACKS`
 .. |BUFFER_SIZE| replace:: :data:`~hypothesis.internal.conjecture.engine.BUFFER_SIZE`
 .. |MAX_SHRINKS| replace:: :data:`~hypothesis.internal.conjecture.engine.MAX_SHRINKS`
 .. |MAX_SHRINKING_SECONDS| replace:: :data:`~hypothesis.internal.conjecture.engine.MAX_SHRINKING_SECONDS`

hypothesis-python/src/hypothesis/control.py

@@ -44,8 +44,8 @@ def reject() -> NoReturn:
         )
     where = _calling_function_location("reject", inspect.currentframe())
     if currently_in_test_context():
-        count = current_build_context().data._observability_predicates[where]
-        count["unsatisfied"] += 1
+        counts = current_build_context().data._observability_predicates[where]
+        counts.update_count(False)
     raise UnsatisfiedAssumption(where)
 
 
@@ -65,8 +65,8 @@ def assume(condition: object) -> bool:
     if TESTCASE_CALLBACKS or not condition:
         where = _calling_function_location("assume", inspect.currentframe())
         if TESTCASE_CALLBACKS and currently_in_test_context():
-            predicates = current_build_context().data._observability_predicates
-            predicates[where]["satisfied" if condition else "unsatisfied"] += 1
+            counts = current_build_context().data._observability_predicates[where]
+            counts.update_count(bool(condition))
         if not condition:
             raise UnsatisfiedAssumption(f"failed to satisfy {where}")
     return True

hypothesis-python/src/hypothesis/core.py

@@ -101,8 +101,9 @@
 from hypothesis.internal.observability import (
     OBSERVABILITY_COLLECT_COVERAGE,
     TESTCASE_CALLBACKS,
-    _system_metadata,
-    deliver_json_blob,
+    InfoObservation,
+    InfoObservationType,
+    deliver_observation,
     make_testcase,
 )
 from hypothesis.internal.reflection import (
@@ -613,14 +614,14 @@ def execute_explicit_examples(state, wrapped_test, arguments, kwargs, original_s
                     )
 
                 tc = make_testcase(
-                    start_timestamp=state._start_timestamp,
-                    test_name_or_nodeid=state.test_identifier,
+                    run_start=state._start_timestamp,
+                    property=state.test_identifier,
                     data=empty_data,
                     how_generated="explicit example",
-                    string_repr=state._string_repr,
+                    representation=state._string_repr,
                     timing=state._timing_features,
                 )
-                deliver_json_blob(tc)
+                deliver_observation(tc)
 
             if fragments_reported:
                 verbose_report(fragments_reported[0].replace("Falsifying", "Trying", 1))
@@ -1225,35 +1226,35 @@ def _execute_once_for_engine(self, data: ConjectureData) -> None:
                     self._string_repr = "<backend failed to realize symbolic arguments>"
 
                 tc = make_testcase(
-                    start_timestamp=self._start_timestamp,
-                    test_name_or_nodeid=self.test_identifier,
+                    run_start=self._start_timestamp,
+                    property=self.test_identifier,
                     data=data,
                     how_generated=f"during {phase} phase{backend_desc}",
-                    string_repr=self._string_repr,
+                    representation=self._string_repr,
                     arguments=data._observability_args,
                     timing=self._timing_features,
                     coverage=tractable_coverage_report(trace) or None,
                     phase=phase,
                     backend_metadata=data.provider.observe_test_case(),
                 )
-                deliver_json_blob(tc)
+                deliver_observation(tc)
                 for msg in data.provider.observe_information_messages(
                     lifetime="test_case"
                 ):
                     self._deliver_information_message(**msg)
             self._timing_features = {}
 
     def _deliver_information_message(
-        self, *, type: str, title: str, content: Union[str, dict]
+        self, *, type: InfoObservationType, title: str, content: Union[str, dict]
     ) -> None:
-        deliver_json_blob(
-            {
-                "type": type,
-                "run_start": self._start_timestamp,
-                "property": self.test_identifier,
-                "title": title,
-                "content": content,
-            }
+        deliver_observation(
+            InfoObservation(
+                type=type,
+                run_start=self._start_timestamp,
+                property=self.test_identifier,
+                title=title,
+                content=content,
+            )
         )
 
     def run_engine(self):
@@ -1421,31 +1422,20 @@ def run_engine(self):
                 raise NotImplementedError("This should be unreachable")
             finally:
                 # log our observability line for the final failing example
-                tc = {
-                    "type": "test_case",
-                    "run_start": self._start_timestamp,
-                    "property": self.test_identifier,
-                    "status": "passed" if sys.exc_info()[0] else "failed",
-                    "status_reason": str(origin or "unexpected/flaky pass"),
-                    "representation": self._string_repr,
-                    "arguments": ran_example._observability_args,
-                    "how_generated": "minimal failing example",
-                    "features": {
-                        **{
-                            f"target:{k}".strip(":"): v
-                            for k, v in ran_example.target_observations.items()
-                        },
-                        **ran_example.events,
-                    },
-                    "timing": self._timing_features,
-                    "coverage": None,  # Not recorded when we're replaying the MFE
-                    "metadata": {
-                        "traceback": tb,
-                        "predicates": dict(ran_example._observability_predicates),
-                        **_system_metadata(),
-                    },
-                }
-                deliver_json_blob(tc)
+                tc = make_testcase(
+                    run_start=self._start_timestamp,
+                    property=self.test_identifier,
+                    data=ran_example,
+                    how_generated="minimal failing example",
+                    representation=self._string_repr,
+                    arguments=ran_example._observability_args,
+                    timing=self._timing_features,
+                    coverage=None,  # Not recorded when we're replaying the MFE
+                    status="passed" if sys.exc_info()[0] else "failed",
+                    status_reason=str(origin or "unexpected/flaky pass"),
+                    metadata={"traceback": tb},
+                )
+                deliver_observation(tc)
                 # Whether or not replay actually raised the exception again, we want
                 # to print the reproduce_failure decorator for the failing example.
                 if self.settings.print_blob:

hypothesis-python/src/hypothesis/internal/conjecture/data.py

@@ -713,7 +713,7 @@ def __init__(
         self.slice_comments: dict[tuple[int, int], str] = {}
         self._observability_args: dict[str, Any] = {}
         self._observability_predicates: defaultdict[str, PredicateCounts] = defaultdict(
-            lambda: {"satisfied": 0, "unsatisfied": 0}
+            PredicateCounts
         )
         self._sampled_from_all_strategies_elements_message: Optional[
             tuple[str, object]

hypothesis-python/src/hypothesis/internal/conjecture/providers.py

@@ -62,6 +62,7 @@
     next_up,
 )
 from hypothesis.internal.intervalsets import IntervalSet
+from hypothesis.internal.observability import InfoObservationType
 
 if TYPE_CHECKING:
     from typing import TypeAlias
@@ -294,7 +295,7 @@ def _get_local_constants() -> Constants:
 
 
 class _BackendInfoMsg(TypedDict):
-    type: str
+    type: InfoObservationType
     title: str
     content: Union[str, dict[str, Any]]