test(integration): wait for every rollup write and bound the disconnect recount

Co-Authored-By: Devin AI <158243242+devin-ai-integration[bot]@users.noreply.github.com>

tests/integration/cost_calculation/conftest.py

@@ -123,7 +123,7 @@ def poll_cost_row(key: str) -> CostRow:
     return result
 
 
-def poll_rows(key: str) -> tuple[CostRow, ...]:
+def poll_rows(key: str, count: int) -> tuple[CostRow, ...]:
     digest: Final = sha256(key.encode()).hexdigest()
 
     def read() -> tuple[CostRow, ...]:
@@ -134,11 +134,11 @@ def poll_rows(key: str) -> tuple[CostRow, ...]:
         )
         return tuple(parsed for row in rows if (parsed := _row(row)) is not None)
 
-    result: Final = eventually(read, lambda rows: len(rows) > 0, seconds=20)
+    result: Final = eventually(read, lambda rows: len(rows) >= count, seconds=60)
     return result
 
 
-def poll_rollups(key: str, team_id: str, user_id: str, end_user_id: str) -> Rollups:
+def poll_rollups(key: str, team_id: str, user_id: str, end_user_id: str, requests: int, spend: float) -> Rollups:
     digest: Final = sha256(key.encode()).hexdigest()
 
     def read() -> Rollups | None:
@@ -170,7 +170,7 @@ def poll_rollups(key: str, team_id: str, user_id: str, end_user_id: str) -> Roll
         )
         if not all((key_rows, team_rows, user_rows, end_user_rows, daily_user_rows, daily_team_rows)):
             return None
-        return Rollups(
+        rollups: Final = Rollups(
             key_spend=float(key_rows[0]["spend"]),
             team_spend=float(team_rows[0]["spend"]),
             user_spend=float(user_rows[0]["spend"]),
@@ -178,8 +178,21 @@ def poll_rollups(key: str, team_id: str, user_id: str, end_user_id: str) -> Roll
             daily_user=DailySpend.model_validate(daily_user_rows[0]),
             daily_team=DailySpend.model_validate(daily_team_rows[0]),
         )
+        if rollups.daily_user.api_requests < requests or rollups.daily_team.api_requests < requests:
+            return None
+        if not all(
+            approx_equal(actual, spend)
+            for actual in (
+                rollups.key_spend,
+                rollups.team_spend,
+                rollups.user_spend,
+                rollups.end_user_spend,
+            )
+        ):
+            return None
+        return rollups
 
-    result: Final = eventually(read, lambda value: value is not None, seconds=20)
+    result: Final = eventually(read, lambda value: value is not None, seconds=60)
     assert result is not None
     return result
 

tests/integration/cost_calculation/cost_tracking_case.py

@@ -178,6 +178,7 @@ class RecountExpected(BaseModel):
     prompt_tokens: int | None = None
     completion_tokens: int | None = None
     min_completion_tokens: int | None = None
+    max_completion_tokens: int | None = None
 
 
 class FailureDetails(BaseModel):

tests/integration/cost_calculation/cost_tracking_cases.json

@@ -30014,7 +30014,10 @@
         "recount": {
           "input_cost_per_token": 1.75e-06,
           "output_cost_per_token": 1.4e-05
-        }
+        },
+        "prompt_tokens": 10,
+        "min_completion_tokens": 9,
+        "max_completion_tokens": 30
       }
     }
   ]

tests/integration/cost_calculation/test_cost_tracking.py

@@ -228,6 +228,10 @@ def _assert_recount(case: CostTrackingTestCase, expected: RecountExpected, row:
         assert row.completion_tokens >= expected.min_completion_tokens, (
             f"{case.name}: completion_tokens {row.completion_tokens} < minimum {expected.min_completion_tokens}"
         )
+    if expected.max_completion_tokens is not None:
+        assert row.completion_tokens <= expected.max_completion_tokens, (
+            f"{case.name}: completion_tokens {row.completion_tokens} > maximum {expected.max_completion_tokens}"
+        )
     recount: Final = row.prompt_tokens * expected.recount.input_cost_per_token + (
         row.completion_tokens * expected.recount.output_cost_per_token
     )
@@ -376,7 +380,7 @@ def test_case_bills_expected_cost(gateway: Gateway, case: CostTrackingTestCase)
         assert response.is_success, f"{case.name}: proxy returned {response.status_code}: {response.text[:400]}"
         if case.response.content_type == "text/event-stream":
             _assert_stream_has_no_error(response.text)
-        rows: Final = poll_rows(key) if len(responses) > 1 else (poll_cost_row(key),)
+        rows: Final = poll_rows(key, len(responses))
         if isinstance(expected, RecountExpected):
             row: Final = rows[0]
             _assert_recount(case, expected, row)
@@ -408,8 +412,8 @@ def test_case_bills_expected_cost(gateway: Gateway, case: CostTrackingTestCase)
         if expected.rollups:
             assert deployment is not None and team_id is not None and user_id is not None
             assert end_user_id is not None
-            rollups: Final = poll_rollups(key, team_id, user_id, end_user_id)
             target_spend: Final = expected.spend * 3
+            rollups: Final = poll_rollups(key, team_id, user_id, end_user_id, requests=3, spend=target_spend)
             assert approx_equal(rollups.key_spend, target_spend)
             assert approx_equal(rollups.team_spend, target_spend)
             assert approx_equal(rollups.user_spend, target_spend)