mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-25 14:53:34 +00:00
07c6ca72e9
* fix: navigate telegram E2E tests to channels subtab wasm_channel extensions (like telegram) are now rendered in the Settings → Channels subtab, not the Extensions subtab. Update test_telegram_hot_activation to navigate there and use the correct card selector. Also mock /api/gateway/status which loadChannelsStatus fetches. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> * fix: select telegram card by name, not first card in channels subtab Built-in channel cards (Web Gateway, HTTP, etc.) render first in the channels subtab content, so .first matches them instead of the telegram extension card. Select by has_text="Telegram" to target the correct card. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> * refactor: make gateway_status_handler parameterizable in mock helper Address review feedback: extract default gateway status handler and accept an optional gateway_status_handler kwarg in mock_extension_lists for test flexibility. Co-Authored-By: Claude Opus 4.6 (1M context) <[email protected]> --------- Co-authored-by: Claude Opus 4.6 (1M context) <[email protected]>
259 lines
9.1 KiB
Python
259 lines
9.1 KiB
Python
"""Telegram hot-activation UI coverage."""
|
|
|
|
import asyncio
|
|
import json
|
|
|
|
from helpers import SEL
|
|
|
|
_CONFIGURE_SECRET_INPUT = "input[type='password']"
|
|
_CONFIGURE_SAVE_BUTTON = ".configure-actions button.btn-ext.activate"
|
|
|
|
|
|
_TELEGRAM_INSTALLED = {
|
|
"name": "telegram",
|
|
"display_name": "Telegram",
|
|
"kind": "wasm_channel",
|
|
"description": "Telegram Bot API channel",
|
|
"url": None,
|
|
"active": False,
|
|
"authenticated": False,
|
|
"has_auth": False,
|
|
"needs_setup": True,
|
|
"tools": [],
|
|
"activation_status": "installed",
|
|
"activation_error": None,
|
|
}
|
|
|
|
_TELEGRAM_ACTIVE = {
|
|
**_TELEGRAM_INSTALLED,
|
|
"active": True,
|
|
"authenticated": True,
|
|
"needs_setup": False,
|
|
"activation_status": "active",
|
|
}
|
|
|
|
|
|
async def go_to_channels(page):
|
|
"""Navigate to Settings → Channels subtab (where wasm_channel extensions live)."""
|
|
await page.locator(SEL["tab_button"].format(tab="settings")).click()
|
|
await page.locator(SEL["settings_subtab"].format(subtab="channels")).click()
|
|
await page.locator(SEL["settings_subpanel"].format(subtab="channels")).wait_for(
|
|
state="visible", timeout=5000
|
|
)
|
|
# Wait for the Telegram card specifically (built-in cards render first)
|
|
await page.locator(SEL["channels_ext_card"], has_text="Telegram").wait_for(
|
|
state="visible", timeout=8000
|
|
)
|
|
|
|
|
|
async def _default_gateway_status_handler(route):
|
|
await route.fulfill(
|
|
status=200,
|
|
content_type="application/json",
|
|
body=json.dumps({"enabled_channels": [], "sse_connections": 0, "ws_connections": 0}),
|
|
)
|
|
|
|
|
|
async def mock_extension_lists(page, ext_handler, *, gateway_status_handler=None):
|
|
async def handle_ext_list(route):
|
|
path = route.request.url.split("?")[0]
|
|
if path.endswith("/api/extensions"):
|
|
await ext_handler(route)
|
|
else:
|
|
await route.continue_()
|
|
|
|
async def handle_tools(route):
|
|
await route.fulfill(
|
|
status=200,
|
|
content_type="application/json",
|
|
body=json.dumps({"tools": []}),
|
|
)
|
|
|
|
async def handle_registry(route):
|
|
await route.fulfill(
|
|
status=200,
|
|
content_type="application/json",
|
|
body=json.dumps({"entries": []}),
|
|
)
|
|
|
|
# Register the broad route first so the specific endpoints below win.
|
|
await page.route("**/api/extensions*", handle_ext_list)
|
|
await page.route("**/api/extensions/tools", handle_tools)
|
|
await page.route("**/api/extensions/registry", handle_registry)
|
|
await page.route(
|
|
"**/api/gateway/status",
|
|
gateway_status_handler or _default_gateway_status_handler,
|
|
)
|
|
|
|
|
|
async def wait_for_toast(page, text: str, *, timeout: int = 5000):
|
|
await page.locator(SEL["toast"], has_text=text).wait_for(
|
|
state="visible", timeout=timeout
|
|
)
|
|
|
|
|
|
async def test_telegram_setup_modal_shows_bot_token_field(page):
|
|
async def handle_ext_list(route):
|
|
await route.fulfill(
|
|
status=200,
|
|
content_type="application/json",
|
|
body=json.dumps({"extensions": [_TELEGRAM_INSTALLED]}),
|
|
)
|
|
|
|
async def handle_setup(route):
|
|
await route.fulfill(
|
|
status=200,
|
|
content_type="application/json",
|
|
body=json.dumps(
|
|
{
|
|
"secrets": [
|
|
{
|
|
"name": "telegram_bot_token",
|
|
"prompt": "Enter your Telegram Bot API token (from @BotFather)",
|
|
"provided": False,
|
|
"optional": False,
|
|
"auto_generate": False,
|
|
}
|
|
]
|
|
}
|
|
),
|
|
)
|
|
|
|
await mock_extension_lists(page, handle_ext_list)
|
|
await page.route("**/api/extensions/telegram/setup", handle_setup)
|
|
await go_to_channels(page)
|
|
|
|
card = page.locator(SEL["channels_ext_card"], has_text="Telegram")
|
|
await card.locator(SEL["ext_configure_btn"], has_text="Setup").click()
|
|
|
|
modal = page.locator(SEL["configure_modal"])
|
|
await modal.wait_for(state="visible", timeout=5000)
|
|
assert "Telegram Bot API token" in await modal.text_content()
|
|
assert "IronClaw will show a one-time code" in (
|
|
await modal.text_content()
|
|
)
|
|
input_el = modal.locator(_CONFIGURE_SECRET_INPUT)
|
|
assert await input_el.count() == 1
|
|
|
|
|
|
async def test_telegram_hot_activation_transitions_installed_to_active(page):
|
|
phase = {"value": "installed"}
|
|
captured_setup_payloads = []
|
|
post_count = {"value": 0}
|
|
second_request_started = asyncio.Event()
|
|
allow_second_response = asyncio.Event()
|
|
|
|
async def handle_ext_list(route):
|
|
extensions = {
|
|
"installed": [_TELEGRAM_INSTALLED],
|
|
"active": [_TELEGRAM_ACTIVE],
|
|
}[phase["value"]]
|
|
await route.fulfill(
|
|
status=200,
|
|
content_type="application/json",
|
|
body=json.dumps({"extensions": extensions}),
|
|
)
|
|
|
|
async def handle_setup(route):
|
|
if route.request.method == "GET":
|
|
await route.fulfill(
|
|
status=200,
|
|
content_type="application/json",
|
|
body=json.dumps(
|
|
{
|
|
"secrets": [
|
|
{
|
|
"name": "telegram_bot_token",
|
|
"prompt": "Enter your Telegram Bot API token (from @BotFather)",
|
|
"provided": False,
|
|
"optional": False,
|
|
"auto_generate": False,
|
|
}
|
|
]
|
|
}
|
|
),
|
|
)
|
|
return
|
|
|
|
payload = json.loads(route.request.post_data or "{}")
|
|
captured_setup_payloads.append(payload)
|
|
post_count["value"] += 1
|
|
await asyncio.sleep(0.05)
|
|
if post_count["value"] == 1:
|
|
await route.fulfill(
|
|
status=200,
|
|
content_type="application/json",
|
|
body=json.dumps(
|
|
{
|
|
"success": True,
|
|
"activated": False,
|
|
"message": "Configuration saved for 'telegram'. Send `/start iclaw-7qk2m9` to @test_hot_bot in Telegram. IronClaw will finish setup automatically.",
|
|
"verification": {
|
|
"code": "iclaw-7qk2m9",
|
|
"instructions": "Send `/start iclaw-7qk2m9` to @test_hot_bot in Telegram. IronClaw will finish setup automatically.",
|
|
"deep_link": "https://t.me/test_hot_bot?start=iclaw-7qk2m9",
|
|
},
|
|
}
|
|
),
|
|
)
|
|
else:
|
|
second_request_started.set()
|
|
await allow_second_response.wait()
|
|
await route.fulfill(
|
|
status=200,
|
|
content_type="application/json",
|
|
body=json.dumps(
|
|
{
|
|
"success": True,
|
|
"activated": True,
|
|
"message": "Configuration saved, Telegram owner verified, and 'telegram' activated. Hot-activated WASM channel",
|
|
}
|
|
),
|
|
)
|
|
|
|
await mock_extension_lists(page, handle_ext_list)
|
|
await page.route("**/api/extensions/telegram/setup", handle_setup)
|
|
await go_to_channels(page)
|
|
|
|
card = page.locator(SEL["channels_ext_card"], has_text="Telegram")
|
|
await card.locator(SEL["ext_configure_btn"], has_text="Setup").click()
|
|
|
|
modal = page.locator(SEL["configure_modal"])
|
|
await modal.wait_for(state="visible", timeout=5000)
|
|
await modal.locator(_CONFIGURE_SECRET_INPUT).fill("123456789:ABCdefGhI")
|
|
await modal.locator(_CONFIGURE_SAVE_BUTTON).click()
|
|
await second_request_started.wait()
|
|
await modal.locator(".configure-inline-status", has_text="Waiting for Telegram owner verification...").wait_for(
|
|
state="visible", timeout=5000
|
|
)
|
|
assert "iclaw-7qk2m9" in (await modal.text_content())
|
|
assert "/start iclaw-7qk2m9" in (await modal.text_content())
|
|
assert await modal.locator(".configure-verification-link").count() == 1
|
|
await modal.locator(_CONFIGURE_SAVE_BUTTON).wait_for(state="hidden", timeout=5000)
|
|
|
|
await page.locator(SEL["configure_overlay"]).click(position={"x": 1, "y": 1})
|
|
assert await page.locator(SEL["configure_overlay"]).is_visible()
|
|
|
|
allow_second_response.set()
|
|
await page.locator(SEL["configure_overlay"]).wait_for(state="hidden", timeout=5000)
|
|
|
|
phase["value"] = "active"
|
|
await page.evaluate(
|
|
"""
|
|
handleAuthCompleted({
|
|
extension_name: 'telegram',
|
|
success: true,
|
|
message: "Configuration saved, Telegram owner verified, and 'telegram' activated. Hot-activated WASM channel",
|
|
});
|
|
"""
|
|
)
|
|
|
|
await wait_for_toast(page, "Telegram owner verified")
|
|
await card.locator(SEL["ext_active_label"]).wait_for(state="visible", timeout=5000)
|
|
assert await card.locator(SEL["ext_pairing_label"]).count() == 0
|
|
|
|
assert captured_setup_payloads == [
|
|
{"secrets": {"telegram_bot_token": "123456789:ABCdefGhI"}},
|
|
{"secrets": {}},
|
|
]
|