test(debugger): align DI hit-limit test with SDK MaxHits semantics (#782)

## Summary

The DI Flask `DIFlaskHitLimitTest` and the matching application target
function had `MaxHits` semantics flipped relative to the SDK. With
`MaxHits=3`, the test expected 2 snapshots, claiming the disable check
was `hit_count >= max_hits`. The SDK's
[increment_hit_count](https://github.com/aws-observability/aws-otel-python-instrumentation/blob/main/aws-opentelemetry-distro/src/amazon/opentelemetry/distro/debugger/instrumentation_manager.py#L631)
actually uses `hit_count > max_hits`, so it emits 3 snapshots and
disables on the 4th hit.

## Why this is surfacing now

The test was passing on main by accident. `BatchLogRecordProcessor`'s
default `schedule_delay_millis` was 5000ms, so the snapshot emitted on
the 3rd hit was still sitting in the batch buffer when the test's
`time.sleep(2)` elapsed and read the collector. The 3rd snapshot never
reached the mock collector within the test's observation window.

OpenTelemetry core 1.41.0 [tightened that default to
1000ms](open-telemetry/opentelemetry-python#4998)
to comply with the OTel spec, so the upcoming dependency bump to 1.42.1
/ 0.63b1 (#762) now sees the 3rd snapshot consistently and the latent
off-by-one assertion fails the test.

## Changes

- Update the test class docstring and both test methods so `MaxHits=3` →
3 snapshots, breakpoint disabled on the 4th hit. Send a 4th request to
actually probe the disabled state.
- Update the `limited_function` docstring in the di-flask sample app to
match.

## Test plan

- [ ] DI Debugger Contract Tests pass on this branch (currently the only
test in the suite that was failing, on the `nightly-dependency-updates`
branch)
- [ ] No SDK behavior change — purely a test/comment alignment with
current SDK semantics

Author: liustve

contract-tests/images/applications/di-fastapi/di_fastapi_server.py

@@ -62,8 +62,8 @@ def calculate_sum(a, b):
 def limited_function(x):
     """Target function with hit limit (MaxHits=3).
 
-    MaxHits=3 means: allow 2 snapshots, disable at 3rd hit.
-    (The check is hit_count >= max_hits, so hit 1&2 pass, hit 3 is blocked.)
+    MaxHits=3 means: allow 3 snapshots, disable on the 4th hit.
+    (The check is hit_count > max_hits, so hits 1-3 pass, hit 4 is blocked.)
     """
     return x * 10
 

contract-tests/images/applications/di-flask/di_flask_server.py

@@ -55,8 +55,8 @@ def calculate_sum(a, b):
 def limited_function(x):
     """Target function with hit limit (MaxHits=3).
 
-    MaxHits=3 means: allow 2 snapshots, disable at 3rd hit.
-    (The check is hit_count >= max_hits, so hit 1&2 pass, hit 3 is blocked.)
+    MaxHits=3 means: allow 3 snapshots, disable on the 4th hit.
+    (The check is hit_count > max_hits, so hits 1-3 pass, hit 4 is blocked.)
     """
     return x * 10
 

contract-tests/tests/test/amazon/di/django_test.py

@@ -273,33 +273,34 @@ def get_application_wait_pattern(self) -> str:
         return "Ready"
 
     def test_breakpoint_generates_snapshots_up_to_limit(self) -> None:
-        """BREAKPOINT generates snapshots up to (max_hits - 1)."""
+        """BREAKPOINT generates a snapshot for each hit up to and including max_hits."""
+        self.send_request("GET", "limited")
         self.send_request("GET", "limited")
         self.send_request("GET", "limited")
 
-        logs = self.wait_for_snapshots(min_count=2)
+        logs = self.wait_for_snapshots(min_count=3)
         limited_logs = self.logs_for_method(logs, "limited_function")
-        self.assertEqual(len(limited_logs), 2, "Expected exactly 2 snapshots (MaxHits=3 allows 2)")
+        self.assertEqual(len(limited_logs), 3, "Expected exactly 3 snapshots (MaxHits=3 allows 3)")
 
     def test_breakpoint_disabled_after_hit_limit(self) -> None:
         """BREAKPOINT stops generating snapshots after hit limit is reached."""
         self.send_request("GET", "limited")
         self.send_request("GET", "limited")
+        self.send_request("GET", "limited")
 
-        logs = self.wait_for_snapshots(min_count=2)
+        logs = self.wait_for_snapshots(min_count=3)
         initial_count = len(self.logs_for_method(logs, "limited_function"))
-        self.assertEqual(initial_count, 2)
+        self.assertEqual(initial_count, 3)
 
         self.send_request("GET", "limited")
-        self.send_request("GET", "limited")
-        time.sleep(2)
+        time.sleep(5)
 
         final_logs = self._peek_snapshots()
         final_count = len(self.logs_for_method(final_logs, "limited_function"))
         self.assertEqual(
             final_count,
-            2,
-            f"Expected 2 snapshots (MaxHits=3), but got {final_count}. "
+            3,
+            f"Expected 3 snapshots (MaxHits=3), but got {final_count}. "
             "BREAKPOINT should be disabled after hitting limit.",
         )
 

contract-tests/tests/test/amazon/di/fastapi_test.py

@@ -278,7 +278,8 @@ class DIFastAPIHitLimitTest(DITestInfrastructure):
     """Test BREAKPOINT hit limit behavior.
 
     BREAKPOINTs have a max_hits limit. With MaxHits=3, the check is
-    hit_count >= max_hits, so hits 1 and 2 generate snapshots, hit 3 is blocked.
+    hit_count > max_hits, so hits 1, 2, and 3 generate snapshots and the
+    breakpoint is disabled on hit 4.
     """
 
     __test__ = True
@@ -293,36 +294,37 @@ def get_application_wait_pattern(self) -> str:
         return "Ready"
 
     def test_breakpoint_generates_snapshots_up_to_limit(self) -> None:
-        """BREAKPOINT generates snapshots up to (max_hits - 1)."""
-        # limited_function has MaxHits=3, so 2 snapshots should be generated
+        """BREAKPOINT generates a snapshot for each hit up to and including max_hits."""
+        # limited_function has MaxHits=3, so 3 snapshots should be generated
+        self.send_request("GET", "limited")
         self.send_request("GET", "limited")
         self.send_request("GET", "limited")
 
-        logs = self.wait_for_snapshots(min_count=2)
+        logs = self.wait_for_snapshots(min_count=3)
         limited_logs = self.logs_for_method(logs, "limited_function")
-        self.assertEqual(len(limited_logs), 2, "Expected exactly 2 snapshots (MaxHits=3 allows 2)")
+        self.assertEqual(len(limited_logs), 3, "Expected exactly 3 snapshots (MaxHits=3 allows 3)")
 
     def test_breakpoint_disabled_after_hit_limit(self) -> None:
         """BREAKPOINT stops generating snapshots after hit limit is reached."""
-        # First 2 calls generate snapshots (hits 1 and 2)
+        # First 3 calls generate snapshots (hits 1, 2, and 3)
+        self.send_request("GET", "limited")
         self.send_request("GET", "limited")
         self.send_request("GET", "limited")
 
-        logs = self.wait_for_snapshots(min_count=2)
+        logs = self.wait_for_snapshots(min_count=3)
         initial_count = len(self.logs_for_method(logs, "limited_function"))
-        self.assertEqual(initial_count, 2)
+        self.assertEqual(initial_count, 3)
 
-        # 3rd and 4th calls hit the limit -- no new snapshots
-        self.send_request("GET", "limited")
+        # 4th call hits the limit -- no new snapshot
         self.send_request("GET", "limited")
-        time.sleep(2)
+        time.sleep(5)
 
         final_logs = self._peek_snapshots()
         final_count = len(self.logs_for_method(final_logs, "limited_function"))
         self.assertEqual(
             final_count,
-            2,
-            f"Expected 2 snapshots (MaxHits=3), but got {final_count}. "
+            3,
+            f"Expected 3 snapshots (MaxHits=3), but got {final_count}. "
             "BREAKPOINT should be disabled after hitting limit.",
         )
 

contract-tests/tests/test/amazon/di/flask_test.py

@@ -270,7 +270,8 @@ class DIFlaskHitLimitTest(DITestInfrastructure):
     """Test BREAKPOINT hit limit behavior.
 
     BREAKPOINTs have a max_hits limit. With MaxHits=3, the check is
-    hit_count >= max_hits, so hits 1 and 2 generate snapshots, hit 3 is blocked.
+    hit_count > max_hits, so hits 1, 2, and 3 generate snapshots and the
+    breakpoint is disabled on hit 4.
     """
 
     __test__ = True
@@ -285,36 +286,37 @@ def get_application_wait_pattern(self) -> str:
         return "Ready"
 
     def test_breakpoint_generates_snapshots_up_to_limit(self) -> None:
-        """BREAKPOINT generates snapshots up to (max_hits - 1)."""
-        # limited_function has MaxHits=3, so 2 snapshots should be generated
+        """BREAKPOINT generates a snapshot for each hit up to and including max_hits."""
+        # limited_function has MaxHits=3, so 3 snapshots should be generated
+        self.send_request("GET", "limited")
         self.send_request("GET", "limited")
         self.send_request("GET", "limited")
 
-        logs = self.wait_for_snapshots(min_count=2)
+        logs = self.wait_for_snapshots(min_count=3)
         limited_logs = self.logs_for_method(logs, "limited_function")
-        self.assertEqual(len(limited_logs), 2, "Expected exactly 2 snapshots (MaxHits=3 allows 2)")
+        self.assertEqual(len(limited_logs), 3, "Expected exactly 3 snapshots (MaxHits=3 allows 3)")
 
     def test_breakpoint_disabled_after_hit_limit(self) -> None:
         """BREAKPOINT stops generating snapshots after hit limit is reached."""
-        # First 2 calls generate snapshots (hits 1 and 2)
+        # First 3 calls generate snapshots (hits 1, 2, and 3)
+        self.send_request("GET", "limited")
         self.send_request("GET", "limited")
         self.send_request("GET", "limited")
 
-        logs = self.wait_for_snapshots(min_count=2)
+        logs = self.wait_for_snapshots(min_count=3)
         initial_count = len(self.logs_for_method(logs, "limited_function"))
-        self.assertEqual(initial_count, 2)
+        self.assertEqual(initial_count, 3)
 
-        # 3rd and 4th calls hit the limit -- no new snapshots
-        self.send_request("GET", "limited")
+        # 4th call hits the limit -- no new snapshot
         self.send_request("GET", "limited")
-        time.sleep(2)
+        time.sleep(5)
 
         final_logs = self._peek_snapshots()
         final_count = len(self.logs_for_method(final_logs, "limited_function"))
         self.assertEqual(
             final_count,
-            2,
-            f"Expected 2 snapshots (MaxHits=3), but got {final_count}. "
+            3,
+            f"Expected 3 snapshots (MaxHits=3), but got {final_count}. "
             "BREAKPOINT should be disabled after hitting limit.",
         )