import time import httpx import pytest from app.config import Settings from app.providers.devin import DevinProvider def make_client(handler): return httpx.AsyncClient(transport=httpx.MockTransport(handler), base_url='https://api.devin.ai') @pytest.mark.asyncio async def test_org_consumption_and_max_limit(): now = int(time.time()) async def handler(request): if request.url.path.endswith('/cycles'): return httpx.Response(200, json={'items': [{'after': now - 10, 'before': now + 1000}]}) if '/daily/organizations/' in request.url.path: return httpx.Response(200, json={'total_acus': 124, 'consumption_by_date': []}) if request.url.path.endswith('/organizations/org/id'): return httpx.Response(200, json={'max_cycle_acu_limit': 320}) return httpx.Response(404) client = make_client(handler) provider = DevinProvider(Settings(devin_org_id='org/id'), client) result = await provider.get_usage() assert result.status == 'ok' limit = result.limits[0] assert limit.used == 124 assert limit.limit == 320 assert limit.remaining == 196 assert limit.used_percent == pytest.approx(38.75) await client.aclose() @pytest.mark.asyncio async def test_user_scope_wins_over_org_scope(): now = int(time.time()) calls = [] async def handler(request): calls.append(request.url.path) if request.url.path.endswith('/cycles'): return httpx.Response(200, json={'items': [{'after': now - 10, 'before': now + 1000}]}) if '/daily/users/' in request.url.path: return httpx.Response(200, json={'total_acus': 12, 'consumption_by_date': []}) if '/organizations/' in request.url.path: return httpx.Response(200, json={'max_cycle_acu_limit': 320}) return httpx.Response(404) client = make_client(handler) provider = DevinProvider(Settings(devin_user_id='user/id', devin_org_id='org/id'), client) result = await provider.get_usage() assert result.status == 'ok' assert result.limits[0].limit is None assert any('/daily/users/' in path for path in calls) assert not any('/v3/enterprise/organizations/' in path for path in calls) await client.aclose() @pytest.mark.asyncio async def test_cycle_pagination_uses_end_cursor(): now = int(time.time()) cycle_queries = [] async def handler(request): if request.url.path.endswith('/cycles'): cycle_queries.append(request.url.params) if len(cycle_queries) == 1: return httpx.Response(200, json={'items': [{'after': now - 1000, 'before': now - 100}], 'has_next_page': True, 'end_cursor': 'next'}) return httpx.Response(200, json={'items': [{'after': now - 10, 'before': now + 1000}]}) return httpx.Response(200, json={'total_acus': 12, 'consumption_by_date': []}) client = make_client(handler) provider = DevinProvider(Settings(), client) result = await provider.get_usage() assert result.status == 'ok' assert len(cycle_queries) == 2 assert cycle_queries[1].get('after') == 'next' await client.aclose() @pytest.mark.asyncio async def test_user_scope_does_not_invent_limit(): now = int(time.time()) async def handler(request): if request.url.path.endswith('/cycles'): return httpx.Response(200, json={'items': [{'after': now - 10, 'before': now + 1000}]}) return httpx.Response(200, json={'total_acus': 12, 'consumption_by_date': []}) client = make_client(handler) provider = DevinProvider(Settings(devin_user_id='user/id'), client) result = await provider.get_usage() assert result.status == 'ok' assert result.limits[0].limit is None assert result.limits[0].used == 12 await client.aclose() @pytest.mark.asyncio async def test_required_forbidden_is_generic(): async def handler(request): return httpx.Response(403, text='secret upstream body') client = make_client(handler) provider = DevinProvider(Settings(), client) result = await provider.get_usage() assert result.status == 'error' assert result.error == 'Unable to retrieve usage information' assert 'secret' not in result.model_dump_json() await client.aclose() @pytest.mark.asyncio async def test_optional_org_forbidden_keeps_consumption(): now = int(time.time()) async def handler(request): if request.url.path.endswith('/cycles'): return httpx.Response(200, json={'items': [{'after': now - 10, 'before': now + 1000}]}) if '/daily/organizations/' in request.url.path: return httpx.Response(200, json={'total_acus': 12, 'consumption_by_date': []}) return httpx.Response(403, text='no org read') client = make_client(handler) provider = DevinProvider(Settings(devin_org_id='org'), client) result = await provider.get_usage() assert result.status == 'ok' assert result.limits[0].used == 12 assert result.limits[0].limit is None await client.aclose() @pytest.mark.asyncio async def test_organization_sessions_sums_pages_and_user_filter(monkeypatch): now = 1_800_000_000 days = 30 monkeypatch.setattr('app.providers.devin.time.time', lambda: now) requests = [] async def handler(request): requests.append(request) assert request.url.raw_path.split(b'?', 1)[0] == b'/v3/organizations/org%2Fid/sessions' assert request.url.params.get('first') == '200' assert int(request.url.params['created_before']) == now assert int(request.url.params['created_after']) == now - days * 86400 assert request.url.params.get_list('user_ids') == ['user-123'] if request.url.params.get('after') == 'next': return httpx.Response(200, json={'items': [ {'session_id': 'session-2', 'created_at': now - 20, 'acus_consumed': 7.6}, ]}) return httpx.Response(200, json={'items': [ {'session_id': 'session-1', 'created_at': now - 10, 'acus_consumed': 12.4}, ], 'has_next_page': True, 'end_cursor': 'next'}) client = make_client(handler) provider = DevinProvider(Settings(devin_mode='organization_sessions', devin_org_id='org/id', devin_user_id=' user-123 ', devin_session_window_days=days), client) result = await provider.get_usage() assert result.status == 'ok' assert result.source == 'Devin Organization Sessions API' assert len(requests) == 2 assert requests[1].url.params.get('after') == 'next' acus, count = result.limits assert acus.id == 'organization-session-acus' assert acus.name == '30-Day Session Usage' assert acus.used == pytest.approx(20) assert acus.unit == 'ACUs' assert acus.window == 'sessions created in the last 30 days' assert all(getattr(acus, field) is None for field in ('used_percent', 'remaining_percent', 'remaining', 'limit', 'reset_at')) assert count.id == 'organization-session-count' assert count.used == 2 assert count.unit == 'sessions' assert count.window == 'created in the last 30 days' assert all(getattr(count, field) is None for field in ('used_percent', 'remaining_percent', 'remaining', 'limit', 'reset_at')) await client.aclose() @pytest.mark.asyncio async def test_organization_sessions_requires_org_without_request(): requests = [] async def handler(request): requests.append(request) return httpx.Response(200, json={'items': []}) client = make_client(handler) provider = DevinProvider(Settings(devin_mode='organization_sessions'), client) result = await provider.get_usage() assert result.status == 'error' assert result.error == 'Unable to retrieve usage information' assert requests == [] await client.aclose() @pytest.mark.asyncio async def test_organization_sessions_forbidden_is_generic(): async def handler(request): return httpx.Response(403, text='secret organization response') client = make_client(handler) provider = DevinProvider(Settings(devin_mode='organization_sessions', devin_org_id='org'), client) result = await provider.get_usage() assert result.status == 'error' assert result.error == 'Unable to retrieve usage information' assert 'secret organization response' not in result.model_dump_json() await client.aclose() @pytest.mark.asyncio @pytest.mark.parametrize('bad_acus', [-1, 'not-a-number']) async def test_organization_sessions_rejects_bad_acus(bad_acus): async def handler(request): return httpx.Response(200, json={'items': [ {'session_id': 'session-1', 'created_at': 1_800_000_000, 'acus_consumed': bad_acus}, ]}) client = make_client(handler) provider = DevinProvider(Settings(devin_mode='organization_sessions', devin_org_id='org'), client) result = await provider.get_usage() assert result.status == 'error' assert result.error == 'Unable to retrieve usage information' await client.aclose()