mirror of
https://github.com/outbackdingo/optimclaw.git
synced 2026-08-25 14:53:34 +00:00
fix: N+1 query pattern in event trigger loop (routine_engine) (#1163)
* fix: N+1 query pattern in event trigger loop (routine_engine) * fix: linter
This commit is contained in:
Nick Pismenkov
@@ -0,0 +1,509 @@
|
||||
//! Tests for batch loading routine concurrent counts (N+1 query fix).
|
||||
//!
|
||||
//! Verifies:
|
||||
//! 1. Batch query returns correct counts for multiple routines
|
||||
//! 2. Concurrent limit enforcement uses batch counts correctly
|
||||
|
||||
#[cfg(feature = "libsql")]
|
||||
mod tests {
|
||||
use std::sync::Arc;
|
||||
|
||||
use chrono::Utc;
|
||||
use uuid::Uuid;
|
||||
|
||||
use ironclaw::agent::routine::{
|
||||
Routine, RoutineAction, RoutineGuardrails, RoutineRun, RunStatus, Trigger,
|
||||
};
|
||||
use ironclaw::db::Database;
|
||||
|
||||
async fn create_test_db() -> (Arc<dyn Database>, tempfile::TempDir) {
|
||||
use ironclaw::db::libsql::LibSqlBackend;
|
||||
|
||||
let temp_dir = tempfile::tempdir().expect("tempdir");
|
||||
let db_path = temp_dir.path().join("test.db");
|
||||
let backend = LibSqlBackend::new_local(&db_path)
|
||||
.await
|
||||
.expect("LibSqlBackend");
|
||||
backend.run_migrations().await.expect("migrations");
|
||||
let db: Arc<dyn Database> = Arc::new(backend);
|
||||
(db, temp_dir)
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Test 1: Batch query returns correct counts for multiple routines
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
#[tokio::test]
|
||||
async fn batch_query_empty_list() {
|
||||
let (db, _tmp) = create_test_db().await;
|
||||
let counts = db
|
||||
.count_running_routine_runs_batch(&[])
|
||||
.await
|
||||
.expect("batch query should not fail");
|
||||
assert!(counts.is_empty(), "Empty input should return empty map");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn batch_query_single_routine() {
|
||||
let (db, _tmp) = create_test_db().await;
|
||||
let routine_id = Uuid::new_v4();
|
||||
|
||||
// Create routine
|
||||
let routine = Routine {
|
||||
id: routine_id,
|
||||
name: "test-routine".to_string(),
|
||||
description: "Test".to_string(),
|
||||
user_id: "default".to_string(),
|
||||
enabled: true,
|
||||
trigger: Trigger::Cron {
|
||||
schedule: "* * * * *".to_string(),
|
||||
timezone: None,
|
||||
},
|
||||
action: RoutineAction::Lightweight {
|
||||
prompt: "test".to_string(),
|
||||
context_paths: vec![],
|
||||
max_tokens: 1000,
|
||||
use_tools: false,
|
||||
max_tool_rounds: 3,
|
||||
},
|
||||
guardrails: RoutineGuardrails {
|
||||
cooldown: std::time::Duration::from_secs(0),
|
||||
max_concurrent: 5,
|
||||
dedup_window: None,
|
||||
},
|
||||
notify: Default::default(),
|
||||
last_run_at: None,
|
||||
next_fire_at: None,
|
||||
run_count: 0,
|
||||
consecutive_failures: 0,
|
||||
state: serde_json::json!({}),
|
||||
created_at: Utc::now(),
|
||||
updated_at: Utc::now(),
|
||||
};
|
||||
db.create_routine(&routine).await.expect("create routine");
|
||||
|
||||
// Create 3 running runs
|
||||
for _ in 0..3 {
|
||||
let run = RoutineRun {
|
||||
id: Uuid::new_v4(),
|
||||
routine_id,
|
||||
trigger_type: "cron".to_string(),
|
||||
trigger_detail: None,
|
||||
started_at: Utc::now(),
|
||||
completed_at: None,
|
||||
status: RunStatus::Running,
|
||||
result_summary: None,
|
||||
tokens_used: None,
|
||||
job_id: None,
|
||||
created_at: Utc::now(),
|
||||
};
|
||||
db.create_routine_run(&run).await.expect("create run");
|
||||
}
|
||||
|
||||
// Batch query for single routine
|
||||
let counts = db
|
||||
.count_running_routine_runs_batch(&[routine_id])
|
||||
.await
|
||||
.expect("batch query should work");
|
||||
|
||||
assert_eq!(counts.len(), 1, "Should return 1 routine");
|
||||
assert_eq!(counts[&routine_id], 3, "Should count 3 running runs");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn batch_query_multiple_routines_different_counts() {
|
||||
let (db, _tmp) = create_test_db().await;
|
||||
|
||||
let r1 = Uuid::new_v4();
|
||||
let r2 = Uuid::new_v4();
|
||||
let r3 = Uuid::new_v4();
|
||||
|
||||
// Create 3 routines
|
||||
for routine_id in [r1, r2, r3] {
|
||||
let routine = Routine {
|
||||
id: routine_id,
|
||||
name: format!("routine-{}", routine_id),
|
||||
description: "Test".to_string(),
|
||||
user_id: "default".to_string(),
|
||||
enabled: true,
|
||||
trigger: Trigger::Cron {
|
||||
schedule: "* * * * *".to_string(),
|
||||
timezone: None,
|
||||
},
|
||||
action: RoutineAction::Lightweight {
|
||||
prompt: "test".to_string(),
|
||||
context_paths: vec![],
|
||||
max_tokens: 1000,
|
||||
use_tools: false,
|
||||
max_tool_rounds: 3,
|
||||
},
|
||||
guardrails: RoutineGuardrails {
|
||||
cooldown: std::time::Duration::from_secs(0),
|
||||
max_concurrent: 5,
|
||||
dedup_window: None,
|
||||
},
|
||||
notify: Default::default(),
|
||||
last_run_at: None,
|
||||
next_fire_at: None,
|
||||
run_count: 0,
|
||||
consecutive_failures: 0,
|
||||
state: serde_json::json!({}),
|
||||
created_at: Utc::now(),
|
||||
updated_at: Utc::now(),
|
||||
};
|
||||
db.create_routine(&routine).await.expect("create routine");
|
||||
}
|
||||
|
||||
// r1: 2 running
|
||||
for _ in 0..2 {
|
||||
let run = RoutineRun {
|
||||
id: Uuid::new_v4(),
|
||||
routine_id: r1,
|
||||
trigger_type: "cron".to_string(),
|
||||
trigger_detail: None,
|
||||
started_at: Utc::now(),
|
||||
completed_at: None,
|
||||
status: RunStatus::Running,
|
||||
result_summary: None,
|
||||
tokens_used: None,
|
||||
job_id: None,
|
||||
created_at: Utc::now(),
|
||||
};
|
||||
db.create_routine_run(&run).await.expect("create run");
|
||||
}
|
||||
|
||||
// r2: 1 running
|
||||
let run = RoutineRun {
|
||||
id: Uuid::new_v4(),
|
||||
routine_id: r2,
|
||||
trigger_type: "cron".to_string(),
|
||||
trigger_detail: None,
|
||||
started_at: Utc::now(),
|
||||
completed_at: None,
|
||||
status: RunStatus::Running,
|
||||
result_summary: None,
|
||||
tokens_used: None,
|
||||
job_id: None,
|
||||
created_at: Utc::now(),
|
||||
};
|
||||
db.create_routine_run(&run).await.expect("create run");
|
||||
|
||||
// r3: 0 running (but has 1 Ok result)
|
||||
let run = RoutineRun {
|
||||
id: Uuid::new_v4(),
|
||||
routine_id: r3,
|
||||
trigger_type: "cron".to_string(),
|
||||
trigger_detail: None,
|
||||
started_at: Utc::now(),
|
||||
completed_at: Some(Utc::now()),
|
||||
status: RunStatus::Ok,
|
||||
result_summary: None,
|
||||
tokens_used: None,
|
||||
job_id: None,
|
||||
created_at: Utc::now(),
|
||||
};
|
||||
db.create_routine_run(&run).await.expect("create run");
|
||||
|
||||
// Single batch query for all 3
|
||||
let counts = db
|
||||
.count_running_routine_runs_batch(&[r1, r2, r3])
|
||||
.await
|
||||
.expect("batch query should work");
|
||||
|
||||
assert_eq!(counts.len(), 3, "Should return 3 routines");
|
||||
assert_eq!(counts[&r1], 2, "r1 should have 2 running");
|
||||
assert_eq!(counts[&r2], 1, "r2 should have 1 running");
|
||||
assert_eq!(
|
||||
counts[&r3], 0,
|
||||
"r3 should have 0 running (Ok status is not running)"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn batch_query_missing_routines_default_to_zero() {
|
||||
let (db, _tmp) = create_test_db().await;
|
||||
|
||||
let r1 = Uuid::new_v4();
|
||||
let r2 = Uuid::new_v4();
|
||||
let r3 = Uuid::new_v4(); // This one won't exist
|
||||
|
||||
// Only create r1
|
||||
let routine = Routine {
|
||||
id: r1,
|
||||
name: "routine-1".to_string(),
|
||||
description: "Test".to_string(),
|
||||
user_id: "default".to_string(),
|
||||
enabled: true,
|
||||
trigger: Trigger::Cron {
|
||||
schedule: "* * * * *".to_string(),
|
||||
timezone: None,
|
||||
},
|
||||
action: RoutineAction::Lightweight {
|
||||
prompt: "test".to_string(),
|
||||
context_paths: vec![],
|
||||
max_tokens: 1000,
|
||||
use_tools: false,
|
||||
max_tool_rounds: 3,
|
||||
},
|
||||
guardrails: RoutineGuardrails {
|
||||
cooldown: std::time::Duration::from_secs(0),
|
||||
max_concurrent: 5,
|
||||
dedup_window: None,
|
||||
},
|
||||
notify: Default::default(),
|
||||
last_run_at: None,
|
||||
next_fire_at: None,
|
||||
run_count: 0,
|
||||
consecutive_failures: 0,
|
||||
state: serde_json::json!({}),
|
||||
created_at: Utc::now(),
|
||||
updated_at: Utc::now(),
|
||||
};
|
||||
db.create_routine(&routine).await.expect("create routine");
|
||||
|
||||
// r1 has 1 running
|
||||
let run = RoutineRun {
|
||||
id: Uuid::new_v4(),
|
||||
routine_id: r1,
|
||||
trigger_type: "cron".to_string(),
|
||||
trigger_detail: None,
|
||||
started_at: Utc::now(),
|
||||
completed_at: None,
|
||||
status: RunStatus::Running,
|
||||
result_summary: None,
|
||||
tokens_used: None,
|
||||
job_id: None,
|
||||
created_at: Utc::now(),
|
||||
};
|
||||
db.create_routine_run(&run).await.expect("create run");
|
||||
|
||||
// Query for r1, r2 (doesn't exist), r3 (doesn't exist)
|
||||
let counts = db
|
||||
.count_running_routine_runs_batch(&[r1, r2, r3])
|
||||
.await
|
||||
.expect("batch query should work");
|
||||
|
||||
assert_eq!(counts.len(), 3, "Should have all 3 routine IDs");
|
||||
assert_eq!(counts[&r1], 1, "r1 should have 1 running");
|
||||
assert_eq!(counts[&r2], 0, "r2 should default to 0");
|
||||
assert_eq!(counts[&r3], 0, "r3 should default to 0");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn batch_query_only_counts_running_status() {
|
||||
let (db, _tmp) = create_test_db().await;
|
||||
let routine_id = Uuid::new_v4();
|
||||
|
||||
// Create routine
|
||||
let routine = Routine {
|
||||
id: routine_id,
|
||||
name: "test-routine".to_string(),
|
||||
description: "Test".to_string(),
|
||||
user_id: "default".to_string(),
|
||||
enabled: true,
|
||||
trigger: Trigger::Cron {
|
||||
schedule: "* * * * *".to_string(),
|
||||
timezone: None,
|
||||
},
|
||||
action: RoutineAction::Lightweight {
|
||||
prompt: "test".to_string(),
|
||||
context_paths: vec![],
|
||||
max_tokens: 1000,
|
||||
use_tools: false,
|
||||
max_tool_rounds: 3,
|
||||
},
|
||||
guardrails: RoutineGuardrails {
|
||||
cooldown: std::time::Duration::from_secs(0),
|
||||
max_concurrent: 5,
|
||||
dedup_window: None,
|
||||
},
|
||||
notify: Default::default(),
|
||||
last_run_at: None,
|
||||
next_fire_at: None,
|
||||
run_count: 0,
|
||||
consecutive_failures: 0,
|
||||
state: serde_json::json!({}),
|
||||
created_at: Utc::now(),
|
||||
updated_at: Utc::now(),
|
||||
};
|
||||
db.create_routine(&routine).await.expect("create routine");
|
||||
|
||||
// Create 5 runs with mixed statuses
|
||||
let statuses = [
|
||||
RunStatus::Running,
|
||||
RunStatus::Running,
|
||||
RunStatus::Ok,
|
||||
RunStatus::Failed,
|
||||
RunStatus::Attention,
|
||||
];
|
||||
|
||||
for status in statuses.iter() {
|
||||
let run = RoutineRun {
|
||||
id: Uuid::new_v4(),
|
||||
routine_id,
|
||||
trigger_type: "cron".to_string(),
|
||||
trigger_detail: None,
|
||||
started_at: Utc::now(),
|
||||
completed_at: Some(Utc::now()),
|
||||
status: *status,
|
||||
result_summary: None,
|
||||
tokens_used: None,
|
||||
job_id: None,
|
||||
created_at: Utc::now(),
|
||||
};
|
||||
db.create_routine_run(&run).await.expect("create run");
|
||||
}
|
||||
|
||||
// Batch query should only count Running status
|
||||
let counts = db
|
||||
.count_running_routine_runs_batch(&[routine_id])
|
||||
.await
|
||||
.expect("batch query should work");
|
||||
|
||||
assert_eq!(
|
||||
counts[&routine_id], 2,
|
||||
"Should only count 2 Running status runs"
|
||||
);
|
||||
}
|
||||
|
||||
// -----------------------------------------------------------------------
|
||||
// Test 2: Concurrent limit enforcement uses batch counts
|
||||
// -----------------------------------------------------------------------
|
||||
|
||||
#[tokio::test]
|
||||
async fn concurrent_limit_enforcement_with_batch_counts() {
|
||||
let (db, _tmp) = create_test_db().await;
|
||||
|
||||
let r1 = Uuid::new_v4();
|
||||
let r2 = Uuid::new_v4();
|
||||
|
||||
// Create 2 routines with max_concurrent=1 (r1) and max_concurrent=2 (r2)
|
||||
for (routine_id, max_concurrent) in [(r1, 1), (r2, 2)] {
|
||||
let routine = Routine {
|
||||
id: routine_id,
|
||||
name: format!("routine-{}", routine_id),
|
||||
description: "Test".to_string(),
|
||||
user_id: "default".to_string(),
|
||||
enabled: true,
|
||||
trigger: Trigger::Cron {
|
||||
schedule: "* * * * *".to_string(),
|
||||
timezone: None,
|
||||
},
|
||||
action: RoutineAction::Lightweight {
|
||||
prompt: "test".to_string(),
|
||||
context_paths: vec![],
|
||||
max_tokens: 1000,
|
||||
use_tools: false,
|
||||
max_tool_rounds: 3,
|
||||
},
|
||||
guardrails: RoutineGuardrails {
|
||||
cooldown: std::time::Duration::from_secs(0),
|
||||
max_concurrent,
|
||||
dedup_window: None,
|
||||
},
|
||||
notify: Default::default(),
|
||||
last_run_at: None,
|
||||
next_fire_at: None,
|
||||
run_count: 0,
|
||||
consecutive_failures: 0,
|
||||
state: serde_json::json!({}),
|
||||
created_at: Utc::now(),
|
||||
updated_at: Utc::now(),
|
||||
};
|
||||
db.create_routine(&routine).await.expect("create routine");
|
||||
}
|
||||
|
||||
// r1: create 1 running run (will hit max_concurrent=1)
|
||||
let run = RoutineRun {
|
||||
id: Uuid::new_v4(),
|
||||
routine_id: r1,
|
||||
trigger_type: "cron".to_string(),
|
||||
trigger_detail: None,
|
||||
started_at: Utc::now(),
|
||||
completed_at: None,
|
||||
status: RunStatus::Running,
|
||||
result_summary: None,
|
||||
tokens_used: None,
|
||||
job_id: None,
|
||||
created_at: Utc::now(),
|
||||
};
|
||||
db.create_routine_run(&run).await.expect("create run");
|
||||
|
||||
// r2: create 2 running runs (will hit max_concurrent=2)
|
||||
for _ in 0..2 {
|
||||
let run = RoutineRun {
|
||||
id: Uuid::new_v4(),
|
||||
routine_id: r2,
|
||||
trigger_type: "cron".to_string(),
|
||||
trigger_detail: None,
|
||||
started_at: Utc::now(),
|
||||
completed_at: None,
|
||||
status: RunStatus::Running,
|
||||
result_summary: None,
|
||||
tokens_used: None,
|
||||
job_id: None,
|
||||
created_at: Utc::now(),
|
||||
};
|
||||
db.create_routine_run(&run).await.expect("create run");
|
||||
}
|
||||
|
||||
// Batch query should return correct counts
|
||||
let counts = db
|
||||
.count_running_routine_runs_batch(&[r1, r2])
|
||||
.await
|
||||
.expect("batch query should work");
|
||||
|
||||
// Verify counts match the limits
|
||||
assert_eq!(
|
||||
counts[&r1], 1,
|
||||
"r1 should have 1 running (at max_concurrent=1)"
|
||||
);
|
||||
assert_eq!(
|
||||
counts[&r2], 2,
|
||||
"r2 should have 2 running (at max_concurrent=2)"
|
||||
);
|
||||
|
||||
// Now verify the limit enforcement logic
|
||||
let r1_routine = db
|
||||
.get_routine(r1)
|
||||
.await
|
||||
.expect("get routine")
|
||||
.expect("routine exists");
|
||||
let r2_routine = db
|
||||
.get_routine(r2)
|
||||
.await
|
||||
.expect("get routine")
|
||||
.expect("routine exists");
|
||||
|
||||
let r1_at_limit = counts[&r1] >= r1_routine.guardrails.max_concurrent as i64;
|
||||
let r2_at_limit = counts[&r2] >= r2_routine.guardrails.max_concurrent as i64;
|
||||
|
||||
assert!(r1_at_limit, "r1 should be detected as at limit");
|
||||
assert!(r2_at_limit, "r2 should be detected as at limit");
|
||||
|
||||
// If we add one more run to r2, it should exceed limit
|
||||
let run = RoutineRun {
|
||||
id: Uuid::new_v4(),
|
||||
routine_id: r2,
|
||||
trigger_type: "cron".to_string(),
|
||||
trigger_detail: None,
|
||||
started_at: Utc::now(),
|
||||
completed_at: None,
|
||||
status: RunStatus::Running,
|
||||
result_summary: None,
|
||||
tokens_used: None,
|
||||
job_id: None,
|
||||
created_at: Utc::now(),
|
||||
};
|
||||
db.create_routine_run(&run).await.expect("create run");
|
||||
|
||||
// Re-query to get updated counts
|
||||
let counts = db
|
||||
.count_running_routine_runs_batch(&[r1, r2])
|
||||
.await
|
||||
.expect("batch query should work");
|
||||
|
||||
let r2_exceeded_limit = counts[&r2] > r2_routine.guardrails.max_concurrent as i64;
|
||||
assert!(r2_exceeded_limit, "r2 should have exceeded its limit");
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,534 @@
|
||||
"""
|
||||
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.
|
||||
"""
|
||||
|
||||
import asyncio
|
||||
import json
|
||||
import pytest
|
||||
from datetime import datetime, timedelta
|
||||
from typing import List, Dict, Any
|
||||
|
||||
from playwright.async_api import async_playwright, Page, Browser, BrowserContext
|
||||
|
||||
|
||||
@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()
|
||||
|
||||
|
||||
class EventTriggerHelper:
|
||||
"""Helper methods for event trigger testing."""
|
||||
|
||||
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 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}",
|
||||
)
|
||||
|
||||
# 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"
|
||||
)
|
||||
and any(
|
||||
e.get("routine_id") == r3_id for e in sse2 if e.get("type") == "routine_fired"
|
||||
)
|
||||
)
|
||||
|
||||
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"
|
||||
|
||||
finally:
|
||||
await page.close()
|
||||
|
||||
|
||||
# =============================================================================
|
||||
# Integration with existing test patterns
|
||||
# =============================================================================
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
# Run tests with: pytest tests/e2e/scenarios/test_routine_event_batch.py -v
|
||||
pytest.main([__file__, "-v", "-s"])
|
||||
Reference in New Issue
Block a user