nessi
792f85fc8b
Add `DEVIN_MODE` configuration supporting `enterprise` (default) and `organization_sessions` modes. Organization sessions mode uses `/v3/organizations/{org_id}/sessions` API with `ViewOrgSessions` permission, suitable for non-Enterprise organizations. Sums session ACUs consumed within configurable rolling window (1-365 days via `DEVIN_SESSION_WINDOW_DAYS`). Mode does not show quota, percentage, credit, or reset fields as these are unav
218 lines
8.9 KiB
Python
218 lines
8.9 KiB
Python
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()
|