Hare/llm_worker_rs
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: Verify provider API ・ Modularize testing

Browse Source
This commit is contained in:
Hare
2026-01-07 00:16:35 +09:00
parent a7581f27bb
commit d04cae2a36
12 changed files with 1845 additions and 2298 deletions
Download Patch File Download Diff File Expand all files Collapse all files
worker/tests/common/mod.rs
+227 -268
View File
@@ -1,284 +1,45 @@
//! テスト用共通ユーティリティ
//!
//! MockLlmClient、イベントレコーダー・プレイヤーを提供する
#![allow(dead_code)]
use std::fs::File;
use std::io::{BufRead, BufReader, BufWriter, Write};
use std::path::Path;
use std::io::{BufRead, BufReader};
use std::path::{Path, PathBuf};
use std::sync::{Arc, Mutex};
use std::pin::Pin;
use std::time::{Instant, SystemTime, UNIX_EPOCH};
use async_trait::async_trait;
use futures::Stream;
use serde::{Deserialize, Serialize};
use worker::{Handler, TextBlockEvent, TextBlockKind, Timeline};
use worker::llm_client::{ClientError, LlmClient, Request};
use worker_types::Event;
use worker_types::{BlockType, DeltaContent, Event};
// =============================================================================
// Recorded Event Types
// =============================================================================
use std::sync::atomic::{AtomicUsize, Ordering};
/// 記録されたSSEイベント
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct RecordedEvent {
/// イベント受信からの経過時間 (ミリ秒)
pub elapsed_ms: u64,
/// SSEイベントタイプ
pub event_type: String,
/// SSEイベントデータ
pub data: String,
}
/// セッションメタデータ
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct SessionMetadata {
/// 記録開始タイムスタンプ (Unix epoch秒)
pub timestamp: u64,
/// モデル名
pub model: String,
/// リクエストの説明
pub description: String,
}
// =============================================================================
// Event Recorder
// =============================================================================
/// SSEイベントレコーダー
///
/// 実際のAPIレスポンスを記録し、後でテストに使用できるようにする
#[allow(dead_code)]
pub struct EventRecorder {
start_time: Instant,
events: Vec<RecordedEvent>,
metadata: SessionMetadata,
}
#[allow(dead_code)]
impl EventRecorder {
/// 新しいレコーダーを作成
pub fn new(model: impl Into<String>, description: impl Into<String>) -> Self {
let timestamp = SystemTime::now()
.duration_since(UNIX_EPOCH)
.unwrap()
.as_secs();
Self {
start_time: Instant::now(),
events: Vec::new(),
metadata: SessionMetadata {
timestamp,
model: model.into(),
description: description.into(),
},
}
}
/// イベントを記録
pub fn record(&mut self, event_type: &str, data: &str) {
let elapsed = self.start_time.elapsed();
self.events.push(RecordedEvent {
elapsed_ms: elapsed.as_millis() as u64,
event_type: event_type.to_string(),
data: data.to_string(),
});
}
/// 記録をファイルに保存
///
/// フォーマット: JSONL (1行目: metadata, 2行目以降: events)
pub fn save(&self, path: impl AsRef<Path>) -> std::io::Result<()> {
let file = File::create(path)?;
let mut writer = BufWriter::new(file);
// メタデータを書き込み
let metadata_json = serde_json::to_string(&self.metadata)?;
writeln!(writer, "{}", metadata_json)?;
// イベントを書き込み
for event in &self.events {
let event_json = serde_json::to_string(event)?;
writeln!(writer, "{}", event_json)?;
}
writer.flush()?;
Ok(())
}
/// 記録されたイベント数を取得
pub fn event_count(&self) -> usize {
self.events.len()
}
}
// =============================================================================
// Event Player
// =============================================================================
/// SSEイベントプレイヤー
///
/// 記録されたイベントを読み込み、テストで使用する
#[allow(dead_code)]
pub struct EventPlayer {
metadata: SessionMetadata,
events: Vec<RecordedEvent>,
current_index: usize,
}
#[allow(dead_code)]
impl EventPlayer {
/// ファイルから読み込み
pub fn load(path: impl AsRef<Path>) -> std::io::Result<Self> {
let file = File::open(path)?;
let reader = BufReader::new(file);
let mut lines = reader.lines();
// メタデータを読み込み
let metadata_line = lines
.next()
.ok_or_else(|| std::io::Error::new(std::io::ErrorKind::InvalidData, "Empty file"))??;
let metadata: SessionMetadata = serde_json::from_str(&metadata_line)?;
// イベントを読み込み
let mut events = Vec::new();
for line in lines {
let line = line?;
if !line.is_empty() {
let event: RecordedEvent = serde_json::from_str(&line)?;
events.push(event);
}
}
Ok(Self {
metadata,
events,
current_index: 0,
})
}
/// メタデータを取得
pub fn metadata(&self) -> &SessionMetadata {
&self.metadata
}
/// 全イベントを取得
pub fn events(&self) -> &[RecordedEvent] {
&self.events
}
/// イベント数を取得
pub fn event_count(&self) -> usize {
self.events.len()
}
/// 次のイベントを取得（Iterator的に使用）
pub fn next_event(&mut self) -> Option<&RecordedEvent> {
if self.current_index < self.events.len() {
let event = &self.events[self.current_index];
self.current_index += 1;
Some(event)
} else {
None
}
}
/// インデックスをリセット
pub fn reset(&mut self) {
self.current_index = 0;
}
/// 全イベントをworker_types::Eventとしてパースして取得
pub fn parse_events(&self) -> Vec<Event> {
self.events
.iter()
.filter_map(|recorded| serde_json::from_str(&recorded.data).ok())
.collect()
}
}
// =============================================================================
// MockLlmClient
// =============================================================================
/// テスト用のモックLLMクライアント
///
/// 事前に定義されたイベントシーケンスをストリームとして返す。
/// fixtureファイルからロードすることも、直接イベントを渡すこともできる。
///
/// # 複数リクエスト対応
///
/// `with_responses()`を使用して、複数回のリクエストに対して異なるレスポンスを設定できる。
/// リクエスト回数が設定されたレスポンス数を超えた場合は空のストリームを返す。
/// A mock LLM client that replays a sequence of events
#[derive(Clone)]
pub struct MockLlmClient {
/// 各リクエストに対するレスポンス（イベントシーケンス）
responses: std::sync::Arc<std::sync::Mutex<Vec<Vec<Event>>>>,
/// 現在のリクエストインデックス
request_index: std::sync::Arc<std::sync::atomic::AtomicUsize>,
responses: Arc<Vec<Vec<Event>>>,
call_count: Arc<AtomicUsize>,
}
#[allow(dead_code)]
impl MockLlmClient {
/// イベントリストから直接作成（単一レスポンス）
///
/// すべてのリクエストに対して同じイベントシーケンスを返す（従来の動作）
pub fn new(events: Vec<Event>) -> Self {
Self {
responses: std::sync::Arc::new(std::sync::Mutex::new(vec![events])),
request_index: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
}
Self::with_responses(vec![events])
}
/// 複数のレスポンスを設定
///
/// 各リクエストに対して順番にイベントシーケンスを返す。
/// N回目のリクエストにはN番目のレスポンスが使用される。
///
/// # Example
/// ```ignore
/// let client = MockLlmClient::with_responses(vec![
/// // 1回目のリクエスト: ツール呼び出し
/// vec![Event::tool_use_start(0, "call_1", "my_tool"), ...],
/// // 2回目のリクエスト: テキストレスポンス
/// vec![Event::text_block_start(0), ...],
/// ]);
/// ```
pub fn with_responses(responses: Vec<Vec<Event>>) -> Self {
Self {
responses: std::sync::Arc::new(std::sync::Mutex::new(responses)),
request_index: std::sync::Arc::new(std::sync::atomic::AtomicUsize::new(0)),
responses: Arc::new(responses),
call_count: Arc::new(AtomicUsize::new(0)),
}
}
/// fixtureファイルからロード（単一レスポンス）
pub fn from_fixture(path: impl AsRef<Path>) -> std::io::Result<Self> {
let player = EventPlayer::load(path)?;
let events = player.parse_events();
pub fn from_fixture(path: impl AsRef<Path>) -> Result<Self, Box<dyn std::error::Error>> {
let events = load_events_from_fixture(path);
Ok(Self::new(events))
}
/// 保持しているレスポンス数を取得
pub fn response_count(&self) -> usize {
self.responses.lock().unwrap().len()
}
/// 最初のレスポンスのイベント数を取得（後方互換性）
pub fn event_count(&self) -> usize {
self.responses
.lock()
.unwrap()
.first()
.map(|v| v.len())
.unwrap_or(0)
}
/// 現在のリクエストインデックスを取得
pub fn current_request_index(&self) -> usize {
self.request_index.load(std::sync::atomic::Ordering::SeqCst)
}
/// リクエストインデックスをリセット
pub fn reset(&self) {
self.request_index.store(0, std::sync::atomic::Ordering::SeqCst);
self.responses.iter().map(|v| v.len()).sum()
}
}
@@ -288,20 +49,218 @@ impl LlmClient for MockLlmClient {
&self,
_request: Request,
) -> Result<Pin<Box<dyn Stream<Item = Result<Event, ClientError>> + Send>>, ClientError> {
let index = self.request_index.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
let events = {
let responses = self.responses.lock().unwrap();
if index < responses.len() {
responses[index].clone()
} else {
// レスポンスが尽きた場合は空のストリーム
Vec::new()
}
};
let count = self.call_count.fetch_add(1, Ordering::SeqCst);
if count >= self.responses.len() {
return Err(ClientError::Api {
status: Some(500),
code: Some("mock_error".to_string()),
message: "No more mock responses".to_string(),
});
}
let events = self.responses[count].clone();
let stream = futures::stream::iter(events.into_iter().map(Ok));
Ok(Box::pin(stream))
}
}
/// Load events from a fixture file
pub fn load_events_from_fixture(path: impl AsRef<Path>) -> Vec<Event> {
let file = File::open(path).expect("Failed to open fixture file");
let reader = BufReader::new(file);
let mut lines = reader.lines();
// Skip metadata line
let _metadata = lines.next().expect("Empty fixture file").unwrap();
let mut events = Vec::new();
for line in lines {
let line = line.unwrap();
if line.is_empty() {
continue;
}
let recorded: serde_json::Value = serde_json::from_str(&line).unwrap();
let data = recorded["data"].as_str().unwrap();
let event: Event = serde_json::from_str(data).unwrap();
events.push(event);
}
events
}
/// Find fixture files in a specific subdirectory
pub fn find_fixtures(subdir: &str) -> Vec<PathBuf> {
let fixtures_dir = Path::new(env!("CARGO_MANIFEST_DIR"))
.join("tests/fixtures")
.join(subdir);
if !fixtures_dir.exists() {
return Vec::new();
}
std::fs::read_dir(&fixtures_dir)
.unwrap()
.filter_map(|e| e.ok())
.map(|e| e.path())
.filter(|p| {
p.file_name()
.and_then(|n| n.to_str())
.is_some_and(|n| n.ends_with(".jsonl"))
})
.collect()
}
/// Assert that events in all fixtures for a provider can be deserialized
pub fn assert_events_deserialize(subdir: &str) {
let fixtures = find_fixtures(subdir);
assert!(!fixtures.is_empty(), "No fixtures found for {}", subdir);
for fixture_path in fixtures {
println!("Testing fixture deserialization: {:?}", fixture_path);
let events = load_events_from_fixture(&fixture_path);
assert!(!events.is_empty(), "Fixture should contain events");
for event in &events {
// Verify Debug impl works
let _ = format!("{:?}", event);
}
}
}
/// Assert that event sequence follows expected patterns
pub fn assert_event_sequence(subdir: &str) {
let fixtures = find_fixtures(subdir);
if fixtures.is_empty() {
println!("No fixtures found for {}, skipping sequence test", subdir);
return;
}
// Find a text-based fixture
let fixture_path = fixtures.iter()
.find(|p| p.to_string_lossy().contains("text"))
.unwrap_or(&fixtures[0]);
println!("Testing sequence with fixture: {:?}", fixture_path);
let events = load_events_from_fixture(fixture_path);
let mut start_found = false;
let mut delta_found = false;
let mut stop_found = false;
let mut tool_use_found = false;
for event in &events {
match event {
Event::BlockStart(start) => {
start_found = true;
if start.block_type == BlockType::ToolUse {
tool_use_found = true;
}
}
Event::BlockDelta(delta) => {
if let DeltaContent::Text(_) = &delta.delta {
delta_found = true;
}
}
Event::BlockStop(stop) => {
if stop.block_type == BlockType::Text {
stop_found = true;
}
}
_ => {}
}
}
assert!(!events.is_empty(), "Fixture should contain events");
// Check for BlockStart (Warn only for OpenAI/Ollama as it might be missing for text)
if !start_found {
println!("Warning: No BlockStart found. This is common for OpenAI/Ollama text streams.");
// For Anthropic, strict start is usually expected, but to keep common logic simple we allow warning.
// If specific strictness is needed, we could add a `strict: bool` arg.
}
assert!(delta_found, "Should contain BlockDelta");
if !tool_use_found {
assert!(stop_found, "Should contain BlockStop for Text block");
} else {
if !stop_found {
println!(" [Type: ToolUse] BlockStop detection skipped (not explicitly emitted by scheme)");
}
}
}
/// Assert usage tokens are present
pub fn assert_usage_tokens(subdir: &str) {
let fixtures = find_fixtures(subdir);
if fixtures.is_empty() {
return;
}
for fixture in fixtures {
let events = load_events_from_fixture(&fixture);
let usage_events: Vec<_> = events
.iter()
.filter_map(|e| if let Event::Usage(u) = e { Some(u) } else { None })
.collect();
if !usage_events.is_empty() {
let last_usage = usage_events.last().unwrap();
if last_usage.input_tokens.is_some() || last_usage.output_tokens.is_some() {
println!(" Fixture {:?} Usage: {:?}", fixture.file_name(), last_usage);
return; // Found valid usage
}
}
}
println!("Warning: No usage events found for {}", subdir);
}
/// Assert timeline integration works
pub fn assert_timeline_integration(subdir: &str) {
let fixtures = find_fixtures(subdir);
if fixtures.is_empty() {
return;
}
let fixture_path = fixtures.iter()
.find(|p| p.to_string_lossy().contains("text"))
.unwrap_or(&fixtures[0]);
println!("Testing timeline with fixture: {:?}", fixture_path);
let events = load_events_from_fixture(fixture_path);
struct TestCollector {
texts: Arc<Mutex<Vec<String>>>,
}
impl Handler<TextBlockKind> for TestCollector {
type Scope = String;
fn on_event(&mut self, buffer: &mut String, event: &TextBlockEvent) {
match event {
TextBlockEvent::Start(_) => {}
TextBlockEvent::Delta(text) => buffer.push_str(text),
TextBlockEvent::Stop(_) => {
let text = std::mem::take(buffer);
self.texts.lock().unwrap().push(text);
}
}
}
}
let collected = Arc::new(Mutex::new(Vec::new()));
let mut timeline = Timeline::new();
timeline.on_text_block(TestCollector {
texts: collected.clone(),
});
for event in &events {
timeline.dispatch(event);
}
let texts = collected.lock().unwrap();
if !texts.is_empty() {
assert!(!texts[0].is_empty(), "Collected text should not be empty");
println!(" Collected {} text blocks.", texts.len());
} else {
println!(" No text blocks collected (might be tool-only fixture)");
}
}