lhk229
1e186369c9
Project CI / AI game creator shell Rust crates (push) Successful in 1m24s
Project CI / AI game creator shell Rust smoke (push) Successful in 1m56s
Project CI / AI game creator shell Rust lane 1/2 (push) Has been cancelled
Project CI / Frontend tests (push) Has been cancelled
Project CI / Backend tests (push) Has been cancelled
Project CI / Repository checks (push) Has been cancelled
Project CI / AI game creator shell web tests (push) Has been cancelled
Project CI / AI game creator shell Rust lane 2/2 (push) Has been cancelled
Project CI / Native shell tests (push) Has been cancelled
Reviewed-on: https://git.genarrative.world/git/GenarrativeAI/Genarrative/pulls/446 Co-authored-by: Linghong <ink29535@proton.me> Co-committed-by: Linghong <ink29535@proton.me>
298 lines
12 KiB
Rust
298 lines
12 KiB
Rust
use super::*;
|
|
use crate::analytics::{
|
|
contract::{Context, EntrySource, EventData, PreviewSource, Route, SessionStart, Source},
|
|
preview::{prepare, Lease, Observation},
|
|
store::AnalyticsWriter,
|
|
};
|
|
|
|
struct Fixture {
|
|
_temp: tempfile::TempDir,
|
|
root: PathBuf,
|
|
config: PathBuf,
|
|
context: Context,
|
|
writer: AnalyticsWriter,
|
|
}
|
|
|
|
impl Fixture {
|
|
fn new() -> Self {
|
|
let temp = tempfile::tempdir().unwrap();
|
|
let root = temp.path().join("project");
|
|
init_local_game_project_at(&root, "preview-analytics", "预览埋点").unwrap();
|
|
fs::create_dir_all(project_game_root(&root)).unwrap();
|
|
fs::write(
|
|
project_game_root(&root).join("index.html"),
|
|
"<html>游戏入口</html>",
|
|
)
|
|
.unwrap();
|
|
let config = temp.path().join("config");
|
|
fs::create_dir_all(&config).unwrap();
|
|
let context = Context {
|
|
route: Route::from_identity(Some("A".into()), Some("https://example.com")),
|
|
editor_session_id: uuid::Uuid::new_v4().to_string(),
|
|
client_version: "1.0.0".into(),
|
|
};
|
|
let writer = AnalyticsWriter::start(config.clone(), context.editor_session_id.clone());
|
|
Self {
|
|
_temp: temp,
|
|
root,
|
|
config,
|
|
context,
|
|
writer,
|
|
}
|
|
}
|
|
|
|
fn capture(&self) -> Option<(Context, AnalyticsWriter)> {
|
|
Some((self.context.clone(), self.writer.clone()))
|
|
}
|
|
|
|
fn observation(&self, source: Source, preview_source: PreviewSource) -> (Lease, Observation) {
|
|
prepare(&self.root, self.capture(), source, preview_source).unwrap()
|
|
}
|
|
|
|
fn events(&self) -> Vec<serde_json::Value> {
|
|
let batches = self
|
|
.config
|
|
.join("analytics/instances")
|
|
.join(&self.context.editor_session_id)
|
|
.join("batches");
|
|
fs::read_dir(batches)
|
|
.into_iter()
|
|
.flatten()
|
|
.flatten()
|
|
.filter(|entry| !entry.file_name().to_string_lossy().starts_with('.'))
|
|
.filter_map(|entry| fs::read_to_string(entry.path().join("events.jsonl")).ok())
|
|
.flat_map(|text| {
|
|
text.lines()
|
|
.filter_map(|line| serde_json::from_str(line).ok())
|
|
.collect::<Vec<_>>()
|
|
})
|
|
.collect()
|
|
}
|
|
|
|
async fn drained(&self) -> Vec<serde_json::Value> {
|
|
let marker = self
|
|
.context
|
|
.capture(
|
|
EventData::EditorSessionStart(SessionStart {
|
|
entry_source: EntrySource::DirectLaunch,
|
|
first_project_id: None,
|
|
}),
|
|
None,
|
|
Source::Editor,
|
|
None,
|
|
)
|
|
.unwrap();
|
|
let marker_id = marker.event_id.clone();
|
|
assert!(self
|
|
.writer
|
|
.try_record(self.context.route.clone(), marker, marker_id.clone()));
|
|
assert!(self.writer.flush());
|
|
let deadline = std::time::Instant::now() + Duration::from_secs(5);
|
|
loop {
|
|
let events = self.events();
|
|
if events.iter().any(|event| event["event_id"] == marker_id) {
|
|
return events;
|
|
}
|
|
assert!(
|
|
std::time::Instant::now() < deadline,
|
|
"preview writer FIFO timeout"
|
|
);
|
|
tokio::time::sleep(Duration::from_millis(5)).await;
|
|
}
|
|
}
|
|
}
|
|
|
|
fn ready(events: &[serde_json::Value]) -> Vec<&serde_json::Value> {
|
|
events
|
|
.iter()
|
|
.filter(|event| event["event_name"] == "preview_ready")
|
|
.collect()
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn preview_analytics_real_entry_and_reopen_have_distinct_instances_and_original_identity() {
|
|
let fixture = Fixture::new();
|
|
let version = read_game_creator_agent_runtime_project_revision(&fixture.root)
|
|
.unwrap()
|
|
.revision;
|
|
let mut current = fixture.context.clone();
|
|
for source in [PreviewSource::User, PreviewSource::Agent] {
|
|
let event_source = if source == PreviewSource::User {
|
|
Source::Editor
|
|
} else {
|
|
Source::Direct
|
|
};
|
|
let (_lease, observation) = fixture.observation(event_source, source);
|
|
current.route.user_id = Some("B".into());
|
|
let (preview, stop) = start_local_game_preview_for_project(&fixture.root).unwrap();
|
|
observation.observe(preview.port).await;
|
|
let _ = stop.send(());
|
|
}
|
|
assert_eq!(current.route.user_id.as_deref(), Some("B"));
|
|
let events = fixture.drained().await;
|
|
let events = ready(&events);
|
|
assert_eq!(events.len(), 2);
|
|
assert_ne!(events[0]["event_id"], events[1]["event_id"]);
|
|
for event in &events {
|
|
assert_eq!(event["user_id"], "A");
|
|
assert_eq!(event["project_id"], "preview-analytics");
|
|
assert_eq!(event["creative_task_id"], "preview-analytics");
|
|
assert_eq!(event["properties"]["preview_version"], version.to_string());
|
|
assert!(event["agent_run_id"].is_null());
|
|
assert!(event["agent_turn_id"].is_null());
|
|
assert!(!event.to_string().contains("127.0.0.1"));
|
|
}
|
|
assert!(events.iter().any(
|
|
|event| event["source"] == "editor" && event["properties"]["preview_source"] == "user"
|
|
));
|
|
assert!(events.iter().any(
|
|
|event| event["source"] == "direct" && event["properties"]["preview_source"] == "agent"
|
|
));
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn preview_analytics_missing_empty_stopped_or_changed_entry_is_not_ready() {
|
|
let fixture = Fixture::new();
|
|
let entry = project_game_root(&fixture.root).join("index.html");
|
|
let (preview, stop) = start_local_game_preview_for_project(&fixture.root).unwrap();
|
|
let (lease, observation) = fixture.observation(Source::Direct, PreviewSource::Agent);
|
|
drop(lease);
|
|
observation.observe(preview.port).await;
|
|
let (_lease, observation) = fixture.observation(Source::Editor, PreviewSource::User);
|
|
fs::write(&entry, b"").unwrap();
|
|
observation.observe(preview.port).await;
|
|
assert!(prepare(
|
|
&fixture.root,
|
|
fixture.capture(),
|
|
Source::Editor,
|
|
PreviewSource::User
|
|
)
|
|
.is_none());
|
|
fs::write(&entry, b"entry").unwrap();
|
|
let (_lease, observation) = fixture.observation(Source::Editor, PreviewSource::User);
|
|
fs::remove_file(&entry).unwrap();
|
|
observation.observe(preview.port).await;
|
|
fs::write(&entry, b"entry").unwrap();
|
|
let (_lease, observation) = fixture.observation(Source::Editor, PreviewSource::User);
|
|
{
|
|
let _lock = acquire_project_write_lock(&fixture.root, "file.write").unwrap();
|
|
advance_agent_runtime_project_revision_locked(&fixture.root).unwrap();
|
|
}
|
|
observation.observe(preview.port).await;
|
|
let _ = stop.send(());
|
|
assert!(ready(&fixture.drained().await).is_empty());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn preview_analytics_checks_version_again_after_the_actual_request() {
|
|
let fixture = Fixture::new();
|
|
let (_lease, observation) = fixture.observation(Source::Editor, PreviewSource::User);
|
|
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
|
let port = listener.local_addr().unwrap().port();
|
|
let (requested, request_received) = tokio::sync::oneshot::channel();
|
|
let (allow_response, response_allowed) = mpsc::channel();
|
|
let server = thread::spawn(move || {
|
|
let (mut stream, _) = listener.accept().unwrap();
|
|
assert!(read_preview_request_line(&mut stream).unwrap().is_some());
|
|
requested.send(()).unwrap();
|
|
response_allowed
|
|
.recv_timeout(Duration::from_secs(3))
|
|
.unwrap();
|
|
stream
|
|
.write_all(b"HTTP/1.1 200 OK\r\nContent-Length: 5\r\nConnection: close\r\n\r\nready")
|
|
.unwrap();
|
|
});
|
|
let probe = tokio::spawn(observation.observe(port));
|
|
tokio::time::timeout(Duration::from_secs(3), request_received)
|
|
.await
|
|
.unwrap()
|
|
.unwrap();
|
|
{
|
|
let _lock = acquire_project_write_lock(&fixture.root, "file.write").unwrap();
|
|
advance_agent_runtime_project_revision_locked(&fixture.root).unwrap();
|
|
}
|
|
allow_response.send(()).unwrap();
|
|
probe.await.unwrap();
|
|
server.join().unwrap();
|
|
assert!(ready(&fixture.drained().await).is_empty());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn preview_analytics_http_failure_or_empty_response_is_not_ready() {
|
|
let fixture = Fixture::new();
|
|
for response in [
|
|
"HTTP/1.1 404 Not Found\r\nContent-Length: 3\r\nConnection: close\r\n\r\nbad",
|
|
"HTTP/1.1 200 OK\r\nContent-Length: 0\r\nConnection: close\r\n\r\n",
|
|
] {
|
|
let (_lease, observation) = fixture.observation(Source::Editor, PreviewSource::User);
|
|
let listener = TcpListener::bind("127.0.0.1:0").unwrap();
|
|
let port = listener.local_addr().unwrap().port();
|
|
let server = thread::spawn(move || {
|
|
let (mut stream, _) = listener.accept().unwrap();
|
|
assert!(read_preview_request_line(&mut stream).unwrap().is_some());
|
|
stream.write_all(response.as_bytes()).unwrap();
|
|
});
|
|
observation.observe(port).await;
|
|
server.join().unwrap();
|
|
}
|
|
assert!(ready(&fixture.drained().await).is_empty());
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn preview_analytics_gui_start_schedules_without_waiting_and_registry_invalidates_old_instance(
|
|
) {
|
|
let fixture = Fixture::new();
|
|
let registry = PreviewRegistry::default();
|
|
let started =
|
|
start_local_game_preview_with_capture(&fixture.root, None, ®istry, fixture.capture())
|
|
.unwrap();
|
|
assert_eq!(registry.status().port, Some(started.port));
|
|
let deadline = std::time::Instant::now() + Duration::from_secs(5);
|
|
loop {
|
|
if !ready(&fixture.drained().await).is_empty() {
|
|
break;
|
|
}
|
|
assert!(
|
|
std::time::Instant::now() < deadline,
|
|
"GUI scheduled ready timeout"
|
|
);
|
|
tokio::time::sleep(Duration::from_millis(10)).await;
|
|
}
|
|
registry.stop();
|
|
let (lease, old_observation) = fixture.observation(Source::Editor, PreviewSource::User);
|
|
let (old_preview, old_stop) = start_local_game_preview_for_project(&fixture.root).unwrap();
|
|
registry.set_running_with_lease(old_preview.clone(), old_stop, Some(lease));
|
|
let (lease, stopped_observation) = fixture.observation(Source::Editor, PreviewSource::User);
|
|
let (new_preview, new_stop) = start_local_game_preview_for_project(&fixture.root).unwrap();
|
|
registry.set_running_with_lease(new_preview.clone(), new_stop, Some(lease));
|
|
old_observation.observe(old_preview.port).await;
|
|
registry.stop();
|
|
stopped_observation.observe(new_preview.port).await;
|
|
assert_eq!(ready(&fixture.drained().await).len(), 1);
|
|
|
|
// 空入口仍遵循原有启动行为，采集跳过不能变成业务启动失败。
|
|
fs::write(project_game_root(&fixture.root).join("index.html"), b"").unwrap();
|
|
assert!(start_local_game_preview_with_capture(
|
|
&fixture.root,
|
|
None,
|
|
®istry,
|
|
fixture.capture()
|
|
)
|
|
.is_ok());
|
|
registry.stop();
|
|
assert_eq!(ready(&fixture.drained().await).len(), 1);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn preview_analytics_cancelled_direct_scope_does_not_publish_ready() {
|
|
let fixture = Fixture::new();
|
|
let (_lease, observation) = fixture.observation(Source::Direct, PreviewSource::Agent);
|
|
let cancelled = Arc::new(std::sync::atomic::AtomicBool::new(false));
|
|
let observation = observation.with_cancellation(Some(cancelled.clone()));
|
|
let (preview, stop) = start_local_game_preview_for_project(&fixture.root).unwrap();
|
|
cancelled.store(true, std::sync::atomic::Ordering::Release);
|
|
observation.observe(preview.port).await;
|
|
let _ = stop.send(());
|
|
assert!(ready(&fixture.drained().await).is_empty());
|
|
}
|