85 lines
3.3 KiB
Python
85 lines
3.3 KiB
Python
from __future__ import annotations
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from beaver.interfaces.web.app import create_app
|
|
from beaver.services.agent_service import AgentService
|
|
|
|
|
|
def test_channel_connection_api_creates_updates_lists_and_revokes(tmp_path) -> None:
|
|
config_path = tmp_path / "config.json"
|
|
config_path.write_text(
|
|
'{"agents": {"defaults": {"workspace": "%s"}}, "providers": {}}' % str(tmp_path),
|
|
encoding="utf-8",
|
|
)
|
|
service = AgentService(config_path=config_path)
|
|
app = create_app(service=service, manage_service_lifecycle=False)
|
|
|
|
try:
|
|
with TestClient(app) as client:
|
|
created = client.post(
|
|
"/api/channel-connections",
|
|
json={
|
|
"kind": "telegram",
|
|
"mode": "polling",
|
|
"displayName": "Telegram Main",
|
|
"authType": "token",
|
|
"secrets": {"botToken": "token-1"},
|
|
"config": {"maxMessageChars": 4096, "requireMentionInGroups": True},
|
|
},
|
|
)
|
|
assert created.status_code == 200
|
|
body = created.json()
|
|
connection_id = body["connection"]["connection_id"]
|
|
assert body["connection"]["kind"] == "telegram"
|
|
assert body["connection"]["status"] == "draft"
|
|
assert "credentials_ref" not in body["connection"]
|
|
assert body["connection"]["runtime_config"] == {
|
|
"max_message_chars": 4096,
|
|
"require_mention_in_groups": True,
|
|
}
|
|
assert body["credentials"] == {"botToken": "***"}
|
|
|
|
patched = client.patch(
|
|
f"/api/channel-connections/{connection_id}",
|
|
json={
|
|
"displayName": "Telegram Ops",
|
|
"config": {"maxMessageChars": 2048},
|
|
"secrets": {"botToken": "token-2"},
|
|
},
|
|
)
|
|
assert patched.status_code == 200
|
|
assert patched.json()["connection"]["display_name"] == "Telegram Ops"
|
|
assert patched.json()["connection"]["runtime_config"] == {"max_message_chars": 2048}
|
|
assert patched.json()["credentials"] == {"botToken": "***"}
|
|
|
|
listed = client.get("/api/channel-connections")
|
|
assert listed.status_code == 200
|
|
assert listed.json()[0]["connection_id"] == connection_id
|
|
assert "credentials_ref" not in listed.json()[0]
|
|
|
|
revoked = client.post(f"/api/channel-connections/{connection_id}/revoke")
|
|
assert revoked.status_code == 200
|
|
assert revoked.json()["connection"]["status"] == "revoked"
|
|
finally:
|
|
service.close()
|
|
|
|
|
|
def test_channel_connectors_api_lists_registered_connectors(tmp_path) -> None:
|
|
config_path = tmp_path / "config.json"
|
|
config_path.write_text(
|
|
'{"agents": {"defaults": {"workspace": "%s"}}, "providers": {}}' % str(tmp_path),
|
|
encoding="utf-8",
|
|
)
|
|
service = AgentService(config_path=config_path)
|
|
app = create_app(service=service, manage_service_lifecycle=False)
|
|
|
|
try:
|
|
with TestClient(app) as client:
|
|
response = client.get("/api/channel-connectors")
|
|
finally:
|
|
service.close()
|
|
|
|
assert response.status_code == 200
|
|
assert response.json() == [{"kind": "feishu"}, {"kind": "telegram"}, {"kind": "weixin"}]
|