mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-26 15:40:18 +00:00
fix: cover staging CI all-features and routine batch regressions (#1256)
* fix staging CI coverage regressions * ci: cover all e2e scenarios in staging * ci: restrict staging PR checks and fix webhook assertions * ci: keep code style checks on PRs * ci: preserve e2e PR coverage * test: stabilize staging e2e coverage * fix: propagate postgres tls builder errors
This commit is contained in:
Henry Park
@@ -5,6 +5,8 @@ on:
|
||||
- cron: "0 6 * * 1" # Weekly Monday 6 AM UTC
|
||||
workflow_dispatch:
|
||||
pull_request:
|
||||
branches:
|
||||
- main
|
||||
paths:
|
||||
- "src/channels/web/**"
|
||||
- "tests/e2e/**"
|
||||
@@ -50,9 +52,11 @@ jobs:
|
||||
- group: core
|
||||
files: "tests/e2e/scenarios/test_connection.py tests/e2e/scenarios/test_chat.py tests/e2e/scenarios/test_sse_reconnect.py tests/e2e/scenarios/test_html_injection.py tests/e2e/scenarios/test_csp.py"
|
||||
- group: features
|
||||
files: "tests/e2e/scenarios/test_skills.py tests/e2e/scenarios/test_tool_approval.py"
|
||||
files: "tests/e2e/scenarios/test_skills.py tests/e2e/scenarios/test_tool_approval.py tests/e2e/scenarios/test_webhook.py"
|
||||
- group: extensions
|
||||
files: "tests/e2e/scenarios/test_extensions.py tests/e2e/scenarios/test_extension_oauth.py tests/e2e/scenarios/test_telegram_token_validation.py tests/e2e/scenarios/test_wasm_lifecycle.py tests/e2e/scenarios/test_tool_execution.py tests/e2e/scenarios/test_pairing.py tests/e2e/scenarios/test_oauth_credential_fallback.py tests/e2e/scenarios/test_routine_oauth_credential_injection.py"
|
||||
files: "tests/e2e/scenarios/test_extensions.py tests/e2e/scenarios/test_extension_oauth.py tests/e2e/scenarios/test_telegram_token_validation.py tests/e2e/scenarios/test_telegram_hot_activation.py tests/e2e/scenarios/test_wasm_lifecycle.py tests/e2e/scenarios/test_tool_execution.py tests/e2e/scenarios/test_pairing.py tests/e2e/scenarios/test_mcp_auth_flow.py tests/e2e/scenarios/test_oauth_credential_fallback.py tests/e2e/scenarios/test_routine_oauth_credential_injection.py"
|
||||
- group: routines
|
||||
files: "tests/e2e/scenarios/test_owner_scope.py tests/e2e/scenarios/test_routine_event_batch.py"
|
||||
steps:
|
||||
- uses: actions/checkout@v6
|
||||
|
||||
|
||||
@@ -17,7 +17,7 @@ jobs:
|
||||
matrix:
|
||||
include:
|
||||
- name: all-features
|
||||
flags: "--features postgres,libsql,html-to-markdown"
|
||||
flags: "--all-features"
|
||||
- name: default
|
||||
flags: ""
|
||||
- name: libsql-only
|
||||
|
||||
+27
-9
@@ -5,13 +5,22 @@
|
||||
//! certificates — the same TLS stack that `reqwest` already uses for HTTP.
|
||||
|
||||
use deadpool_postgres::{Pool, Runtime};
|
||||
use thiserror::Error;
|
||||
use tokio_postgres::NoTls;
|
||||
use tokio_postgres_rustls::MakeRustlsConnect;
|
||||
|
||||
use crate::config::SslMode;
|
||||
|
||||
#[derive(Debug, Error)]
|
||||
pub enum CreatePoolError {
|
||||
#[error("{0}")]
|
||||
Pool(#[from] deadpool_postgres::CreatePoolError),
|
||||
#[error("postgres TLS configuration failed: {0}")]
|
||||
TlsConfig(#[from] rustls::Error),
|
||||
}
|
||||
|
||||
/// Build a rustls-based TLS connector using the platform's root certificate store.
|
||||
fn make_rustls_connector() -> MakeRustlsConnect {
|
||||
fn make_rustls_connector() -> Result<MakeRustlsConnect, rustls::Error> {
|
||||
let mut root_store = rustls::RootCertStore::empty();
|
||||
let native = rustls_native_certs::load_native_certs();
|
||||
for e in &native.errors {
|
||||
@@ -25,10 +34,15 @@ fn make_rustls_connector() -> MakeRustlsConnect {
|
||||
if root_store.is_empty() {
|
||||
tracing::error!("no system root certificates found -- TLS connections will fail");
|
||||
}
|
||||
let config = rustls::ClientConfig::builder()
|
||||
.with_root_certificates(root_store)
|
||||
.with_no_client_auth();
|
||||
MakeRustlsConnect::new(config)
|
||||
// `--all-features` brings in both aws-lc-rs and ring-backed rustls providers.
|
||||
// Pick the same ring provider reqwest already uses so postgres TLS setup stays deterministic.
|
||||
let config = rustls::ClientConfig::builder_with_provider(
|
||||
rustls::crypto::ring::default_provider().into(),
|
||||
)
|
||||
.with_safe_default_protocol_versions()?
|
||||
.with_root_certificates(root_store)
|
||||
.with_no_client_auth();
|
||||
Ok(MakeRustlsConnect::new(config))
|
||||
}
|
||||
|
||||
/// Create a [`deadpool_postgres::Pool`] with the appropriate TLS connector.
|
||||
@@ -45,12 +59,16 @@ fn make_rustls_connector() -> MakeRustlsConnect {
|
||||
pub fn create_pool(
|
||||
config: &deadpool_postgres::Config,
|
||||
ssl_mode: SslMode,
|
||||
) -> Result<Pool, deadpool_postgres::CreatePoolError> {
|
||||
) -> Result<Pool, CreatePoolError> {
|
||||
match ssl_mode {
|
||||
SslMode::Disable => config.create_pool(Some(Runtime::Tokio1), NoTls),
|
||||
SslMode::Disable => config
|
||||
.create_pool(Some(Runtime::Tokio1), NoTls)
|
||||
.map_err(CreatePoolError::from),
|
||||
SslMode::Prefer | SslMode::Require => {
|
||||
let tls = make_rustls_connector();
|
||||
config.create_pool(Some(Runtime::Tokio1), tls)
|
||||
let tls = make_rustls_connector()?;
|
||||
config
|
||||
.create_pool(Some(Runtime::Tokio1), tls)
|
||||
.map_err(CreatePoolError::from)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
+17
-17
@@ -319,15 +319,14 @@ async def http_channel_server(ironclaw_server, server_ports):
|
||||
|
||||
|
||||
@pytest.fixture(scope="session")
|
||||
async def ironclaw_server_with_webhook_secret(ironclaw_binary, mock_llm_server, wasm_tools_dir):
|
||||
"""Start ironclaw with HTTP_WEBHOOK_SECRET configured for webhook tests.
|
||||
|
||||
Yields a dict with:
|
||||
- 'url': base URL of the gateway
|
||||
- 'secret': the webhook secret value
|
||||
"""
|
||||
async def http_channel_server_without_secret(
|
||||
ironclaw_binary,
|
||||
mock_llm_server,
|
||||
wasm_tools_dir,
|
||||
):
|
||||
"""Start the HTTP webhook channel without a configured secret."""
|
||||
gateway_port = _find_free_port()
|
||||
webhook_secret = "test-webhook-secret-e2e-12345"
|
||||
http_port = _find_free_port()
|
||||
env = {
|
||||
# Minimal env: PATH for process spawning, HOME for Rust/cargo defaults
|
||||
"PATH": os.environ.get("PATH", "/usr/bin:/bin"),
|
||||
@@ -339,13 +338,14 @@ async def ironclaw_server_with_webhook_secret(ironclaw_binary, mock_llm_server,
|
||||
"GATEWAY_PORT": str(gateway_port),
|
||||
"GATEWAY_AUTH_TOKEN": AUTH_TOKEN,
|
||||
"GATEWAY_USER_ID": "e2e-tester",
|
||||
"HTTP_WEBHOOK_SECRET": webhook_secret,
|
||||
"HTTP_HOST": "127.0.0.1",
|
||||
"HTTP_PORT": str(http_port),
|
||||
"CLI_ENABLED": "false",
|
||||
"LLM_BACKEND": "openai_compatible",
|
||||
"LLM_BASE_URL": mock_llm_server,
|
||||
"LLM_MODEL": "mock-model",
|
||||
"DATABASE_BACKEND": "libsql",
|
||||
"LIBSQL_PATH": os.path.join(_DB_TMPDIR.name, "e2e-webhook.db"),
|
||||
"LIBSQL_PATH": os.path.join(_DB_TMPDIR.name, "e2e-webhook-no-secret.db"),
|
||||
"SANDBOX_ENABLED": "false",
|
||||
"SKILLS_ENABLED": "true",
|
||||
"ROUTINES_ENABLED": "false",
|
||||
@@ -375,13 +375,12 @@ async def ironclaw_server_with_webhook_secret(ironclaw_binary, mock_llm_server,
|
||||
stderr=asyncio.subprocess.PIPE,
|
||||
env=env,
|
||||
)
|
||||
base_url = f"http://127.0.0.1:{gateway_port}"
|
||||
gateway_url = f"http://127.0.0.1:{gateway_port}"
|
||||
http_base_url = f"http://127.0.0.1:{http_port}"
|
||||
try:
|
||||
await wait_for_ready(f"{base_url}/api/health", timeout=60)
|
||||
yield {
|
||||
"url": base_url,
|
||||
"secret": webhook_secret,
|
||||
}
|
||||
await wait_for_ready(f"{gateway_url}/api/health", timeout=60)
|
||||
await wait_for_ready(f"{http_base_url}/health", timeout=30)
|
||||
yield http_base_url
|
||||
except TimeoutError:
|
||||
# Dump stderr so CI logs show why the server failed to start
|
||||
returncode = proc.returncode
|
||||
@@ -394,7 +393,8 @@ async def ironclaw_server_with_webhook_secret(ironclaw_binary, mock_llm_server,
|
||||
stderr_text = stderr_bytes.decode("utf-8", errors="replace")
|
||||
proc.kill()
|
||||
pytest.fail(
|
||||
f"ironclaw server with webhook secret failed to start on port {gateway_port} "
|
||||
f"ironclaw server without webhook secret failed to start on ports "
|
||||
f"gateway={gateway_port}, http={http_port} "
|
||||
f"(returncode={returncode}).\nstderr:\n{stderr_text}"
|
||||
)
|
||||
finally:
|
||||
|
||||
@@ -12,11 +12,17 @@ scenarios/test_csp.py
|
||||
scenarios/test_extension_oauth.py
|
||||
scenarios/test_extensions.py
|
||||
scenarios/test_html_injection.py
|
||||
scenarios/test_mcp_auth_flow.py
|
||||
scenarios/test_oauth_credential_fallback.py
|
||||
scenarios/test_owner_scope.py
|
||||
scenarios/test_pairing.py
|
||||
scenarios/test_routine_event_batch.py
|
||||
scenarios/test_routine_oauth_credential_injection.py
|
||||
scenarios/test_skills.py
|
||||
scenarios/test_sse_reconnect.py
|
||||
scenarios/test_telegram_hot_activation.py
|
||||
scenarios/test_telegram_token_validation.py
|
||||
scenarios/test_tool_approval.py
|
||||
scenarios/test_tool_execution.py
|
||||
scenarios/test_wasm_lifecycle.py
|
||||
scenarios/test_wasm_lifecycle.py
|
||||
scenarios/test_webhook.py
|
||||
@@ -55,6 +55,25 @@ TOOL_CALL_PATTERNS = [
|
||||
"action_type": "full_job",
|
||||
},
|
||||
),
|
||||
(
|
||||
re.compile(
|
||||
r"create event routine (?P<name>[a-z0-9][a-z0-9_-]*) "
|
||||
r"channel (?P<channel>[a-z0-9_-]+) pattern (?P<pattern>[a-z0-9_|-]+)",
|
||||
re.IGNORECASE,
|
||||
),
|
||||
"routine_create",
|
||||
lambda m: {
|
||||
"name": m.group("name"),
|
||||
"description": f"Event routine {m.group('name')}",
|
||||
"trigger_type": "event",
|
||||
"event_channel": None if m.group("channel").lower() == "any" else m.group("channel"),
|
||||
"event_pattern": m.group("pattern"),
|
||||
"prompt": f"Acknowledge that {m.group('name')} fired.",
|
||||
"action_type": "lightweight",
|
||||
"use_tools": False,
|
||||
"cooldown_secs": 0,
|
||||
},
|
||||
),
|
||||
(
|
||||
re.compile(r"list owner routines", re.IGNORECASE),
|
||||
"routine_list",
|
||||
|
||||
@@ -1,534 +1,317 @@
|
||||
"""
|
||||
E2E tests for event-triggered routines with batch loading.
|
||||
|
||||
These tests verify that the N+1 query fix correctly:
|
||||
1. Fires event-triggered routines on matching messages
|
||||
2. Enforces concurrent limits via batch-loaded counts
|
||||
3. Maintains performance with multiple simultaneous triggers
|
||||
4. Works correctly through the full UI and agent loop
|
||||
|
||||
Playwright-based UI tests + SSE verification.
|
||||
"""
|
||||
"""E2E tests for event-triggered routines over the HTTP channel."""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import uuid
|
||||
|
||||
import httpx
|
||||
import pytest
|
||||
from datetime import datetime, timedelta
|
||||
from typing import List, Dict, Any
|
||||
|
||||
from playwright.async_api import async_playwright, Page, Browser, BrowserContext
|
||||
from helpers import AUTH_TOKEN, SEL, signed_http_webhook_headers
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
async def browser_and_context():
|
||||
"""Create a Playwright browser and context for testing."""
|
||||
async with async_playwright() as p:
|
||||
browser = await p.chromium.launch(headless=True)
|
||||
context = await browser.new_context()
|
||||
yield browser, context
|
||||
await context.close()
|
||||
await browser.close()
|
||||
async def _send_chat_message(page, message: str) -> None:
|
||||
"""Send a chat message and wait for the assistant turn to appear."""
|
||||
chat_input = page.locator(SEL["chat_input"])
|
||||
await chat_input.wait_for(state="visible", timeout=5000)
|
||||
assistant_messages = page.locator(SEL["message_assistant"])
|
||||
before_count = await assistant_messages.count()
|
||||
|
||||
await chat_input.fill(message)
|
||||
await chat_input.press("Enter")
|
||||
|
||||
await page.wait_for_function(
|
||||
"""({ selector, expectedCount }) => {
|
||||
return document.querySelectorAll(selector).length >= expectedCount;
|
||||
}""",
|
||||
arg={
|
||||
"selector": SEL["message_assistant"],
|
||||
"expectedCount": before_count + 1,
|
||||
},
|
||||
timeout=30000,
|
||||
)
|
||||
|
||||
|
||||
class EventTriggerHelper:
|
||||
"""Helper methods for event trigger testing."""
|
||||
async def _create_event_routine(
|
||||
page,
|
||||
base_url: str,
|
||||
*,
|
||||
name: str,
|
||||
pattern: str,
|
||||
channel: str = "http",
|
||||
) -> dict:
|
||||
"""Create an event routine through chat and return its API record."""
|
||||
await _send_chat_message(
|
||||
page,
|
||||
f"create event routine {name} channel {channel} pattern {pattern}",
|
||||
)
|
||||
return await _wait_for_routine(base_url, name)
|
||||
|
||||
def __init__(self, page: Page):
|
||||
self.page = page
|
||||
|
||||
async def navigate_to_routines(self):
|
||||
"""Navigate to the routines page."""
|
||||
await self.page.goto("http://localhost:8000/routines")
|
||||
await self.page.wait_for_load_state("networkidle")
|
||||
async def _post_http_message(
|
||||
http_channel_server: str,
|
||||
*,
|
||||
content: str,
|
||||
sender_id: str | None = None,
|
||||
thread_id: str | None = None,
|
||||
) -> dict:
|
||||
"""Send a signed HTTP-channel message and return the JSON body."""
|
||||
payload = {
|
||||
"user_id": sender_id or f"sender-{uuid.uuid4().hex[:8]}",
|
||||
"thread_id": thread_id or f"thread-{uuid.uuid4().hex[:8]}",
|
||||
"content": content,
|
||||
"wait_for_response": True,
|
||||
}
|
||||
body = json.dumps(payload).encode("utf-8")
|
||||
|
||||
async def create_event_routine(
|
||||
self,
|
||||
name: str,
|
||||
trigger_regex: str,
|
||||
channel: str = "slack",
|
||||
max_concurrent: int = 1,
|
||||
) -> str:
|
||||
"""
|
||||
Create an event-triggered routine via UI.
|
||||
Returns the routine ID.
|
||||
"""
|
||||
await self.navigate_to_routines()
|
||||
|
||||
# Click "New Routine" button
|
||||
await self.page.click('button:has-text("New Routine")')
|
||||
await self.page.wait_for_selector('input[name="routine_name"]')
|
||||
|
||||
# Fill routine details
|
||||
await self.page.fill('input[name="routine_name"]', name)
|
||||
await self.page.fill(
|
||||
'textarea[name="routine_description"]',
|
||||
f"Test routine: {name}",
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.post(
|
||||
f"{http_channel_server}/webhook",
|
||||
content=body,
|
||||
headers=signed_http_webhook_headers(body),
|
||||
timeout=90,
|
||||
)
|
||||
|
||||
# Select "Event Trigger" type
|
||||
await self.page.click('label:has-text("Event Trigger")')
|
||||
await self.page.wait_for_selector('input[name="trigger_regex"]')
|
||||
|
||||
# Fill trigger details
|
||||
await self.page.fill('input[name="trigger_regex"]', trigger_regex)
|
||||
await self.page.select_option('select[name="trigger_channel"]', channel)
|
||||
|
||||
# Set guardrails
|
||||
await self.page.fill('input[name="max_concurrent"]', str(max_concurrent))
|
||||
|
||||
# Select lightweight action
|
||||
await self.page.click('label:has-text("Lightweight")')
|
||||
await self.page.fill(
|
||||
'textarea[name="lightweight_prompt"]',
|
||||
"Acknowledge the message and confirm trigger worked.",
|
||||
)
|
||||
|
||||
# Save routine
|
||||
await self.page.click('button:has-text("Save Routine")')
|
||||
await self.page.wait_for_selector('text=Routine created successfully')
|
||||
|
||||
# Extract routine ID from success message or URL
|
||||
routine_id = await self.page.locator('data-testid=routine-id').text_content()
|
||||
return routine_id.strip() if routine_id else None
|
||||
|
||||
async def create_multiple_routines(
|
||||
self, base_name: str, count: int, trigger_regex: str = None
|
||||
) -> List[str]:
|
||||
"""Create multiple event-triggered routines."""
|
||||
routine_ids = []
|
||||
for i in range(count):
|
||||
name = f"{base_name}_{i}"
|
||||
regex = trigger_regex or f"({i}|{base_name})"
|
||||
routine_id = await self.create_event_routine(name, regex)
|
||||
routine_ids.append(routine_id)
|
||||
await asyncio.sleep(0.1) # Small delay between creations
|
||||
return routine_ids
|
||||
|
||||
async def send_chat_message(self, message: str) -> List[str]:
|
||||
"""
|
||||
Send a chat message and return SSE events received.
|
||||
Captures all routine firing events.
|
||||
"""
|
||||
await self.page.goto("http://localhost:8000/chat")
|
||||
await self.page.wait_for_selector('input[placeholder*="message"]', timeout=5000)
|
||||
|
||||
# Collect SSE events
|
||||
sse_events = []
|
||||
|
||||
async def capture_sse(response):
|
||||
"""Intercept SSE events."""
|
||||
if "event-stream" in response.headers.get("content-type", ""):
|
||||
text = await response.text()
|
||||
for line in text.split("\n"):
|
||||
if line.startswith("data:"):
|
||||
try:
|
||||
event = json.loads(line[5:])
|
||||
sse_events.append(event)
|
||||
except json.JSONDecodeError:
|
||||
pass
|
||||
|
||||
self.page.on("response", capture_sse)
|
||||
|
||||
# Send message
|
||||
await self.page.fill('input[placeholder*="message"]', message)
|
||||
await self.page.press('input[placeholder*="message"]', "Enter")
|
||||
|
||||
# Wait for response
|
||||
await self.page.wait_for_selector('text=Message processed', timeout=10000)
|
||||
await asyncio.sleep(0.5) # Allow time for SSE events
|
||||
|
||||
self.page.remove_listener("response", capture_sse)
|
||||
return sse_events
|
||||
|
||||
async def get_routine_execution_log(self, routine_id: str) -> List[Dict]:
|
||||
"""Get execution log entries for a routine."""
|
||||
await self.page.goto(f"http://localhost:8000/routines/{routine_id}/executions")
|
||||
await self.page.wait_for_load_state("networkidle")
|
||||
|
||||
# Extract log entries from table
|
||||
rows = await self.page.locator("tbody tr").all()
|
||||
executions = []
|
||||
|
||||
for row in rows:
|
||||
cells = await row.locator("td").all()
|
||||
if len(cells) >= 3:
|
||||
execution = {
|
||||
"timestamp": await cells[0].text_content(),
|
||||
"status": await cells[1].text_content(),
|
||||
"details": await cells[2].text_content(),
|
||||
}
|
||||
executions.append(execution)
|
||||
|
||||
return executions
|
||||
|
||||
async def check_database_queries_in_logs(
|
||||
self, max_queries_expected: int = 1
|
||||
) -> int:
|
||||
"""Check debug logs for database query count."""
|
||||
await self.page.goto("http://localhost:8000/debug/logs?filter=database")
|
||||
await self.page.wait_for_load_state("networkidle")
|
||||
|
||||
# Count batch queries
|
||||
log_lines = await self.page.locator("tr:has-text('batch')").all()
|
||||
batch_count = len(log_lines)
|
||||
|
||||
# Count individual COUNT queries (should be 0 after fix)
|
||||
count_queries = await self.page.locator("tr:has-text('COUNT')").all()
|
||||
count_query_count = len(count_queries)
|
||||
|
||||
return batch_count, count_query_count
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Tests
|
||||
# =============================================================================
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_event_trigger_routine(browser_and_context):
|
||||
"""Test creating an event-triggered routine via UI."""
|
||||
browser, context = browser_and_context
|
||||
page = await context.new_page()
|
||||
helper = EventTriggerHelper(page)
|
||||
|
||||
try:
|
||||
routine_id = await helper.create_event_routine(
|
||||
name="Test Trigger",
|
||||
trigger_regex="test|demo",
|
||||
channel="slack",
|
||||
max_concurrent=1,
|
||||
)
|
||||
|
||||
assert routine_id is not None, "Routine ID should be returned"
|
||||
assert len(routine_id) > 0, "Routine ID should not be empty"
|
||||
|
||||
finally:
|
||||
await page.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_event_trigger_fires_on_matching_message(browser_and_context):
|
||||
"""Test that event-triggered routine fires when message matches."""
|
||||
browser, context = browser_and_context
|
||||
page = await context.new_page()
|
||||
helper = EventTriggerHelper(page)
|
||||
|
||||
try:
|
||||
# Create routine
|
||||
routine_id = await helper.create_event_routine(
|
||||
name="Alert Handler",
|
||||
trigger_regex="urgent|critical|alert",
|
||||
channel="slack",
|
||||
)
|
||||
|
||||
# Send matching message
|
||||
sse_events = await helper.send_chat_message("URGENT: Server down!")
|
||||
|
||||
# Verify routine fired (look for event in SSE stream)
|
||||
routine_fired = any(
|
||||
event.get("type") == "routine_fired" and event.get("routine_id") == routine_id
|
||||
for event in sse_events
|
||||
)
|
||||
assert routine_fired, "Routine should fire on matching message"
|
||||
|
||||
# Check execution log
|
||||
executions = await helper.get_routine_execution_log(routine_id)
|
||||
assert len(executions) > 0, "Execution should be logged"
|
||||
assert "success" in executions[0]["status"].lower()
|
||||
|
||||
finally:
|
||||
await page.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_event_trigger_skips_non_matching_message(browser_and_context):
|
||||
"""Test that event-triggered routine skips when message doesn't match."""
|
||||
browser, context = browser_and_context
|
||||
page = await context.new_page()
|
||||
helper = EventTriggerHelper(page)
|
||||
|
||||
try:
|
||||
# Create routine
|
||||
routine_id = await helper.create_event_routine(
|
||||
name="Alert Handler",
|
||||
trigger_regex="urgent|critical|alert",
|
||||
channel="slack",
|
||||
)
|
||||
|
||||
# Send non-matching message
|
||||
sse_events = await helper.send_chat_message("Hello, how are you?")
|
||||
|
||||
# Verify routine did NOT fire
|
||||
routine_fired = any(
|
||||
event.get("type") == "routine_fired" and event.get("routine_id") == routine_id
|
||||
for event in sse_events
|
||||
)
|
||||
assert not routine_fired, "Routine should not fire on non-matching message"
|
||||
|
||||
finally:
|
||||
await page.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multiple_routines_fire_on_matching_message(browser_and_context):
|
||||
"""Test that multiple event-triggered routines fire on same message."""
|
||||
browser, context = browser_and_context
|
||||
page = await context.new_page()
|
||||
helper = EventTriggerHelper(page)
|
||||
|
||||
try:
|
||||
# Create 3 overlapping routines
|
||||
routine_ids = await helper.create_multiple_routines(
|
||||
base_name="Handler", count=3, trigger_regex="alert|warning|error"
|
||||
)
|
||||
|
||||
# Send matching message
|
||||
sse_events = await helper.send_chat_message("ERROR: Database connection failed")
|
||||
|
||||
# Verify all 3 routines fired
|
||||
fired_count = sum(
|
||||
1
|
||||
for event in sse_events
|
||||
if event.get("type") == "routine_fired" and event.get("routine_id") in routine_ids
|
||||
)
|
||||
|
||||
assert (
|
||||
fired_count >= 3
|
||||
), f"Expected all 3 routines to fire, got {fired_count}"
|
||||
|
||||
finally:
|
||||
await page.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_concurrent_limit_prevents_additional_fires(browser_and_context):
|
||||
"""Test that concurrent limit is enforced via batch counts."""
|
||||
browser, context = browser_and_context
|
||||
page = await context.new_page()
|
||||
helper = EventTriggerHelper(page)
|
||||
|
||||
try:
|
||||
# Create routine with max_concurrent=1
|
||||
routine_id = await helper.create_event_routine(
|
||||
name="Limited Handler",
|
||||
trigger_regex="process|task",
|
||||
max_concurrent=1,
|
||||
)
|
||||
|
||||
# Trigger first message
|
||||
await helper.send_chat_message("Process message 1")
|
||||
await asyncio.sleep(1)
|
||||
|
||||
# Check first execution logged
|
||||
executions_1 = await helper.get_routine_execution_log(routine_id)
|
||||
assert len(executions_1) >= 1
|
||||
|
||||
# Trigger second message while first is still running
|
||||
sse_events = await helper.send_chat_message("Process message 2")
|
||||
|
||||
# Second routine should be skipped (concurrent limit)
|
||||
routine_skipped = any(
|
||||
event.get("type") == "routine_skipped"
|
||||
and event.get("reason") == "max_concurrent_reached"
|
||||
and event.get("routine_id") == routine_id
|
||||
for event in sse_events
|
||||
)
|
||||
assert routine_skipped, "Routine should be skipped when concurrent limit reached"
|
||||
|
||||
finally:
|
||||
await page.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_rapid_messages_with_multiple_triggers_efficiency(browser_and_context):
|
||||
"""Test efficiency of batch loading with multiple rapid messages."""
|
||||
browser, context = browser_and_context
|
||||
page = await context.new_page()
|
||||
helper = EventTriggerHelper(page)
|
||||
|
||||
try:
|
||||
# Create 5 overlapping routines
|
||||
routine_ids = await helper.create_multiple_routines(
|
||||
base_name="Rapid", count=5, trigger_regex="test|demo|check"
|
||||
)
|
||||
|
||||
# Send 10 matching messages rapidly
|
||||
for i in range(10):
|
||||
message = f"test message {i}"
|
||||
await helper.send_chat_message(message)
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
# Check database logs for query efficiency
|
||||
batch_count, count_query_count = await helper.check_database_queries_in_logs()
|
||||
|
||||
# After fix: should have ~10 batch queries (1 per message)
|
||||
# Before fix: would have ~50 individual COUNT queries (5 routines × 10 messages)
|
||||
assert (
|
||||
count_query_count == 0
|
||||
), f"Should have 0 individual COUNT queries after fix, got {count_query_count}"
|
||||
assert (
|
||||
batch_count <= 15
|
||||
), f"Should have <=15 batch queries for 10 messages, got {batch_count}"
|
||||
|
||||
finally:
|
||||
await page.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_channel_filter_applied_correctly(browser_and_context):
|
||||
"""Test that channel filter prevents non-matching messages."""
|
||||
browser, context = browser_and_context
|
||||
page = await context.new_page()
|
||||
helper = EventTriggerHelper(page)
|
||||
|
||||
try:
|
||||
# Create routine for Slack channel
|
||||
slack_routine_id = await helper.create_event_routine(
|
||||
name="Slack Handler",
|
||||
trigger_regex="alert",
|
||||
channel="slack",
|
||||
)
|
||||
|
||||
# Simulate message from Telegram channel
|
||||
# (Note: In real UI, would need to change channel context)
|
||||
page.goto(
|
||||
"http://localhost:8000/chat?channel=telegram"
|
||||
) # Switch channel
|
||||
await helper.send_chat_message("alert: something urgent")
|
||||
|
||||
# Routine should not fire (different channel)
|
||||
executions = await helper.get_routine_execution_log(slack_routine_id)
|
||||
|
||||
# Check if any recent execution (last 5 min) exists
|
||||
recent = [
|
||||
e
|
||||
for e in executions
|
||||
if (datetime.now() - datetime.fromisoformat(e["timestamp"])).total_seconds()
|
||||
< 300
|
||||
]
|
||||
assert (
|
||||
len(recent) == 0
|
||||
), "Routine should not fire for different channel"
|
||||
|
||||
finally:
|
||||
await page.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_batch_query_failure_handling(browser_and_context):
|
||||
"""Test graceful handling of batch query failures."""
|
||||
browser, context = browser_and_context
|
||||
page = await context.new_page()
|
||||
helper = EventTriggerHelper(page)
|
||||
|
||||
try:
|
||||
# Create routine
|
||||
routine_id = await helper.create_event_routine(
|
||||
name="Error Handler",
|
||||
trigger_regex="test",
|
||||
)
|
||||
|
||||
# Simulate database error in logs (if possible with test hooks)
|
||||
# For now, just verify error handling doesn't crash UI
|
||||
await helper.send_chat_message("test message")
|
||||
|
||||
# Check that UI remains responsive
|
||||
assert await page.locator("text=Message processed").is_visible()
|
||||
|
||||
finally:
|
||||
await page.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_routine_execution_history_display(browser_and_context):
|
||||
"""Test that execution history correctly displays routine firings."""
|
||||
browser, context = browser_and_context
|
||||
page = await context.new_page()
|
||||
helper = EventTriggerHelper(page)
|
||||
|
||||
try:
|
||||
# Create routine
|
||||
routine_id = await helper.create_event_routine(
|
||||
name="History Test",
|
||||
trigger_regex="test",
|
||||
)
|
||||
|
||||
# Trigger routine 3 times
|
||||
for i in range(3):
|
||||
await helper.send_chat_message(f"test message {i}")
|
||||
await asyncio.sleep(0.2)
|
||||
|
||||
# Check execution log
|
||||
executions = await helper.get_routine_execution_log(routine_id)
|
||||
assert len(executions) >= 3, "Should have at least 3 executions logged"
|
||||
|
||||
# Verify all are recent (within last 5 minutes)
|
||||
for execution in executions[:3]:
|
||||
timestamp = datetime.fromisoformat(execution["timestamp"])
|
||||
age = datetime.now() - timestamp
|
||||
assert age < timedelta(minutes=5), "Execution should be recent"
|
||||
|
||||
finally:
|
||||
await page.close()
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_concurrent_batch_loads_independent(browser_and_context):
|
||||
"""Test that concurrent messages each get independent batch queries."""
|
||||
browser, context = browser_and_context
|
||||
page = await context.new_page()
|
||||
helper = EventTriggerHelper(page)
|
||||
|
||||
try:
|
||||
# Create 5 routines matching different patterns
|
||||
r1_id = await helper.create_event_routine(
|
||||
name="Pattern A", trigger_regex="alpha|alpha_only"
|
||||
)
|
||||
r2_id = await helper.create_event_routine(
|
||||
name="Pattern B", trigger_regex="beta|beta_only"
|
||||
)
|
||||
r3_id = await helper.create_event_routine(
|
||||
name="Pattern AB", trigger_regex="alpha|beta|common"
|
||||
)
|
||||
|
||||
# Send overlapping messages
|
||||
# Message 1: matches r1, r3
|
||||
sse1 = await helper.send_chat_message("alpha common")
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
# Message 2: matches r2, r3
|
||||
sse2 = await helper.send_chat_message("beta common")
|
||||
await asyncio.sleep(0.1)
|
||||
|
||||
# Verify correct routines fired
|
||||
r1_fired_msg1 = any(
|
||||
e.get("routine_id") == r1_id for e in sse1 if e.get("type") == "routine_fired"
|
||||
)
|
||||
r2_fired_msg2 = any(
|
||||
e.get("routine_id") == r2_id for e in sse2 if e.get("type") == "routine_fired"
|
||||
)
|
||||
r3_fired_both = (
|
||||
any(
|
||||
e.get("routine_id") == r3_id for e in sse1 if e.get("type") == "routine_fired"
|
||||
assert response.status_code == 200, (
|
||||
f"HTTP webhook failed: {response.status_code} {response.text[:400]}"
|
||||
)
|
||||
return response.json()
|
||||
|
||||
|
||||
async def _wait_for_routine(base_url: str, name: str, timeout: float = 20.0) -> dict:
|
||||
"""Poll the routines API until the named routine exists."""
|
||||
async with httpx.AsyncClient() as client:
|
||||
for _ in range(int(timeout * 2)):
|
||||
response = await client.get(
|
||||
f"{base_url}/api/routines",
|
||||
headers={"Authorization": f"Bearer {AUTH_TOKEN}"},
|
||||
timeout=10,
|
||||
)
|
||||
and any(
|
||||
e.get("routine_id") == r3_id for e in sse2 if e.get("type") == "routine_fired"
|
||||
response.raise_for_status()
|
||||
for routine in response.json()["routines"]:
|
||||
if routine["name"] == name:
|
||||
return routine
|
||||
await asyncio.sleep(0.5)
|
||||
raise AssertionError(f"Routine '{name}' was not created within {timeout}s")
|
||||
|
||||
|
||||
async def _get_routine_runs(base_url: str, routine_id: str) -> list[dict]:
|
||||
"""Fetch recent routine runs from the web API."""
|
||||
async with httpx.AsyncClient() as client:
|
||||
response = await client.get(
|
||||
f"{base_url}/api/routines/{routine_id}/runs",
|
||||
headers={"Authorization": f"Bearer {AUTH_TOKEN}"},
|
||||
timeout=10,
|
||||
)
|
||||
response.raise_for_status()
|
||||
return response.json()["runs"]
|
||||
|
||||
|
||||
async def _wait_for_run_count(
|
||||
base_url: str,
|
||||
routine_id: str,
|
||||
*,
|
||||
expected_at_least: int,
|
||||
timeout: float = 20.0,
|
||||
) -> list[dict]:
|
||||
"""Poll until the routine has at least the expected run count."""
|
||||
for _ in range(int(timeout * 2)):
|
||||
runs = await _get_routine_runs(base_url, routine_id)
|
||||
if len(runs) >= expected_at_least:
|
||||
return runs
|
||||
await asyncio.sleep(0.5)
|
||||
raise AssertionError(
|
||||
f"Routine '{routine_id}' did not reach {expected_at_least} runs within {timeout}s"
|
||||
)
|
||||
|
||||
|
||||
async def _wait_for_completed_run(
|
||||
base_url: str,
|
||||
routine_id: str,
|
||||
*,
|
||||
timeout: float = 30.0,
|
||||
) -> dict:
|
||||
"""Poll until the newest run is no longer marked running."""
|
||||
for _ in range(int(timeout * 2)):
|
||||
runs = await _get_routine_runs(base_url, routine_id)
|
||||
if runs and runs[0]["status"].lower() != "running":
|
||||
return runs[0]
|
||||
await asyncio.sleep(0.5)
|
||||
raise AssertionError(f"Routine '{routine_id}' did not complete within {timeout}s")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_create_event_trigger_routine(page, ironclaw_server):
|
||||
"""Event routines can be created through the supported chat flow."""
|
||||
name = f"evt-{uuid.uuid4().hex[:8]}"
|
||||
routine = await _create_event_routine(
|
||||
page,
|
||||
ironclaw_server,
|
||||
name=name,
|
||||
pattern="test|demo",
|
||||
)
|
||||
|
||||
assert routine["id"]
|
||||
assert routine["trigger_type"] == "event"
|
||||
assert "test|demo" in routine["trigger_summary"]
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_event_trigger_fires_on_matching_message(
|
||||
page,
|
||||
ironclaw_server,
|
||||
http_channel_server,
|
||||
):
|
||||
"""Matching HTTP-channel messages create routine runs."""
|
||||
name = f"evt-{uuid.uuid4().hex[:8]}"
|
||||
routine = await _create_event_routine(
|
||||
page,
|
||||
ironclaw_server,
|
||||
name=name,
|
||||
pattern="urgent|critical|alert",
|
||||
)
|
||||
|
||||
response = await _post_http_message(
|
||||
http_channel_server,
|
||||
content="urgent: server down",
|
||||
)
|
||||
assert response["status"] == "accepted"
|
||||
|
||||
await _wait_for_run_count(
|
||||
ironclaw_server,
|
||||
routine["id"],
|
||||
expected_at_least=1,
|
||||
)
|
||||
completed_run = await _wait_for_completed_run(ironclaw_server, routine["id"])
|
||||
|
||||
assert completed_run["status"].lower() == "attention"
|
||||
assert completed_run["trigger_type"] == "event"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_event_trigger_skips_non_matching_message(
|
||||
page,
|
||||
ironclaw_server,
|
||||
http_channel_server,
|
||||
):
|
||||
"""Non-matching messages do not create routine runs."""
|
||||
name = f"evt-{uuid.uuid4().hex[:8]}"
|
||||
routine = await _create_event_routine(
|
||||
page,
|
||||
ironclaw_server,
|
||||
name=name,
|
||||
pattern="urgent|critical|alert",
|
||||
)
|
||||
|
||||
await _post_http_message(
|
||||
http_channel_server,
|
||||
content="hello there",
|
||||
)
|
||||
await asyncio.sleep(2)
|
||||
|
||||
assert await _get_routine_runs(ironclaw_server, routine["id"]) == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_multiple_routines_fire_on_matching_message(
|
||||
page,
|
||||
ironclaw_server,
|
||||
http_channel_server,
|
||||
):
|
||||
"""A single matching message can fire multiple event routines."""
|
||||
routines = []
|
||||
for _ in range(3):
|
||||
name = f"evt-{uuid.uuid4().hex[:8]}"
|
||||
routines.append(
|
||||
await _create_event_routine(
|
||||
page,
|
||||
ironclaw_server,
|
||||
name=name,
|
||||
pattern="error|warning|alert",
|
||||
)
|
||||
)
|
||||
|
||||
assert r1_fired_msg1, "Routine 1 should fire on message 1"
|
||||
assert r2_fired_msg2, "Routine 2 should fire on message 2"
|
||||
assert r3_fired_both, "Routine 3 should fire on both messages"
|
||||
await _post_http_message(
|
||||
http_channel_server,
|
||||
content="error: database connection failed",
|
||||
)
|
||||
|
||||
finally:
|
||||
await page.close()
|
||||
for routine in routines:
|
||||
await _wait_for_run_count(
|
||||
ironclaw_server,
|
||||
routine["id"],
|
||||
expected_at_least=1,
|
||||
)
|
||||
completed_run = await _wait_for_completed_run(ironclaw_server, routine["id"])
|
||||
assert completed_run["status"].lower() == "attention"
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Integration with existing test patterns
|
||||
# =============================================================================
|
||||
@pytest.mark.asyncio
|
||||
async def test_channel_filter_applied_correctly(
|
||||
page,
|
||||
ironclaw_server,
|
||||
http_channel_server,
|
||||
):
|
||||
"""Channel filters prevent HTTP messages from firing non-HTTP routines."""
|
||||
http_routine = await _create_event_routine(
|
||||
page,
|
||||
ironclaw_server,
|
||||
name=f"evt-{uuid.uuid4().hex[:8]}",
|
||||
pattern="alert",
|
||||
channel="http",
|
||||
)
|
||||
telegram_routine = await _create_event_routine(
|
||||
page,
|
||||
ironclaw_server,
|
||||
name=f"evt-{uuid.uuid4().hex[:8]}",
|
||||
pattern="alert",
|
||||
channel="telegram",
|
||||
)
|
||||
|
||||
await _post_http_message(
|
||||
http_channel_server,
|
||||
content="alert from webhook",
|
||||
)
|
||||
|
||||
await _wait_for_run_count(
|
||||
ironclaw_server,
|
||||
http_routine["id"],
|
||||
expected_at_least=1,
|
||||
)
|
||||
http_run = await _wait_for_completed_run(ironclaw_server, http_routine["id"])
|
||||
await asyncio.sleep(2)
|
||||
telegram_runs = await _get_routine_runs(ironclaw_server, telegram_routine["id"])
|
||||
|
||||
assert http_run["status"].lower() == "attention"
|
||||
assert telegram_runs == []
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Run tests with: pytest tests/e2e/scenarios/test_routine_event_batch.py -v
|
||||
pytest.main([__file__, "-v", "-s"])
|
||||
@pytest.mark.asyncio
|
||||
async def test_routine_execution_history_is_available(
|
||||
page,
|
||||
ironclaw_server,
|
||||
http_channel_server,
|
||||
):
|
||||
"""Routine run history is exposed by the routines runs API."""
|
||||
routine = await _create_event_routine(
|
||||
page,
|
||||
ironclaw_server,
|
||||
name=f"evt-{uuid.uuid4().hex[:8]}",
|
||||
pattern="history",
|
||||
)
|
||||
|
||||
await _post_http_message(
|
||||
http_channel_server,
|
||||
content="history event",
|
||||
)
|
||||
|
||||
await _wait_for_run_count(
|
||||
ironclaw_server,
|
||||
routine["id"],
|
||||
expected_at_least=1,
|
||||
)
|
||||
completed_run = await _wait_for_completed_run(ironclaw_server, routine["id"])
|
||||
|
||||
assert completed_run["id"]
|
||||
assert completed_run["started_at"]
|
||||
assert completed_run["status"].lower() == "attention"
|
||||
|
||||
+118
-255
@@ -7,7 +7,7 @@ import json
|
||||
import httpx
|
||||
import pytest
|
||||
|
||||
from helpers import AUTH_TOKEN
|
||||
from helpers import HTTP_WEBHOOK_SECRET
|
||||
|
||||
|
||||
def compute_signature(secret: str, body: bytes) -> str:
|
||||
@@ -16,325 +16,188 @@ def compute_signature(secret: str, body: bytes) -> str:
|
||||
return f"sha256={mac.hexdigest()}"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_requires_http_webhook_secret_configured(ironclaw_server):
|
||||
"""
|
||||
Webhook endpoint rejects requests when HTTP_WEBHOOK_SECRET is not configured.
|
||||
This tests the fail-closed security posture.
|
||||
"""
|
||||
headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
|
||||
async def _post_webhook(
|
||||
base_url: str,
|
||||
body_data: dict,
|
||||
*,
|
||||
signature: str | None = None,
|
||||
content_type: str = "application/json",
|
||||
) -> httpx.Response:
|
||||
"""Send a raw webhook request with optional signature."""
|
||||
body_bytes = json.dumps(body_data).encode()
|
||||
headers = {"Content-Type": content_type}
|
||||
if signature is not None:
|
||||
headers["X-Hub-Signature-256"] = signature
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
# When no webhook secret is configured on the server, all requests fail
|
||||
r = await client.post(
|
||||
f"{ironclaw_server}/webhook",
|
||||
json={"content": "test message"},
|
||||
return await client.post(
|
||||
f"{base_url}/webhook",
|
||||
content=body_bytes,
|
||||
headers=headers,
|
||||
)
|
||||
# Server should reject with 503 Service Unavailable (fail closed)
|
||||
assert r.status_code in (401, 503)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_hmac_signature_valid(ironclaw_server_with_webhook_secret):
|
||||
async def test_webhook_requires_http_webhook_secret_configured(
|
||||
http_channel_server_without_secret,
|
||||
):
|
||||
"""Webhook fails closed when no secret is configured."""
|
||||
response = await _post_webhook(
|
||||
http_channel_server_without_secret,
|
||||
{"content": "test message"},
|
||||
)
|
||||
|
||||
assert response.status_code == 503
|
||||
data = response.json()
|
||||
assert data["status"] == "error"
|
||||
assert "Webhook authentication not configured" in data.get("response", "")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_hmac_signature_valid(http_channel_server):
|
||||
"""Valid X-Hub-Signature-256 HMAC signature is accepted."""
|
||||
secret = ironclaw_server_with_webhook_secret["secret"]
|
||||
base_url = ironclaw_server_with_webhook_secret["url"]
|
||||
body = {"content": "hello from webhook"}
|
||||
signature = compute_signature(HTTP_WEBHOOK_SECRET, json.dumps(body).encode())
|
||||
|
||||
headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
|
||||
body_data = {"content": "hello from webhook"}
|
||||
body_bytes = json.dumps(body_data).encode()
|
||||
signature = compute_signature(secret, body_bytes)
|
||||
response = await _post_webhook(http_channel_server, body, signature=signature)
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
r = await client.post(
|
||||
f"{base_url}/webhook",
|
||||
content=body_bytes,
|
||||
headers={
|
||||
**headers,
|
||||
"Content-Type": "application/json",
|
||||
"X-Hub-Signature-256": signature,
|
||||
},
|
||||
)
|
||||
assert r.status_code == 200, f"Expected 200, got {r.status_code}: {r.text}"
|
||||
resp = r.json()
|
||||
assert resp["status"] == "ok"
|
||||
assert response.status_code == 200, (
|
||||
f"Expected 200, got {response.status_code}: {response.text}"
|
||||
)
|
||||
data = response.json()
|
||||
assert data["status"] == "accepted"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_invalid_hmac_signature_rejected(
|
||||
ironclaw_server_with_webhook_secret,
|
||||
):
|
||||
async def test_webhook_invalid_hmac_signature_rejected(http_channel_server):
|
||||
"""Invalid X-Hub-Signature-256 signature is rejected with 401."""
|
||||
base_url = ironclaw_server_with_webhook_secret["url"]
|
||||
response = await _post_webhook(
|
||||
http_channel_server,
|
||||
{"content": "hello"},
|
||||
signature="sha256=0000000000000000000000000000000000000000000000000000000000000000",
|
||||
)
|
||||
|
||||
headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
|
||||
body_data = {"content": "hello"}
|
||||
body_bytes = json.dumps(body_data).encode()
|
||||
invalid_signature = "sha256=0000000000000000000000000000000000000000000000000000000000000000"
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
r = await client.post(
|
||||
f"{base_url}/webhook",
|
||||
content=body_bytes,
|
||||
headers={
|
||||
**headers,
|
||||
"Content-Type": "application/json",
|
||||
"X-Hub-Signature-256": invalid_signature,
|
||||
},
|
||||
)
|
||||
assert r.status_code == 401, f"Expected 401, got {r.status_code}"
|
||||
resp = r.json()
|
||||
assert resp["status"] == "error"
|
||||
assert "Invalid webhook signature" in resp.get("response", "")
|
||||
assert response.status_code == 401
|
||||
data = response.json()
|
||||
assert data["status"] == "error"
|
||||
assert "Invalid webhook signature" in data.get("response", "")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_wrong_secret_rejected(ironclaw_server_with_webhook_secret):
|
||||
async def test_webhook_wrong_secret_rejected(http_channel_server):
|
||||
"""Signature computed with wrong secret is rejected."""
|
||||
base_url = ironclaw_server_with_webhook_secret["url"]
|
||||
body = {"content": "hello"}
|
||||
signature = compute_signature("wrong-secret", json.dumps(body).encode())
|
||||
|
||||
headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
|
||||
body_data = {"content": "hello"}
|
||||
body_bytes = json.dumps(body_data).encode()
|
||||
# Compute signature with wrong secret
|
||||
wrong_signature = compute_signature("wrong-secret", body_bytes)
|
||||
response = await _post_webhook(http_channel_server, body, signature=signature)
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
r = await client.post(
|
||||
f"{base_url}/webhook",
|
||||
content=body_bytes,
|
||||
headers={
|
||||
**headers,
|
||||
"Content-Type": "application/json",
|
||||
"X-Hub-Signature-256": wrong_signature,
|
||||
},
|
||||
)
|
||||
assert r.status_code == 401
|
||||
resp = r.json()
|
||||
assert resp["status"] == "error"
|
||||
assert response.status_code == 401
|
||||
assert response.json()["status"] == "error"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_malformed_signature_rejected(
|
||||
ironclaw_server_with_webhook_secret,
|
||||
):
|
||||
"""Malformed X-Hub-Signature-256 header is rejected."""
|
||||
base_url = ironclaw_server_with_webhook_secret["url"]
|
||||
async def test_webhook_missing_signature_header_rejected(http_channel_server):
|
||||
"""Missing X-Hub-Signature-256 header is rejected when no body secret is provided."""
|
||||
response = await _post_webhook(http_channel_server, {"content": "hello"})
|
||||
|
||||
headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
|
||||
body_data = {"content": "hello"}
|
||||
body_bytes = json.dumps(body_data).encode()
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
# Missing sha256= prefix
|
||||
r = await client.post(
|
||||
f"{base_url}/webhook",
|
||||
content=body_bytes,
|
||||
headers={
|
||||
**headers,
|
||||
"Content-Type": "application/json",
|
||||
"X-Hub-Signature-256": "deadbeef",
|
||||
},
|
||||
)
|
||||
assert r.status_code == 401
|
||||
assert response.status_code == 401
|
||||
data = response.json()
|
||||
assert "Webhook authentication required" in data.get("response", "")
|
||||
assert "X-Hub-Signature-256" in data.get("response", "")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_missing_signature_header_rejected(
|
||||
ironclaw_server_with_webhook_secret,
|
||||
):
|
||||
"""Missing X-Hub-Signature-256 header is rejected when no body secret provided."""
|
||||
base_url = ironclaw_server_with_webhook_secret["url"]
|
||||
async def test_webhook_deprecated_body_secret_still_works(http_channel_server):
|
||||
"""Deprecated body secret support still accepts old clients."""
|
||||
response = await _post_webhook(
|
||||
http_channel_server,
|
||||
{"content": "hello", "secret": HTTP_WEBHOOK_SECRET},
|
||||
)
|
||||
|
||||
headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
|
||||
body_data = {"content": "hello"}
|
||||
body_bytes = json.dumps(body_data).encode()
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
# No X-Hub-Signature-256 header and no body secret
|
||||
r = await client.post(
|
||||
f"{base_url}/webhook",
|
||||
content=body_bytes,
|
||||
headers={
|
||||
**headers,
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
)
|
||||
assert r.status_code == 401
|
||||
resp = r.json()
|
||||
assert "Webhook authentication required" in resp.get("response", "")
|
||||
assert "X-Hub-Signature-256" in resp.get("response", "")
|
||||
assert response.status_code == 200, (
|
||||
f"Expected 200, got {response.status_code}: {response.text}"
|
||||
)
|
||||
assert response.json()["status"] == "accepted"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_deprecated_body_secret_still_works(
|
||||
ironclaw_server_with_webhook_secret,
|
||||
):
|
||||
"""
|
||||
Deprecated: body 'secret' field still works for backward compatibility.
|
||||
This test ensures we don't break existing clients during the migration period.
|
||||
"""
|
||||
secret = ironclaw_server_with_webhook_secret["secret"]
|
||||
base_url = ironclaw_server_with_webhook_secret["url"]
|
||||
async def test_webhook_header_takes_precedence_over_body_secret(http_channel_server):
|
||||
"""Header signature wins when both header and body secret are provided."""
|
||||
body = {"content": "hello", "secret": "wrong-secret-in-body"}
|
||||
signature = compute_signature(HTTP_WEBHOOK_SECRET, json.dumps(body).encode())
|
||||
|
||||
headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
|
||||
# Old-style request with secret in body
|
||||
body_data = {"content": "hello", "secret": secret}
|
||||
body_bytes = json.dumps(body_data).encode()
|
||||
response = await _post_webhook(http_channel_server, body, signature=signature)
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
r = await client.post(
|
||||
f"{base_url}/webhook",
|
||||
content=body_bytes,
|
||||
headers={
|
||||
**headers,
|
||||
"Content-Type": "application/json",
|
||||
},
|
||||
)
|
||||
# Should succeed (backward compatibility)
|
||||
assert r.status_code == 200, f"Expected 200, got {r.status_code}: {r.text}"
|
||||
resp = r.json()
|
||||
assert resp["status"] == "ok"
|
||||
assert response.status_code == 200
|
||||
assert response.json()["status"] == "accepted"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_header_takes_precedence_over_body_secret(
|
||||
ironclaw_server_with_webhook_secret,
|
||||
):
|
||||
"""
|
||||
When both X-Hub-Signature-256 header and body secret are provided,
|
||||
header takes precedence.
|
||||
"""
|
||||
secret = ironclaw_server_with_webhook_secret["secret"]
|
||||
base_url = ironclaw_server_with_webhook_secret["url"]
|
||||
|
||||
headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
|
||||
body_data = {"content": "hello", "secret": "wrong-secret-in-body"}
|
||||
body_bytes = json.dumps(body_data).encode()
|
||||
# Compute signature with correct secret
|
||||
signature = compute_signature(secret, body_bytes)
|
||||
async def test_webhook_case_insensitive_header_lookup(http_channel_server):
|
||||
"""HTTP headers are treated case-insensitively."""
|
||||
body = {"content": "hello"}
|
||||
body_bytes = json.dumps(body).encode()
|
||||
signature = compute_signature(HTTP_WEBHOOK_SECRET, body_bytes)
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
r = await client.post(
|
||||
f"{base_url}/webhook",
|
||||
response = await client.post(
|
||||
f"{http_channel_server}/webhook",
|
||||
content=body_bytes,
|
||||
headers={
|
||||
**headers,
|
||||
"Content-Type": "application/json",
|
||||
"X-Hub-Signature-256": signature,
|
||||
},
|
||||
)
|
||||
# Should succeed because header signature is valid (takes precedence)
|
||||
assert r.status_code == 200
|
||||
resp = r.json()
|
||||
assert resp["status"] == "ok"
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_case_insensitive_header_lookup(
|
||||
ironclaw_server_with_webhook_secret,
|
||||
):
|
||||
"""HTTP headers are case-insensitive. Test with different cases."""
|
||||
secret = ironclaw_server_with_webhook_secret["secret"]
|
||||
base_url = ironclaw_server_with_webhook_secret["url"]
|
||||
|
||||
headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
|
||||
body_data = {"content": "hello"}
|
||||
body_bytes = json.dumps(body_data).encode()
|
||||
signature = compute_signature(secret, body_bytes)
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
# Try with lowercase
|
||||
r = await client.post(
|
||||
f"{base_url}/webhook",
|
||||
content=body_bytes,
|
||||
headers={
|
||||
**headers,
|
||||
"Content-Type": "application/json",
|
||||
"x-hub-signature-256": signature,
|
||||
},
|
||||
)
|
||||
assert r.status_code == 200
|
||||
|
||||
assert response.status_code == 200
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_wrong_content_type_rejected(
|
||||
ironclaw_server_with_webhook_secret,
|
||||
):
|
||||
async def test_webhook_wrong_content_type_rejected(http_channel_server):
|
||||
"""Webhook only accepts application/json Content-Type."""
|
||||
secret = ironclaw_server_with_webhook_secret["secret"]
|
||||
base_url = ironclaw_server_with_webhook_secret["url"]
|
||||
body = {"content": "hello"}
|
||||
signature = compute_signature(HTTP_WEBHOOK_SECRET, json.dumps(body).encode())
|
||||
|
||||
headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
|
||||
body_data = {"content": "hello"}
|
||||
body_bytes = json.dumps(body_data).encode()
|
||||
signature = compute_signature(secret, body_bytes)
|
||||
response = await _post_webhook(
|
||||
http_channel_server,
|
||||
body,
|
||||
signature=signature,
|
||||
content_type="text/plain",
|
||||
)
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
r = await client.post(
|
||||
f"{base_url}/webhook",
|
||||
content=body_bytes,
|
||||
headers={
|
||||
**headers,
|
||||
"Content-Type": "text/plain",
|
||||
"X-Hub-Signature-256": signature,
|
||||
},
|
||||
)
|
||||
assert r.status_code == 415 # Unsupported Media Type
|
||||
resp = r.json()
|
||||
assert "application/json" in resp.get("response", "")
|
||||
assert response.status_code == 415
|
||||
assert "application/json" in response.json().get("response", "")
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_invalid_json_rejected(ironclaw_server_with_webhook_secret):
|
||||
async def test_webhook_invalid_json_rejected(http_channel_server):
|
||||
"""Invalid JSON in body is rejected."""
|
||||
secret = ironclaw_server_with_webhook_secret["secret"]
|
||||
base_url = ironclaw_server_with_webhook_secret["url"]
|
||||
|
||||
headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
|
||||
body_bytes = b"not valid json"
|
||||
signature = compute_signature(secret, body_bytes)
|
||||
signature = compute_signature(HTTP_WEBHOOK_SECRET, body_bytes)
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
r = await client.post(
|
||||
f"{base_url}/webhook",
|
||||
response = await client.post(
|
||||
f"{http_channel_server}/webhook",
|
||||
content=body_bytes,
|
||||
headers={
|
||||
**headers,
|
||||
"Content-Type": "application/json",
|
||||
"X-Hub-Signature-256": signature,
|
||||
},
|
||||
)
|
||||
assert r.status_code == 401 or r.status_code == 400
|
||||
|
||||
assert response.status_code in (400, 401)
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_webhook_message_queued_for_processing(
|
||||
ironclaw_server_with_webhook_secret,
|
||||
):
|
||||
"""Message via webhook is queued and can be retrieved."""
|
||||
secret = ironclaw_server_with_webhook_secret["secret"]
|
||||
base_url = ironclaw_server_with_webhook_secret["url"]
|
||||
async def test_webhook_message_queued_for_processing(http_channel_server):
|
||||
"""Accepted webhook requests return a real message id."""
|
||||
body = {"content": "webhook test message 12345"}
|
||||
signature = compute_signature(HTTP_WEBHOOK_SECRET, json.dumps(body).encode())
|
||||
|
||||
headers = {"Authorization": f"Bearer {AUTH_TOKEN}"}
|
||||
test_message = "webhook test message 12345"
|
||||
body_data = {"content": test_message}
|
||||
body_bytes = json.dumps(body_data).encode()
|
||||
signature = compute_signature(secret, body_bytes)
|
||||
response = await _post_webhook(http_channel_server, body, signature=signature)
|
||||
|
||||
async with httpx.AsyncClient() as client:
|
||||
r = await client.post(
|
||||
f"{base_url}/webhook",
|
||||
content=body_bytes,
|
||||
headers={
|
||||
**headers,
|
||||
"Content-Type": "application/json",
|
||||
"X-Hub-Signature-256": signature,
|
||||
},
|
||||
)
|
||||
assert r.status_code == 200
|
||||
resp = r.json()
|
||||
assert resp["status"] == "ok"
|
||||
# Message ID should be present
|
||||
assert "message_id" in resp
|
||||
assert resp["message_id"] != "00000000-0000-0000-0000-000000000000"
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert data["status"] == "accepted"
|
||||
assert "message_id" in data
|
||||
assert data["message_id"] != "00000000-0000-0000-0000-000000000000"
|
||||
|
||||
Reference in New Issue
Block a user