xinyin025/LangBot
1
0
Fork 0
mirror of https://github.com/langbot-app/LangBot.git synced 2026-10-22 16:35:20 +08:00
Code Issues Packages Projects Releases Wiki Activity

test(auth): verify revocation and authorization ordering

Browse Source
This commit is contained in:
huanghuoguoguo
2026-10-17 15:29:39 +08:00
parent bbf517edbc
commit efe1e4a758
4 changed files with 77 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/integration/api/test_support_admin_launch.py
+11 -3
View File
@@ -405,7 +405,7 @@ async def test_support_admin_persisted_expiry_and_revocation_are_enforced(suppor
async def test_support_admin_websocket_preserves_actor_and_revalidates(support_admin_api):
app, _client, _engine, private_key = support_admin_api
app, _client, engine, private_key = support_admin_api
launch = await _issue_support_token(app, private_key)
captured_contexts = []
@@ -414,7 +414,6 @@ async def test_support_admin_websocket_preserves_actor_and_revalidates(support_a
del data
captured_contexts.append(connection.execution_context)
await connection.send_queue.put({'type': 'handled'})
connection.is_active = False
app.pipeline_service = SimpleNamespace(get_pipeline=AsyncMock(return_value=SimpleNamespace(uuid='pipeline-1')))
app.platform_mgr = SimpleNamespace(
@@ -439,8 +438,17 @@ async def test_support_admin_websocket_preserves_actor_and_revalidates(support_a
await websocket.send(json.dumps({'type': 'message', 'message': [{'type': 'text', 'text': 'hi'}]}))
handled = json.loads(await websocket.receive())
assert handled['type'] == 'handled'
async with engine.begin() as connection:
await connection.execute(
sqlalchemy.update(SupportAdminTemporarySession)
.where(SupportAdminTemporarySession.grant_jti_hash == launch['grant_jti_hash'])
.values(revoked_at=datetime.datetime.now(datetime.UTC).replace(tzinfo=None))
)
await websocket.send(json.dumps({'type': 'message', 'message': [{'type': 'text', 'text': 'after revoke'}]}))
rejected = json.loads(await websocket.receive())
assert rejected == {'type': 'error', 'message': 'Unauthorized'}
assert captured_contexts
assert len(captured_contexts) == 1
principal = captured_contexts[0].trigger_principal
assert principal is not None
assert principal.principal_type == PrincipalType.SUPPORT_ADMIN