trending/treg
1
0
Fork 0
mirror of https://github.com/superdesigndev/treg.git synced 2026-10-02 03:24:35 +08:00
Code Issues Packages Projects Releases Wiki Activity

refactor(tests): import moved helpers from their real homes; retire the api.py test-compat surface

Browse Source 
Every test that reached a moved helper through treg.api - direct imports, module-attr
reads, and the alias forms A / api / treg_api - now imports it from the module that owns
it: routers.auth/orgs/web/catalog/onboard/call, application.call.{resolve,settle,
evidence,idempotency,intake,reserve}, application.signup, bootstrap_http, signup_cookies,
domain.identity.access, timeutil, and sqlalchemy.exc for PoolTimeoutError. Dead api
aliases are dropped with the last use.

That empties the test half of api.py's compatibility block, so it shrinks to the two
things bootstrap still addresses through this module at runtime: the dotted-string
startup check treg.api._backfill_provider_extra_tools and the static dirs _mount_static
reads off api_module. test_router_dependencies.py retires with the contract it pinned -
its assertions said 'treg.api re-exports these', which stops being true on purpose here -
and super-admin.md's note about the transitional re-export is updated to match.

Verified: route dump still byte-identical to main; full suite green (2167 passed - the
two retired contract tests account for the delta).
This commit is contained in:
SToneX
2026-08-28 20:39:07 +08:00
parent a529ab724b
commit 65f5088949
29 changed files with 203 additions and 297 deletions
Download Patch File Download Diff File Expand all files Collapse all files
docs/context/architecture/super-admin.md
+2 -2
View File
@@ -26,8 +26,8 @@ A caller is a super-admin if EITHER:
Otherwise 403. The env key bootstraps; `POST /admin/users/{id}/superadmin` then grants named users the
flag (so a web portal can log in with either). Returns a principal string (for audit).
`api.py` re-exports this dependency during the staged router migration, so compatibility imports refer
to the same function object.
The dependency lives in `domain.identity.access` and every consumer imports it from there; the
transitional `api.py` re-export retired with the rest of the stage-3 compatibility surface.
The cross-tenant read, mutation, and reconciliation handlers live in three ordered blocks in
`routers.admin`. The mutation block shares the org deletion and member-rule cleanup helpers from
src/treg/api.py
+5 -82
View File
@@ -62,89 +62,12 @@ from .routers.web import LOCAL_USER_EMAIL, _WEB_DIR
from .timeutil import utcnow_naive as _utcnow_naive
# ---- compatibility re-exports --------------------------------------------------------------
# api.py WAS the whole server before the refactor, and some callers still address moved code
# through this module: tests import helpers as `treg.api.<name>` (or patch them on the module),
# and bootstrap's role startup manifests resolve "treg.api._backfill_provider_extra_tools" by
# dotted string. Everything below exists only for those callers - new code imports from the real
# home, and a name leaves this block when its last external reference does.
from sqlalchemy.exc import TimeoutError as PoolTimeoutError # noqa: F401
from . import ledger, oauth_providers # noqa: F401
from .application.call.evidence import ( # noqa: F401
_ERROR_CALLER_BODY_MAX,
_ERROR_MASKING_FAILED,
_ERROR_RESPONSE_MAX,
_decode_error_body,
_secret_renderings,
)
from .application.call.idempotency import ( # noqa: F401
_IDEM_SCOPE_SEP,
_request_fingerprint,
_scoped_idempotency_key,
)
from .application.call.intake import CallMeta # noqa: F401
from .application.call.reserve import _enforce_tag_budgets # noqa: F401
from .application.call.resolve import ( # noqa: F401
MarketplaceCall,
_PLATFORM_PAGE_DEFAULT,
_PLATFORM_PAGE_MAX,
_billed_endpoint_match,
_marketplace_pricing,
_marketplace_upstream,
_oauth_billed_estimate,
_platform_bindings,
_platform_estimate_micro,
_post_has_link,
_resolve_call,
_usd_to_micro,
)
from .application.call.settle import _observed_cost_micro, _platform_billable # noqa: F401
# bootstrap addresses these through this module: the role startup manifests resolve
# "treg.api._backfill_provider_extra_tools" by dotted string, and _mount_static reads the
# static dirs off `api_module`. Everything else moved code once re-exported here now lives at -
# and is imported from - its real home.
from .application.connect import _backfill_provider_extra_tools # noqa: F401
from .application.signup import _grant_signup_promo # noqa: F401
from .bootstrap_http import ( # noqa: F401
_BodyDecodeMiddleware,
_LegacyHostRedirectMiddleware,
_SecurityHeadersMiddleware,
_decode_request_body,
)
from .domain.identity.access import ( # noqa: F401
_resolve_org,
require_identity,
require_superadmin,
)
from .routers.admin import _ERROR_EVIDENCE_EXPIRED, _ERROR_EVIDENCE_TTL_DAYS # noqa: F401
from .routers.auth import ( # noqa: F401
CLI_APPROVE_MAX_TRIES,
HANDSHAKE_TTL,
MAX_OTP_ATTEMPTS,
OTP_START_MAX_PER_EMAIL,
OTP_START_MAX_PER_IP,
_cli_pending,
_cli_results,
_cli_states,
_effective_mcp_resource,
_login_callback_base,
_prune_handshakes,
_same_mcp_resource,
_wrong_resource,
)
from .routers.auth_helpers import _is_https # noqa: F401
from .routers.call import _parse_call_meta # noqa: F401
from .routers.catalog import _platform_rows # noqa: F401
from .routers.onboard import SANDBOX_RATE_MAX # noqa: F401
from .routers.orgs import _ORG_SCOPED_MODELS, _deny_match, count_today # noqa: F401
from .routers.signup_cookies import REFERRAL_COOKIE # noqa: F401
from .routers.web import ( # noqa: F401
_LOGO_DIR, # bootstrap mounts the static dirs through this module
_MEDIA_DIR,
_TOUR_DIR,
_VENDOR_DIR,
_provider_rows,
_related_link,
_usd_short,
_use_case_page_for,
use_case_job_page,
)
from .timeutil import as_naive as _as_naive # noqa: F401
from .routers.web import _LOGO_DIR, _MEDIA_DIR, _TOUR_DIR, _VENDOR_DIR # noqa: F401
LOCAL_ORG_NAME = "personal"
tests/callmatrix/test_matrix.py
+3 -2
View File
@@ -8,7 +8,8 @@ from collections import Counter
from httpx import AsyncClient
from sqlmodel import select
from treg import api as A
from sqlalchemy.exc import TimeoutError as PoolTimeoutError
from treg.application.call import service as call_service
from treg.routers import call as call_routes
from treg import audit, ledger
@@ -907,7 +908,7 @@ async def test_h1_saturation_503_still_carries_a_call_id_and_a_row(
await _register_echo(matrix_clients)
async def _no_slot(*args, **kwargs):
raise A.PoolTimeoutError("QueuePool limit of size 5 overflow 10 reached, connection timed out")
raise PoolTimeoutError("QueuePool limit of size 5 overflow 10 reached, connection timed out")
monkeypatch.setattr(call_service, "_resolve_call", _no_slot)
before = await snapshot(matrix_clients, fake_provider)
tests/test_agent_pages.py
+6 -6
View File
@@ -183,8 +183,8 @@ async def test_reliability_section_appears_only_with_traffic(clients: AsyncClien
html = (await clients.get(USECASE)).text
assert "Which one is the most reliable" not in html
import inspect
from treg import api
src = inspect.getsource(api.use_case_job_page)
from treg.routers.web import use_case_job_page
src = inspect.getsource(use_case_job_page)
assert "not a controlled benchmark" in src
@@ -205,7 +205,7 @@ async def test_use_case_page_prices_come_from_the_catalog(clients: AsyncClient):
if e["kind"] not in catalog_store.HIDDEN_KINDS]
lowest = min(c["usd"] for e in eps
if (c := cat.cost_view(e.get("cost"), e.get("provider"))) and c["usd"])
from treg.api import _usd_short
from treg.routers.web import _usd_short
html = (await clients.get(USECASE)).text
# the price sits in the hero kicker and the economics block, not the title: a title that fits a
# search result has no room for it
@@ -327,8 +327,8 @@ async def test_no_agent_or_job_specific_string_is_hardcoded_in_the_route():
"""Everything job-specific comes from the page spec, and the example agent from one constant,
so writing page 2 is data entry."""
import inspect
from treg import api
src = inspect.getsource(api.use_case_job_page)
from treg.routers.web import use_case_job_page
src = inspect.getsource(use_case_job_page)
for bad in ("email finder", "found addresses", "an address is found", "email address"):
assert bad not in src.lower(), bad
assert src.count("ChatGPT") == 0, "the example agent must come from DEFAULT_AGENT"
@@ -410,7 +410,7 @@ def test_related_cards_resolve_to_the_job_s_own_category():
"""Four categories carry fewer than five jobs, so `related` has to cross categories there.
Resolving inside the current page's category sent those cards to the wrong anchor under a
caption naming the wrong category, and no test noticed because the label still existed."""
from treg.api import _related_link, _use_case_page_for
from treg.routers.web import _related_link, _use_case_page_for
owner = {lbl: c for c, jobs in agent_pages.USE_CASES for lbl, _ in jobs}
for key, spec in agent_pages.USE_CASE_PAGES.items():
for lbl in spec["related"]:
tests/test_asgi_middleware.py
+1 -1
View File
@@ -7,7 +7,7 @@ import base64
from httpx import ASGITransport, AsyncClient
from starlette.responses import PlainTextResponse
from treg.api import (
from treg.bootstrap_http import (
_BodyDecodeMiddleware,
_LegacyHostRedirectMiddleware,
_SecurityHeadersMiddleware,
tests/test_auth_email.py
+3 -3
View File
@@ -59,7 +59,7 @@ async def test_code_is_one_time(client):
async def test_start_is_rate_limited_per_email(client):
from treg.api import OTP_START_MAX_PER_EMAIL
from treg.routers.auth import OTP_START_MAX_PER_EMAIL
email = "flood@matrix.io"
for _ in range(OTP_START_MAX_PER_EMAIL):
assert (await client.post("/auth/email/start", json={"email": email})).status_code == 200
@@ -68,7 +68,7 @@ async def test_start_is_rate_limited_per_email(client):
async def test_start_rate_limit_is_per_email_not_global(client):
from treg.api import OTP_START_MAX_PER_EMAIL
from treg.routers.auth import OTP_START_MAX_PER_EMAIL
for _ in range(OTP_START_MAX_PER_EMAIL + 2): # drive one inbox past its cap
await client.post("/auth/email/start", json={"email": "victim@matrix.io"})
other = await client.post("/auth/email/start", json={"email": "bystander@matrix.io"})
@@ -76,7 +76,7 @@ async def test_start_rate_limit_is_per_email_not_global(client):
async def test_start_is_rate_limited_per_ip(client):
from treg.api import OTP_START_MAX_PER_IP
from treg.routers.auth import OTP_START_MAX_PER_IP
for i in range(OTP_START_MAX_PER_IP): # distinct emails so the per-email cap never trips first
assert (await client.post("/auth/email/start", json={"email": f"u{i}@matrix.io"})).status_code == 200
blocked = await client.post("/auth/email/start", json={"email": "late@matrix.io"})
tests/test_body_encoding.py
+1 -1
View File
@@ -16,7 +16,7 @@ import json
import pytest
from httpx import AsyncClient
from treg.api import _decode_request_body
from treg.bootstrap_http import _decode_request_body
# ---- the pure decoder --------------------------------------------------------------------
tests/test_bughunt_server.py
+10 -9
View File
@@ -113,7 +113,7 @@ async def test_session_key_not_hardcoded_constant():
# ---- OTP brute-force ----------------------------------------------------------------------
async def test_otp_code_dies_after_max_wrong_attempts(c):
from treg.api import MAX_OTP_ATTEMPTS
from treg.routers.auth import MAX_OTP_ATTEMPTS
good = (await c.post("/auth/email/start", json={"email": "trinity@matrix.io"})).json()["dev_code"]
for _ in range(MAX_OTP_ATTEMPTS):
bad = await c.post("/auth/email/verify", json={"email": "trinity@matrix.io", "code": "000001"})
@@ -382,7 +382,7 @@ async def test_bad_binding_format_rejected_at_create(c):
async def test_numeric_slug_resolves_as_slug_not_id():
from treg.api import _resolve_org
from treg.domain.identity.access import _resolve_org
from treg.models import Org
await reset_db()
async with session_maker() as db:
@@ -637,13 +637,14 @@ def crypto_encrypt(v):
def test_prune_handshakes_evicts_stale():
from datetime import timedelta
import treg.api as api
old = api._utcnow_naive() - timedelta(seconds=api.HANDSHAKE_TTL + 60)
api._cli_states["stale"] = ("lid", old)
api._cli_results["lidX"] = ({"token": "T"}, old)
api._cli_pending["lidP"] = ("CODE", 8, old) # (pairing_code, attempts_left, created_at)
api._prune_handshakes()
assert "stale" not in api._cli_states and "lidX" not in api._cli_results and "lidP" not in api._cli_pending
from treg.routers import auth as auth_routes
from treg.timeutil import utcnow_naive
old = utcnow_naive() - timedelta(seconds=auth_routes.HANDSHAKE_TTL + 60)
auth_routes._cli_states["stale"] = ("lid", old)
auth_routes._cli_results["lidX"] = ({"token": "T"}, old)
auth_routes._cli_pending["lidP"] = ("CODE", 8, old) # (pairing_code, attempts_left, created_at)
auth_routes._prune_handshakes()
assert "stale" not in auth_routes._cli_states and "lidX" not in auth_routes._cli_results and "lidP" not in auth_routes._cli_pending
# ---- more invite / admin / health coverage ------------------------------------------------
tests/test_call_application_contract.py
+1 -2
View File
@@ -8,7 +8,6 @@ from typing import Awaitable, Callable, Literal, Protocol
import pytest
from treg import api as A
from treg.application.call import authorize, evidence, idempotency, intake, reserve, resolve, service, settle
from treg.application.call.types import (
AuthorizationFailed,
@@ -150,7 +149,7 @@ def test_gateway_failure_mapping_is_one_source_of_truth() -> None:
def test_provider_responses_are_data_and_billability_is_independent(
status: int, cost_type: str, billable: bool,
) -> None:
assert A._platform_billable(status, cost_type) is billable
assert settle._platform_billable(status, cost_type) is billable
def test_compatibility_surface_stays_literal_during_boundary_extraction() -> None:
tests/test_call_cancellation.py
+2 -2
View File
@@ -15,7 +15,7 @@ from httpx import AsyncClient
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from treg import api as A, ledger
from treg import ledger
from treg.application.call import service as call_service
from treg.application.call.types import GatewayFailed
from treg.routers import call as call_routes
@@ -306,7 +306,7 @@ async def test_cancellation_after_claim_before_reserve_releases_the_label(
key = "cancel-before-reserve"
resolve_reached = asyncio.Event()
never_resolve = asyncio.Event()
original_resolve = A._resolve_call
original_resolve = call_service._resolve_call
async def _blocked_resolve(*args, **kwargs):
resolve_reached.set()
tests/test_call_pool_discipline.py
+1 -1
View File
@@ -38,7 +38,7 @@ from treg.models import CallRecord, Hold
from test_marketplace_call import EP, EP_MICRO, platform_on # noqa: F401 — fixture reuse
from test_mcp import _call_tool, mcp_session
PoolTimeoutError = A.PoolTimeoutError
from sqlalchemy.exc import TimeoutError as PoolTimeoutError
@pytest.fixture
tests/test_deny_rules.py
+2 -1
View File
@@ -20,7 +20,8 @@ from sqlmodel import select
from conftest import make_upstream
from treg import crypto
from treg.api import _deny_match, app
from treg.api import app
from treg.routers.orgs import _deny_match
from treg.db import reset_db, session_maker
from treg.models import DenyRule, Membership, Org, User
tests/test_error_capture.py
+13 -13
View File
@@ -24,10 +24,10 @@ import pytest
from httpx import AsyncClient
from sqlalchemy import select
from treg import api as A
from treg.application.call import evidence as call_evidence
from treg.application.call import service as call_service
from treg.application.call.types import ReservationFailed, UpstreamResponse
from treg.routers import admin as admin_routes
from treg.routers import call as call_routes
from treg.config import get_settings
from treg.db import session_maker
@@ -59,7 +59,7 @@ def test_constant_binding_format_is_not_a_secret_rendering(monkeypatch):
"name": "x-api-version",
"format": "2025-11-01",
}
assert A._secret_renderings(SimpleNamespace(bindings=[constant]), {}) == []
assert call_evidence._secret_renderings(SimpleNamespace(bindings=[constant]), {}) == []
tool = SimpleNamespace(bindings=[
constant,
@@ -72,7 +72,7 @@ def test_constant_binding_format_is_not_a_secret_rendering(monkeypatch):
},
])
renderings = A._secret_renderings(tool, {})
renderings = call_evidence._secret_renderings(tool, {})
assert "2025-11-01" not in renderings
assert key in renderings
@@ -275,8 +275,8 @@ async def test_masking_render_failure_is_redacted_not_a_500(clients: AsyncClient
r = await clients.get("/call/masking-fails/fail?case=render")
assert r.status_code == 400
row = await _row(clients)
assert row["error_request"] == A._ERROR_MASKING_FAILED
assert row["error_response"] == A._ERROR_MASKING_FAILED
assert row["error_request"] == call_evidence._ERROR_MASKING_FAILED
assert row["error_response"] == call_evidence._ERROR_MASKING_FAILED
async def test_streaming_4xx_reaches_caller_byte_for_byte_and_keeps_evidence(
@@ -307,7 +307,7 @@ async def test_streaming_4xx_reaches_caller_byte_for_byte_and_keeps_evidence(
async def test_large_unmetered_body_keeps_query_only(clients: AsyncClient, monkeypatch):
await _own_tool(clients, name="large-own")
monkeypatch.setattr(call_service, "relay", _fake_relay(400, b'{"error":"too large"}'))
body = b"body-marker-" + b"x" * A._ERROR_CALLER_BODY_MAX
body = b"body-marker-" + b"x" * call_evidence._ERROR_CALLER_BODY_MAX
r = await clients.post("/call/large-own/fail?request_id=query-only", content=body)
assert r.status_code == 400
row = await _row(clients)
@@ -506,7 +506,7 @@ async def test_expired_evidence_is_a_state_not_content(clients: AsyncClient, pla
async with session_maker() as db:
row = (await db.execute(
select(CallRecord).order_by(CallRecord.id.desc()).limit(1))).scalars().first()
row.created_at = row.created_at - timedelta(days=A._ERROR_EVIDENCE_TTL_DAYS + 1)
row.created_at = row.created_at - timedelta(days=admin_routes._ERROR_EVIDENCE_TTL_DAYS + 1)
db.add(row)
await db.commit()
d = (await clients.get("/admin/errors?days=30", headers=ADMIN)).json()
@@ -571,7 +571,7 @@ async def test_a_huge_error_page_is_truncated(clients: AsyncClient, platform_on,
r = await clients.get(f"/call/{EP}?aweme_id=7")
assert r.status_code == 500
row = await _row(clients)
assert len(row["error_response"]) <= A._ERROR_RESPONSE_MAX + 1
assert len(row["error_response"]) <= call_evidence._ERROR_RESPONSE_MAX + 1
def test_a_compression_bomb_does_not_expand_without_bound():
@@ -586,8 +586,8 @@ def test_a_compression_bomb_does_not_expand_without_bound():
"""
bomb = gzip.compress(b"A" * 20_000_000)
assert len(bomb) < 32_000, "sanity: the bomb really is small compressed"
out = A._decode_error_body(bomb, "gzip")
assert len(out) <= A._ERROR_RESPONSE_MAX * 4 + 1, "decompression ran unbounded"
out = call_evidence._decode_error_body(bomb, "gzip")
assert len(out) <= call_evidence._ERROR_RESPONSE_MAX * 4 + 1, "decompression ran unbounded"
def test_unknown_telemetry_costs_a_column_not_the_whole_row():
@@ -636,7 +636,7 @@ async def test_admin_errors_lists_all_failed_tiers_and_filters_them(clients: Asy
await audit.drain()
d = (await clients.get("/admin/errors", headers=ADMIN)).json()
assert d["retention_days"] == A._ERROR_EVIDENCE_TTL_DAYS
assert d["retention_days"] == admin_routes._ERROR_EVIDENCE_TTL_DAYS
assert len(d["errors"]) == 3, "only failures carry evidence, across every tier"
assert {e["tier"] for e in d["errors"]} == {"platform", "credential", None}
@@ -659,7 +659,7 @@ async def test_evidence_ages_out_but_the_audit_row_survives(clients: AsyncClient
async with session_maker() as db:
row = (await db.execute(
select(CallRecord).order_by(CallRecord.id.desc()).limit(1))).scalars().first()
row.created_at = row.created_at - timedelta(days=A._ERROR_EVIDENCE_TTL_DAYS + 1)
row.created_at = row.created_at - timedelta(days=admin_routes._ERROR_EVIDENCE_TTL_DAYS + 1)
db.add(row)
await db.commit()
call_id, status = row.id, row.status_code
@@ -667,6 +667,6 @@ async def test_evidence_ages_out_but_the_audit_row_survives(clients: AsyncClient
assert (await clients.get("/admin/errors", headers=ADMIN)).json()["expired_rows_purged"] == 1
async with session_maker() as db:
aged = await db.get(CallRecord, call_id)
assert aged.error_response == A._ERROR_EVIDENCE_EXPIRED, "aged out, not silently NULL"
assert aged.error_response == admin_routes._ERROR_EVIDENCE_EXPIRED, "aged out, not silently NULL"
assert aged.status_code == status, "the rest of the audit row is untouched"
assert aged.endpoint_id == EP
tests/test_ledger.py
+2 -2
View File
@@ -475,7 +475,7 @@ async def test_a_grant_failure_after_staging_still_returns_the_signup(c: AsyncCl
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.api import REFERRAL_COOKIE
from treg.routers.signup_cookies import REFERRAL_COOKIE
from treg.models import Referral
_, ann_token = await _org(c, "ref-ann@superdesign.dev")
@@ -894,6 +894,6 @@ async def test_demo_orgs_get_no_promo_credit(c: AsyncClient):
await ledger._add_balance(db, org_id, -org.balance_micro) # zero it as a demo org would be
await db.commit()
# the hook is what enforces this — a demo org that somehow reaches it gets nothing
from treg.api import _grant_signup_promo
from treg.application.signup import _grant_signup_promo
await _grant_signup_promo(db, await db.get(Org, org_id))
assert await ledger.balance_of(db, org_id) == 0
tests/test_legacy_host_redirect.py
+4 -3
View File
@@ -136,7 +136,7 @@ async def test_canonical_and_legacy_resources_are_the_same_server(raw_client):
# A grant consented on one name must stay exchangeable/refreshable by a client re-based onto
# the other — in BOTH directions, and regardless of slash spelling. (Round-2's refactor of
# this helper silently dropped the cross-name rule; round-3 review caught it.)
from treg.api import _same_mcp_resource
from treg.routers.auth import _same_mcp_resource
canon, legacy = "https://treg.to/mcp/", "https://treg.superdesign.dev/mcp/"
assert _same_mcp_resource(canon, legacy)
assert _same_mcp_resource(legacy, canon)
@@ -151,7 +151,7 @@ async def test_login_round_trip_is_anchored_to_the_host_it_started_on(raw_client
# its callback exchange must keep naming that host, not public_url.
from starlette.requests import Request as StarletteRequest
from treg.api import _login_callback_base
from treg.routers.auth import _login_callback_base
def req(host: str) -> StarletteRequest:
return StarletteRequest({"type": "http", "method": "GET", "path": "/",
@@ -170,7 +170,8 @@ async def test_env_revert_is_a_complete_rollback(monkeypatch):
from httpx import ASGITransport, AsyncClient
from treg import mcp, mcp_oauth
from treg.api import _login_callback_base, app
from treg.api import app
from treg.routers.auth import _login_callback_base
monkeypatch.setenv("TREG_PUBLIC_URL", "https://treg.superdesign.dev")
get_settings.cache_clear()
tests/test_login_page.py
+1 -1
View File
@@ -267,7 +267,7 @@ async def test_approve_requires_a_started_login_and_matching_code(web):
async def test_wrong_code_attempts_are_capped(web):
"""Brute-forcing the short code is bounded: after CLI_APPROVE_MAX_TRIES misses the pending login is
discarded, so the real code can no longer be ground down (and the correct code then also fails)."""
from treg.api import CLI_APPROVE_MAX_TRIES
from treg.routers.auth import CLI_APPROVE_MAX_TRIES
uid = await _seed_user()
web.cookies.set("treg_session", sess.make(uid))
lid, code = await _start(web)
tests/test_marketplace_call.py
+98 -97
View File
@@ -22,8 +22,9 @@ from datetime import datetime, timezone
import pytest
from httpx import AsyncClient
from treg import api as A, audit, catalog_store, oauth_providers
from treg import api as A, audit, catalog_store, ledger, oauth_providers
from treg.application.call import resolve as call_resolution
from treg.application.call import settle as call_settle
from treg.application.call import service as call_service
from treg.application.call.types import ResolutionFailed, UpstreamResponse
from treg.routers import call as call_routes
@@ -201,7 +202,7 @@ def test_path_placeholders_fill_from_query_and_are_consumed():
def test_gtm_catalog_builds_hierarchy_from_atomic_ids_without_encoded_slashes():
ep = catalog_store.load().by_id["google-tag-manager.workspaces"]
url, consumed = A._marketplace_upstream(
url, consumed = call_resolution._marketplace_upstream(
ep,
oauth_providers.GOOGLE_TAG_MANAGER,
{"account_id": "123", "container_id": "456", "pageToken": "next"},
@@ -326,7 +327,7 @@ async def test_empty_balance_is_a_402_an_agent_can_act_on(clients: AsyncClient,
"""Out of money is not "no credential" — it names the balance, the price, and the way to fix it."""
org_id = (await clients.get("/orgs")).json()[0]["org_id"]
async with session_maker() as db: # spend the whole promo through the ledger's own front door
await A.ledger.reserve(db, org_id, "drain", 1_000_000)
await ledger.reserve(db, org_id, "drain", 1_000_000)
r = await clients.get(f"/call/{EP}?aweme_id=7")
assert r.status_code == 402, r.text
d = r.json()["detail"]
@@ -351,7 +352,7 @@ async def test_402_with_autotopup_on_names_the_policy_not_a_missing_card(clients
org.autotopup_amount_micro = 20_000_000
org.autotopup_threshold_micro = 5_000_000
await db.commit()
await A.ledger.reserve(db, org_id, "drain", 1_000_000)
await ledger.reserve(db, org_id, "drain", 1_000_000)
r = await clients.get(f"/call/{EP}?aweme_id=7")
assert r.status_code == 402, r.text
d = r.json()["detail"]
@@ -478,44 +479,44 @@ async def test_scrapecreators_settles_on_the_credits_it_charged(clients: AsyncCl
assert (await _telemetry(clients))["cost_observed_micro"] == 3 * EP_CALL_MICRO
def _mk(provider: str, **kw) -> A.MarketplaceCall:
def _mk(provider: str, **kw) -> call_resolution.MarketplaceCall:
"""A minimal MarketplaceCall for observed-cost tests — only the fields the settle math reads."""
kw.setdefault("tier", "platform")
kw.setdefault("endpoint_id", "ep") # hunter's derived cost is keyed on the endpoint, not just the provider
return A.MarketplaceCall(tool=None, upstream="", consumed=set(), provider=provider, **kw)
return call_resolution.MarketplaceCall(tool=None, upstream="", consumed=set(), provider=provider, **kw)
def test_observed_cost_only_trusts_a_real_number():
"""A missing, non-numeric or negative charge means "we never learned it" — settle at the estimate.
A reported ZERO is different: the provider is saying it did not charge, and is honoured."""
assert A._observed_cost_micro(_mk("dataforseo"), b'{"cost": 0}') == 0
assert A._observed_cost_micro(_mk("dataforseo"), b'{"cost": "0.5"}') is None
assert A._observed_cost_micro(_mk("dataforseo"), b'{"cost": -1}') is None
assert A._observed_cost_micro(_mk("dataforseo"), b"not json") is None
assert A._observed_cost_micro(_mk("dataforseo"), b"[1,2,3]") is None
assert A._observed_cost_micro(_mk("tikhub"), b'{"cost": 0.5}') is None, "tikhub doesn't report a charge"
assert call_settle._observed_cost_micro(_mk("dataforseo"), b'{"cost": 0}') == 0
assert call_settle._observed_cost_micro(_mk("dataforseo"), b'{"cost": "0.5"}') is None
assert call_settle._observed_cost_micro(_mk("dataforseo"), b'{"cost": -1}') is None
assert call_settle._observed_cost_micro(_mk("dataforseo"), b"not json") is None
assert call_settle._observed_cost_micro(_mk("dataforseo"), b"[1,2,3]") is None
assert call_settle._observed_cost_micro(_mk("tikhub"), b'{"cost": 0.5}') is None, "tikhub doesn't report a charge"
# exa reports dollars one level down; a 20-result search is the base plus ten $0.001 riders
assert A._observed_cost_micro(_mk("exa"), b'{"costDollars": {"total": 0.016, "search": {"neural": 0.016}}}') == 16_000
assert A._observed_cost_micro(_mk("exa"), b'{"costDollars": {"total": 0}}') == 0
assert A._observed_cost_micro(_mk("exa"), b'{"costDollars": {"total": "0.007"}}') is None
assert A._observed_cost_micro(_mk("exa"), b'{"costDollars": 0.007}') is None
assert A._observed_cost_micro(_mk("exa"), b'{"results": []}') is None
assert A._observed_cost_micro(_mk("scrapecreators"), b'{"credits_charged": 2}') == 2 * EP_CALL_MICRO
assert A._observed_cost_micro(_mk("scrapecreators"), b'{"success": true}') is None
assert call_settle._observed_cost_micro(_mk("exa"), b'{"costDollars": {"total": 0.016, "search": {"neural": 0.016}}}') == 16_000
assert call_settle._observed_cost_micro(_mk("exa"), b'{"costDollars": {"total": 0}}') == 0
assert call_settle._observed_cost_micro(_mk("exa"), b'{"costDollars": {"total": "0.007"}}') is None
assert call_settle._observed_cost_micro(_mk("exa"), b'{"costDollars": 0.007}') is None
assert call_settle._observed_cost_micro(_mk("exa"), b'{"results": []}') is None
assert call_settle._observed_cost_micro(_mk("scrapecreators"), b'{"credits_charged": 2}') == 2 * EP_CALL_MICRO
assert call_settle._observed_cost_micro(_mk("scrapecreators"), b'{"success": true}') is None
# akta reports `credits_consumed` — the field that makes its per-section enrich billable at
# actuals rather than the catalog's upper-bound estimate. $0.05/credit (fx.yaml).
assert A._observed_cost_micro(_mk("akta"), b'{"credits_consumed": 0.5}') == 25_000
assert A._observed_cost_micro(_mk("akta"), b'{"credits_consumed": 0}') == 0, "a reported zero is honoured"
assert A._observed_cost_micro(_mk("akta"), b'{"credits_charged": 2}') is None, "wrong field name means we never learned it"
assert call_settle._observed_cost_micro(_mk("akta"), b'{"credits_consumed": 0.5}') == 25_000
assert call_settle._observed_cost_micro(_mk("akta"), b'{"credits_consumed": 0}') == 0, "a reported zero is honoured"
assert call_settle._observed_cost_micro(_mk("akta"), b'{"credits_charged": 2}') is None, "wrong field name means we never learned it"
def test_crustdata_settles_from_the_response_credit_header():
"""Crustdata's body has no billing field; X-Credits-Used is the exact call charge."""
mk = _mk("crustdata", endpoint_id="crustdata.companies.search")
assert A._observed_cost_micro(
assert call_settle._observed_cost_micro(
mk, b'{"rows": []}', httpx.Headers({"X-Credits-Used": "0.03"})) == 9_000
assert A._observed_cost_micro(mk, b'{"rows": []}', httpx.Headers()) is None
assert A._observed_cost_micro(
assert call_settle._observed_cost_micro(mk, b'{"rows": []}', httpx.Headers()) is None
assert call_settle._observed_cost_micro(
mk, b'{"rows": []}', httpx.Headers({"X-Credits-Used": "not-a-number"})) is None
@@ -525,7 +526,7 @@ def test_aviato_conditional_prices_follow_live_balance_deltas():
def price(endpoint_id, query=None, body=None):
ep = cat.by_id[endpoint_id]
cv = cat.cost_view(ep["cost"], "aviato")
return A._marketplace_pricing(
return call_resolution._marketplace_pricing(
"aviato", endpoint_id, cv, query or {}, json.dumps(body or {}).encode())
assert price("aviato.companies.enrich", {"preview": "true"}) == (0, 0)
@@ -544,19 +545,19 @@ def test_aviato_conditional_prices_follow_live_balance_deltas():
def test_aviato_bulk_settles_from_counts_and_simple_search_releases_unbilled_rider():
companies = _mk("aviato", endpoint_id="aviato.companies.enrich.bulk", unit_micro=150_000)
assert A._observed_cost_micro(companies, b'{"companies": [{"id": "1"}, null]}') == 150_000
assert call_settle._observed_cost_micro(companies, b'{"companies": [{"id": "1"}, null]}') == 150_000
people = _mk("aviato", endpoint_id="aviato.people.enrich.bulk", unit_micro=70_000)
assert A._observed_cost_micro(people, b'[{"id": "1"}, null]') == 70_000
assert call_settle._observed_cost_micro(people, b'[{"id": "1"}, null]') == 70_000
simple = _mk("aviato", endpoint_id="aviato.people.search.simple", unit_micro=0)
assert A._observed_cost_micro(simple, b'{"items": [{"id":"1"},{"id":"2"},{"id":"3"},'
assert call_settle._observed_cost_micro(simple, b'{"items": [{"id":"1"},{"id":"2"},{"id":"3"},'
b'{"id":"4"},{"id":"5"}]}') == 2_500
def test_aviato_single_enrich_releases_documented_but_live_unbilled_riders():
company = _mk("aviato", endpoint_id="aviato.companies.enrich", unit_micro=150_000)
assert A._observed_cost_micro(company, b'{"id":"company"}') == 150_000
assert call_settle._observed_cost_micro(company, b'{"id":"company"}') == 150_000
person = _mk("aviato", endpoint_id="aviato.people.enrich", unit_micro=80_000)
assert A._observed_cost_micro(person, b'{"id":"person"}') == 80_000
assert call_settle._observed_cost_micro(person, b'{"id":"person"}') == 80_000
def test_observed_cost_counts_resources_for_billed_oauth_reads():
@@ -565,25 +566,25 @@ def test_observed_cost_counts_resources_for_billed_oauth_reads():
zero, and a single-object `data` (a profile read) at one. Anything unparseable falls back to
the estimate (None), and a non-per_result billed call never counts."""
x = _mk("x", tier="tool", billed_oauth=True, cost_type="per_result", unit_micro=5_000)
assert A._observed_cost_micro(x, b'{"data": [{}, {}, {}]}') == 15_000
assert A._observed_cost_micro(x, b'{"data": []}') == 0
assert A._observed_cost_micro(x, b'{"data": {"id": "1"}}') == 5_000
assert A._observed_cost_micro(x, b'{"errors": [{}]}') == 0, "no data key = nothing served"
assert A._observed_cost_micro(x, b"not json") is None, "unreadable body settles at the estimate"
assert call_settle._observed_cost_micro(x, b'{"data": [{}, {}, {}]}') == 15_000
assert call_settle._observed_cost_micro(x, b'{"data": []}') == 0
assert call_settle._observed_cost_micro(x, b'{"data": {"id": "1"}}') == 5_000
assert call_settle._observed_cost_micro(x, b'{"errors": [{}]}') == 0, "no data key = nothing served"
assert call_settle._observed_cost_micro(x, b"not json") is None, "unreadable body settles at the estimate"
write = _mk("x", tier="tool", billed_oauth=True, cost_type="per_call", unit_micro=0)
assert A._observed_cost_micro(write, b'{"data": {"id": "1"}}') is None, "per_call settles at the estimate"
assert call_settle._observed_cost_micro(write, b'{"data": {"id": "1"}}') is None, "per_call settles at the estimate"
# leadmagic reports `credits_consumed` too — including 0 on a 2xx miss (observed at verify
# time) and fractions (email verify = 0.25 credits). $0.025/credit (fx.yaml).
assert A._observed_cost_micro(_mk("leadmagic"), b'{"credits_consumed": 1}') == 25_000
assert A._observed_cost_micro(_mk("leadmagic"), b'{"credits_consumed": 0}') == 0, "a 2xx miss is free"
assert A._observed_cost_micro(_mk("leadmagic"), b'{"credits_consumed": 0.25}') == 6_250
assert call_settle._observed_cost_micro(_mk("leadmagic"), b'{"credits_consumed": 1}') == 25_000
assert call_settle._observed_cost_micro(_mk("leadmagic"), b'{"credits_consumed": 0}') == 0, "a 2xx miss is free"
assert call_settle._observed_cost_micro(_mk("leadmagic"), b'{"credits_consumed": 0.25}') == 6_250
# lusha nests the same contract one level down: billing.creditsCharged — 0 on a 2xx miss
# (the captured people.enrich example is one), 2 credits on a company enrich. $0.1248/credit.
assert A._observed_cost_micro(_mk("lusha"), b'{"billing": {"creditsCharged": 1, "resultsReturned": 10}}') == 124_800
assert A._observed_cost_micro(_mk("lusha"), b'{"billing": {"creditsCharged": 0, "resultsReturned": 0}}') == 0, "a 2xx miss is free"
assert A._observed_cost_micro(_mk("lusha"), b'{"billing": {"creditsCharged": 2}}') == 249_600
assert A._observed_cost_micro(_mk("lusha"), b'{"requestId": "x"}') is None, "no billing block means we never learned it"
assert call_settle._observed_cost_micro(_mk("lusha"), b'{"billing": {"creditsCharged": 1, "resultsReturned": 10}}') == 124_800
assert call_settle._observed_cost_micro(_mk("lusha"), b'{"billing": {"creditsCharged": 0, "resultsReturned": 0}}') == 0, "a 2xx miss is free"
assert call_settle._observed_cost_micro(_mk("lusha"), b'{"billing": {"creditsCharged": 2}}') == 249_600
assert call_settle._observed_cost_micro(_mk("lusha"), b'{"requestId": "x"}') is None, "no billing block means we never learned it"
def test_apollo_settles_a_2xx_miss_at_zero():
@@ -593,12 +594,12 @@ def test_apollo_settles_a_2xx_miss_at_zero():
carrying neither documented shape (people enrichment's 1-9 credit range) stays at the
estimate — deriving is only safe where the rule is flat."""
credit = 26_000 # $0.026/credit (fx.yaml, Basic $65/mo / 2,500 credits)
assert A._observed_cost_micro(_mk("apollo"), b'{"organization": {"name": "Apple"}}') == credit
assert A._observed_cost_micro(_mk("apollo"), b'{"organization": null}') == 0, "a 2xx miss is free"
assert A._observed_cost_micro(_mk("apollo"), b'{"organizations": [{"name": "Apple"}], "pagination": {}}') == credit
assert A._observed_cost_micro(_mk("apollo"), b'{"organizations": [], "pagination": {}}') == 0, "an empty page is free"
assert A._observed_cost_micro(_mk("apollo"), b'{"person": {"id": "x"}}') is None, "1-9 credit range: estimate, not a guess"
assert A._observed_cost_micro(_mk("apollo"), b"not json") is None
assert call_settle._observed_cost_micro(_mk("apollo"), b'{"organization": {"name": "Apple"}}') == credit
assert call_settle._observed_cost_micro(_mk("apollo"), b'{"organization": null}') == 0, "a 2xx miss is free"
assert call_settle._observed_cost_micro(_mk("apollo"), b'{"organizations": [{"name": "Apple"}], "pagination": {}}') == credit
assert call_settle._observed_cost_micro(_mk("apollo"), b'{"organizations": [], "pagination": {}}') == 0, "an empty page is free"
assert call_settle._observed_cost_micro(_mk("apollo"), b'{"person": {"id": "x"}}') is None, "1-9 credit range: estimate, not a guess"
assert call_settle._observed_cost_micro(_mk("apollo"), b"not json") is None
def test_hunter_domain_search_settles_on_the_emails_it_returned():
@@ -609,19 +610,19 @@ def test_hunter_domain_search_settles_on_the_emails_it_returned():
and one email costs the same whole credit ten do."""
credit = 24_500 # $0.0245/credit (fx.yaml, Starter $49/mo / 2,000 credits)
h = _mk("hunter", endpoint_id="hunter.companies.emails", cost_type="per_result")
assert A._observed_cost_micro(h, b'{"data": {"domain": "x.com", "emails": []}}') == 0, \
assert call_settle._observed_cost_micro(h, b'{"data": {"domain": "x.com", "emails": []}}') == 0, \
"a domain with no results is free — the catalog says so and Hunter bills so"
assert A._observed_cost_micro(h, b'{"data": {"emails": [{"value": "a@x.com"}]}}') == credit, \
assert call_settle._observed_cost_micro(h, b'{"data": {"emails": [{"value": "a@x.com"}]}}') == credit, \
"one email costs a whole search credit, not a tenth of one"
def _emails(n: int) -> bytes:
return json.dumps({"data": {"emails": [{"value": f"p{i}@x.com"} for i in range(n)]}}).encode()
assert A._observed_cost_micro(h, _emails(10)) == credit, "ten still fit in one credit"
assert A._observed_cost_micro(h, _emails(11)) == 2 * credit, "the 11th rounds up to a second credit"
assert A._observed_cost_micro(h, b'{"errors": [{"code": "wrong_params"}]}') is None, \
assert call_settle._observed_cost_micro(h, _emails(10)) == credit, "ten still fit in one credit"
assert call_settle._observed_cost_micro(h, _emails(11)) == 2 * credit, "the 11th rounds up to a second credit"
assert call_settle._observed_cost_micro(h, b'{"errors": [{"code": "wrong_params"}]}') is None, \
"no emails key at all: we never learned the count, settle at the estimate"
assert A._observed_cost_micro(h, b"not json") is None
assert call_settle._observed_cost_micro(h, b"not json") is None
other = _mk("hunter", endpoint_id="hunter.people.email.verify", cost_type="per_call")
assert A._observed_cost_micro(other, b'{"data": {"emails": []}}') is None, \
assert call_settle._observed_cost_micro(other, b'{"data": {"emails": []}}') is None, \
"only domain search bills per 10 returned; every other hunter route settles at its estimate"
@@ -632,14 +633,14 @@ def test_hunter_email_finder_miss_is_free():
nothing on. The body is the only place the found/missed distinction exists."""
credit = 24_500 # $0.0245/credit (fx.yaml, Starter $49/mo / 2,000 credits)
f = _mk("hunter", endpoint_id="hunter.people.email.find", cost_type="per_success")
assert A._observed_cost_micro(f, b'{"data": {"email": "a@x.com", "score": 92}}') == credit
assert A._observed_cost_micro(f, b'{"data": {"email": null, "score": null}}') == 0, \
assert call_settle._observed_cost_micro(f, b'{"data": {"email": "a@x.com", "score": 92}}') == credit
assert call_settle._observed_cost_micro(f, b'{"data": {"email": null, "score": null}}') == 0, \
"a miss is free — the catalog says so and Hunter bills so"
assert A._observed_cost_micro(f, b'{"data": {"email": "", "score": null}}') == 0, \
assert call_settle._observed_cost_micro(f, b'{"data": {"email": "", "score": null}}') == 0, \
"an empty string is a miss too"
assert A._observed_cost_micro(f, b'{"errors": [{"code": "wrong_params"}]}') is None, \
assert call_settle._observed_cost_micro(f, b'{"errors": [{"code": "wrong_params"}]}') is None, \
"no email key at all: we never learned the outcome, settle at the estimate"
assert A._observed_cost_micro(f, b"not json") is None
assert call_settle._observed_cost_micro(f, b"not json") is None
def test_tikhub_envelope_no_charge_settles_at_zero():
@@ -648,14 +649,14 @@ def test_tikhub_envelope_no_charge_settles_at_zero():
(verified live 2026-07-30), so those settle at the estimate, faithfully. Only the explicit
no-charge phrasing settles at zero."""
t = _mk("tikhub", cost_type="per_success")
assert A._observed_cost_micro(t, b'{"code": 200, "message": "Request successful. This request will incur a charge.", "data": {}}') is None, \
assert call_settle._observed_cost_micro(t, b'{"code": 200, "message": "Request successful. This request will incur a charge.", "data": {}}') is None, \
"a billed answer settles at the estimate — that IS what TikHub takes"
assert A._observed_cost_micro(t, b'{"code": 200, "message": "Request successful. This request will incur a charge.", "data": {"error": "dead_page"}}') is None, \
assert call_settle._observed_cost_micro(t, b'{"code": 200, "message": "Request successful. This request will incur a charge.", "data": {"error": "dead_page"}}') is None, \
"a dead page TikHub bills us for is passed through, not eaten"
assert A._observed_cost_micro(t, b'{"code": 400, "message": "Request failed. You won\'t be charged for this request.", "data": null}') == 0
assert A._observed_cost_micro(t, b'{"code": 200, "message": "This request will not incur charges.", "data": {}}') == 0
assert A._observed_cost_micro(t, b'{"code": 200, "data": {}}') is None, "no message: estimate"
assert A._observed_cost_micro(t, b"not json") is None
assert call_settle._observed_cost_micro(t, b'{"code": 400, "message": "Request failed. You won\'t be charged for this request.", "data": null}') == 0
assert call_settle._observed_cost_micro(t, b'{"code": 200, "message": "This request will not incur charges.", "data": {}}') == 0
assert call_settle._observed_cost_micro(t, b'{"code": 200, "data": {}}') is None, "no message: estimate"
assert call_settle._observed_cost_micro(t, b"not json") is None
async def test_hunter_zero_result_search_costs_nothing(clients: AsyncClient, platform_on, monkeypatch):
@@ -727,7 +728,7 @@ async def test_daily_cap_refuses_when_it_cannot_be_verified(clients: AsyncClient
async def _boom(db, org_id):
raise RuntimeError("ledger unavailable")
monkeypatch.setattr(A.ledger, "spent_today", _boom)
monkeypatch.setattr(ledger, "spent_today", _boom)
r = await clients.get(f"/call/{EP}?aweme_id=7")
assert r.status_code == 429
assert "refusing to spend" in r.json()["detail"]
@@ -814,9 +815,9 @@ def test_local_run_cannot_export_a_platform_binding():
from treg import localrun
from treg.models import Tool
provider = A.oauth_providers.get("tikhub")
provider = oauth_providers.get("tikhub")
tool = Tool(org_id=1, name=EP, base_url=provider.base_url, host="api.tikhub.io",
bindings=A._platform_bindings(provider),
bindings=call_resolution._platform_bindings(provider),
cli={"enabled": True, "bin": "sh", "inject": [{"via": "env", "name": "TIKHUB_API_KEY"}]})
assert all(b.get("secret_id") is None for b in tool.bindings)
assert localrun._resolve_secret_id(tool.cli["inject"][0], tool) is None
@@ -826,14 +827,14 @@ def test_platform_estimate_normalizes_per_result_pricing():
"""A per-row price needs a row count: the caller's own limit param, else a page, and capped so one
call can't reserve an org's whole balance."""
per_call = {"type": "per_call", "usd": 0.002}
assert A._platform_estimate_micro(per_call, {}) == 2_000
assert call_resolution._platform_estimate_micro(per_call, {}) == 2_000
per_row = {"type": "per_result", "usd": 0.0001}
assert A._platform_estimate_micro(per_row, {}) == 0.0001 * A._PLATFORM_PAGE_DEFAULT * 1_000_000
assert A._platform_estimate_micro(per_row, {"limit": "5"}) == 500
assert A._platform_estimate_micro(per_row, {"limit": "100000"}) == 0.0001 * A._PLATFORM_PAGE_MAX * 1_000_000
assert A._platform_estimate_micro({"type": "per_call", "usd": None}, {}) == 0
assert call_resolution._platform_estimate_micro(per_row, {}) == 0.0001 * call_resolution._PLATFORM_PAGE_DEFAULT * 1_000_000
assert call_resolution._platform_estimate_micro(per_row, {"limit": "5"}) == 500
assert call_resolution._platform_estimate_micro(per_row, {"limit": "100000"}) == 0.0001 * call_resolution._PLATFORM_PAGE_MAX * 1_000_000
assert call_resolution._platform_estimate_micro({"type": "per_call", "usd": None}, {}) == 0
# rounds UP — a sub-micro fraction must never round to free
assert A._platform_estimate_micro({"type": "per_call", "usd": 0.0000005}, {}) == 1
assert call_resolution._platform_estimate_micro({"type": "per_call", "usd": 0.0000005}, {}) == 1
def test_brightdata_platform_key_injects_as_bearer(platform_on):
@@ -841,14 +842,14 @@ def test_brightdata_platform_key_injects_as_bearer(platform_on):
field is found by name (`platform_key_for`) and the header shape comes from the registry entry,
so this is the regression guard on the generic path staying generic."""
assert get_settings().platform_key_for("brightdata") == PLATFORM_KEYS["BRIGHTDATA"]
assert A._platform_bindings(A.oauth_providers.get("brightdata")) == [
assert call_resolution._platform_bindings(oauth_providers.get("brightdata")) == [
{"platform_setting": "platform_key_brightdata", "injector": "env", "location": "header",
"name": "Authorization", "format": "Bearer {secret}"}]
def test_crustdata_platform_key_keeps_the_required_version_header():
"""Tier 4 must speak the same provider protocol as BYOK, not only inject the key."""
assert A._platform_bindings(A.oauth_providers.get("crustdata")) == [
assert call_resolution._platform_bindings(oauth_providers.get("crustdata")) == [
{"platform_setting": "platform_key_crustdata", "injector": "env", "location": "header",
"name": "Authorization", "format": "Bearer {secret}"},
{"platform_setting": "platform_key_crustdata", "injector": "env", "location": "header",
@@ -876,9 +877,9 @@ def test_brightdata_estimate_counts_the_body_array():
"""Bright Data bills per record delivered and takes its targets as a bare JSON array, so the
reserve has to scale with the array's LENGTH — there is no limit param in the query to read."""
cost = {"type": "per_result", "usd": 0.0015}
assert A._platform_estimate_micro(cost, {}, json.dumps([{"url": "a"}]).encode()) == 1_500
assert call_resolution._platform_estimate_micro(cost, {}, json.dumps([{"url": "a"}]).encode()) == 1_500
five = json.dumps([{"url": u} for u in "abcde"]).encode()
assert A._platform_estimate_micro(cost, {}, five) == 7_500
assert call_resolution._platform_estimate_micro(cost, {}, five) == 7_500
def test_brightdata_documented_prices_are_billable(platform_on):
@@ -960,7 +961,7 @@ async def test_one_caller_cannot_reuse_a_key_twice(clients: AsyncClient):
async def test_deleting_a_team_takes_its_remembered_answers(clients: AsyncClient):
"""A stored response belongs to the team that paid for it. Left behind it is a dangling row
holding someone's data after they asked to be gone."""
from treg.api import _ORG_SCOPED_MODELS
from treg.routers.orgs import _ORG_SCOPED_MODELS
from treg.models import IdempotentCall
assert IdempotentCall in _ORG_SCOPED_MODELS
@@ -1296,7 +1297,7 @@ def test_the_billability_truth_table():
(302, "per_call", False),
]
for status, cost_type, expected in cases:
got = A._platform_billable(status, cost_type)
got = call_settle._platform_billable(status, cost_type)
assert got is expected, f"({status}, {cost_type}) -> {got}, expected {expected}"
@@ -1450,8 +1451,8 @@ def test_x_catalog_price_equals_what_the_meter_charges():
x = oauth_providers.get("x")
for ep in _x_endpoints():
method = (ep.get("method") or "GET").upper()
est, ctype, _ = A._oauth_billed_estimate(x, ep, method, {}, b"")
published = A._platform_estimate_micro(
est, ctype, _ = call_resolution._oauth_billed_estimate(x, ep, method, {}, b"")
published = call_resolution._platform_estimate_micro(
A.catalog_store.load().cost_view(ep["cost"], "x"), {}, b"")
assert est == published and ctype == ep["cost"]["type"], (
f"{ep['id']}: catalog says {published} micro ({ep['cost']['type']}), "
@@ -1465,8 +1466,8 @@ def test_a_zero_price_on_a_billed_provider_falls_back_rather_than_billing_zero()
x = oauth_providers.get("x")
ep = {"id": "x.x.stale", "provider": "x", "method": "GET", "path": "/2/tweets",
"cost": {"type": "free", "value": 0, "currency": "USD", "unit": "call"}}
est, ctype, unit = A._oauth_billed_estimate(x, ep, "GET", {}, b"")
assert est > 0 and ctype == "per_result" and unit == A._usd_to_micro(x.billed_read_usd)
est, ctype, unit = call_resolution._oauth_billed_estimate(x, ep, "GET", {}, b"")
assert est > 0 and ctype == "per_result" and unit == call_resolution._usd_to_micro(x.billed_read_usd)
# ---- X end to end: the published price is the price the balance loses --------------------------
@@ -1572,30 +1573,30 @@ def test_observed_cost_counts_brightdata_records():
consumed upstream vs $0.35 billed over three weeks (2026-08-24)."""
bd = _mk("brightdata", cost_type="per_result", unit_micro=1500)
# sync scrape / snapshot download, format=json: an array, one element per record
assert A._observed_cost_micro(bd, b'[{"url": "a"}, {"url": "b"}, {"url": "c"}]') == 4500
assert A._observed_cost_micro(bd, b"[]") == 0
assert call_settle._observed_cost_micro(bd, b'[{"url": "a"}, {"url": "b"}, {"url": "c"}]') == 4500
assert call_settle._observed_cost_micro(bd, b"[]") == 0
# the >60s sync fallback and /trigger hand back a snapshot id: zero records HERE — they bill
# when the snapshot is downloaded
assert A._observed_cost_micro(bd, b'{"snapshot_id": "sd_x"}') == 0
assert call_settle._observed_cost_micro(bd, b'{"snapshot_id": "sd_x"}') == 0
# an early snapshot download answers the job's state, not rows: nothing delivered, nothing billed
assert A._observed_cost_micro(bd, b'{"status": "running", "message": "not ready"}') == 0
assert call_settle._observed_cost_micro(bd, b'{"status": "running", "message": "not ready"}') == 0
# ndjson: one record per line
assert A._observed_cost_micro(bd, b'{"url": "a"}\n{"url": "b"}\n') == 3000
assert call_settle._observed_cost_micro(bd, b'{"url": "a"}\n{"url": "b"}\n') == 3000
# csv: header + rows
assert A._observed_cost_micro(bd, b"url,name\na,x\nb,y\n") == 3000
assert call_settle._observed_cost_micro(bd, b"url,name\na,x\nb,y\n") == 3000
# a payload the 8MB metered buffer truncated must settle at the estimate, never a partial count
assert A._observed_cost_micro(bd, b'[{"url": "a"}, {"url"') is None
assert call_settle._observed_cost_micro(bd, b'[{"url": "a"}, {"url"') is None
# gzipped (compress=true): can't count, estimate wins
assert A._observed_cost_micro(bd, b"\x1f\x8b\x08\x00junk") is None
assert call_settle._observed_cost_micro(bd, b"\x1f\x8b\x08\x00junk") is None
# a free management route (progress polls) never reaches the counter
assert A._observed_cost_micro(_mk("brightdata", cost_type="free"), b'{"status": "ready"}') is None
assert call_settle._observed_cost_micro(_mk("brightdata", cost_type="free"), b'{"status": "ready"}') is None
def test_marketplace_resolution_carries_the_per_row_price():
"""`unit_micro` must ride the MarketplaceCall on every tier for per_result endpoints — a settle
that can't see the row price can only ever bill the estimate."""
cv = {"type": "per_result", "usd": 0.0015}
assert A._usd_to_micro(cv["usd"]) == 1500
assert call_resolution._usd_to_micro(cv["usd"]) == 1500
async def test_brightdata_sync_scrape_settles_per_record_through_the_ledger(
tests/test_mcp_directory.py
+3 -3
View File
@@ -10,8 +10,8 @@ from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from starlette.routing import Mount
from treg import api as treg_api
from treg import mcp, mcp_oauth
from treg.routers import auth as auth_routes
from treg.bootstrap import create_app
from treg.config import Settings, get_settings
@@ -114,7 +114,7 @@ async def test_v2_feature_flag_disables_mount_metadata_grants_and_catalog_route(
assert "not enabled" in metadata.json()["detail"]
resource = mcp_oauth.mcp_resource_url("v2")
assert "not enabled" in treg_api._wrong_resource(resource)
assert "not enabled" in auth_routes._wrong_resource(resource)
direct = await clients.get("/catalog/call/tikhub.tiktok.video.comments?aweme_id=7")
assert direct.status_code == 404
@@ -150,7 +150,7 @@ async def test_v2_feature_flag_enables_mount_metadata_and_resource(monkeypatch):
challenge.headers["www-authenticate"]
resource = mcp_oauth.mcp_resource_url("v2")
assert treg_api._wrong_resource(resource) is None
assert auth_routes._wrong_resource(resource) is None
finally:
get_settings.cache_clear()
tests/test_mcp_oauth.py
+2 -2
View File
@@ -16,8 +16,8 @@ import time
import pytest
from treg import api as treg_api
from treg import mcp, mcp_oauth, session
from treg.routers import auth as auth_routes
from treg.config import Settings, get_settings
# The MCP transport helpers live with the MCP tests; a token is only interesting here because it can
@@ -444,7 +444,7 @@ async def test_v2_scope_selects_v2_when_claude_omits_the_resource(clients):
def test_explicit_resource_wins_over_the_v2_scope_marker():
v1 = mcp_oauth.mcp_resource_url("v1")
assert treg_api._effective_mcp_resource(v1, mcp_oauth.DIRECTORY_SCOPE) == v1
assert auth_routes._effective_mcp_resource(v1, mcp_oauth.DIRECTORY_SCOPE) == v1
async def test_a_code_can_be_redeemed_only_ONCE(clients):
tests/test_oauth_billed.py
+10 -10
View File
@@ -17,7 +17,7 @@ import pytest
from httpx import AsyncClient
from sqlalchemy import update
from treg import api as A
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
@@ -205,20 +205,20 @@ async def test_empty_balance_is_an_actionable_402(clients: AsyncClient, billed_o
# ---- pricing units -----------------------------------------------------------------------------
def test_post_has_link_sniffs_only_the_text_field():
assert A._post_has_link(b'{"text": "see https://a.example"}')
assert A._post_has_link(b'{"text": "see www.example.com"}')
assert not A._post_has_link(b'{"text": "no links here"}')
assert not A._post_has_link(b'{"text": "plain", "quote_tweet_id": "123"}')
assert not A._post_has_link(b'not json')
assert not A._post_has_link(b"")
assert call_resolution._post_has_link(b'{"text": "see https://a.example"}')
assert call_resolution._post_has_link(b'{"text": "see www.example.com"}')
assert not call_resolution._post_has_link(b'{"text": "no links here"}')
assert not call_resolution._post_has_link(b'{"text": "plain", "quote_tweet_id": "123"}')
assert not call_resolution._post_has_link(b'not json')
assert not call_resolution._post_has_link(b"")
def test_billed_endpoint_match_prefers_exact_over_template():
ep = A._billed_endpoint_match("x", "GET", "/2/users/me")
ep = call_resolution._billed_endpoint_match("x", "GET", "/2/users/me")
assert ep and ep["id"] == "x.x.user.profile"
ep = A._billed_endpoint_match("x", "GET", "/2/users/44196397/tweets")
ep = call_resolution._billed_endpoint_match("x", "GET", "/2/users/44196397/tweets")
assert ep and ep["id"] == "x.x.user.posts"
assert A._billed_endpoint_match("x", "GET", "/2/no/such/route") is None
assert call_resolution._billed_endpoint_match("x", "GET", "/2/no/such/route") is None
def _stub_relay(status_code: int, body: bytes):
tests/test_orgs.py
+1 -1
View File
@@ -16,7 +16,7 @@ async def test_org_delete_clears_EVERY_org_scoped_table(clients):
from sqlmodel import SQLModel
from treg import models as m
from treg.api import _ORG_SCOPED_MODELS
from treg.routers.orgs import _ORG_SCOPED_MODELS
covered = {model.__name__ for model in _ORG_SCOPED_MODELS}
missing = []
tests/test_provider_pages.py
+4 -4
View File
@@ -13,7 +13,7 @@ import re
from httpx import AsyncClient
from treg import api
from treg.routers import web as web_routes
from treg.config import get_settings
@@ -23,13 +23,13 @@ def _ld(html: str) -> list[dict]:
def _first_provider() -> str:
rows = api._provider_rows()
rows = web_routes._provider_rows()
assert rows, "catalog has no providers"
return rows[0]["service"]
async def test_every_provider_page_renders(clients: AsyncClient):
for row in api._provider_rows():
for row in web_routes._provider_rows():
r = await clients.get(f"/tools/{row['service']}")
assert r.status_code == 200, (row["service"], r.status_code)
assert f'href="/tools/{row["service"]}"' in r.text or "canonical" in r.text
@@ -75,7 +75,7 @@ async def test_pricing_page(clients: AsyncClient):
async def test_sitemap_and_catalog_link_every_provider_page(clients: AsyncClient):
services = [r["service"] for r in api._provider_rows()]
services = [r["service"] for r in web_routes._provider_rows()]
sm = (await clients.get("/sitemap.xml")).text
cat = (await clients.get("/catalog")).text
for s in services:
tests/test_referrals.py
+1 -1
View File
@@ -69,7 +69,7 @@ async def _signup(c: AsyncClient, email: str, *, ref: str = "") -> tuple[int, st
def referrals_cookie() -> str:
from treg.api import REFERRAL_COOKIE
from treg.routers.signup_cookies import REFERRAL_COOKIE
return REFERRAL_COOKIE
tests/test_router_dependencies.py
-27
View File
@@ -1,27 +0,0 @@
"""Compatibility checks for the identity-domain dependency extraction."""
from treg import api, timeutil
from treg.domain.identity import access
from treg.routers import auth_helpers
def test_api_reexports_shared_http_dependencies() -> None:
names = (
"Caller",
"_membership_by_token",
"_resolve_org",
"_role_at_least",
"_user_from_identity_token",
"_user_from_session",
"require_identity",
"require_member",
"require_superadmin",
)
for name in names:
assert getattr(api, name) is getattr(access, name)
assert api._is_https is auth_helpers._is_https
def test_api_reexports_shared_time_convention() -> None:
assert api._utcnow_naive is timeutil.utcnow_naive
assert api._as_naive is timeutil.as_naive
tests/test_sandbox.py
+3 -2
View File
@@ -15,7 +15,8 @@ from sqlalchemy import select
from conftest import make_upstream
from treg import api, sandbox
from treg import sandbox
from treg.routers.onboard import SANDBOX_RATE_MAX
from treg.api import app
from treg.db import reset_db, session_maker
from treg.models import Secret, Tool, User
@@ -141,7 +142,7 @@ async def test_export_skill_rejects_non_sandbox(anon):
async def test_rate_limited_per_ip(anon):
for _ in range(api.SANDBOX_RATE_MAX):
for _ in range(SANDBOX_RATE_MAX):
assert (await anon.post("/demo/sandbox")).status_code == 200
rejected = await anon.post("/demo/sandbox")
assert rejected.status_code == 429
tests/test_seo.py
+2 -2
View File
@@ -321,7 +321,7 @@ def test_no_shelf_is_published_that_the_app_grid_hides():
would still render and the sitemap would still publish it — but nothing in the app's own grid
would link to it. Give the new platform a label and category in capabilities.yaml.
"""
from treg.api import _platform_rows
from treg.routers.catalog import _platform_rows
orphans = [r["slug"] for r in _platform_rows() if r["category"] == "Other"]
assert not orphans, (
f"{orphans} have endpoints but no capabilities.yaml `platforms:` entry — the sitemap will "
@@ -333,7 +333,7 @@ def test_no_shelf_is_published_that_the_app_grid_hides():
# suite reads as text (see tests/test_dashboard_markup.py). Both were reported from the browser.
def _spa() -> str:
from treg.api import _WEB_DIR
from treg.routers.web import _WEB_DIR
return (_WEB_DIR / "index.html").read_text(encoding="utf-8")
tests/test_tag_billing.py
+6 -5
View File
@@ -17,8 +17,9 @@ import pytest
from httpx import AsyncClient
from sqlmodel import select
from treg import api as A, audit, crypto, ledger
from treg import audit, crypto, 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
@@ -779,10 +780,10 @@ async def test_a_caller_input_4xx_is_still_billed_on_a_per_call_endpoint(clients
r = await clients.get(f"/call/{EP}?aweme_id=7", headers={"X-Treg-Meta": "customer=cust_A"})
assert r.status_code == 400
# tikhub comments is per_success, so 400 releases; assert the RULE directly for per_call.
assert A._platform_billable(400, "per_call") is True
assert A._platform_billable(404, "per_call") is True
assert A._platform_billable(402, "per_call") is False
assert A._platform_billable(401, "per_result") is False
assert call_settle._platform_billable(400, "per_call") is True
assert call_settle._platform_billable(404, "per_call") is True
assert call_settle._platform_billable(402, "per_call") is False
assert call_settle._platform_billable(401, "per_result") is False
# ---- per-dimension defaults with overrides -------------------------------------------------------
tests/test_tag_billing_adversarial.py
+14 -11
View File
@@ -14,8 +14,11 @@ from sqlalchemy.ext.asyncio import AsyncSession
from sqlmodel import select
from starlette.requests import Request
from treg import api as A, audit, crypto, ledger, localproxy
from treg import audit, crypto, ledger, localproxy
from treg.application.call import idempotency as call_idem
from treg.application.call import reserve as call_reserve
from treg.application.call import service as call_service
from treg.application.call.intake import CallMeta
from treg.application.call.types import UpstreamResponse
from treg.routers import call as call_routes
from treg.config import get_settings
@@ -64,7 +67,7 @@ def _request_with_meta(value: str) -> Request:
def _assert_meta_rejected(value: str) -> None:
with pytest.raises(HTTPException) as exc:
A._parse_call_meta(_request_with_meta(value))
call_routes._parse_call_meta(_request_with_meta(value))
assert exc.value.status_code == 422
@@ -91,8 +94,8 @@ async def test_attack_1_all_ingress_paths_reject_storage_key_delimiters(
budget_policy._validate_tag_pair("customer", value)
assert exc.value.status_code == 422
meta = A._parse_call_meta(_request_with_meta("customer=safe_value"))
stored_key = A._scoped_idempotency_key("retry-1", meta)
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", ","))
@@ -205,7 +208,7 @@ async def test_attack_4_concurrent_prechecks_overshoot_is_bounded_not_exact(
f"/orgs/{org_id}/budgets/customer/race", json={"daily_cap_micro": cap_micro})
assert budget.status_code == 200, budget.text
original = A._enforce_tag_budgets
original = call_reserve._enforce_tag_budgets
ready = 0
all_prechecked = asyncio.Event()
@@ -445,15 +448,15 @@ async def test_review_replay_keeps_the_original_call_id(clients: AsyncClient, pl
def test_review_idempotency_separator_and_post_fingerprint_are_unambiguous():
scoped = {
A._scoped_idempotency_key(key, A.CallMeta(tags={"customer": customer}))
call_idem._scoped_idempotency_key(key, CallMeta(tags={"customer": customer}))
for customer in ("A", "AB", "A-B", "A:B")
for key in ("C", f"B{A._IDEM_SCOPE_SEP}C", f"{A._IDEM_SCOPE_SEP}C")
for key in ("C", f"B{call_idem._IDEM_SCOPE_SEP}C", f"{call_idem._IDEM_SCOPE_SEP}C")
}
assert len(scoped) == 12
original = A._request_fingerprint("POST", "endpoint", b'{"amount":1}', "mode=fast")
assert original == A._request_fingerprint("post", "endpoint", b'{"amount":1}', "mode=fast")
assert original != A._request_fingerprint("POST", "endpoint", b'{"amount":2}', "mode=fast")
assert original != A._request_fingerprint("POST", "endpoint", b'{"amount":1}', "mode=slow")
original = call_idem._request_fingerprint("POST", "endpoint", b'{"amount":1}', "mode=fast")
assert original == call_idem._request_fingerprint("post", "endpoint", b'{"amount":1}', "mode=fast")
assert original != call_idem._request_fingerprint("POST", "endpoint", b'{"amount":2}', "mode=fast")
assert original != call_idem._request_fingerprint("POST", "endpoint", b'{"amount":1}', "mode=slow")
@pytest.mark.anyio
tests/test_usage_caps.py
+2 -1
View File
@@ -12,7 +12,8 @@ from httpx import AsyncClient
from sqlmodel import select
from treg import audit
from treg.api import _utcnow_naive, count_today
from treg.routers.orgs import count_today
from treg.timeutil import utcnow_naive as _utcnow_naive
from treg.db import session_maker
from treg.models import CallRecord, Membership, RunRecord, User