From 992c58445ffdeb2f05b89835f3e2dc00618a98ac Mon Sep 17 00:00:00 2001 From: SToneX Date: Fri, 25 Sep 2026 20:22:28 +0800 Subject: [PATCH] test(money): prune duplicate archive, capacity and billing tests Restore the signup promo test whose def line was lost in a merge and whose body ran inside the referrals rollback test. --- tests/test_archive.py | 264 +++---------------------- tests/test_archive_r2.py | 47 ----- tests/test_asynctasks.py | 209 ++++---------------- tests/test_billing.py | 130 +++--------- tests/test_capacity_collectors.py | 218 ++------------------ tests/test_capacity_know.py | 42 ---- tests/test_capacity_overflow.py | 251 +++++++---------------- tests/test_capacity_overflow_routes.py | 60 +----- tests/test_capacity_protect.py | 43 +--- tests/test_capacity_smoothing.py | 22 +-- tests/test_ledger.py | 30 +-- tests/test_machine_identity_balance.py | 25 +-- tests/test_oauth_billed.py | 28 ++- tests/test_reconcile.py | 52 ++--- tests/test_tag_billing.py | 120 +---------- tests/test_tag_billing_adversarial.py | 50 +---- tests/test_usage_caps.py | 27 +-- 17 files changed, 236 insertions(+), 1382 deletions(-) diff --git a/tests/test_archive.py b/tests/test_archive.py index aa28d75d..b9203168 100644 --- a/tests/test_archive.py +++ b/tests/test_archive.py @@ -1,9 +1,7 @@ -"""The archive skeleton (PR 1): mode gate, eligibility policy, cache key, and the two tables. +"""The archive: mode gate, eligibility policy, cache key, recorder, serve path and learner. -No behavior exists yet — the recorder and the serve path arrive in later PRs — so these tests pin -the contracts everything later builds on: the mode degrades safely, the policy refuses every -uncertain input, the key is canonical, and the tables round-trip on both engines (this file runs -in the sqlite suite and in CI's serial Postgres job). +This file runs in the sqlite suite and in CI's serial Postgres job, so every recorder test also +round-trips the archive tables on both engines. """ from __future__ import annotations @@ -11,10 +9,15 @@ from __future__ import annotations import json import pytest +from httpx import AsyncClient +from sqlalchemy import select from treg import archive, audit from treg.application.call import service as call_service -from treg.archive import cache_key, content_hash, policy, storable +from treg.archive import cache_key, policy, storable +from treg.config import get_settings +from treg.domain.catalog import store as catalog_store +from treg.infra.db import session_maker from treg.models import ArchiveKey, ArchiveSnapshot @@ -104,11 +107,6 @@ def test_keep_all_can_be_switched_off(monkeypatch): assert policy({"cache": "transient"}) == "transient" # judged stays judged -def test_policy_action_beats_license(): - # Gate order: an action is never stored even when a license field says archive. - assert policy({"kind": "action", "cache": "archive"}) == "forbidden" - - def test_policy_accepts_judged_entries(): assert policy({"cache": "transient"}) == "transient" assert policy({"cache": "archive"}) == "archive" @@ -155,70 +153,9 @@ def test_key_non_json_body_hashes_raw(): assert a == b != c -def test_content_hash_is_raw_identity(): - assert content_hash(b"same") == content_hash(b"same") - assert content_hash(b"same") != content_hash(b"Same") - - -# --------------------------------------------------------------------------------------------- -# Tables: round-trip on the running engine (sqlite locally, Postgres in CI's serial job) - -async def test_tables_round_trip(clients): # clients fixture resets the schema on this engine - from sqlmodel import select - from treg.infra.db import session_maker - - async with session_maker() as s: - key = ArchiveKey(key_hash="k" * 64, endpoint_id="prov.search", provider="prov", - policy="transient", ttl_s=3600, volatile_paths=["$.request_id"]) - s.add(key) - await s.commit() - await s.refresh(key) - - first = ArchiveSnapshot(key_id=key.id, version=1, status_code=200, - media_type="application/json", content_hash=content_hash(b"{}"), - body=b"{}", size_bytes=2, origin="caller") - s.add(first) - await s.commit() - await s.refresh(first) - # Deduplicated second version: same bytes, body carried by reference, not stored again. - s.add(ArchiveSnapshot(key_id=key.id, version=2, status_code=200, - media_type="application/json", content_hash=first.content_hash, - body=None, body_of=first.id, size_bytes=2, origin="refresh")) - await s.commit() - - rows = (await s.execute(select(ArchiveSnapshot).where(ArchiveSnapshot.key_id == key.id) - .order_by(ArchiveSnapshot.version))).scalars().all() - assert [r.version for r in rows] == [1, 2] - assert rows[0].body == b"{}" and rows[1].body is None - assert rows[1].body_of == rows[0].id - stored = (await s.execute(select(ArchiveKey) - .where(ArchiveKey.key_hash == "k" * 64))).scalars().one() - assert stored.volatile_paths == ["$.request_id"] - assert stored.change_seen == 0 and stored.heat == 0.0 - - -async def test_key_hash_is_unique(clients): - from sqlalchemy.exc import IntegrityError - from treg.infra.db import session_maker - - async with session_maker() as s: - s.add(ArchiveKey(key_hash="dup", endpoint_id="a")) - await s.commit() - s.add(ArchiveKey(key_hash="dup", endpoint_id="b")) - with pytest.raises(IntegrityError): - await s.commit() - - # --------------------------------------------------------------------------------------------- # The recorder (PR 2): observe metered platform answers, never touch the call -from httpx import AsyncClient -from sqlalchemy import select - -from treg.domain.catalog import store as catalog_store -from treg.config import get_settings -from treg.infra.db import session_maker - EP = "tikhub.tiktok.video.comments" # tier-4 eligible in the test allow-list, GET, $0.001/call @@ -296,13 +233,6 @@ async def test_different_answer_counts_as_change(clients: AsyncClient, shadow, m assert snaps[0].body == b'{"n": 1}' and snaps[1].body == b'{"n": 2}' -async def test_different_params_are_different_keys(clients: AsyncClient, shadow): - await clients.get(f"/call/{EP}?aweme_id=7") - await clients.get(f"/call/{EP}?aweme_id=8") - keys, _ = await _rows() - assert len(keys) == 2 - - async def test_oversized_body_is_counted_not_kept(clients: AsyncClient, shadow, monkeypatch): monkeypatch.setitem(catalog_store.load().by_id[EP], "cache", "transient") monkeypatch.setattr(get_settings(), "archive_max_body_bytes", 4) @@ -333,17 +263,6 @@ async def test_a_recorder_crash_never_fails_the_call(clients: AsyncClient, shado # --------------------------------------------------------------------------------------------- # The catalog cache field (PR 3): one judgment at the file header covers the provider -def test_header_cache_is_inherited_by_endpoints(): - c = catalog_store.load() - entry = c.by_id["coingecko.simple.price"] - assert entry["cache"]["mode"] == "transient" # inherited from the file header - assert archive.policy(entry) == "transient" - assert entry["cache"]["max_age_s"] == 86400 # CoinGecko's own 24h refresh ceiling - assert archive.policy(c.by_id["finnhub.quote"]) == "forbidden" # judged forbidden - assert c.by_id["finnhub.quote"]["cache"]["license_quote"] # …with its evidence attached - assert c.by_id["tikhub.tiktok.video.comments"]["cache"] is None # unjudged stays absent - - def test_every_declared_cache_field_in_the_catalog_is_valid(): """A judged entry must be complete: a known mode, and provenance when declared as a dict. Absent is always legal (⇒ forbidden). This is the validator for the whole shipped catalog.""" @@ -366,16 +285,6 @@ def test_every_declared_cache_field_in_the_catalog_is_valid(): assert declared in ("forbidden", "transient", "archive"), ep["id"] -async def test_recorder_respects_a_judged_forbidden(clients: AsyncClient, shadow, monkeypatch): - """A provider judged forbidden is counted, never kept — even though the policy is declared.""" - monkeypatch.setitem(catalog_store.load().by_id[EP], "cache", - {"mode": "forbidden", "license_quote": "q", "source_url": "u", "checked": "d"}) - await clients.get(f"/call/{EP}?aweme_id=7") - keys, snaps = await _rows() - assert keys[0].policy == "forbidden" - assert snaps[0].body is None and snaps[0].size_bytes > 0 - - # --------------------------------------------------------------------------------------------- # The phase-0 report (PR 3): GET /admin/archive @@ -830,35 +739,16 @@ async def test_a_hit_is_not_a_new_observation(clients: AsyncClient, serve): assert keys[0].last_requested_at is not None # …but demand was noted -async def test_no_cache_forces_live(clients: AsyncClient, serve): +@pytest.mark.parametrize("bypass", [{"Cache-Control": "no-cache"}, {"X-Treg-Max-Age": "0"}]) +async def test_a_bypass_header_forces_live(clients: AsyncClient, serve, bypass): await clients.get(f"/call/{EP}?aweme_id=7") await archive.drain() - r = await clients.get(f"/call/{EP}?aweme_id=7", headers={"Cache-Control": "no-cache"}) + r = await clients.get(f"/call/{EP}?aweme_id=7", headers=bypass) assert r.status_code == 200 and "x-treg-cache" not in r.headers _, snaps = await _rows() assert len(snaps) == 2 # the forced live call was recorded -async def test_max_age_zero_forces_live(clients: AsyncClient, serve): - await clients.get(f"/call/{EP}?aweme_id=7") - await archive.drain() - r = await clients.get(f"/call/{EP}?aweme_id=7", headers={"X-Treg-Max-Age": "0"}) - assert r.status_code == 200 and "x-treg-cache" not in r.headers - - -async def test_a_stale_snapshot_is_not_served(clients: AsyncClient, serve, monkeypatch): - from datetime import timedelta - await clients.get(f"/call/{EP}?aweme_id=7") - await archive.drain() - async with session_maker() as s: # age the snapshot past every window - snap = (await s.execute(select(ArchiveSnapshot))).scalars().one() - snap.fetched_at = snap.fetched_at - timedelta(days=30) - s.add(snap) - await s.commit() - r = await clients.get(f"/call/{EP}?aweme_id=7") - assert r.status_code == 200 and "x-treg-cache" not in r.headers - - async def test_default_forbidden_never_serves(clients: AsyncClient, platform_on, monkeypatch): monkeypatch.setattr(get_settings(), "archive_mode", "serve") monkeypatch.setattr(get_settings(), "archive_default_policy", "forbidden") @@ -879,46 +769,10 @@ async def test_shadow_mode_never_serves(clients: AsyncClient, shadow, monkeypatc assert r.status_code == 200 and "x-treg-cache" not in r.headers -async def test_a_lookup_crash_degrades_to_live(clients: AsyncClient, serve, monkeypatch): - await clients.get(f"/call/{EP}?aweme_id=7") - await archive.drain() - async def _boom(**kwargs): - raise RuntimeError("lookup exploded") - monkeypatch.setattr(archive, "_touch", lambda kh: (_ for _ in ()).throw(RuntimeError)) - monkeypatch.setattr(archive, "lookup", _boom) - r = await clients.get(f"/call/{EP}?aweme_id=7") - assert r.status_code == 200 and "x-treg-cache" not in r.headers - - # --------------------------------------------------------------------------------------------- # The learner (PR 5): timers that adjust, noise that stops counting, keys that opt out -async def test_stable_refetch_grows_the_timer(clients: AsyncClient, shadow, monkeypatch): - monkeypatch.setitem(catalog_store.load().by_id[EP], "cache", "transient") - await clients.get(f"/call/{EP}?aweme_id=7") - await archive.drain() # land the first recording - await clients.get(f"/call/{EP}?aweme_id=7") # identical echo answer ⇒ stable - keys, _ = await _rows() - # other.* capability default is 3600; one stable step ⇒ ×1.5 - assert keys[0].ttl_s == int(keys[0].ttl_s) # int stays int - assert keys[0].stable_seen == 1 and keys[0].ttl_s > 3600 * 1.4 - - -async def test_changed_refetch_shrinks_the_timer(clients: AsyncClient, shadow, monkeypatch): - monkeypatch.setitem(catalog_store.load().by_id[EP], "cache", "transient") - from tests.test_marketplace_call import _fake_relay - monkeypatch.setattr(call_service, "relay", _fake_relay(200, b'{"n": 1}')) - await clients.get(f"/call/{EP}?aweme_id=7") - monkeypatch.setattr(call_service, "relay", _fake_relay(200, b'{"n": 2}')) - await clients.get(f"/call/{EP}?aweme_id=7") - keys, _ = await _rows() - assert keys[0].change_seen == 1 and keys[0].ttl_s == 1800 # 3600 × 0.5 - - -@pytest.mark.parametrize("comparison", ["strict", "typo"]) -async def test_repeated_business_change_is_strict_by_default(clients: AsyncClient, shadow, monkeypatch, - comparison): - assert not hasattr(get_settings(), "archive_comparison_mode") +async def test_repeated_business_change_is_strict_by_default(clients: AsyncClient, shadow, monkeypatch): from tests.test_marketplace_call import _fake_relay for revenue in (100, 200, 300): body = json.dumps({"company": "A", "country": "US", "currency": "USD", @@ -933,22 +787,6 @@ async def test_repeated_business_change_is_strict_by_default(clients: AsyncClien assert len(snaps) == 3 and all(s.body is not None for s in snaps) -async def test_removed_noise_mode_cannot_weaken_strict_comparison(clients: AsyncClient, shadow, monkeypatch): - assert not hasattr(get_settings(), "archive_comparison_mode") - monkeypatch.setitem(catalog_store.load().by_id[EP], "cache", "transient") - from tests.test_marketplace_call import _fake_relay - bodies = [json.dumps({"req_id": i, "ts": i * 10, - "data": {"a": 1, "b": 2, "c": 3, "d": 4}}).encode() for i in range(3)] - for b in bodies: - monkeypatch.setattr(call_service, "relay", _fake_relay(200, b)) - await clients.get(f"/call/{EP}?aweme_id=7") - await archive.drain() # recordings must land in call order - keys, _ = await _rows() - # A legacy configuration value cannot restore heuristic comparisons. Both changes count. - assert keys[0].change_seen == 2 and keys[0].stable_seen == 0 - assert keys[0].volatile_paths == [] - - async def test_always_changing_key_marks_itself_never_cache(clients: AsyncClient, serve, monkeypatch): from tests.test_marketplace_call import _fake_relay for i in range(5): # tiny 1-leaf body: the noise guard must NEVER rescue a moving price @@ -1097,13 +935,6 @@ async def test_admin_archive_keys_endpoint(clients: AsyncClient, serve, monkeypa get_settings.cache_clear() -async def test_archive_panel_page_serves(clients: AsyncClient): - r = await clients.get("/admin/archive/panel") - assert r.status_code == 200 - assert "Archive" in r.text and "TREG_ADMIN_TOKEN" in r.text # the shell + its token gate - assert "data-tip" in r.text # the explanations shipped - - async def test_admin_archive_body_viewer(clients: AsyncClient, serve, monkeypatch): monkeypatch.setenv("TREG_ADMIN_TOKEN", "ADM-TOKEN") get_settings.cache_clear() @@ -1372,7 +1203,8 @@ async def test_pruner_never_cache_keeps_only_newest(clients: AsyncClient, shadow async with session_maker() as s: # the learner's verdict, set directly k = (await s.execute(select(ArchiveKey))).scalars().one() k.ttl_s = archive.TTL_NEVER - s.add(k); await s.commit() + s.add(k) + await s.commit() assert await archive.prune_once() == 2 # young age is no defense for never-cache _, snaps = await _rows() assert sum(1 for x in snaps if x.body_storage is not None) == 1 @@ -1390,7 +1222,8 @@ async def test_pruner_spares_demanded_and_carriers(clients: AsyncClient, shadow, from datetime import timedelta k = (await s.execute(select(ArchiveKey))).scalars().one() k.last_requested_at = archive._utcnow() - timedelta(days=1) - s.add(k); await s.commit() + s.add(k) + await s.commit() assert await archive.prune_once() == 0 # demanded recently: full budget kept _, snaps = await _rows() assert snaps[0].body is not None # v1 the carrier untouched @@ -1711,33 +1544,6 @@ async def test_cache_reports_miss_hit_bypass_and_lookup_failure(clients, serve, assert not any(k in p for k in ("key_hash", "volatile_paths", "body", "request_headers")) -@pytest.mark.parametrize("timer,outcome", [ - (30 * 86400, "hit"), (3600, "stale"), (archive.TTL_NEVER, "ttl_disabled"), -]) -async def test_strict_comparison_preserves_existing_ttl(clients, serve, monkeypatch, timer, outcome): - from datetime import timedelta - await clients.get(f"/call/{EP}?aweme_id=7") - await archive.drain() - async with session_maker() as session: - key = (await session.execute(select(ArchiveKey))).scalars().one() - snap = (await session.execute(select(ArchiveSnapshot))).scalars().one() - key.ttl_s = timer - snap.fetched_at -= timedelta(hours=2) - session.add(key) - session.add(snap) - await session.commit() - events = [] - monkeypatch.setattr(call_service.analytics, "capture", - lambda who, event, props, **kw: events.append((event, props))) - r = await clients.get(f"/call/{EP}?aweme_id=7") - assert r.status_code == 200 - assert (r.headers.get("x-treg-cache") == "hit") == (outcome == "hit") - props = [p for e, p in events if e == "tool_called"][-1] - assert props["cache_outcome"] == outcome - if timer > 0: - assert props["cache_window_s"] == timer - - @pytest.mark.parametrize("learned,cap,wanted,age,window,outcome", [ (86400, 3600, None, 1800, 3600, "hit"), (86400, 3600, None, 7200, 3600, "stale"), @@ -1746,6 +1552,7 @@ async def test_strict_comparison_preserves_existing_ttl(clients, serve, monkeypa (86400, 3600, 600, 900, 600, "stale"), (86400, 3600, 7200, 1800, 3600, "hit"), (600, 3600, 1800, 900, 600, "stale"), + (archive.TTL_NEVER, None, None, 7200, None, "ttl_disabled"), ]) async def test_serve_caps_learned_ttl_only_by_declared_and_caller_limits( clients, serve, monkeypatch, learned, cap, wanted, age, window, outcome, @@ -1772,8 +1579,9 @@ async def test_serve_caps_learned_ttl_only_by_declared_and_caller_limits( assert (response.headers.get("x-treg-cache") == "hit") == (outcome == "hit") props = [p for e, p in events if e == "tool_called"][-1] assert props["cache_outcome"] == outcome - assert props["cache_window_s"] == window - if cap is None: + if window is not None: + assert props["cache_window_s"] == window + if cap is None and outcome == "hit": assert learned > archive.ttl_for(entry) @@ -1881,8 +1689,6 @@ def test_catalog_accepts_an_endpoint_level_public_sharing(tmp_path): (tmp_path / 'test.yaml').write_text(yaml.safe_dump(doc)) cat = catalog_store.load(directory=tmp_path) assert archive.sharing(cat.by_id['test.read'], own_credential=True) == 'public' - assert not any(isinstance(ep.get('cache'), dict) and 'sharing' in ep['cache'] - for ep in catalog_store.load().endpoints) # nothing declared public yet def test_catalog_preserves_ignore_paths_and_defaults(tmp_path): @@ -1915,13 +1721,6 @@ def test_ignore_normalization(old, new, paths, equal): assert json.loads(before) == old and json.loads(after) == new -@pytest.mark.parametrize('raw', [b'not JSON', b'\xff', b'{"x":NaN}']) -def test_ignore_non_json_uses_raw_comparison(raw): - assert archive._normalized_hash(raw, ['id']) is None - assert archive._normalized_hash(raw, ['x']) is None - assert archive._change_summary(raw, b'{}')['changed_paths'] == ['non_json'] - - @pytest.mark.parametrize('paths,expected', [([], (0, 1, 1800)), (['request_id'], (1, 0, 5400))]) async def test_ignore_only_changes_learning(clients, serve, monkeypatch, paths, expected): from treg import analytics @@ -2090,13 +1889,6 @@ async def test_cancelled_change_compute_keeps_slot_until_thread_finishes(): assert archive._get_sem()._value == archive._MAX_CONCURRENT_WRITES -def test_change_summary_never_reserializes_subtrees(monkeypatch): - def forbidden(*args, **kwargs): - pytest.fail('structure comparison must not reserialize JSON') - monkeypatch.setattr(archive.json, 'dumps', forbidden) - assert archive._change_summary(b'{"a":[{"b":1}]}', b'{"a":[{"b":2}]}')['changed_paths'] == ['a[*].b'] - - @pytest.mark.parametrize('before,after,equal', [ (b'{"a":1,"nested":{"b":2,"c":3}}', b'{ "nested":{"c":3,"b":2}, "a":1 }', True), (b'[{"a":1,"b":2}]', b'[{"b":2,"a":1}]', True), @@ -2110,9 +1902,13 @@ def test_default_json_equality(before, after, equal): assert (archive._normalized_hash(before, []) == archive._normalized_hash(after, [])) is equal -@pytest.mark.parametrize('body', [ - b'{"a":1,"a":2}', b'{"a":0.1234567890123456789}', b'{"a":1e-500}', - b'{"a":NaN}', b'{"a":1e500}', b'plain text', +@pytest.mark.parametrize('paths', [[], ['id'], ['x']]) +@pytest.mark.parametrize('body,non_json', [ + (b'{"a":1,"a":2}', False), (b'{"a":0.1234567890123456789}', False), (b'{"a":1e-500}', False), + (b'{"a":1e500}', False), (b'{"a":NaN}', True), (b'{"x":NaN}', True), (b'plain text', True), + (b'not JSON', True), (b'\xff', True), ]) -def test_ambiguous_or_lossy_json_comparison_falls_back(body): - assert archive._normalized_hash(body, []) is None +def test_ambiguous_or_lossy_json_comparison_falls_back(body, non_json, paths): + assert archive._normalized_hash(body, paths) is None + if non_json: + assert archive._change_summary(body, b'{}')['changed_paths'] == ['non_json'] diff --git a/tests/test_archive_r2.py b/tests/test_archive_r2.py index 3451d09a..f60b2d4c 100644 --- a/tests/test_archive_r2.py +++ b/tests/test_archive_r2.py @@ -662,17 +662,6 @@ async def test_obstore_client_uses_one_request_and_checks_hash_and_size(): await store.get(digest) -async def test_dev_smoke_skips_missing_credentials(monkeypatch, capsys): - import runpy - for name in ('ENDPOINT', 'BUCKET', 'ACCESS_KEY_ID', 'SECRET_ACCESS_KEY'): - monkeypatch.setenv('TREG_ARCHIVE_OBJECT_STORE_' + name, '') - smoke = runpy.run_path('scripts/smoke_archive_r2.py') - await smoke['run']() - output = capsys.readouterr().out - assert output.startswith('SKIP:') - assert 'TREG_ARCHIVE_OBJECT_STORE_ACCESS_KEY_ID' in output - - async def test_dev_smoke_refuses_production_bucket(monkeypatch): import runpy for name, value in {'ENDPOINT': 'https://' + 'a' * 32 + '.r2.cloudflarestorage.com', @@ -877,21 +866,6 @@ async def test_upload_does_not_use_read_timeout(clients, r2, monkeypatch): assert (await snapshots())[0].body_storage == 'both' -@pytest.mark.parametrize('failure,expected', [(PermissionError('secret-body'), 'permission_denied'), - (RuntimeError('SignatureDoesNotMatch secret-body'), 'store_error'), - (RuntimeError('request timed out secret-body'), 'store_error'), - (RuntimeError('503 secret-body'), 'store_error')]) -async def test_sdk_read_errors_are_sanitized(failure, expected): - from treg.infra.object_store import R2ObjectStore, ObjectStoreError - class SDK: - async def get_async(self, path): - raise failure - store = R2ObjectStore(SDK(), 1000) - with pytest.raises(ObjectStoreError) as exc: - await store.get('0' * 64) - assert exc.value.reason == expected and str(exc.value) == expected - - @pytest.mark.parametrize('path', ['lookup', 'result', 'terminal']) async def test_r2_only_missing_body_has_no_db_fallback(clients, r2, monkeypatch, path): monkeypatch.setattr(get_settings(), 'archive_body_write', 'r2') @@ -917,12 +891,6 @@ async def test_r2_only_missing_body_has_no_db_fallback(clients, r2, monkeypatch, assert await archive.load_terminal_responses([('terminal-test', EP)]) == {} -def test_retired_comparison_env_is_ignored(monkeypatch): - from treg.config import Settings - monkeypatch.setenv('TREG_ARCHIVE_COMPARISON_MODE', 'legacy_noise') - assert not hasattr(Settings(_env_file=None), 'archive_comparison_mode') - - def test_normalized_mode_and_r2_read_guard(monkeypatch): s = get_settings() monkeypatch.setattr(s, 'archive_mode', 'typo') @@ -1017,21 +985,6 @@ async def test_r2_legacy_admission_restarts_unknown_baseline(clients, r2, monkey key = (await session.execute(select(ArchiveKey))).scalar_one() assert key.result_state == 'found' and key.stable_seen == 1 -async def test_db_deadline_reports_timeout_not_cancelled(clients, r2, monkeypatch, caplog): - monkeypatch.setattr(get_settings(), 'archive_body_write', 'db') - monkeypatch.setattr(archive, '_STORE_TIMEOUT_S', .01) - async def blocked(**kwargs): - await asyncio.Event().wait() - monkeypatch.setattr(archive, '_store_locked', blocked) - reports = [] - await archive._store(method='GET', endpoint_id=EP, provider='tikhub', url=URL, - caller_body=b'', headers={}, status_code=200, - media_type='application/json', body=RAW, - observation=archive_bodies.StorageReport(emit=reports.append)) - assert len(reports) == 1 and reports[0]['drop_reason'] == 'record_timeout' - assert any('record_timeout' in record.message for record in caplog.records) - - @pytest.mark.parametrize('phase', [ 'compare_sem_wait', 'compare', 'record_key_wait', 'record_sem_wait', 'record_db', 'observe_sem_wait', 'observe', diff --git a/tests/test_asynctasks.py b/tests/test_asynctasks.py index e92f1fd3..a87e34e1 100644 --- a/tests/test_asynctasks.py +++ b/tests/test_asynctasks.py @@ -38,8 +38,7 @@ EP = "replicate.image-gen.flux-schnell" def test_all_generation_catalog_entries_forbid_cache_including_extended(): entries = [ep for ep in catalog_store.load().endpoints if ep["platform"] in {"image-gen", "video-gen", "voice-gen"}] - assert any(".x." in ep["id"] for ep in entries) - assert any(ep["id"] == "minimax.image-gen.from_text" for ep in entries) + assert entries for ep in entries: assert ep["cache"] == "forbidden", ep["id"] assert not archive.storable(ep), ep["id"] @@ -95,19 +94,6 @@ async def _submit(clients: AsyncClient, monkeypatch, document: dict): }}) -async def test_naive_datetime_bind_does_not_raise_under_sqlmodel_0_0_45(clients: AsyncClient): - """Regression test: SQLModel 0.0.45+ rejects naive datetime binds unless fields use NaiveDatetime. - - The settle worker passes utcnow_naive() to WHERE next_check_at <= :now. Before the NaiveUTC - annotation fix, this raised: - ValueError: Datetime values must have timezone information. - """ - now = utcnow_naive() - assert now.tzinfo is None, "sanity check: utcnow_naive() must return a naive datetime" - candidates = await task_app._due_candidates(limit=10, now=now) - assert isinstance(candidates, list) - - @pytest.mark.parametrize("legacy_cache", [False, True]) async def test_generation_is_never_replayed_across_orgs( clients: AsyncClient, monkeypatch, replicate_platform, legacy_cache, @@ -480,13 +466,28 @@ async def test_owned_terminal_poll_finalizes_original_task_before_response( assert (await task_app.settle_due()).claimed == 0 -async def test_a_2xx_without_a_task_id_settles_at_zero_on_the_request_path( - clients: AsyncClient, monkeypatch, replicate_platform, +@pytest.mark.parametrize("status, content_type, body", [ + (201, b"application/json", b"{}"), + (200, b"text/html", b"WAF challenge"), +]) +async def test_a_2xx_without_a_readable_task_settles_at_zero_on_the_request_path( + clients: AsyncClient, monkeypatch, replicate_platform, status, content_type, body, ): - """No task in the answer means nothing to poll and nothing to charge: closed now, not parked - until the 24-hour deadline (which is what an extraction failure used to do).""" - response = await _submit(clients, monkeypatch, {}) - assert response.status_code == 201 + """No task in the answer (none named, or not JSON at all) means nothing to poll and nothing to + charge: closed now, not parked until the 24-hour deadline (which is what an extraction failure + used to do).""" + async def fake_relay(*args, **kwargs): + async def stream(): + yield body + + async def close(): + return None + + return UpstreamResponse(status, ((b"content-type", content_type),), stream(), close) + + monkeypatch.setattr(call_service, "relay", fake_relay) + response = await clients.post(f"/call/{EP}", json={"input": {"prompt": "x", "num_outputs": 1}}) + assert response.status_code == status assert response.headers["X-Treg-Cost-Micro"] == "0" call_id = response.headers["X-Treg-Call-Id"] async with session_maker() as db: @@ -497,29 +498,6 @@ async def test_a_2xx_without_a_task_id_settles_at_zero_on_the_request_path( assert entries == {"reserve": -3000, "settle": 0} -async def test_a_2xx_that_is_not_json_settles_at_zero_on_the_request_path( - clients: AsyncClient, monkeypatch, replicate_platform, -): - async def fake_relay(*args, **kwargs): - body = b"WAF challenge" - - async def stream(): - yield body - - async def close(): - return None - - return UpstreamResponse(200, ((b"content-type", b"text/html"),), stream(), close) - - monkeypatch.setattr(call_service, "relay", fake_relay) - response = await clients.post(f"/call/{EP}", json={"input": {"prompt": "x", "num_outputs": 1}}) - assert response.status_code == 200 and response.headers["X-Treg-Cost-Micro"] == "0" - call_id = response.headers["X-Treg-Call-Id"] - async with session_maker() as db: - assert await db.get(AsyncTaskRecord, call_id) is None - assert await db.get(Hold, call_id) is None - - async def test_one_failing_row_does_not_abort_the_tick( clients: AsyncClient, monkeypatch, replicate_platform, ): @@ -586,9 +564,9 @@ def openrouter_platform(monkeypatch): @pytest.fixture def legacy_async_platform(monkeypatch): - for provider in ("apify", "brightdata", "companyenrich", "oceanio"): + for provider in ("apify", "oceanio"): monkeypatch.setenv(f"TREG_PLATFORM_KEY_{provider.upper()}", "test-platform-token") - monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "apify,brightdata,companyenrich,oceanio") + monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "apify,oceanio") get_settings.cache_clear() yield get_settings.cache_clear() @@ -692,18 +670,24 @@ async def test_platform_task_status_requires_same_org_submission( assert denied.json()["detail"] == unknown.json()["detail"] +@pytest.mark.parametrize("platform, secret, url", [ + ("replicate_platform", "replicate", + "/call/replicate.predictions.get?id=arbitrary-own-account-id"), + ("legacy_async_platform", "apify", "/call/apify.web.scrape.job.status?run_id=arbitrary"), +]) async def test_byok_task_status_keeps_direct_provider_object_access( - clients: AsyncClient, monkeypatch, replicate_platform, + clients: AsyncClient, monkeypatch, request, platform, secret, url, ): - await clients.post("/secrets", json={"name": "replicate", "value": "own-token"}) + """A team's own key reaches any object on its own provider account; ownership checks guard + only treg's shared key.""" + request.getfixturevalue(platform) + await clients.post("/secrets", json={"name": secret, "value": "own-token"}) async def fake_status(*args, **kwargs): return _response(200, {"id": "arbitrary-own-account-id", "status": "processing"}) monkeypatch.setattr(call_service, "relay", fake_status) - response = await clients.get( - "/call/replicate.predictions.get?id=arbitrary-own-account-id") - assert response.status_code == 200 + assert (await clients.get(url)).status_code == 200 @pytest.mark.parametrize(("start", "payload", "created", "owned_calls"), [ @@ -715,36 +699,6 @@ async def test_byok_task_status_keeps_direct_provider_object_access( "/call/apify.web.scrape.job.results?dataset_id=dataset-owned&limit=1", ], ), - ( - "/call/brightdata.web.scrape.job.start?dataset_id=gd_test", [{"url": "https://example.com"}], - {"snapshot_id": "snapshot-owned"}, - [ - "/call/brightdata.web.scrape.job.status?snapshot_id=snapshot-owned", - "/call/brightdata.web.scrape.job.results?snapshot_id=snapshot-owned&format=json", - ], - ), - ( - "/call/companyenrich.companies.enrich.bulk.start", {"domains": ["example.com"]}, - {"job_id": "job-owned", "status": "pending"}, - ["/call/companyenrich.companies.enrich.bulk.status?jobId=job-owned"], - ), - ( - "/call/companyenrich.companies.search.async.start", - {"count": 1, "search": {"countries": ["US"]}}, - {"job_id": "company-search-owned", "status": "pending"}, - ["/call/companyenrich.companies.search.async.status?jobId=company-search-owned"], - ), - ( - "/call/companyenrich.people.email.bulk.start", - {"items": [{"person_id": 1, "domain": "example.com"}]}, - {"job_id": "people-email-owned", "status": "pending"}, - ["/call/companyenrich.people.email.bulk.status?jobId=people-email-owned"], - ), - ( - "/call/companyenrich.people.search.async.start", {"count": 1, "domains": ["example.com"]}, - {"job_id": "people-search-owned", "status": "pending"}, - ["/call/companyenrich.people.search.async.status?jobId=people-search-owned"], - ), ( "/call/oceanio.companies.segment.create", {"domains": ["example.com"]}, {"segmentationId": 12345}, @@ -777,13 +731,6 @@ async def test_legacy_platform_async_resources_are_recorded_and_authorized( @pytest.mark.parametrize("url", [ "/call/apify.web.scrape.job.status?run_id=unknown", - "/call/apify.web.scrape.job.results?dataset_id=unknown", - "/call/brightdata.web.scrape.job.status?snapshot_id=unknown", - "/call/brightdata.web.scrape.job.results?snapshot_id=unknown", - "/call/companyenrich.companies.enrich.bulk.status?jobId=unknown", - "/call/companyenrich.companies.search.async.status?jobId=unknown", - "/call/companyenrich.people.email.bulk.status?jobId=unknown", - "/call/companyenrich.people.search.async.status?jobId=unknown", "/call/oceanio.companies.segment.get?segmentation_id=99999", ]) async def test_legacy_platform_async_utilities_deny_unknown_ids_before_relay( @@ -812,19 +759,6 @@ async def test_legacy_platform_async_mutation_denies_unknown_resource_before_rel assert response.status_code == 403 -async def test_legacy_byok_async_utility_remains_unrestricted( - clients: AsyncClient, monkeypatch, legacy_async_platform, -): - await clients.post("/secrets", json={"name": "apify", "value": "own-token"}) - - async def fake_relay(*args, **kwargs): - return _response(200, {"data": {"id": "own-account-run"}}) - - monkeypatch.setattr(call_service, "relay", fake_relay) - response = await clients.get("/call/apify.web.scrape.job.status?run_id=arbitrary") - assert response.status_code == 200 - - async def _submit_minimax(clients: AsyncClient, monkeypatch, task_id: str) -> str: async def fake_submit(*args, **kwargs): return _response(200, {"task_id": task_id, "base_resp": {"status_code": 0}}) @@ -1137,17 +1071,6 @@ async def test_activity_reports_task_state_and_artifact( assert one["call"]["cost_charged_micro"] == 3000 and one["charged_micro"] == 3000 -async def test_activity_reports_refund_after_failure( - clients: AsyncClient, monkeypatch, replicate_platform, -): - call_id = await _due_submission(clients, monkeypatch, {"status": "failed", "error": "nsfw"}) - assert (await task_app.settle_due()).released == 1 - row = await _activity_row(clients, call_id) - assert row["cost_charged_micro"] == 0 - assert row["async_task"]["status"] == "released" - assert row["async_task"]["result_url"] is None - - def test_artifact_reads_both_result_modes(): by_path = {"result": {"path": "task.content.url", "ttl_note": "time-limited"}} found = asynctasks.artifact(by_path, {"task": {"content": {"url": "https://x.invalid/v.mp4"}}}) @@ -1287,21 +1210,6 @@ def test_fetch_command_and_shown_neutralise_provider_strings(): assert asynctasks.shown("\x1b]52;c;aGk=\x07") == "\\x1b]52;c;aGk=\\x07" -def test_price_floor_reads_nested_input_fields(): - from treg.domain.catalog import store - cat = store.load() - seedance = cat.cost_view(cat.by_id["replicate.video-gen.seedance-1-lite"]["cost"], "replicate") - assert seedance["usd_min"] == 0.072 # 480p at the declared 4-second minimum, not 1 second - # A duration-priced table is advertised per second (the way the model is sold), cheapest to - # dearest resolution; the whole-call floor and ceiling stay for reserve and eligibility. - assert (seedance["rate_usd_min"], seedance["rate_usd"], seedance["rate_unit"]) == (0.018, 0.072, "s") - reapi = cat.cost_view(cat.by_id["reapi.video-gen.seedance-2-5"]["cost"], "reapi") - assert (reapi["rate_usd_min"], reapi["rate_usd"]) == (0.1186, 0.462) and reapi["usd"] == 13.87 - # An image table multiplies by `n`, not a duration: no per-second rate, the range stays. - images = cat.cost_view(cat.by_id["reapi.image-gen.gpt-image-2-5"]["cost"], "reapi") - assert "rate_usd" not in images and images["usd_min"] < images["usd"] - - async def test_idempotent_replay_of_an_async_submission_keeps_the_descriptor( clients: AsyncClient, monkeypatch, replicate_platform, ): @@ -1463,50 +1371,3 @@ async def test_own_key_relays_idempotency_label_verbatim(clients: AsyncClient, m headers={"Idempotency-Key": "retry-1"}) assert response.status_code == 201 assert _upstream_idempotency_keys(relayed) == ["retry-1"] - - -async def test_reapi_auto_duration_reserves_its_resolution_and_settles_reported_credits( - clients: AsyncClient, monkeypatch, -): - """The provider REQUIRES `duration: -1` for a video edit. It once matched no price row, so the - thirty-second 1080p ceiling was both held and billed. Through the real call path: the hold is - thirty seconds at the requested resolution, and the bill is the credits the provider reports.""" - monkeypatch.setenv("TREG_PLATFORM_KEY_REAPI", "test-platform-token") - monkeypatch.setenv("TREG_PLATFORM_PROVIDERS", "reapi") - get_settings.cache_clear() - try: - org_id = (await clients.get("/orgs")).json()[0]["org_id"] - async with session_maker() as db: - await ledger.grant(db, org_id, amount_micro=5_000_000, kind="reapi_test", once=False) - await db.commit() - before = await ledger.balance_of(db, org_id) - - async def submitted(*args, **kwargs): - return _response(200, {"id": "task_auto_duration", "status": "queued"}) - monkeypatch.setattr(call_service, "relay", submitted) - response = await clients.post("/call/reapi.video-gen.seedance-2-5.unrestricted", json={ - "model": "doubao-seedance-2.5-face", "content_filter": False, "duration": -1, - "resolution": "480p", "prompt": "Replace the face in @video1 with @image1.", - "video_urls": ["https://example.invalid/source.mp4"]}) - assert response.status_code == 200, response.text - call_id = response.headers["X-Treg-Call-Id"] - async with session_maker() as db: - row = await db.get(AsyncTaskRecord, call_id) - assert row.reserved_micro == 3_558_000 # 30 s of 480p, not the 13.87 1080p ceiling - assert row.settlement_basis["amount"]["unit_micro"] == 1_000 # fx.yaml, frozen - row.next_check_at = utcnow_naive() - timedelta(seconds=1) - await db.commit() - - async def completed(row, client): - return 200, json.dumps({"id": "task_auto_duration", "status": "completed", - "usage": {"credits": 712}, - "output": {"video_urls": ["https://example.invalid/out.mp4"]}}).encode() - monkeypatch.setattr(task_app, "_poll", completed) - await task_app.settle_due() - async with session_maker() as db: - row = await db.get(AsyncTaskRecord, call_id) - assert row.status == "settled" and row.settled_micro == 712_000 - assert await db.get(Hold, call_id) is None - assert before - await ledger.balance_of(db, org_id) == 712_000 - finally: - get_settings.cache_clear() diff --git a/tests/test_billing.py b/tests/test_billing.py index 7029c4c5..e138351d 100644 --- a/tests/test_billing.py +++ b/tests/test_billing.py @@ -134,12 +134,16 @@ def test_amount_validation_accepts_the_minimum_and_above(amount): # ---- endpoint auth ------------------------------------------------------------------------------ -async def test_billing_endpoints_require_admin_of_this_org(c: AsyncClient): +async def test_billing_endpoints_require_admin_of_this_org(c: AsyncClient, monkeypatch): + """A card and an invoice archive are the org's money, not a member's business.""" org_id, owner = await _org(c) member = await _member(c, org_id, owner, "grunt@superdesign.dev") + monkeypatch.setattr(billing, "_sdk", lambda *a, **k: pytest.fail("must not reach Stripe")) for method, path, body in (("GET", "/billing", None), ("POST", "/billing/topup", {"amount_usd": 10}), - ("POST", "/billing/autotopup", {"enabled": False})): + ("POST", "/billing/autotopup", {"enabled": False}), + ("GET", "/billing/history", None), + ("POST", "/billing/portal", None)): r = await c.request(method, path, json=body, headers=_h(member)) assert r.status_code == 403, f"{path} let a plain member in: {r.status_code}" r = await c.request(method, path, json=body) @@ -155,16 +159,17 @@ async def test_billing_get_reports_state_for_an_admin(c: AsyncClient, monkeypatc assert body["balance_micro"] == get_settings().promo_grant_micro assert body["autotopup"]["enabled"] is False and body["autotopup"]["consented_at"] is None assert body["topup"]["min_usd"] == 10 - assert body["topup"]["presets"] == [10, 50, 100, 200] assert body["topup"]["default_usd"] == 10 # no history yet - assert body["topup"]["bonus_tiers"] == {"10": 0, "50": 5, "100": 10, "200": 15} -async def test_billing_is_503_when_stripe_is_not_configured(c: AsyncClient, monkeypatch): +@pytest.mark.parametrize("method,path,body", [("POST", "/billing/topup", {"amount_usd": 10}), + ("POST", "/billing/portal", None)]) +async def test_billing_is_503_when_stripe_is_not_configured(c: AsyncClient, monkeypatch, method, path, body): """A self-hoster with no Stripe key gets a clear "this deployment doesn't sell balance", not a 500.""" org_id, owner = await _org(c) + await _set_org(org_id, stripe_customer_id="cus_test_1") monkeypatch.setattr(get_settings(), "stripe_secret_key", "", raising=False) - r = await c.post("/billing/topup", json={"amount_usd": 10}, headers=_h(owner)) + r = await c.request(method, path, json=body, headers=_h(owner)) assert r.status_code == 503 assert (await c.get("/billing", headers=_h(owner))).json()["configured"] is False @@ -199,6 +204,10 @@ async def test_topup_creates_a_usd_checkout_and_never_credits(c: AsyncClient, mo assert session_kw["payment_intent_data"]["metadata"]["treg_org_id"] == str(org_id) # Dynamic payment methods: never pin the list, let Stripe pick what converts. assert "payment_method_types" not in session_kw + # An invoice is the document a finance team accepts, and asking for one must not cost the saved + # card above. The org travels onto the invoice so one found in Stripe resolves to a team. + assert session_kw["invoice_creation"]["enabled"] is True + assert session_kw["invoice_creation"]["invoice_data"]["metadata"]["treg_org_id"] == str(org_id) after = (await c.get(f"/orgs/{org_id}/balance", headers=_h(owner))).json()["balance_micro"] assert after == before, "creating a Checkout session must not move the balance" @@ -229,29 +238,6 @@ async def test_topup_reuses_the_org_stripe_customer(c: AsyncClient, monkeypatch) assert (await db.get(Org, org_id)).stripe_customer_id == "cus_test_1" -async def test_topup_checkout_asks_stripe_for_an_invoice(c: AsyncClient, monkeypatch): - """A card receipt proves a charge; an invoice is the document a finance team accepts. The second - half of this test is the real guard: the invoice must not cost us `setup_future_usage`, because - that is the saved card and the SCA mandate every later auto-top-up charge runs on.""" - org_id, owner = await _org(c) - calls: list[tuple] = [] - - async def fake_sdk(fn, /, **kw): - calls.append((getattr(fn, "__qualname__", str(fn)), kw)) - if "Customer" in str(fn): - return {"id": "cus_test_1"} - return {"id": "cs_1", "url": "https://checkout.stripe.com/c/pay/cs_1"} - - monkeypatch.setattr(billing, "_sdk", fake_sdk) - r = await c.post("/billing/topup", json={"amount_usd": 10}, headers=_h(owner)) - assert r.status_code == 200, r.text - session_kw = [kw for name, kw in calls if "Session" in name][0] - assert session_kw["invoice_creation"]["enabled"] is True - # The org travels onto the invoice too, so one found in the Stripe dashboard resolves to a team. - assert session_kw["invoice_creation"]["invoice_data"]["metadata"]["treg_org_id"] == str(org_id) - assert session_kw["payment_intent_data"]["setup_future_usage"] == "off_session" - - async def test_invoice_events_are_acknowledged_but_never_credit(c: AsyncClient): """`invoice_creation` makes Stripe emit invoice.* for every top-up. Crediting on those as well as on the PaymentIntent would be a second door onto the same money.""" @@ -285,6 +271,7 @@ async def test_portal_returns_a_one_time_url_for_a_paying_org(c: AsyncClient, mo assert "billing_portal" in name.lower() or "Session" in name assert kw["customer"] == "cus_test_1" assert kw["return_url"].endswith("/app#billing") + assert (await c.get("/billing", headers=_h(owner))).json()["portal"] is True async def test_portal_refuses_an_org_with_no_stripe_customer(c: AsyncClient, monkeypatch): @@ -297,19 +284,6 @@ async def test_portal_refuses_an_org_with_no_stripe_customer(c: AsyncClient, mon assert (await c.get("/billing", headers=_h(owner))).json()["portal"] is False -async def test_portal_is_advertised_once_the_org_has_a_customer(c: AsyncClient): - org_id, owner = await _org(c) - await _set_org(org_id, stripe_customer_id="cus_test_1") - assert (await c.get("/billing", headers=_h(owner))).json()["portal"] is True - - -async def test_portal_is_503_when_stripe_is_not_configured(c: AsyncClient, monkeypatch): - org_id, owner = await _org(c) - await _set_org(org_id, stripe_customer_id="cus_test_1") - monkeypatch.setattr(get_settings(), "stripe_secret_key", "", raising=False) - assert (await c.post("/billing/portal", headers=_h(owner))).status_code == 503 - - # ---- payment history ---------------------------------------------------------------------------- def _charge(pi: str, *, invoice: str | None = None, receipt: str = "https://pay.stripe.com/r/1") -> dict: return {"id": f"ch_{pi}", "payment_intent": pi, "receipt_url": receipt, "invoice": invoice} @@ -385,30 +359,6 @@ async def test_history_survives_a_stripe_outage(c: AsyncClient, monkeypatch): assert body["items"][0]["invoice_pdf"] == "" and body["items"][0]["receipt_url"] == "" -async def test_history_never_moves_money(c: AsyncClient, monkeypatch): - org_id, owner = await _org(c) - await _set_org(org_id, stripe_customer_id="cus_test_1") - async with session_maker() as db: - await ledger.topup(db, org_id, 10_000_000, "pi_manual", meta={"source": "stripe"}) - await db.commit() - monkeypatch.setattr(billing, "_sdk", _docs_sdk([_charge("pi_manual", invoice="in_1")], [_invoice("in_1")])) - before = (await c.get(f"/orgs/{org_id}/balance", headers=_h(owner))).json()["balance_micro"] - await c.get("/billing/history", headers=_h(owner)) - after = (await c.get(f"/orgs/{org_id}/balance", headers=_h(owner))).json()["balance_micro"] - assert after == before - - -async def test_history_and_portal_need_admin_of_this_org(c: AsyncClient, monkeypatch): - """A card and an invoice archive are the org's money, not a member's business — the same gate as - the rest of /billing.""" - org_id, owner = await _org(c) - member = await _member(c, org_id, owner, "grunt@superdesign.dev") - monkeypatch.setattr(billing, "_sdk", lambda *a, **k: pytest.fail("must not reach Stripe")) - for method, path in (("GET", "/billing/history"), ("POST", "/billing/portal")): - assert (await c.request(method, path, headers=_h(member))).status_code == 403 - assert (await c.request(method, path)).status_code in (401, 403) - - async def test_history_of_a_team_that_never_paid_is_empty_not_an_error(c: AsyncClient, monkeypatch): org_id, owner = await _org(c) monkeypatch.setattr(billing, "_sdk", lambda *a, **k: pytest.fail("no customer — must not ask Stripe")) @@ -647,27 +597,20 @@ def test_the_idempotency_key_collapses_a_burst_but_not_a_changed_card(): assert billing._idempotency_key(1, 25_000_000, 0, "pm_a") not in burst # a different amount -async def test_auto_topup_refuses_without_recorded_consent(c: AsyncClient, monkeypatch): +@pytest.mark.parametrize("over,reason", [ + (lambda: {"autotopup_consented_at": None}, "no_consent"), + (lambda: {"stripe_default_pm": None}, "no_card"), + # A burst of calls noticing the same low balance must not each fire a charge; the cooldown is + # stamped in the DB, so it also holds across web workers. + (lambda: {"autotopup_last_attempt_at": billing._now() - timedelta(seconds=60)}, "cooldown"), + # Re-read under the lock: a manual top-up (or the winner of a race) may have already funded it. + (lambda: {"balance_micro": 50_000_000}, "above_threshold"), +], ids=["no_consent", "no_card", "cooldown", "above_threshold"]) +async def test_auto_topup_refuses_without_charging(c: AsyncClient, monkeypatch, over, reason): org_id, _ = await _org(c) - await _armed(org_id, autotopup_consented_at=None) + await _armed(org_id, **over()) monkeypatch.setattr(billing, "_sdk", _no_sdk) - assert (await _attempt(org_id))["reason"] == "no_consent" - - -async def test_auto_topup_refuses_without_a_card(c: AsyncClient, monkeypatch): - org_id, _ = await _org(c) - await _armed(org_id, stripe_default_pm=None) - monkeypatch.setattr(billing, "_sdk", _no_sdk) - assert (await _attempt(org_id))["reason"] == "no_card" - - -async def test_auto_topup_respects_the_cooldown(c: AsyncClient, monkeypatch): - """A burst of calls all noticing the same low balance must not each fire a charge; the cooldown is - stamped in the DB, so it also holds across web workers.""" - org_id, _ = await _org(c) - await _armed(org_id, autotopup_last_attempt_at=billing._now() - timedelta(seconds=60)) - monkeypatch.setattr(billing, "_sdk", _no_sdk) - assert (await _attempt(org_id))["reason"] == "cooldown" + assert (await _attempt(org_id))["reason"] == reason async def test_auto_topup_stops_at_the_monthly_cap(c: AsyncClient, monkeypatch): @@ -692,14 +635,6 @@ async def test_monthly_spend_counts_only_automatic_topups(c: AsyncClient, monkey assert await billing.monthly_autotopup_spend(db, org_id) == 10_000_000 -async def test_auto_topup_skips_when_the_balance_recovered(c: AsyncClient, monkeypatch): - """Re-read under the lock: a manual top-up (or the winner of a race) may have already funded it.""" - org_id, _ = await _org(c) - await _armed(org_id, balance_micro=50_000_000) - monkeypatch.setattr(billing, "_sdk", _no_sdk) - assert (await _attempt(org_id))["reason"] == "above_threshold" - - async def test_authentication_required_disables_autotopup_with_a_reason(c: AsyncClient, monkeypatch): """3-D Secure cannot be satisfied off-session, so retrying is pointless — it disables itself and keeps the PaymentIntent so the dashboard can offer an on-session recovery.""" @@ -875,15 +810,6 @@ async def test_analytics_outage_cannot_500_the_webhook(c: AsyncClient, monkeypat analytics._queue.clear() -async def test_no_posthog_key_means_no_events(c: AsyncClient, monkeypatch): - from treg import analytics - org_id, _owner = await _org(c) - monkeypatch.setattr(billing, "_sdk", _no_sdk) - analytics._queue.clear() # default settings: no key - assert (await _deliver(c, _pi_event(org_id, pi="pi_no_key", cents=500))).status_code == 200 - assert analytics._queue == [] - - # ---- Google Ads conversion tracking: first top-up ------------------------------------------------ async def test_first_topup_queues_exactly_one_ad_conversion(c, monkeypatch): """Stripe delivers at least once; a redelivery must not double-count the conversion.""" diff --git a/tests/test_capacity_collectors.py b/tests/test_capacity_collectors.py index 5a6944db..f6965229 100644 --- a/tests/test_capacity_collectors.py +++ b/tests/test_capacity_collectors.py @@ -34,11 +34,6 @@ async def test_fishaudio_balance_uses_workspace_wallet(monkeypatch): collectors.get_settings.cache_clear() assert row["value"] == 99.95754 assert row["unit"] == "USD" - capacity = policy.default_policy("fishaudio", has_key=True) - assert capacity.capacity_type == "cash" - assert capacity.funding_mode == "manual" - assert capacity.source == "api" - assert capacity.rate_limit is None async def test_fishaudio_balance_is_unknown_without_workspace_and_skips_request(monkeypatch): @@ -71,7 +66,7 @@ async def test_fishaudio_balance_rejects_invalid_credit(monkeypatch, credit): assert row["value"] is None -async def test_openmart_balance_collector_and_policy(): +async def test_openmart_balance_collector(): def probe(request): assert request.method == "GET" assert request.url.path == "/api/v2/credit-balance" @@ -89,13 +84,9 @@ async def test_openmart_balance_collector_and_policy(): "unit": "credits", "note": "Monthly subscription balance; current period ends 2026-10-01T00:00:00Z.", } - capacity = policy.default_policy("openmart", has_key=True) - assert capacity.capacity_type == "credits" - assert capacity.funding_mode == "subscription" - assert capacity.rate_limit == {"limit": 15, "window_s": 1, "source": "docs"} -async def test_tavily_capacity_uses_key_credit_remainder_and_conservative_rate(): +async def test_tavily_capacity_uses_key_credit_remainder(): def probe(request): assert request.method == "GET" assert request.url == "https://api.tavily.com/usage" @@ -113,14 +104,9 @@ async def test_tavily_capacity_uses_key_credit_remainder_and_conservative_rate() "unit": "API credits", "note": "key usage 125 of 1000; account pools are informational", } - capacity = policy.default_policy("tavily", has_key=True) - assert capacity.capacity_type == "credits" - assert capacity.funding_mode == "manual" - assert capacity.source == "api" - assert capacity.rate_limit == {"limit": 100, "window_s": 60, "source": "docs"} -async def test_serper_capacity_uses_free_account_balance_and_live_rate(): +async def test_serper_capacity_uses_free_account_balance(): def probe(request): assert request.method == "GET" assert request.url == "https://google.serper.dev/account" @@ -134,12 +120,6 @@ async def test_serper_capacity_uses_free_account_balance_and_live_rate(): "unit": "credits", "note": "account rate limit 50 queries/s", } - capacity = policy.default_policy("serper", has_key=True) - assert capacity.capacity_type == "credits" - assert capacity.funding_mode == "auto_recharge" - assert capacity.auto_funding_enabled is True - assert capacity.source == "api" - assert capacity.rate_limit == {"limit": 50, "window_s": 1, "source": "api"} @pytest.mark.parametrize("balance", [None, True, "bad", "NaN", "Infinity", -1]) @@ -150,7 +130,7 @@ async def test_serper_capacity_rejects_invalid_balance(balance): await collectors._serper(client, "test") -async def test_fetchin_capacity_uses_free_subscription_balance_and_safe_rate(): +async def test_fetchin_capacity_uses_free_subscription_balance(): def probe(request): assert request.method == "GET" assert request.url == "https://api.fetchin.io/api/v1/subscription" @@ -165,11 +145,6 @@ async def test_fetchin_capacity_uses_free_subscription_balance_and_safe_rate(): assert row["value"] == 51_000 assert row["unit"] == "credits" assert "account limit 5 requests/s" in row["note"] - capacity = policy.default_policy("fetchinio", has_key=True) - assert capacity.capacity_type == "credits" - assert capacity.funding_mode == "manual" - assert capacity.source == "api" - assert capacity.rate_limit == {"limit": 2, "window_s": 1, "source": "policy"} @pytest.mark.parametrize("remaining", [None, True, -1, "51000", float("nan")]) @@ -258,15 +233,6 @@ async def test_zerobounce_balance_rejects_uncertain_values_without_exposing_key( collectors.get_settings.cache_clear() -def test_zerobounce_capacity_policy_stays_manual_until_vendor_auto_pay_is_verified(): - row = policy.default_policy("zerobounce", has_key=True) - assert row.capacity_type == "credits" - assert row.funding_mode == "manual" - assert row.source == "api" - assert row.auto_funding_enabled is False - assert row.rate_limit == {"limit": 25, "window_s": 1, "source": "policy"} - - @pytest.mark.parametrize("balance", [0, 465]) async def test_millionverifier_balance_uses_query_key_without_double_counting(monkeypatch, balance): monkeypatch.setenv("TREG_PLATFORM_KEY_MILLIONVERIFIER", "private-test-key") @@ -306,13 +272,17 @@ async def test_bounceban_balance_uses_raw_authorization_header(monkeypatch, bala collectors.get_settings.cache_clear() -@pytest.mark.parametrize("balance", [None, -1, True, "9997"]) +@pytest.mark.parametrize("balance", [None, -1, True, "9997", float("inf")]) async def test_bounceban_balance_rejects_uncertain_values_without_exposing_key(monkeypatch, balance): + def probe(_request): + if balance == float("inf"): + return httpx.Response(200, content=b'{"available_credits": Infinity}') + return httpx.Response(200, json={"available_credits": balance}) + monkeypatch.setenv("TREG_PLATFORM_KEY_BOUNCEBAN", "private-test-key") collectors.get_settings.cache_clear() try: - async with httpx.AsyncClient(transport=httpx.MockTransport( - lambda request: httpx.Response(200, json={"available_credits": balance}))) as client: + async with httpx.AsyncClient(transport=httpx.MockTransport(probe)) as client: row = await collectors.provider_balance("bounceban", client) assert row["value"] is None assert "valid verification-credit balance" in row["note"] @@ -321,38 +291,6 @@ async def test_bounceban_balance_rejects_uncertain_values_without_exposing_key(m collectors.get_settings.cache_clear() -async def test_bounceban_balance_rejects_non_finite_value(monkeypatch): - class Response: - def raise_for_status(self): - return None - - def json(self): - return {"available_credits": float("inf")} - - class Client: - async def get(self, *args, **kwargs): - return Response() - - monkeypatch.setenv("TREG_PLATFORM_KEY_BOUNCEBAN", "private-test-key") - collectors.get_settings.cache_clear() - try: - row = await collectors.provider_balance("bounceban", Client()) - assert row["value"] is None - assert "valid verification-credit balance" in row["note"] - assert "private-test-key" not in str(row) - finally: - collectors.get_settings.cache_clear() - - -def test_bounceban_capacity_policy_uses_manual_prepaid_credits(): - row = policy.default_policy("bounceban", has_key=True) - assert row.capacity_type == "credits" - assert row.funding_mode == "manual" - assert row.source == "api" - assert row.auto_funding_enabled is False - assert row.rate_limit == {"limit": 25, "window_s": 1, "source": "docs"} - - @pytest.mark.parametrize("status,body", [(200, {"error": "apikey_not_found"}), (401, {}), (200, {})]) async def test_millionverifier_balance_errors_do_not_expose_key(monkeypatch, status, body): monkeypatch.setenv("TREG_PLATFORM_KEY_MILLIONVERIFIER", "private-test-key") @@ -412,14 +350,6 @@ async def test_brightdata_collector_parses_balance_and_pending(): assert "pending $12.34" in result["note"] -async def test_brightdata_collector_handles_missing_pending(): - resp = MockResponse({"balance": 100.0, "credit": 0, "prepayment": 0}) - client = MockClient(get_response=resp) - result = await collectors._brightdata(client, "test-key") - assert result["value"] == 100.0 - assert "pending $0.00" in result["note"] - - # ---- crustdata -------------------------------------------------------------------------- async def test_crustdata_collector_parses_credits_and_recurring(): @@ -439,21 +369,6 @@ async def test_crustdata_collector_parses_credits_and_recurring(): assert "2026-09-01" in result["note"] -async def test_crustdata_collector_handles_no_recurring_grant(): - resp = MockResponse({ - "account": { - "credits": 1234.0, - "recurring_credits": None, - "recurring_credits_frequency": None, - "recurring_credits_refresh_date": None - } - }) - client = MockClient(get_response=resp) - result = await collectors._crustdata(client, "test-key") - assert result["value"] == 1234.0 - assert "no recurring grant" in result["note"] - - # ---- akta ------------------------------------------------------------------------------- async def test_akta_collector_parses_credits_and_tier(): @@ -474,46 +389,9 @@ async def test_akta_collector_parses_credits_and_tier(): assert "lifetime 100.0 used" in result["note"] -async def test_akta_collector_marks_enterprise_accounts(): - resp = MockResponse({ - "credit_balance": 50000.0, - "package_type": "scale", - "is_enterprise": True, - "lifetime_consumed_credits": 0 - }) - client = MockClient(get_response=resp) - result = await collectors._akta(client, "test-key") - assert result["value"] == 50000.0 - assert "(enterprise)" in result["note"] - - # ---- NO_BALANCE_API / BALANCE_ROUTES registration --------------------------------------- -def test_no_balance_api_includes_expected_providers(): - """Verify the vendors that have no free balance API are documented.""" - expected = { - "adyntel", "aviato", "coresignal", "exa", "financialdatasets", "finnhub", - "justoneapi", "keenable", "limadata", "marketstack", "scrubby", "tiingo", "trestleiq", - } - assert expected == set(collectors.NO_BALANCE_API.keys()) - - -async def test_keenable_capacity_is_portal_only_with_documented_rate_limit(monkeypatch): - monkeypatch.setenv("TREG_PLATFORM_KEY_KEENABLE", "test") - collectors.get_settings.cache_clear() - try: - row = await collectors.provider_balance("keenable") - assert row["value"] is None and row["no_api"] is True - capacity = policy.default_policy("keenable", has_key=True) - assert capacity.capacity_type == "requests" - assert capacity.funding_mode == "manual" - assert capacity.source == "manual" - assert capacity.rate_limit == {"limit": 10, "window_s": 1, "source": "docs"} - finally: - collectors.get_settings.cache_clear() - - -async def test_olostep_balance_and_conservative_shared_key_rate(monkeypatch): +async def test_olostep_balance(monkeypatch): monkeypatch.setenv("TREG_PLATFORM_KEY_OLOSTEP", "test-key") collectors.get_settings.cache_clear() try: @@ -535,16 +413,11 @@ async def test_olostep_balance_and_conservative_shared_key_rate(monkeypatch): "unit": "credits", "note": "plan Free; usage allowed", } - capacity = policy.default_policy("olostep", has_key=True) - assert capacity.capacity_type == "credits" - assert capacity.funding_mode == "manual" - assert capacity.source == "api" - assert capacity.rate_limit == {"limit": 5, "window_s": 1, "source": "policy"} finally: collectors.get_settings.cache_clear() -async def test_scrapegraphai_balance_and_shared_key_policy(monkeypatch): +async def test_scrapegraphai_balance(monkeypatch): monkeypatch.setenv("TREG_PLATFORM_KEY_SCRAPEGRAPHAI", "test-key") collectors.get_settings.cache_clear() try: @@ -570,11 +443,6 @@ async def test_scrapegraphai_balance_and_shared_key_policy(monkeypatch): "unit": "credits", "note": "plan Free Plan; used 25; crawl jobs 0/1; monitors 0/1", } - capacity = policy.default_policy("scrapegraphai", has_key=True) - assert capacity.capacity_type == "credits" - assert capacity.funding_mode == "subscription" - assert capacity.source == "api" - assert capacity.rate_limit == {"limit": 500, "window_s": 60, "source": "policy"} finally: collectors.get_settings.cache_clear() @@ -591,48 +459,6 @@ async def test_scrapegraphai_balance_rejects_uncertain_values(remaining): await collectors._scrapegraphai(client, "test") -async def test_adyntel_capacity_is_dashboard_only_and_rate_limited(monkeypatch): - monkeypatch.setenv("TREG_PLATFORM_KEY_ADYNTEL", "PLATFORM-ADYNTEL") - collectors.get_settings.cache_clear() - try: - row = await collectors.provider_balance("adyntel") - assert row["no_api"] is True and row["value"] is None - assert "dashboard only" in row["note"] - capacity = policy.default_policy("adyntel", has_key=True) - assert capacity.capacity_type == "credits" - assert capacity.funding_mode == "manual" - assert capacity.auto_funding_enabled is False - assert capacity.source == "manual" - assert capacity.rate_limit == {"limit": 5, "window_s": 1, "source": "docs"} - finally: - collectors.get_settings.cache_clear() - - -def test_limadata_policy_uses_auto_recharge_and_the_documented_rate(): - row = policy.default_policy("limadata", has_key=True) - assert row.capacity_type == "credits" - assert row.funding_mode == "auto_recharge" - assert row.auto_funding_enabled is True - assert row.source == "manual" - assert row.rate_limit == {"limit": 1, "window_s": 1, "source": "docs"} - - -async def test_trestleiq_capacity_is_portal_only_with_manually_verified_auto_recharge(monkeypatch): - monkeypatch.setenv("TREG_PLATFORM_KEY_TRESTLEIQ", "PLATFORM-TRESTLEIQ") - collectors.get_settings.cache_clear() - try: - row = await collectors.provider_balance("trestleiq") - assert row["no_api"] is True and row["value"] is None - capacity = policy.default_policy("trestleiq", has_key=True) - assert capacity.capacity_type == "cash" - assert capacity.funding_mode == "auto_recharge" - assert capacity.auto_funding_enabled is True - assert capacity.source == "manual" - assert capacity.rate_limit == {"limit": 10, "window_s": 1, "source": "docs"} - finally: - collectors.get_settings.cache_clear() - - def test_implemented_collectors_are_registered_and_do_not_overlap_absent_list(): """A collector that parses a vendor must be on BALANCE_ROUTES, and a provider cannot be both 'we collect' and 'there is no balance API'.""" @@ -713,14 +539,6 @@ async def test_aiark_balance_collector_uses_total(remaining, expected): assert "roll over" in row["note"] -def test_aiark_policy_uses_subscription_and_documented_rate(): - row = policy.default_policy("aiark", has_key=True) - assert row.capacity_type == "monthly_quota" - assert row.funding_mode == "quota_reset" - assert row.auto_funding_enabled is False - assert row.rate_limit == {"limit": 5, "window_s": 1, "source": "docs"} - - @pytest.mark.parametrize("remaining,expected", [(997, 997), (0, 0), (-1, None), (True, None)]) async def test_getleadsio_balance_collector_uses_fair_use_credits(remaining, expected): def serve(request): @@ -739,16 +557,6 @@ async def test_getleadsio_balance_collector_uses_fair_use_credits(remaining, exp assert "Live Leads wallet is not included" in row["note"] -def test_getleadsio_policy_uses_the_documented_default_rate(): - row = policy.default_policy("getleadsio", has_key=True) - assert row.rate_limit == {"limit": 100, "window_s": 60, "source": "docs"} - - -def test_prospeo_policy_smooths_at_the_stricter_shared_key_rate(): - row = policy.default_policy("prospeo", has_key=True) - assert row.rate_limit == {"limit": 1, "window_s": 1, "source": "docs"} - - @pytest.mark.parametrize('remaining,expected', [(300, 300), (0, 0), (None, None), (-1, None), ('unlimited', None), (True, None)]) async def test_quickenrich_subscription_allowance_from_free_discovery(remaining, expected): def reply(request): diff --git a/tests/test_capacity_know.py b/tests/test_capacity_know.py index 4652501a..08a5d13d 100644 --- a/tests/test_capacity_know.py +++ b/tests/test_capacity_know.py @@ -6,7 +6,6 @@ from __future__ import annotations from datetime import timedelta import httpx -import pytest from sqlalchemy import select from treg import ratestore @@ -44,8 +43,6 @@ async def test_import_creates_one_policy_per_account_and_flags_unknowns_without_ assert rows["crustdata"].capacity_type == "unknown" and "crustdata" in unknown assert rows["findymail"].enabled and not rows["dataforseo"].enabled # enabled ⇔ a key exists assert rows["overflow:orthogonal"].source == "manual" - assert rows["hunter"].quota["period"] == "billing" - assert rows["leadsforge"].rate_limit == {"limit": 120, "window_s": 60, "source": "headers"} # a second import is a no-op async with session_maker() as db: assert await ensure_policies(db, has_key=lambda p: False) == [] @@ -130,42 +127,3 @@ def test_capacity_view_getters_are_sync_and_io_free(): import inspect assert not inspect.iscoroutinefunction(LatestStateView.get) assert not inspect.iscoroutinefunction(LatestStateView.is_exhausted) - - -def test_worker_cli_parses_the_sweep_command(monkeypatch): - from treg import worker - seen = {} - - async def fake(args): - seen["only"] = args.only - return 0 - - monkeypatch.setattr(worker, "_capacity_sweep", fake) - assert worker.main(["capacity", "sweep", "--only", "hunter,lusha"]) == 0 - assert seen["only"] == "hunter,lusha" - with pytest.raises(SystemExit): - worker.main(["capacity"]) - - -@pytest.mark.parametrize('remaining,health', [(300, 'ok'), (0, 'exhausted'), (None, 'stale')]) -def test_quickenrich_capacity_uses_reported_allowance(remaining, health): - from treg.domain.capacity.sweep import snapshot_from - from treg.domain.capacity.policy import default_policy, latest_state - from treg.timeutil import utcnow_naive - now = utcnow_naive() - policy = default_policy('quickenrich', has_key=True) - assert policy.capacity_type == 'monthly_quota' - assert policy.funding_mode == 'quota_reset' - assert not policy.auto_funding_enabled - snap = snapshot_from('quickenrich', {'value': remaining, 'unit': 'credits'}, observed_at=now) - state = latest_state(policy, snap, now) - assert state.health == health - assert state.is_exhausted(now) == (remaining == 0) - - -@pytest.mark.parametrize("balance,exhausted", [(0, True), (9.992, False)]) -def test_trykitt_paid_balance_uses_common_exhaustion_rule(balance, exhausted): - state = latest_state(default_policy("trykitt", has_key=True), CapacitySnapshot( - provider="trykitt", remaining=balance, unit="USD", - observed_at=utcnow_naive(), confidence="exact")) - assert state.is_exhausted() is exhausted diff --git a/tests/test_capacity_overflow.py b/tests/test_capacity_overflow.py index a005e28e..8ce7d095 100644 --- a/tests/test_capacity_overflow.py +++ b/tests/test_capacity_overflow.py @@ -11,15 +11,11 @@ import pytest from httpx import AsyncClient from sqlmodel import select -from dataclasses import replace -from datetime import datetime, timedelta -from treg.domain.capacity import marks, policy, signatures +from datetime import timedelta from treg.domain.capacity import routes as R -from treg.domain.catalog import store from treg import audit, ratestore from treg.application.call import overflow as O from treg.application.call import service as call_service -from treg.application.call.types import UpstreamResponse from treg.config import get_settings from treg.infra.db import session_maker from treg.domain.capacity.policy import LatestState @@ -454,17 +450,6 @@ async def test_org_opt_out_is_honoured_before_any_aggregator_is_contacted(client assert r.status_code == 200 and r.headers["X-Treg-Served-Via"] == "overflow:orthogonal" and len(seen) == 1 -def test_cli_org_overflow_parses(monkeypatch): - from treg import cli - seen = {} - monkeypatch.setattr(cli, "cmd_org_overflow", lambda args, cfg: seen.update(vars(args))) - parser = cli.build_parser() if hasattr(cli, "build_parser") else None - if parser is None: - pytest.skip("no exposed parser builder") - args = parser.parse_args(["org", "overflow", "off"]) - assert args.state == "off" and args.fn is not None - - async def test_child_with_no_reported_cost_settles_at_the_aggregator_reserve( clients: AsyncClient, overflow_on, monkeypatch, ): @@ -575,40 +560,42 @@ async def test_aggregator_relaying_the_vendors_own_out_of_credits_dialect_is_the assert r2.status_code == 422 and again == [] and "X-Treg-Served-Via" not in r2.headers -async def test_a_malformed_relay_marks_the_aggregator_for_that_vendor_only( - clients: AsyncClient, overflow_on, monkeypatch): - """2026-09-17: Orthogonal's Apollo relay timed out ("timeout of 30000ms exceeded", a 5xx with no - vendor status) and the `malformed` strike took overflow:orthogonal offline for EVERY provider - for 15 minutes. Only the aggregator's own key or account is out for everyone.""" - await _route(endpoint_id=APOLLO_SEARCH, provider="apollo", method="POST", path=APOLLO_SEARCH_PATH, price_micro=10_000, ratio=0.38) - monkeypatch.setattr(call_service, "relay", _fake_relay(422, APOLLO_OUT_OF_CREDITS)) - monkeypatch.setattr(O, "_send", _orthogonal([(500, {"success": False, "error": "timeout of 30000ms exceeded"})], [])) - before = await _balance(clients) - r = await clients.post(f"/call/{APOLLO_SEARCH}", json={"q_organization_name": "x"}) - assert r.status_code == 503 and await _balance(clients) == before and await _holds() == [] - async with session_maker() as db: - whole = await ratestore.kv_get(db, LOCK_NS, "overflow:orthogonal") - mine = Lock.from_json(await ratestore.kv_get(db, LOCK_NS, "overflow:orthogonal:apollo")) - assert whole is None and mine.is_active(), "one vendor's broken relay is not the aggregator's outage" +APOLLO_ROUTE = dict(endpoint_id=APOLLO_SEARCH, provider="apollo", method="POST", path=APOLLO_SEARCH_PATH, + price_micro=10_000, ratio=0.38) -async def test_a_vendors_period_quota_through_the_aggregator_is_that_vendors_answer_not_an_aggregator_outage( - clients: AsyncClient, overflow_on, monkeypatch): - """Apollo's daily cap on Orthogonal's Apollo account: the aggregator is dry for APOLLO. The - child is released and Apollo skips Orthogonal for a while; hunter, lusha and everyone else - keep overflowing through it.""" - await _route(endpoint_id=APOLLO_SEARCH, provider="apollo", method="POST", path=APOLLO_SEARCH_PATH, price_micro=10_000, ratio=0.38) - monkeypatch.setattr(call_service, "relay", _fake_relay(422, APOLLO_OUT_OF_CREDITS)) - relayed = {"success": False, "error": "Upstream returned status 429", - "data": {"error": "You have exceeded the rate limit per day"}, "priceCents": 0} - monkeypatch.setattr(O, "_send", _orthogonal([(429, relayed)], [])) +@pytest.mark.parametrize("route,vendor,answer", [ + # Orthogonal's Apollo relay timed out ("timeout of 30000ms exceeded", a 5xx with no vendor + # status) and the `malformed` strike used to take overflow:orthogonal offline for EVERY provider. + (APOLLO_ROUTE, (422, APOLLO_OUT_OF_CREDITS), + (500, {"success": False, "error": "timeout of 30000ms exceeded"})), + # Apollo's daily cap on Orthogonal's Apollo account: the aggregator is dry for APOLLO only; + # hunter, lusha and everyone else keep overflowing through it. + (APOLLO_ROUTE, (422, APOLLO_OUT_OF_CREDITS), + (429, {"success": False, "error": "Upstream returned status 429", + "data": {"error": "You have exceeded the rate limit per day"}, "priceCents": 0})), + # A plain 5xx on one vendor's relay. + ({"price_micro": 3_000}, (402, b'{"detail":"nope"}'), + (503, {"success": False, "error": "upstream gateway timeout"})), +], ids=["malformed-relay", "vendor-period-quota", "orthogonal-5xx"]) +async def test_a_failed_relay_marks_the_aggregator_for_that_vendor_only( + clients: AsyncClient, overflow_on, monkeypatch, route, vendor, answer): + """Only the aggregator's own key or account is out for everyone: the child is released, the + caller is not charged, and only this vendor skips the aggregator for a while.""" + await _route(**route) + provider = route.get("provider", "tikhub") + monkeypatch.setattr(call_service, "relay", _fake_relay(*vendor)) + monkeypatch.setattr(O, "_send", _orthogonal([answer], [])) before = await _balance(clients) - r = await clients.post(f"/call/{APOLLO_SEARCH}", json={"q_organization_name": "x"}) - assert r.status_code == 503 and await _balance(clients) == before + if provider == "apollo": + r = await clients.post(f"/call/{APOLLO_SEARCH}", json={"q_organization_name": "x"}) + else: + r = await clients.get(f"/call/{EP}?aweme_id=7") + assert r.status_code == 503 and r.json()["detail"]["error"] == "provider_capacity_unavailable" + assert await _balance(clients) == before and await _holds() == [] async with session_maker() as db: - whole = await ratestore.kv_get(db, LOCK_NS, "overflow:orthogonal") - mine = Lock.from_json(await ratestore.kv_get(db, LOCK_NS, "overflow:orthogonal:apollo")) - assert whole is None and mine.is_active(), "one vendor's quota is not the aggregator's outage" + mine = Lock.from_json(await ratestore.kv_get(db, LOCK_NS, f"overflow:orthogonal:{provider}")) + assert await _orthogonal_lock() is None and mine.is_active(), "one vendor's failure is not the aggregator's outage" async def test_apollo_validation_422_is_the_callers_and_never_overflows(clients: AsyncClient, overflow_on, monkeypatch): @@ -631,58 +618,51 @@ async def _exhausted(provider: str = "tikhub") -> None: capacity_view.invalidate() -async def test_a_call_rescued_by_overflow_is_one_ok_event_not_a_refusal(clients: AsyncClient, overflow_on, monkeypatch, posthog_events): - """Skip-direct: the resolver knew the account was dry and went straight to the aggregator. The - caller got a 200; the dashboard used to see a 503 refused_by=capacity and nothing else.""" +OK_ANSWER = (200, {"success": True, "data": VENDOR_BODY, "priceCents": 0.3}) +CONTRACT_REFUSAL = (400, {"success": False, "error": "x", + "_orthogonal": {"error": "orthogonal_endpoint_contract", "message": "missing"}}) + + +@pytest.mark.parametrize("skip_direct,vendor_body,answer,status,props", [ + # Skip-direct: the resolver knew the account was dry and went straight to the aggregator. The + # caller got a 200; the dashboard used to see a 503 refused_by=capacity and nothing else. + (True, None, OK_ANSWER, 200, + {"outcome": "ok", "refused_by": None, "tier": "platform-overflow", + "served_via": "overflow:orthogonal", "charged_micro": 3_000}), + (True, None, (402, {"success": False, "error": "insufficient balance"}), 503, + {"outcome": "treg_refused", "refused_by": "capacity"}), + # The post-failure path: the vendor answered 402, the child served. One event, the child's, + # and the strike is still on it. + (False, b'{"detail":"Insufficient balance"}', OK_ANSWER, 200, + {"outcome": "ok", "tier": "platform-overflow", "served_via": "overflow:orthogonal", + "capacity_signal": "balance"}), + (False, b'{"detail":"nope"}', CONTRACT_REFUSAL, 402, {"outcome": "vendor_error", "tier": "platform"}), +], ids=["skip-direct-rescued", "skip-direct-not-rescued", "vendor-402-rescued", "vendor-402-not-rescued"]) +async def test_an_overflow_attempt_is_one_event_saying_what_the_caller_got( + clients: AsyncClient, overflow_on, monkeypatch, posthog_events, + skip_direct, vendor_body, answer, status, props, +): await _route(price_micro=3_000) - await _exhausted() - monkeypatch.setattr(O, "_send", _orthogonal([(200, {"success": True, "data": VENDOR_BODY, "priceCents": 0.3})], [])) + if skip_direct: + await _exhausted() + else: + monkeypatch.setattr(call_service, "relay", _fake_relay(402, vendor_body)) + monkeypatch.setattr(O, "_send", _orthogonal([answer], [])) r = await clients.get(f"/call/{EP}?aweme_id=7") - assert r.status_code == 200 and r.headers["X-Treg-Served-Via"] == "overflow:orthogonal" + assert r.status_code == status (e,) = await posthog_events() p = e["properties"] - assert p["status_code"] == 200 and p["outcome"] == "ok" and p["refused_by"] is None - assert p["tier"] == "platform-overflow" and p["served_via"] == "overflow:orthogonal" - assert p["charged_micro"] == 3_000 and p["call_ref"] == r.headers["X-Treg-Call-Id"] - await audit.drain() - rows = [x for x in (await clients.get("/calls")).json() if x["tool_name"] == EP] - assert {x.get("credential_tier") for x in rows} == {"platform", "platform-overflow"}, "both DB rows stay" - - -async def test_a_call_overflow_could_not_rescue_is_still_one_refusal_event(clients: AsyncClient, overflow_on, monkeypatch, posthog_events): - await _route(price_micro=3_000) - await _exhausted() - monkeypatch.setattr(O, "_send", _orthogonal([(402, {"success": False, "error": "insufficient balance"})], [])) - r = await clients.get(f"/call/{EP}?aweme_id=7") - assert r.status_code == 503 - (e,) = await posthog_events() - p = e["properties"] - assert p["status_code"] == 503 and p["outcome"] == "treg_refused" and p["refused_by"] == "capacity" - assert "served_via" not in p - - -async def test_a_vendor_402_rescued_by_overflow_is_one_ok_event(clients: AsyncClient, overflow_on, monkeypatch, posthog_events): - """The post-failure path: the vendor answered 402, the child served. One event, the child's.""" - await _route(price_micro=3_000) - monkeypatch.setattr(call_service, "relay", _fake_relay(402, b'{"detail":"Insufficient balance"}')) - monkeypatch.setattr(O, "_send", _orthogonal([(200, {"success": True, "data": VENDOR_BODY, "priceCents": 0.3})], [])) - r = await clients.get(f"/call/{EP}?aweme_id=7") - assert r.status_code == 200 - (e,) = await posthog_events() - p = e["properties"] - assert p["status_code"] == 200 and p["outcome"] == "ok" and p["tier"] == "platform-overflow" - assert p["served_via"] == "overflow:orthogonal" and p["capacity_signal"] == "balance", "the strike is still on the event" - - -async def test_a_vendor_402_overflow_did_not_rescue_keeps_the_vendor_error_event(clients: AsyncClient, overflow_on, monkeypatch, posthog_events): - await _route(price_micro=3_000) - monkeypatch.setattr(call_service, "relay", _fake_relay(402, b'{"detail":"nope"}')) - monkeypatch.setattr(O, "_send", _orthogonal([(400, {"success": False, "error": "x", "_orthogonal": {"error": "orthogonal_endpoint_contract", "message": "missing"}})], [])) - r = await clients.get(f"/call/{EP}?aweme_id=7") - assert r.status_code == 402 - (e,) = await posthog_events() - p = e["properties"] - assert p["status_code"] == 402 and p["outcome"] == "vendor_error" and p["tier"] == "platform" and "served_via" not in p + assert p["status_code"] == status + assert {k: p[k] for k in props} == props + if "served_via" in props: + assert r.headers["X-Treg-Served-Via"] == props["served_via"] + else: + assert "served_via" not in p + if skip_direct and status == 200: + assert p["call_ref"] == r.headers["X-Treg-Call-Id"] + await audit.drain() + rows = [x for x in (await clients.get("/calls")).json() if x["tool_name"] == EP] + assert {x.get("credential_tier") for x in rows} == {"platform", "platform-overflow"}, "both DB rows stay" # ---- the weekly verify: renewals are held to their own cap and the run to a budget ------------- @@ -766,76 +746,6 @@ def contactout_on(monkeypatch, overflow_on): get_settings.cache_clear() -def test_contactout_credit_exhaustion_is_endpoint_scoped_and_access_refusal_is_not_capacity(): - signal = signatures.classify('contactout', 403, {}, CONTACTOUT_QUOTA) - assert signal.kind == 'quota' and signatures.is_exhausting(signal) - assert marks.lock_key('contactout', CONTACTOUT_EP, signal.kind) == CONTACTOUT_EP - assert signatures.classify('contactout', 403, {}, b'{"message":"No access to endpoint"}') is None - assert signatures.classify('contactout', 429, httpx.Headers({'Retry-After':'2'}), b'').kind == 'burst' - assert policy.default_policy('contactout', has_key=True).overflow_allowed - - -@pytest.mark.parametrize('status,body', [(403,CONTACTOUT_QUOTA),(429,b'{"message":"Rate limit reached"}')]) -async def test_contactout_capacity_failure_uses_existing_child_billing(clients, contactout_on, monkeypatch, status, body): - await _route(endpoint_id=CONTACTOUT_EP, provider='contactout', path=CONTACTOUT_PATH, price_micro=550000) - fake = _fake_relay(status, body) - async def relay(*args, **kwargs): - response = await fake(*args, **kwargs) - return replace(response, raw_headers=((b'retry-after', b'0'),)) if status == 429 else response - monkeypatch.setattr(call_service, 'relay', relay) - seen=[] - monkeypatch.setattr(O, '_send', _orthogonal([(200,{'success':True,'data':CONTACTOUT_BODY,'priceCents':55})],seen)) - before=await _balance(clients) - r=await clients.get('/call/'+CONTACTOUT_EP,params=CONTACTOUT_QUERY) - assert r.status_code==200 - assert r.json()==CONTACTOUT_BODY - assert r.headers['X-Treg-Served-Via']=='overflow:orthogonal' - assert before-await _balance(clients)==550000 - assert not await _holds() - assert seen[0].json['query']['email_type']=='work' - - -@pytest.mark.parametrize('own,optout,status,body', [ - (True,False,403,CONTACTOUT_QUOTA), (False,True,403,CONTACTOUT_QUOTA), - (False,False,404,b'{}'), (False,False,403,b'{"message":"No access to endpoint"}')]) -async def test_contactout_byok_optout_and_noncapacity_errors_never_overflow(clients,contactout_on,monkeypatch,own,optout,status,body): - await _route(endpoint_id=CONTACTOUT_EP,provider='contactout',path=CONTACTOUT_PATH,price_micro=550000) - if own: - await clients.post('/secrets',json={'name':'contactout','value':'OWN-TEST'}) - if optout: - org=(await clients.get('/orgs')).json()[0]['org_id'] - await clients.patch(f'/orgs/{org}/settings',json={'platform_overflow':False}) - monkeypatch.setattr(call_service,'relay',_fake_relay(status,body)) - seen=[] - monkeypatch.setattr(O,'_send',_orthogonal([],seen)) - r=await clients.get('/call/'+CONTACTOUT_EP,params=CONTACTOUT_QUERY) - assert r.status_code==status and seen==[] - assert not await _holds() - - -def test_contactout_only_verified_compatible_candidates_enable(): - enabled=[] - cat=store.load() - for row in R.load_seed(): - if row['provider']!='contactout': continue - ep=cat.by_id[row['endpoint_id']];cv=cat.cost_view(ep['cost'],'contactout') - route=OverflowRoute(**{k:row[k] for k in ('endpoint_id','aggregator','provider','method','path','agg_slug','agg_path','agg_unit')},agg_price_micro=round(row['agg_price_usd']*1e6),ratio=R.price_ratio(row['agg_price_usd'],R.our_event_usd(cv)),last_verified_at=datetime.fromisoformat(row['verified_at']) if row['verified_at'] else None) - verdict=R.eligible( - route, our_cost=ep['cost'], platform_eligible=True, policy=None, - our_usd=cv['usd'], now=route.last_verified_at, - ) - if verdict.enabled: - enabled.append((row['endpoint_id'],row['aggregator'])) - assert row['verified_at'] - expired=R.eligible(route,our_cost=ep['cost'],platform_eligible=True,policy=None,our_usd=cv['usd'],now=route.last_verified_at+timedelta(days=8)) - assert not expired.enabled - assert len(enabled)==13 - assert (CONTACTOUT_EP,'orthogonal') in enabled - assert (CONTACTOUT_EP,'monid') not in enabled - assert ('contactout.companies.enrich','orthogonal') not in enabled - - - @pytest.mark.parametrize('budget,expected_calls', [(0.5,0),(1.0,1)]) async def test_contactout_renewal_budget_includes_direct_cost_and_uses_ephemeral_profile(monkeypatch,budget,expected_calls): import runpy @@ -968,19 +878,6 @@ async def test_orthogonals_own_422_on_the_skip_direct_ladder_is_a_typed_503_nami assert len(seen) == 1 -async def test_orthogonal_5xx_still_marks_the_aggregator_unhealthy_for_that_vendor(clients: AsyncClient, overflow_on, monkeypatch): - await _route(price_micro=3_000) - monkeypatch.setattr(call_service, "relay", _fake_relay(402, b'{"detail":"nope"}')) - monkeypatch.setattr(O, "_send", _orthogonal([(503, {"success": False, "error": "upstream gateway timeout"})], [])) - r = await clients.get(f"/call/{EP}?aweme_id=7") - assert r.status_code == 503 and r.json()["detail"]["error"] == "provider_capacity_unavailable" - assert await _orthogonal_lock() is None, "a 5xx on one vendor's relay is not the aggregator's outage (2026-09-17)" - async with session_maker() as db: - mine = Lock.from_json(await ratestore.kv_get(db, LOCK_NS, "overflow:orthogonal:tikhub")) - assert mine.is_active() - assert await _holds() == [] - - # ---- the catalog discloses what a relayed call bills ------------------------------------------- async def test_catalog_get_discloses_the_overflow_price_of_a_free_endpoint(clients: AsyncClient, overflow_on): diff --git a/tests/test_capacity_overflow_routes.py b/tests/test_capacity_overflow_routes.py index 28c8f7ab..b49d0a9b 100644 --- a/tests/test_capacity_overflow_routes.py +++ b/tests/test_capacity_overflow_routes.py @@ -20,7 +20,6 @@ from treg.domain.catalog import store as catalog_store from treg.infra.upstream.aggregators import by_name, monid, orthogonal from treg.models import CapacityPolicy, OverflowRoute from treg.timeutil import utcnow_naive -from treg import worker FIX = Path(__file__).parent / "fixtures" / "aggregators" @@ -100,7 +99,6 @@ async def test_sync_reproduces_the_verified_set_and_never_enables_a_bad_ratio(mo seed = [{**x, "verified_at": None} if x["provider"] in ("influencersclub", "contactout") else x for x in R.load_seed()] verified = {(x["endpoint_id"], x["aggregator"]) for x in seed if x["verified_at"]} - assert len(verified) == 145, "the 2026-08-26 verified set (131 ROUTE + 11 tomba + 2 phone + hunter domain-search)" # Freeze "now" at the mapping date so the seed's stamps are within the 7-day window. now = R._dt("2026-08-27T00:00:00") cat = catalog_store.load() @@ -118,18 +116,12 @@ async def test_sync_reproduces_the_verified_set_and_never_enables_a_bad_ratio(mo assert all((by[k].ratio is not None and by[k].ratio <= R.MAX_RATIO) or (by[k].ratio is None and by[k].agg_price_micro <= R.FREE_ROUTE_MAX_USD * 1_000_000) for k in on) assert not any(k[0].startswith(("scrapecreators.", "tikhub.")) for k in on) - assert ("hunter.companies.emails", "orthogonal") in on, "§10 correction: per-10 credit compares as a call" - assert ("findymail.search.business-profile", "orthogonal") in on # ratio 3.54 ≤ 4, the #1 402 source + assert ("findymail.search.business-profile", "orthogonal") in on # the removal check below needs it on # What was verified but is NOT on, and why — every reason is one the rule names. off = {k: by[k].disabled_reason for k in verified - on} allowed = ("ratio ", "unit mismatch", "endpoint not platform-eligible", "policy for scrapecreators disallows overflow", "no price on one side", "free for us") assert all(r.startswith(allowed) for r in off.values()), off - # Recorded 2026-08-28: 113 on. The 32 verified-but-off are the per-result-vs-per-call unit - # question (23, plan §7), not platform-eligible (7), a $0.50 aggregator price on a free route, - # a 56× ratio, scrapecreators policy, and rows with no aggregator price. - assert len(on) == 113, (len(on), sorted(off.items())) - assert ("tomba.companies.emails.count", "orthogonal") in on # free for us, 1¢ there, 155 402s/30d # a route with ratio 6.5 in the seed never enables, and a re-sync without it disables it bad = {**seed[0], "endpoint_id": "findymail.search.business-profile", "aggregator": "monid", "agg_price_usd": 0.0198 * 6.5, "agg_unit": "call", "verified_at": "2026-08-26"} @@ -208,33 +200,6 @@ def test_every_recorded_phrase_arms_the_tripwire(): assert sig is not None and sig.kind == "unrecorded", f"{provider}'s phrase {pattern!r} does not arm the tripwire" -def test_moz_spent_row_quota_is_a_quota_mark(): - """Moz answers a spent period allowance with 403 {"issue": "insufficient-quota"} — 115 of one - org's calls went upstream to a dead key on 2026-09-04 because no row matched a 403. It is a - `quota` exhaustion (resets on Moz's billing day, which the body does not name → default lock); - Moz's caller-fault 4xx stay None.""" - body = (b'{"error":"The account does not have enough quota remaining for current period.",' - b'"data":{"explanation":"account does not have sufficient quota","issue":"insufficient-quota"}}') - sig = S.classify("moz", 403, None, body) - assert sig is not None and sig.kind == "quota" and sig.resets_at is None - assert S.classify("moz", 400, None, b'{"error":"target is required"}') is None - assert S.classify("moz", 403, None, b'{"error":"forbidden"}') is None - - -def test_tavily_documents_separate_plan_and_paygo_quota_statuses(): - plan = S.classify( - "tavily", 432, None, - b'{"detail":{"error":"This request exceeds your plan\'s set usage limit."}}', - ) - paygo = S.classify( - "tavily", 433, None, - b'{"detail":{"error":"This request exceeds the pay-as-you-go limit."}}', - ) - assert plan is not None and plan.kind == "quota" and S.is_exhausting(plan) - assert paygo is not None and paygo.kind == "quota" and S.is_exhausting(paygo) - assert S.classify("tavily", 432, None, b'{"detail":{"error":"bad query"}}') is None - - def test_an_unrecorded_vendor_phrase_is_a_tripwire_never_a_mark(): """The next Apollo: a 4xx no row matched whose body still names credits/quota/balance. It is logged and counted (`capacity_signal=unrecorded`) and does nothing else.""" @@ -544,29 +509,6 @@ async def test_verify_route_marks_same_shape_and_polls_async_runs(): assert res.failure == "pending" -def test_worker_cli_parses_overflow_commands(monkeypatch): - seen = {} - async def fake(args): - seen.update(vars(args)); return 0 - monkeypatch.setattr(worker, "_overflow_sync", fake) - monkeypatch.setattr(worker, "_overflow_verify", fake) - assert worker.main(["overflow", "sync", "--live"]) == 0 and seen["live"] is True - assert worker.main(["overflow", "verify", "--max-usd", "0.05"]) == 0 and seen["max_usd"] == 0.05 - assert seen["renew_max_usd"] == worker.RENEW_MAX_USD and seen["budget_usd"] == worker.VERIFY_BUDGET_USD - assert worker.main(["overflow", "verify", "--renew-max-usd", "0.7", "--budget-usd", "3"]) == 0 - assert seen["renew_max_usd"] == 0.7 and seen["budget_usd"] == 3.0 - - -def test_trykitt_throttle_is_not_exhaustion(): - s=S.classify('trykitt',418,body=json.dumps({'message': 'temporarily throttled', 'response_code': 418})) - assert s.kind=='burst' and not S.is_exhausting(s) - assert S.classify('trykitt',402,body='rate limit').kind=='unknown' - assert S.classify('trykitt',402,body='insufficient funds').kind=='balance' - - assert S.classify("trykitt", 418, headers={"retry-after": "5"}, body="temporarily throttled").retry_after_s == 5 - - - def test_pdl_operation_allowance_does_not_lock_other_pdl_products(): from treg.domain.capacity.marks import lock_key body = b'{"status":402,"error":{"message":"You have hit your account maximum for person_identify (all matches used)"}}' diff --git a/tests/test_capacity_protect.py b/tests/test_capacity_protect.py index d29094e7..d12e8aea 100644 --- a/tests/test_capacity_protect.py +++ b/tests/test_capacity_protect.py @@ -15,7 +15,7 @@ from treg import archive, audit, ratestore from treg.config import get_settings from treg.application.call import service as call_service from treg.application.call import settle as call_settle -from treg.application.call.types import CallFailure, UpstreamResponse +from treg.application.call.types import UpstreamResponse from treg.infra.db import session_maker from treg.domain.capacity import marks as capacity_marks from treg.domain.capacity.marks import DEFAULT_LOCK, LOCK_NS, MAX_LOCK, Lock @@ -26,7 +26,7 @@ from treg.models import Hold, LedgerEntry from treg.timeutil import utcnow_naive from test_marketplace_call import ( # noqa: F401 - EP, EP_MICRO, PLATFORM_KEYS, _balance, _fake_relay, getleadsio_trial_on, platform_on, + EP, EP_MICRO, PLATFORM_KEYS, _balance, _fake_relay, platform_on, ) OUT = b'{"detail":"Insufficient balance"}' # matches the bare-402 balance signature @@ -102,27 +102,6 @@ async def test_own_key_is_never_affected_by_an_exhausted_platform_account(client assert r.json()["auth"] == "Bearer MKKEY" -async def test_getleadsio_zero_capacity_is_clear_and_byok_still_works( - clients: AsyncClient, getleadsio_trial_on): - await _publish("getleadsio", exhausted=True) - before = await _balance(clients) - body = {"filters": {"domains": ["example.com"]}, "limit": 1} - - refused = await clients.post("/call/getleadsio.people.search", json=body) - assert refused.status_code == 503, refused.text - detail = refused.json()["detail"] - assert detail["error"] == "provider_capacity_unavailable" - assert detail["provider"] == "getleadsio" - assert "own key" in detail["message"] - assert await _balance(clients) == before - assert await _rows(Hold) == [] - - await clients.post("/secrets", json={"name": "getleadsio", "value": "OWN-GETLEADSIO"}) - own = await clients.post("/call/getleadsio.people.search", json=body) - assert own.status_code == 200, own.text - assert own.json()["auth"] == "Bearer OWN-GETLEADSIO" - - async def test_a_stale_or_ok_view_never_refuses(clients: AsyncClient, platform_on, monkeypatch): monkeypatch.setattr(call_service, "relay", _fake_relay(200, b'{"ok":true}')) await _publish("tikhub", exhausted=False) @@ -288,21 +267,3 @@ async def test_a_failed_strike_never_fails_the_call(clients: AsyncClient, platfo # strike itself swallows; simulate the seam above it raising to prove the guard r = await clients.get(f"/call/{EP}?aweme_id=7") assert r.status_code == 402 and r.headers["X-Treg-Cost-Micro"] == "0" - - -def test_provider_capacity_is_a_treg_blamed_typed_failure(): - exc = CallFailure("provider_capacity", status_code=503, detail={"error": "x"}) - assert exc.blame == "treg" and exc.status_code == 503 - - -async def test_pdl_identify_quota_breaker_preserves_enrichment(clients, monkeypatch): - from types import SimpleNamespace - monkeypatch.setattr(capacity_marks, 'STRIKE_MIN_GAP', timedelta(0)) - mk = SimpleNamespace(tier='platform', provider='pdl', endpoint_id='pdl.x.person-identify') - body = b'{"error":{"message":"You have hit your account maximum for person_identify (all matches used)"}}' - for _ in range(2): - assert await call_settle._note_capacity_signal(mk, 402, {}, body) == 'quota' - await capacity_view.load() - assert capacity_view.is_exhausted('pdl', 'pdl.x.person-identify') - assert not capacity_view.is_exhausted('pdl', 'pdl.people.enrich') - assert not capacity_view.is_exhausted('pdl', 'pdl.companies.enrich') diff --git a/tests/test_capacity_smoothing.py b/tests/test_capacity_smoothing.py index 9cb4ed43..07d48e6b 100644 --- a/tests/test_capacity_smoothing.py +++ b/tests/test_capacity_smoothing.py @@ -88,15 +88,6 @@ async def test_burst_429_with_a_short_retry_after_is_re_sent_once_on_the_same_ho assert (await db.execute(select(Hold))).scalars().all() == [], "one hold, closed once" -async def test_still_429_after_the_retry_is_relayed_as_is(clients: AsyncClient, platform_on, monkeypatch): - seen = [] - monkeypatch.setattr(call_service, "relay", _relay_script( - [(429, ((b"retry-after", b"0"),), b"x"), (429, ((b"retry-after", b"0"),), b"y")], seen)) - r = await clients.get(f"/call/{EP}?aweme_id=7") - assert r.status_code == 429 and len(seen) == 2 and r.text == "y" - assert r.headers["X-Treg-Cost-Micro"] == "0" - - async def test_tool_called_reads_a_vendor_429_as_the_vendors_burst( clients: AsyncClient, platform_on, monkeypatch, posthog_events, ): @@ -105,7 +96,8 @@ async def test_tool_called_reads_a_vendor_429_as_the_vendors_burst( monkeypatch.setattr(call_service, "relay", _relay_script( [(429, ((b"retry-after", b"0"),), b"x"), (429, ((b"retry-after", b"0"),), b"y")], seen)) r = await clients.get(f"/call/{EP}?aweme_id=7", headers={"User-Agent": "Python-urllib/3.12"}) - assert r.status_code == 429 + assert r.status_code == 429 and len(seen) == 2 and r.text == "y" + assert r.headers["X-Treg-Cost-Micro"] == "0" (e,) = await posthog_events() p = e["properties"] assert p["outcome"] == "vendor_error" and p["refused_by"] is None @@ -161,18 +153,17 @@ async def test_own_key_calls_are_never_smoothed(clients: AsyncClient, platform_o assert len(seen) == 2 and time.monotonic() - t0 < 1.0, "no bucket wait, no re-send on an org's own key" -async def test_concurrent_platform_calls_over_the_limit_relay_no_429_and_hold_no_db( +async def test_concurrent_platform_calls_over_the_limit_relay_no_429( clients: AsyncClient, platform_on, monkeypatch, ): """The leadsforge case (plan §6), scaled to a 2-per-second limit so the test runs in seconds: five calls at once, a provider that counts calls per window (as providers do) and 429s the - third in any second → the bucket spaces them, the provider sees no burst, added latency stays - ≤ 2 s, and no DB connection is held while a call waits or relays.""" + third in any second → the bucket spaces them, the provider sees no burst, and added latency + stays ≤ 2 s. The pool counter is process-wide, so concurrent siblings in their DB phase make it + nonzero here; test_a_smoothed_call_holds_no_db_connection_while_it_waits proves the discipline.""" await _publish_rate("tikhub", 2, 1.0) stamps: list[float] = [] - pool_seen = [] async def provider(request, upstream_url, tool, secrets, client, drop_params=None, force_identity=False): - pool_seen.append(_engine.pool.checkedout()) now = time.monotonic() stamps.append(now) status = 429 if sum(1 for t in stamps if now - t < 0.9) > 2 else 200 @@ -184,7 +175,6 @@ async def test_concurrent_platform_calls_over_the_limit_relay_no_429_and_hold_no monkeypatch.setattr(call_service, "relay", provider) assert (await clients.get(f"/call/{EP}?aweme_id=99")).status_code == 200 # warm the process (see below) stamps.clear() - pool_seen.clear() limiter.reset() t0 = time.monotonic() rs = await asyncio.gather(*(clients.get(f"/call/{EP}?aweme_id={i}") for i in range(4))) diff --git a/tests/test_ledger.py b/tests/test_ledger.py index f73c9dce..47f8ea3f 100644 --- a/tests/test_ledger.py +++ b/tests/test_ledger.py @@ -113,8 +113,7 @@ async def test_two_deliveries_of_one_payment_credit_once(c: AsyncClient): Each task needs its OWN session: two coroutines sharing one AsyncSession is a different bug. """ - from treg.models import AdConversion, User - from treg.timeutil import utcnow_naive + from treg.models import AdConversion org_id, _ = await _org(c) promo = get_settings().promo_grant_micro @@ -410,6 +409,9 @@ async def test_referrals_page_survives_a_sweep_rollback(c: AsyncClient, monkeypa summary = await referrals_app.get_referral_summary(user_id) # must not raise assert "referral sweep failed for user" in caplog.text assert summary["code"] # the page still renders + + +async def test_signup_promo_and_its_conversion_land_or_fail_together(c: AsyncClient, monkeypatch): """The one-transaction property, in both directions: a failed commit loses the grant AND the queued ad conversion (and does not raise - the never-500-the-signup contract), and the retry makes both durable in one commit.""" @@ -461,26 +463,12 @@ async def test_referrals_page_survives_a_sweep_rollback(c: AsyncClient, monkeypa assert await _assert_invariant(org_id) == get_settings().promo_grant_micro -async def test_a_grant_failure_cannot_fail_signup(c: AsyncClient, monkeypatch): - """The promo is a nicety: a broken grant must cost the team its $1, never their signup.""" - async def boom(*a, **kw): - raise RuntimeError("grant broke") - - monkeypatch.setattr(ledger, "grant", boom) - r = await verified_signup(c, json={"email": "promo-fails@superdesign.dev"}) - assert r.status_code == 200, r.text - org_id = r.json()["org_id"] - async with session_maker() as db: - assert await db.get(Org, org_id) is not None - assert await ledger.balance_of(db, org_id) == 0 - assert await ledger.blocks_of(db, org_id) == [] - - async def test_a_grant_failure_after_staging_still_returns_the_signup(c: AsyncClient, monkeypatch): - """The sharper variant of the test above: the grant fails AFTER its SQL has staged, so the - recovery rollback expires every object the session tracks. Both signup doors must still answer - with the fields they promised, and the referral must still be attributed - the never-500-the- - signup contract does not stop at objects that now need a reload.""" + """A broken promo grant costs the team its credit, never their signup. The grant fails AFTER + its SQL has staged, so the recovery rollback expires every object the session tracks. Both + signup doors must still answer with the fields they promised, and the referral must still be + attributed - the never-500-the-signup contract does not stop at objects that now need a + reload.""" from treg.routers.signup_cookies import REFERRAL_COOKIE from treg.models import Referral diff --git a/tests/test_machine_identity_balance.py b/tests/test_machine_identity_balance.py index 7f6dbfb6..b10b4eaa 100644 --- a/tests/test_machine_identity_balance.py +++ b/tests/test_machine_identity_balance.py @@ -31,31 +31,15 @@ async def test_a_machine_identity_can_learn_its_own_org(clients: AsyncClient): assert me.json()["org_id"] == org_id # …but it can learn this -async def test_a_member_can_read_the_balance_it_spends(clients: AsyncClient): +async def test_a_machine_identity_reads_the_wallet_but_not_the_purchase_history(clients: AsyncClient): """Every agent is told to run `treg balance` after a call, and a 402 already hands the caller - `balance_micro` — refusing the same number here while shipping it in an error was incoherent.""" + `balance_micro` - refusing the same number here while shipping it in an error was incoherent. + The wallet is everyone's; what was bought, when, and what is left of each block is not.""" token, org_id = await _agent_token(clients) r = await clients.get(f"/orgs/{org_id}/balance", headers={"X-Treg-Token": token}) assert r.status_code == 200, r.text assert "balance_micro" in r.json() - - -async def test_but_a_member_does_not_see_the_purchase_history(clients: AsyncClient): - """The wallet is everyone's; what was bought, when, and what is left of each block is not.""" - token, org_id = await _agent_token(clients) - member = (await clients.get(f"/orgs/{org_id}/balance", - headers={"X-Treg-Token": token})).json() - assert member["blocks"] == [] and member["entries"]["items"] == [] - - admin = (await clients.get(f"/orgs/{org_id}/balance")).json() # the owner sees the detail - assert admin["blocks"], "an admin must still get the funding detail" - - -async def test_the_balance_is_not_readable_across_orgs(clients: AsyncClient): - _, org_id = await _agent_token(clients) - r = await clients.post("/users", json={"email": "stranger@elsewhere.dev"}) - other = {"X-Treg-Token": r.json()["token"]} - assert (await clients.get(f"/orgs/{org_id}/balance", headers=other)).status_code == 403 + assert r.json()["blocks"] == [] and r.json()["entries"]["items"] == [] async def test_a_missing_org_is_indistinguishable_from_one_you_cannot_see(clients: AsyncClient): @@ -76,7 +60,6 @@ def test_an_invalid_token_is_named_as_such_not_as_a_missing_org(monkeypatch, cap """`_active_org_id` returns None both when the token is bad and when there is genuinely no org. 21 commands turned that into a bare "no active org", sending the reader to fix org config when the real problem was authentication.""" - import httpx from treg import cli class _Resp: diff --git a/tests/test_oauth_billed.py b/tests/test_oauth_billed.py index 5d96a394..ac1e74bd 100644 --- a/tests/test_oauth_billed.py +++ b/tests/test_oauth_billed.py @@ -19,7 +19,6 @@ from sqlalchemy import update from treg.application.call import resolve as call_resolution from treg.application.call import service as call_service -from treg.routers import call as call_routes from treg import crypto from treg.config import get_settings from treg.infra.db import session_maker @@ -170,24 +169,19 @@ async def test_unmatched_passthrough_gets_the_default_read_rate( # ---- writes ------------------------------------------------------------------------------------ -async def test_post_create_is_per_call(clients: AsyncClient, billed_on, monkeypatch): +@pytest.mark.parametrize("text, reply, micro", [ + ("Shipping something new today.", b'{"data": {"id": "1", "text": "hi"}}', CREATE_MICRO), + # $0.015 -> $0.20 when the text carries a URL - the single biggest mispricing risk, sniffed + # from the body at estimate time. + ("read this: https://example.com/post", b'{"data": {"id": "1"}}', CREATE_LINK_MICRO), +]) +async def test_post_create_is_per_call_and_a_link_prices_13x(clients: AsyncClient, billed_on, + monkeypatch, text, reply, micro): await _connect_x(clients) - monkeypatch.setattr(call_service, "relay", _stub_relay(201, b'{"data": {"id": "1", "text": "hi"}}')) - r = await clients.post("/call/x.x.post.create", - json={"text": "Shipping something new today."}) + monkeypatch.setattr(call_service, "relay", _stub_relay(201, reply)) + r = await clients.post("/call/x.x.post.create", json={"text": text}) assert r.status_code == 201, r.text - assert int(r.headers["x-treg-cost-micro"]) == CREATE_MICRO - - -async def test_post_with_url_prices_13x(clients: AsyncClient, billed_on, monkeypatch): - """$0.015 → $0.20 when the text carries a URL — the single biggest mispricing risk, sniffed - from the body at estimate time.""" - await _connect_x(clients) - monkeypatch.setattr(call_service, "relay", _stub_relay(201, b'{"data": {"id": "1"}}')) - r = await clients.post("/call/x.x.post.create", - json={"text": "read this: https://example.com/post"}) - assert r.status_code == 201, r.text - assert int(r.headers["x-treg-cost-micro"]) == CREATE_LINK_MICRO + assert int(r.headers["x-treg-cost-micro"]) == micro # ---- the money fences -------------------------------------------------------------------------- diff --git a/tests/test_reconcile.py b/tests/test_reconcile.py index f6203941..f1c2e0eb 100644 --- a/tests/test_reconcile.py +++ b/tests/test_reconcile.py @@ -82,25 +82,21 @@ def _since(): # ---- drift math -------------------------------------------------------------------------------- -async def test_drift_flags_a_ten_percent_climb(c: AsyncClient): - await _calls([{"est": 1000, "obs": 1100}] * 4) +@pytest.mark.parametrize("obs,calls,ratio,flagged", [ + (1100, 4, 0.1, True), # a 10% climb + (1020, 5, 0.02, False), # 2%, under the 5% tolerance +]) +async def test_drift_flags_only_beyond_tolerance(c: AsyncClient, obs, calls, ratio, flagged): + await _calls([{"est": 1000, "obs": obs}] * calls) async with session_maker() as db: rows = await reconcile.price_drift(db, _since()) assert len(rows) == 1 r = rows[0] - assert (r["endpoint_id"], r["provider"], r["calls"]) == ("dataforseo.serp.google", "dataforseo", 4) - assert r["estimate_mean_micro"] == 1000 and r["observed_mean_micro"] == 1100 - assert r["drift_ratio"] == pytest.approx(0.1) - assert r["drift_micro"] == 100 - assert r["flagged"] is True - - -async def test_drift_within_tolerance_is_not_flagged(c: AsyncClient): - await _calls([{"est": 1000, "obs": 1020}] * 5) # 2% — under the 5% tolerance - async with session_maker() as db: - rows = await reconcile.price_drift(db, _since()) - assert rows[0]["drift_ratio"] == pytest.approx(0.02) - assert rows[0]["flagged"] is False + assert (r["endpoint_id"], r["provider"], r["calls"]) == ("dataforseo.serp.google", "dataforseo", calls) + assert r["estimate_mean_micro"] == 1000 and r["observed_mean_micro"] == obs + assert r["drift_ratio"] == pytest.approx(ratio) + assert r["drift_micro"] == obs - 1000 + assert r["flagged"] is flagged async def test_drift_needs_min_calls_before_it_flags(c: AsyncClient): @@ -213,12 +209,6 @@ async def test_repeat_rate_top_repeated_only_lists_actual_repeats(c: AsyncClient assert hot["endpoint_id"] == "dataforseo.serp.google" -async def test_repeat_rate_is_empty_without_telemetry(c: AsyncClient): - async with session_maker() as db: - out = await reconcile.repeat_rate(db, _since()) - assert out == {"calls": 0, "repeat_calls": 0, "repeat_ratio": 0.0, "providers": [], "top_repeated": []} - - # ---- the endpoints + their gate ---------------------------------------------------------------- @pytest.mark.parametrize("report", ["drift", "spend", "repeats"]) async def test_reconcile_endpoints_require_superadmin(c: AsyncClient, report: str): @@ -241,15 +231,6 @@ async def test_drift_endpoint_splits_out_the_flagged_rows(c: AsyncClient): assert {x["endpoint_id"] for x in body["endpoints"]} == {"ep.bad", "ep.ok"} -async def test_spend_and_repeats_endpoints_report_their_windows(c: AsyncClient): - await _settles([{"amount": 2000, "observed": 2000}]) - await _calls([{"hash": "aaa", "est": 100, "obs": 100}] * 2) - spend = (await c.get("/admin/reconcile/spend", headers=_a())).json() - assert spend["charged_micro"] == 2000 and spend["providers"][0]["reported_calls"] == 1 - repeats = (await c.get("/admin/reconcile/repeats?top=1", headers=_a())).json() - assert repeats["repeat_calls"] == 1 and len(repeats["top_repeated"]) == 1 - - async def test_window_start_is_clamped(c: AsyncClient): """A hostile `since_days` can't turn the report into a full-table scan of all time (or the future).""" now = datetime.now(timezone.utc).replace(tzinfo=None) @@ -306,14 +287,3 @@ async def test_recovery_ignores_vendor_priced_providers(c: AsyncClient): async with session_maker() as db: out = await reconcile.shared_plan_recovery(db, _since()) assert all(p["provider"] != "dataforseo" for p in out["providers"]) - - -async def test_price_drift_never_sees_a_shared_plan_provider(c: AsyncClient): - """Pinning a NATURAL property before someone breaks it: drift compares our estimate against the - provider's own reported charge, and a flat-fee provider never reports one — there is no number - to drift from. If a future `_observed_cost_micro` parser is added for such a provider, this test - is the alarm that the drift report now polices a price treg itself set.""" - await _calls([{"endpoint_id": "alphavantage.quote", "provider": "alphavantage", "est": 1000}] * 3) - async with session_maker() as db: - rows = await reconcile.price_drift(db, _since(), min_calls=1) - assert all(r["provider"] != "alphavantage" for r in rows) diff --git a/tests/test_tag_billing.py b/tests/test_tag_billing.py index 7549b249..26eb99d5 100644 --- a/tests/test_tag_billing.py +++ b/tests/test_tag_billing.py @@ -22,7 +22,6 @@ from treg.domain import money as ledger from treg.application.call import service as call_service from treg.application.call import settle as call_settle from treg.application.call.types import UpstreamResponse -from treg.routers import call as call_routes from treg.config import get_settings from treg.infra.db import session_maker from treg.models import Membership, Org, TagSpend, User @@ -235,28 +234,6 @@ async def test_in_flight_spend_counts_toward_a_cap_but_not_an_invoice(clients: A assert await ledger.tag_invoice_since(db, org_id, "customer", "cust_A", _EPOCH) == 2_000 -async def test_tag_totals_reconcile_with_org_spend(clients: AsyncClient, platform_on): - """The identity a builder's invoice depends on: for ANY key, the per-value totals plus whatever - could not be attributed equal the org's own settled spend for the window. It must hold whichever - dimension you slice by — that is what proves stacked reports agree with each other.""" - org_id = await _org_id(clients) - for tags in ("customer=cust_A, workspace=ws_1", - "customer=cust_B, workspace=ws_1", - "customer=cust_C, workspace=ws_2"): - assert (await clients.get(f"/call/{EP}?aweme_id=7", - headers={"X-Treg-Meta": tags})).status_code == 200 - await clients.get(f"/call/{EP}?aweme_id=7") # one untagged call - - async with session_maker() as db: - org_total = (await ledger.spend_since(db, org_id, _EPOCH))["spend_micro"] - for dim, expected_values in (("customer", 3), ("workspace", 2)): - by_value = await ledger.spend_by_tag(db, org_id, dim, _EPOCH) - assert len(by_value) == expected_values - unattributed = org_total - sum(by_value.values()) - assert sum(by_value.values()) + unattributed == org_total - assert unattributed == EP_MICRO, "the untagged call must show up as unattributed" - - async def test_caller_tags_cannot_overwrite_ledger_provenance(clients: AsyncClient, platform_on): """A hostile bag must not rewrite the money journal. treg's own keys merge LAST now — before this change a caller could zero `charged_micro` or forge the `tier` that reconcile.py reads.""" @@ -302,13 +279,6 @@ async def _declare_dims(org_id: int, *dims: str) -> None: await db.commit() -async def test_no_budget_row_means_unlimited(clients: AsyncClient, platform_on): - """Builders never pre-register a user: the first call for an unknown id just works.""" - org_id = await _org_id(clients) - r = await clients.get(f"/call/{EP}?aweme_id=7", headers={"X-Treg-Meta": "customer=brand_new"}) - assert r.status_code == 200, r.text - - async def test_a_blocked_tag_is_refused(clients: AsyncClient, platform_on): org_id = await _org_id(clients) await _set_budget(clients, org_id, "customer", "cust_A", status="blocked") @@ -446,14 +416,6 @@ async def test_one_user_still_replays_their_own_label(clients: AsyncClient, plat assert await ledger.tag_invoice_since(db, org_id, "customer", "cust_A", _EPOCH) == EP_MICRO -async def test_untagged_idempotency_is_unchanged(clients: AsyncClient, platform_on): - """An untagged caller keeps exactly today's behaviour — the key is stored unscoped.""" - hdr = {"Idempotency-Key": "retry-1"} - first = await clients.get(f"/call/{EP}?aweme_id=7", headers=hdr) - again = await clients.get(f"/call/{EP}?aweme_id=7", headers=hdr) - assert first.status_code == 200 and again.headers.get("X-Treg-Idempotent-Replay") == "true" - - async def test_a_reused_label_on_a_different_request_still_says_so(clients: AsyncClient, platform_on): """The fingerprint refusal survives scoping, and the message quotes the label the CALLER wrote — not treg's internal scoped form.""" @@ -471,31 +433,6 @@ async def _mint_agent(c: AsyncClient, org_id: int, name: str, **kw) -> str: return r.json()["token"] -async def test_a_pinned_token_cannot_bill_another_customer(clients: AsyncClient, platform_on): - """The whole point of handing a scoped token to a customer's own machine. If the header won, that - customer could retag their calls and walk straight out of their own budget.""" - org_id = await _org_id(clients) - token = await _mint_agent(clients, org_id, "cust-a-bot", pinned_tags={"customer": "cust_A"}) - hdr = {"X-Treg-Token": token} - - bad = await clients.get(f"/call/{EP}?aweme_id=7", - headers={**hdr, "X-Treg-Meta": "customer=cust_B"}) - assert bad.status_code == 403, bad.text - async with session_maker() as db: - assert (await db.execute(select(TagSpend))).scalars().all() == [], \ - "a refused pin must not have reserved anything" - - # Naming its OWN customer is fine, so a builder can send the header unconditionally... - ok = await clients.get(f"/call/{EP}?aweme_id=7", - headers={**hdr, "X-Treg-Meta": "customer=cust_A"}) - assert ok.status_code == 200, ok.text - # ...and sending no header at all still attributes to the pin. - bare = await clients.get(f"/call/{EP}?aweme_id=7", headers=hdr) - assert bare.status_code == 200, bare.text - async with session_maker() as db: - assert await ledger.tag_invoice_since(db, org_id, "customer", "cust_A", _EPOCH) == EP_MICRO * 2 - - async def test_rotating_a_pinned_token_keeps_its_pin(clients: AsyncClient): """A rotate replaces the token, never the limits — re-minting must not silently unpin.""" org_id = await _org_id(clients) @@ -528,15 +465,6 @@ async def test_a_team_sets_its_daily_cap_in_either_direction(clients: AsyncClien assert neg.status_code == 422 -async def test_the_team_cap_actually_refuses_spend(clients: AsyncClient, platform_on): - org_id = await _org_id(clients) - await clients.patch(f"/orgs/{org_id}/settings", json={"daily_cap_micro": EP_MICRO}) - assert (await clients.get(f"/call/{EP}?aweme_id=7")).status_code == 200 - over = await clients.get(f"/call/{EP}?aweme_id=7") - assert over.status_code == 429 - assert over.json()["detail"]["error"] == "platform_daily_cap_reached" - - async def test_declaring_more_than_three_budget_dimensions_is_refused(clients: AsyncClient): org_id = await _org_id(clients) r = await clients.patch(f"/orgs/{org_id}/settings", @@ -603,17 +531,6 @@ async def test_the_join_key_is_scoped_to_the_team(clients: AsyncClient, platform assert denied.status_code == 404 -async def test_calls_can_be_windowed_and_paged(clients: AsyncClient): - await _mk_echo_tool(clients) - for _ in range(3): - await clients.get("/call/echo/x", headers={"X-Treg-Meta": "customer=cust_A"}) - await audit.drain() - rows = (await clients.get("/calls?days=1")).json() - assert len(rows) >= 3 - older = (await clients.get(f"/calls?before_id={rows[0]['id']}")).json() - assert all(r["id"] < rows[0]["id"] for r in older) - - async def _make_org(name: str, slug: str) -> int: async with session_maker() as db: org = Org(name=name, slug=slug) @@ -658,21 +575,13 @@ async def test_ordinary_customer_ids_still_pass(clients: AsyncClient): assert r.status_code == 200, f"{good!r} should be accepted: {r.text}" -async def test_the_collision_is_closed_end_to_end(clients: AsyncClient, platform_on): - """The attack, driven the way an attacker would: one user tries to reach another's stored answer - by crafting the key, and is refused at the door.""" - crafted = await clients.get( - f"/call/{EP}?aweme_id=7", - headers={"X-Treg-Meta": "customer=A\x1fB", "Idempotency-Key": "C"}) - assert crafted.status_code == 422 - - async def test_a_pin_cannot_smuggle_what_the_header_cannot(clients: AsyncClient): """`pinned_tags` arrives as JSON on the mint endpoint and never passes the header parser, so it was a second door onto the same storage keys. Both doors must enforce one rule.""" org_id = await _org_id(clients) for hostile in ({"customer": "A\x1fB"}, {"bad key!": "x"}, {"customer": "a b"}, - {"customer": "jane@example.com"}): + {"customer": "jane@example.com"}, + {f"tag{i}": f"v{i}" for i in range(6)}): # the header's five-pair limit r = await clients.post(f"/orgs/{org_id}/agents", json={"name": "bot", "pinned_tags": hostile}) assert r.status_code == 422, f"{hostile!r} must be refused, got {r.status_code}" @@ -785,7 +694,6 @@ async def test_a_caller_input_4xx_may_bill_only_on_a_per_call_endpoint(clients: `per_call` and only at the charge the provider reports for it (an unreported 400 releases — `test_a_4xx_bills_only_what_the_provider_reports` in test_marketplace_call.py). The status gate itself is asserted here because tikhub comments is per_success and always releases.""" - org_id = await _org_id(clients) monkeypatch.setattr(call_service, "relay", _fake_relay(400, b'{"error":"bad param"}')) r = await clients.get(f"/call/{EP}?aweme_id=7", headers={"X-Treg-Meta": "customer=cust_A"}) assert r.status_code == 400 @@ -797,15 +705,6 @@ async def test_a_caller_input_4xx_may_bill_only_on_a_per_call_endpoint(clients: # ---- per-dimension defaults with overrides ------------------------------------------------------- -async def test_unlimited_until_a_default_is_set(clients: AsyncClient, platform_on): - """The shipped state: no default, no override, no limit. A team that never opens this page keeps - behaving exactly as before.""" - org_id = await _org_id(clients) - for _ in range(3): - r = await clients.get(f"/call/{EP}?aweme_id=7", headers={"X-Treg-Meta": "customer=anyone"}) - assert r.status_code == 200, r.text - - async def test_a_default_applies_to_every_value_without_an_override(clients: AsyncClient, platform_on): """One setting covers a customer base of any size — the whole point. A builder with 10k customers cannot write 10k rows.""" @@ -903,18 +802,3 @@ async def test_the_dimension_bound_still_holds(clients: AsyncClient): d = r.json()["detail"] assert d["error"] == "too_many_budget_dimensions" and d["limit"] == 3 assert sorted(d["declared"]) == ["customer", "project", "workspace"] - - -async def test_the_integration_skill_is_served_and_templated(clients: AsyncClient): - """`/integrate.md` is a front door: a builder pastes it into their repo and points a coding agent - at it, so a stale `{BASE}` or a 404 breaks an integration before it starts.""" - r = await clients.get("/integrate.md") - assert r.status_code == 200, r.text - body = r.text - assert "{BASE}" not in body, "the serving host must be templated in" - # The load-bearing claims — if any of these names drift, the skill teaches an integration that - # does not work, and nothing else in the suite would notice. - for must in ("X-Treg-Meta", "X-Treg-Call-Id", "X-Treg-Cost-Micro", "X-Treg-Error", - "usage/by-tag", "/budgets/", "attributed", "unattributed", - "Idempotency-Key", "--pin"): - assert must in body, f"integrate.md no longer mentions {must!r}" diff --git a/tests/test_tag_billing_adversarial.py b/tests/test_tag_billing_adversarial.py index 032de958..f958fa83 100644 --- a/tests/test_tag_billing_adversarial.py +++ b/tests/test_tag_billing_adversarial.py @@ -25,7 +25,7 @@ from treg.routers import call as call_routes from treg.config import get_settings from treg.infra.db import session_maker from treg.domain.governance import budgets as budget_policy -from treg.models import CallRecord, CreditBlock, Hold, LedgerEntry, Membership, Org, TagSpend, User +from treg.models import CreditBlock, Hold, LedgerEntry, Membership, Org, TagSpend, User EP = "tikhub.tiktok.video.comments" @@ -95,13 +95,11 @@ async def test_attack_1_all_ingress_paths_reject_storage_key_delimiters( assert exc.value.status_code == 422 meta = call_routes._parse_call_meta(_request_with_meta("customer=safe_value")) - stored_key = call_idem._scoped_idempotency_key("retry-1", meta) - assert stored_key == "safe_value\x1fretry-1" assert all(bad not in meta.primary_val for bad in ("\x1f", "\n", ",")) async def test_attack_2_usage_identity_survives_five_tags_zero_release_settle_and_org_overlap( - clients: AsyncClient, platform_on, monkeypatch, + clients: AsyncClient, platform_on, ): org_id = await _org_id(clients) tags = { @@ -123,37 +121,6 @@ async def test_attack_2_usage_identity_survives_five_tags_zero_release_settle_an zero_id = await ledger.reserve(db, org_id, EP, 500, tags={"customer": "zero"}) assert await ledger.settle(db, zero_id, 0) == 0 - raced_id = await ledger.reserve(db, org_id, EP, 500, tags={"customer": "raced"}) - - # Force both lifecycle operations to observe the same Hold before either is allowed to proceed. - original_get = AsyncSession.get - readers = 0 - both_loaded = asyncio.Event() - - async def synchronized_get(self, entity, ident, **kwargs): - nonlocal readers - row = await original_get(self, entity, ident, **kwargs) - if entity is Hold and ident == raced_id: - readers += 1 - if readers == 2: - both_loaded.set() - await asyncio.wait_for(both_loaded.wait(), timeout=5) - return row - - monkeypatch.setattr(AsyncSession, "get", synchronized_get) - - async def attempt_settle(): - async with session_maker() as db: - return await ledger.settle(db, raced_id, 500) - - async def attempt_release(): - async with session_maker() as db: - return await ledger.release(db, raced_id, reason="concurrent release") - - # A losing SQLite writer may surface a lock error; the accounting identity must still hold after - # both attempts have completed or rolled back. - await asyncio.gather(attempt_settle(), attempt_release(), return_exceptions=True) - other_org_id = await _make_org("Other adversarial org", "other-adversarial-org") async with session_maker() as db: other_call = await ledger.reserve( @@ -386,21 +353,16 @@ async def test_review_pin_bypass_matrix_never_attributes_a_different_value( budget_policy._validate_tag_pair("customer", "cust_\u00c0") assert exc.value.status_code == 422 async with session_maker() as db: + refused = (await db.execute(select(TagSpend).where( + TagSpend.org_id == org_id, TagSpend.dim == "customer", + TagSpend.val != "cust_A"))).scalars().all() + assert refused == [], "a refused pin must not have reserved anything" assert await ledger.tag_invoice_since( db, org_id, "customer", "cust_B", datetime(2000, 1, 1)) == 0 assert await ledger.tag_invoice_since( db, org_id, "customer", "cust_A", datetime(2000, 1, 1)) == successes * EP_MICRO -async def test_review_pins_cannot_bypass_the_five_pair_limit(clients: AsyncClient): - org_id = await _org_id(clients) - response = await clients.post( - f"/orgs/{org_id}/agents", - json={"name": "six-pins", "pinned_tags": {f"tag{i}": f"v{i}" for i in range(6)}}, - ) - assert response.status_code == 422, response.text - - async def test_review_distinct_memberships_never_share_idempotent_bodies( clients: AsyncClient, platform_on, monkeypatch, ): diff --git a/tests/test_usage_caps.py b/tests/test_usage_caps.py index d953a66e..2bbba228 100644 --- a/tests/test_usage_caps.py +++ b/tests/test_usage_caps.py @@ -7,7 +7,6 @@ Soft by design (best-effort audit → fails open), so these tests seed records d from __future__ import annotations from datetime import date, timedelta -from pathlib import Path from httpx import AsyncClient from sqlmodel import select @@ -49,22 +48,6 @@ async def _mk_echo_tool(c: AsyncClient, name: str = "echo") -> None: assert r.status_code == 200, r.text -async def test_call_blocked_once_the_cap_is_reached(clients: AsyncClient): - await _mk_echo_tool(clients) - org_id = await _set_cap(1) - assert (await clients.get("/call/echo/anything")).status_code == 200 # 1st call: under cap - await audit.drain() # its CallRecord is fire-and-forget - blocked = await clients.get("/call/echo/anything") - assert blocked.status_code == 429 and "daily usage limit" in blocked.json()["detail"] - _ = org_id - - -async def test_unlimited_by_default(clients: AsyncClient): - await _mk_echo_tool(clients) # cap defaults to -1 → no cap query, never blocked - for _ in range(5): - assert (await clients.get("/call/echo/anything")).status_code == 200 - - async def _counter(org_id: int, email: str) -> tuple[int, date | None]: async with session_maker() as s: uid = (await s.execute(select(User.id).where(User.email == email))).scalar_one() @@ -83,11 +66,8 @@ async def _set_counter(org_id: int, email: str, n: int, day: date) -> None: async def test_runs_and_calls_share_one_gate(clients: AsyncClient): - """A member can't dodge the cap by switching path: both run handlers in api.py go through the - same `_enforce_daily_cap` door as `/call/` (authorize.py), and that door is what moves the - counter. Pinned statically because the run surfaces need a bundle to exercise end to end.""" - src = (Path(__file__).parents[1] / "src" / "treg" / "api.py").read_text() - assert src.count("await _enforce_daily_cap(caller, db)") == 2 # local-run grant + server run + """Prior events already on today's counter (from runs or calls alike) block the next call, and + a refused call does not move the counter.""" await _mk_echo_tool(clients) org_id = await _set_cap(2) today = _utcnow_naive().date() @@ -118,7 +98,8 @@ async def test_yesterdays_usage_does_not_count_today(clients: AsyncClient): await _set_counter(org_id, "tim@superdesign.dev", 1, today - timedelta(days=1)) # yesterday's assert (await clients.get("/call/echo/anything")).status_code == 200 # a new day starts from 0 assert await _counter(org_id, "tim@superdesign.dev") == (1, today) # ...and this was its first - assert (await clients.get("/call/echo/anything")).status_code == 429 + blocked = await clients.get("/call/echo/anything") + assert blocked.status_code == 429 and "daily usage limit" in blocked.json()["detail"] async def test_setting_a_cap_seeds_the_counter_from_todays_journal(clients: AsyncClient):