Write observability in fuzz_one_input

hypothesis-python/RELEASE.rst

@@ -0,0 +1,3 @@
+RELEASE_TYPE: patch
+
+:ref:`fuzz_one_input <fuzz_one_input>` now writes :ref:`observability reports <observability>` if observability is enabled, bringing it in line with the behavior of other standard ways to invoke a Hypothesis test.

hypothesis-python/src/hypothesis/core.py

@@ -1911,7 +1911,12 @@ def fuzz_one_input(
                 )
                 try:
                     state.execute_once(data)
-                except (StopTest, UnsatisfiedAssumption):
+                    status = Status.VALID
+                except StopTest:
+                    status = data.status
+                    return None
+                except UnsatisfiedAssumption:
+                    status = Status.INVALID
                     return None
                 except BaseException:
                     known = minimal_failures.get(data.interesting_origin)
@@ -1922,7 +1927,25 @@ def fuzz_one_input(
                             database_key, choices_to_bytes(data.choices)
                         )
                         minimal_failures[data.interesting_origin] = data.nodes
+                    status = Status.INTERESTING
                     raise
+                finally:
+                    if TESTCASE_CALLBACKS:
+                        tc = make_testcase(
+                            run_start=state._start_timestamp,
+                            property=state.test_identifier,
+                            data=data,
+                            how_generated="fuzz_one_input",
+                            representation=state._string_repr,
+                            arguments=data._observability_args,
+                            timing=state._timing_features,
+                            coverage=None,
+                            status=status,
+                            backend_metadata=data.provider.observe_test_case(),
+                        )
+                        deliver_observation(tc)
+                        state._timing_features = {}
+
                 assert isinstance(data.provider, BytestringProvider)
                 return bytes(data.provider.drawn)
 

hypothesis-python/src/hypothesis/internal/observability.py

@@ -26,7 +26,7 @@
 if TYPE_CHECKING:
     from typing import TypeAlias
 
-    from hypothesis.internal.conjecture.data import ConjectureData
+    from hypothesis.internal.conjecture.data import ConjectureData, Status
 
 
 @dataclass
@@ -115,7 +115,9 @@ def make_testcase(
     coverage: Optional[dict[str, list[int]]] = None,
     phase: Optional[str] = None,
     backend_metadata: Optional[dict[str, Any]] = None,
-    status: Optional[TestCaseStatus] = None,  # overrides automatic calculation
+    status: Optional[
+        Union[TestCaseStatus, "Status"]
+    ] = None,  # overrides automatic calculation
     status_reason: Optional[str] = None,  # overrides automatic calculation
     # added to calculated metadata. If keys overlap, the value from this `metadata`
     # is used
@@ -140,6 +142,9 @@ def make_testcase(
         Status.INTERESTING: "failed",
     }
 
+    if status is not None and isinstance(status, Status):
+        status = status_map[status]
+
     return TestCaseObservation(
         type="test_case",
         status=status if status is not None else status_map[data.status],

hypothesis-python/tests/conjecture/test_engine.py

@@ -45,6 +45,7 @@
 from hypothesis.internal.conjecture.junkdrawer import startswith
 from hypothesis.internal.conjecture.pareto import DominanceRelation, dominance
 from hypothesis.internal.conjecture.shrinker import Shrinker
+from hypothesis.internal.coverage import IN_COVERAGE_TESTS
 from hypothesis.internal.entropy import deterministic_PRNG
 
 from tests.common.debug import minimal
@@ -160,6 +161,9 @@ def recur(i, data):
 
 
 @pytest.mark.skipif(PYPY, reason="stack tricks only work reliably on CPython")
+@pytest.mark.skipif(
+    IN_COVERAGE_TESTS, reason="flaky under coverage instrumentation? see #4391"
+)
 def test_recursion_error_is_not_flaky():
     def tf(data):
         i = data.draw_integer(0, 2**16 - 1)

hypothesis-python/tests/cover/test_fuzz_one_input.py

@@ -108,22 +108,19 @@ def test(s):
     assert test.hypothesis.fuzz_one_input(b"deadbeef") == b"dead"
 
 
-STRAT = st.builds(object)
-
-
-@given(x=STRAT)
-def addx(x, y):
-    pass
-
+def test_can_access_strategy_for_wrapped_test():
+    strategy = st.builds(object)
 
-@given(STRAT)
-def addy(x, y):
-    pass
+    @given(x=strategy)
+    def addx(x, y):
+        pass
 
+    @given(strategy)
+    def addy(x, y):
+        pass
 
-def test_can_access_strategy_for_wrapped_test():
-    assert addx.hypothesis._given_kwargs == {"x": STRAT}
-    assert addy.hypothesis._given_kwargs == {"y": STRAT}
+    assert addx.hypothesis._given_kwargs == {"x": strategy}
+    assert addy.hypothesis._given_kwargs == {"y": strategy}
 
 
 @pytest.mark.parametrize(

hypothesis-python/tests/cover/test_observability.py

@@ -9,6 +9,7 @@
 # obtain one at https://mozilla.org/MPL/2.0/.
 
 import textwrap
+from contextlib import nullcontext
 
 import pytest
 
@@ -298,3 +299,36 @@ def test_observability_captures_stateful_reprs():
         has_inc = "generate:rule:inc" in t and "execute:rule:inc" in t
         has_dec = "generate:rule:dec" in t and "execute:rule:dec" in t
         assert has_inc or has_dec
+
+
+# BytestringProvider.draw_boolean divides [0, 127] as False and [128, 255]
+# as True
+@pytest.mark.parametrize(
+    "buffer, expected_status",
+    [
+        # Status.OVERRUN
+        (b"", "gave_up"),
+        # Status.INVALID
+        (b"\x00" + bytes([255]), "gave_up"),
+        # Status.VALID
+        (b"\x00\x00", "passed"),
+        # Status.INTERESTING
+        (bytes([255]) + b"\x00", "failed"),
+    ],
+)
+def test_fuzz_one_input_status(buffer, expected_status):
+    @given(st.booleans(), st.booleans())
+    def test_fails(should_fail, should_fail_assume):
+        if should_fail:
+            raise AssertionError
+        if should_fail_assume:
+            assume(False)
+
+    with (
+        capture_observations() as ls,
+        pytest.raises(AssertionError) if expected_status == "failed" else nullcontext(),
+    ):
+        test_fails.hypothesis.fuzz_one_input(buffer)
+    assert len(ls) == 1
+    assert ls[0].status == expected_status
+    assert ls[0].how_generated == "fuzz_one_input"