nessi
58eab24258
Add FastAPI-based monitoring dashboard with support for Codex CLI and Devin API v3 Enterprise consumption tracking. Includes Docker deployment, systemd service configuration, kiosk mode launcher, comprehensive documentation, and demo mode for testing without credentials.
126 lines
5.0 KiB
Python
126 lines
5.0 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()
|