mirror of
https://github.com/tinyhumansai/openhuman.git
synced 2026-07-27 21:08:00 +00:00
test(memory): expand sync, vault, and artifact rust coverage (#2594)
This commit is contained in:
Steven Enamakel
@@ -55,6 +55,31 @@ impl WorkspaceEnvGuard {
|
||||
}
|
||||
}
|
||||
|
||||
struct HomeEnvGuard {
|
||||
previous: Option<std::ffi::OsString>,
|
||||
}
|
||||
|
||||
impl HomeEnvGuard {
|
||||
fn set(path: &Path) -> Self {
|
||||
let previous = std::env::var_os("HOME");
|
||||
unsafe {
|
||||
std::env::set_var("HOME", path);
|
||||
}
|
||||
Self { previous }
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for HomeEnvGuard {
|
||||
fn drop(&mut self) {
|
||||
unsafe {
|
||||
match self.previous.take() {
|
||||
Some(value) => std::env::set_var("HOME", value),
|
||||
None => std::env::remove_var("HOME"),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for WorkspaceEnvGuard {
|
||||
fn drop(&mut self) {
|
||||
unsafe {
|
||||
@@ -800,10 +825,12 @@ async fn execute_tool_per_call_factory_means_no_baked_client() {
|
||||
|
||||
let tmp = tempfile::tempdir().unwrap();
|
||||
let _workspace_guard = WorkspaceEnvGuard::set(tmp.path());
|
||||
let _home_guard = HomeEnvGuard::set(tmp.path());
|
||||
|
||||
let mut config = crate::openhuman::config::Config::default();
|
||||
config.config_path = tmp.path().join("config.toml");
|
||||
config.workspace_dir = tmp.path().join("workspace");
|
||||
std::fs::create_dir_all(&config.workspace_dir).expect("create workspace dir");
|
||||
config.composio.mode = crate::openhuman::config::schema::COMPOSIO_MODE_DIRECT.to_string();
|
||||
// No api_key here — direct-mode factory must reject.
|
||||
config.save().await.expect("save fake config to disk");
|
||||
@@ -846,10 +873,12 @@ async fn list_toolkits_in_direct_mode_returns_empty_without_hitting_backend() {
|
||||
|
||||
let tmp = tempfile::tempdir().expect("tempdir");
|
||||
let _workspace_guard = WorkspaceEnvGuard::set(tmp.path());
|
||||
let _home_guard = HomeEnvGuard::set(tmp.path());
|
||||
|
||||
let mut config = crate::openhuman::config::Config::default();
|
||||
config.config_path = tmp.path().join("config.toml");
|
||||
config.workspace_dir = tmp.path().join("workspace");
|
||||
std::fs::create_dir_all(&config.workspace_dir).expect("create workspace dir");
|
||||
config.composio.mode = crate::openhuman::config::schema::COMPOSIO_MODE_DIRECT.to_string();
|
||||
config.composio.api_key = Some("test-direct-key".to_string());
|
||||
config.save().await.expect("save fake config to disk");
|
||||
@@ -913,10 +942,12 @@ async fn authorize_in_direct_mode_refuses_with_app_composio_dev_hint() {
|
||||
|
||||
let tmp = tempfile::tempdir().expect("tempdir");
|
||||
let _workspace_guard = WorkspaceEnvGuard::set(tmp.path());
|
||||
let _home_guard = HomeEnvGuard::set(tmp.path());
|
||||
|
||||
let mut config = crate::openhuman::config::Config::default();
|
||||
config.config_path = tmp.path().join("config.toml");
|
||||
config.workspace_dir = tmp.path().join("workspace");
|
||||
std::fs::create_dir_all(&config.workspace_dir).expect("create workspace dir");
|
||||
config.composio.mode = crate::openhuman::config::schema::COMPOSIO_MODE_DIRECT.to_string();
|
||||
config.composio.api_key = Some("test-direct-key".to_string());
|
||||
config.save().await.expect("save fake config to disk");
|
||||
|
||||
@@ -3,6 +3,30 @@ use crate::openhuman::credentials::session_support::local_session_user_id;
|
||||
use serde_json::json;
|
||||
use tempfile::TempDir;
|
||||
|
||||
struct EnvVarGuard {
|
||||
key: &'static str,
|
||||
previous: Option<std::ffi::OsString>,
|
||||
}
|
||||
|
||||
impl EnvVarGuard {
|
||||
fn set_to_path(key: &'static str, path: &std::path::Path) -> Self {
|
||||
let previous = std::env::var_os(key);
|
||||
unsafe { std::env::set_var(key, path) };
|
||||
Self { key, previous }
|
||||
}
|
||||
}
|
||||
|
||||
impl Drop for EnvVarGuard {
|
||||
fn drop(&mut self) {
|
||||
unsafe {
|
||||
match self.previous.take() {
|
||||
Some(value) => std::env::set_var(self.key, value),
|
||||
None => std::env::remove_var(self.key),
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn test_config(tmp: &TempDir) -> Config {
|
||||
Config {
|
||||
workspace_dir: tmp.path().join("workspace"),
|
||||
@@ -81,7 +105,12 @@ fn sanitize_stored_session_user_discards_empty_objects() {
|
||||
/// user from an API response.
|
||||
#[tokio::test]
|
||||
async fn store_session_local_token_rejects_missing_user_payload() {
|
||||
let _env_guard = crate::openhuman::config::TEST_ENV_LOCK
|
||||
.lock()
|
||||
.unwrap_or_else(|e| e.into_inner());
|
||||
let tmp = TempDir::new().unwrap();
|
||||
std::fs::create_dir_all(tmp.path().join("workspace")).unwrap();
|
||||
let _home = EnvVarGuard::set_to_path("HOME", tmp.path());
|
||||
let config = test_config(&tmp);
|
||||
let local_token = "header.payload.local";
|
||||
let err = store_session(&config, local_token, None, None)
|
||||
@@ -99,7 +128,12 @@ async fn store_session_local_token_rejects_missing_user_payload() {
|
||||
/// summary.
|
||||
#[tokio::test]
|
||||
async fn store_session_local_token_succeeds_without_network_and_forces_local_user_id() {
|
||||
let _env_guard = crate::openhuman::config::TEST_ENV_LOCK
|
||||
.lock()
|
||||
.unwrap_or_else(|e| e.into_inner());
|
||||
let tmp = TempDir::new().unwrap();
|
||||
std::fs::create_dir_all(tmp.path().join("workspace")).unwrap();
|
||||
let _home = EnvVarGuard::set_to_path("HOME", tmp.path());
|
||||
let config = test_config(&tmp);
|
||||
let local_token = "header.payload.local";
|
||||
let user = serde_json::json!({
|
||||
|
||||
@@ -1670,8 +1670,11 @@ mod tests {
|
||||
use crate::openhuman::composio::providers::sync_state::KV_NAMESPACE;
|
||||
use crate::openhuman::embeddings::NoopEmbedding;
|
||||
use crate::openhuman::memory::ingest_pipeline::ingest_chat;
|
||||
use crate::openhuman::memory_queue::drain_until_idle;
|
||||
use crate::openhuman::memory_store::unified::UnifiedMemory;
|
||||
use crate::openhuman::memory_sync::canonicalize::chat::{ChatBatch, ChatMessage};
|
||||
use crate::openhuman::memory_sync::composio::providers::slack::ingest::ingest_page_into_memory_tree as ingest_slack_page;
|
||||
use crate::openhuman::memory_sync::composio::providers::slack::SlackMessage;
|
||||
use chrono::{TimeZone, Utc};
|
||||
use rusqlite::params;
|
||||
use std::sync::Arc;
|
||||
@@ -1710,6 +1713,35 @@ mod tests {
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
async fn seed_slack_chunk_with_raw_archive(cfg: &Config) -> String {
|
||||
let msg = SlackMessage {
|
||||
channel_id: "C123".into(),
|
||||
channel_name: "engineering".into(),
|
||||
is_private: false,
|
||||
author: "alice".into(),
|
||||
author_id: "U123".into(),
|
||||
text: "Phoenix migration launch window is Friday at 22:00 UTC.".into(),
|
||||
timestamp: Utc.timestamp_opt(1_700_000_000, 0).single().unwrap(),
|
||||
ts_raw: "1700000000.000100".into(),
|
||||
thread_ts: None,
|
||||
permalink: Some("https://slack.example.test/archives/C123/p1700000000000100".into()),
|
||||
};
|
||||
ingest_slack_page(cfg, "alice", "conn-slack-1", &[msg])
|
||||
.await
|
||||
.expect("seed slack ingest");
|
||||
drain_until_idle(cfg).await.expect("drain slack ingest");
|
||||
|
||||
list_chunks_rpc(cfg, ChunkFilter::default())
|
||||
.await
|
||||
.expect("list chunks")
|
||||
.value
|
||||
.chunks
|
||||
.into_iter()
|
||||
.find(|chunk| chunk.source_id == "slack:conn-slack-1")
|
||||
.expect("seeded slack chunk")
|
||||
.id
|
||||
}
|
||||
|
||||
fn update_chunk_timestamp(cfg: &Config, chunk_id: &str, timestamp_ms: i64) {
|
||||
with_connection(cfg, |conn| {
|
||||
conn.execute(
|
||||
@@ -2155,6 +2187,93 @@ mod tests {
|
||||
assert!(row.content_preview.as_deref().unwrap_or("").contains(body));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn flush_now_enqueues_once_and_reports_stale_buffers() {
|
||||
let (_tmp, cfg) = test_config();
|
||||
seed_chat_chunk(
|
||||
&cfg,
|
||||
"slack:#eng",
|
||||
"Phoenix migration ships Friday after the release checklist closes.",
|
||||
)
|
||||
.await;
|
||||
drain_until_idle(&cfg).await.expect("drain jobs");
|
||||
|
||||
let first = flush_now_rpc(&cfg).await.expect("flush_now first");
|
||||
assert!(first.value.enqueued, "first flush should enqueue work");
|
||||
assert!(
|
||||
first.value.stale_buffers >= 1,
|
||||
"expected at least one stale buffer after ingest"
|
||||
);
|
||||
|
||||
let second = flush_now_rpc(&cfg).await.expect("flush_now second");
|
||||
assert!(
|
||||
!second.value.enqueued,
|
||||
"same 3-hour window should dedupe duplicate flush triggers"
|
||||
);
|
||||
assert!(
|
||||
second.value.stale_buffers >= 1,
|
||||
"deduped flush should still report current stale buffer count"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn reset_tree_preserves_raw_archive_and_source_registry() {
|
||||
let (_tmp, cfg) = test_config();
|
||||
let chunk_id = seed_slack_chunk_with_raw_archive(&cfg).await;
|
||||
let content_root = cfg.memory_tree_content_root();
|
||||
let raw_file = content_root
|
||||
.join("raw")
|
||||
.join("slack-conn-slack-1")
|
||||
.join("chats")
|
||||
.join("1700000000000_1700000000.000100.md");
|
||||
let source_file = content_root
|
||||
.join("raw")
|
||||
.join("slack-conn-slack-1")
|
||||
.join("_source.md");
|
||||
assert!(raw_file.exists(), "raw archive should exist before reset");
|
||||
assert!(
|
||||
source_file.exists(),
|
||||
"source registry should exist before reset"
|
||||
);
|
||||
|
||||
let stale_summary = content_root
|
||||
.join("wiki")
|
||||
.join("summaries")
|
||||
.join("source-slack-conn-slack-1")
|
||||
.join("L1")
|
||||
.join("summary-stale.md");
|
||||
std::fs::create_dir_all(
|
||||
stale_summary
|
||||
.parent()
|
||||
.expect("stale summary parent should exist"),
|
||||
)
|
||||
.expect("create stale summary dir");
|
||||
std::fs::write(&stale_summary, "stale summary body").expect("write stale summary");
|
||||
assert!(stale_summary.exists(), "stale summary fixture should exist");
|
||||
|
||||
let outcome = reset_tree_rpc(&cfg).await.expect("reset_tree");
|
||||
assert_eq!(outcome.value.chunks_requeued, 1);
|
||||
assert_eq!(outcome.value.jobs_enqueued, 1);
|
||||
assert!(
|
||||
outcome.value.tree_rows_deleted >= 1,
|
||||
"buffer/tree rows should be removed during reset"
|
||||
);
|
||||
|
||||
let row = read_chunk_row(&cfg, &chunk_id)
|
||||
.expect("read chunk row")
|
||||
.expect("chunk row present after reset");
|
||||
assert_eq!(row.lifecycle_status, "pending_extraction");
|
||||
assert!(raw_file.exists(), "raw archive must survive reset_tree");
|
||||
assert!(
|
||||
source_file.exists(),
|
||||
"source registry must survive reset_tree"
|
||||
);
|
||||
assert!(
|
||||
!content_root.join("wiki").join("summaries").exists(),
|
||||
"derived wiki summaries should be removed"
|
||||
);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn read_chunk_row_returns_none_for_missing_chunk() {
|
||||
let (_tmp, cfg) = test_config();
|
||||
|
||||
@@ -158,3 +158,124 @@ impl EventHandler for MemorySyncStageBridge {
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use std::sync::{Arc, Mutex, OnceLock};
|
||||
|
||||
use crate::core::event_bus::{self, init_global, subscribe_global};
|
||||
|
||||
fn test_mutex() -> &'static std::sync::Mutex<()> {
|
||||
static LOCK: OnceLock<std::sync::Mutex<()>> = OnceLock::new();
|
||||
LOCK.get_or_init(|| std::sync::Mutex::new(()))
|
||||
}
|
||||
|
||||
#[derive(Clone, Default)]
|
||||
struct StageCollector {
|
||||
events: Arc<Mutex<Vec<DomainEvent>>>,
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
impl EventHandler for StageCollector {
|
||||
fn name(&self) -> &str {
|
||||
"memory::sync::tests::stage_collector"
|
||||
}
|
||||
|
||||
fn domains(&self) -> Option<&[&str]> {
|
||||
Some(&["memory"])
|
||||
}
|
||||
|
||||
async fn handle(&self, event: &DomainEvent) {
|
||||
if matches!(event, DomainEvent::MemorySyncStageChanged { .. }) {
|
||||
self.events.lock().unwrap().push(event.clone());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn document_canonicalized_emits_stored_and_queued_stages() {
|
||||
let _guard = test_mutex()
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner());
|
||||
init_global(event_bus::DEFAULT_CAPACITY);
|
||||
|
||||
let collector = StageCollector::default();
|
||||
let _subscription =
|
||||
subscribe_global(Arc::new(collector.clone())).expect("event bus initialized");
|
||||
|
||||
let bridge = MemorySyncStageBridge;
|
||||
bridge
|
||||
.handle(&DomainEvent::DocumentCanonicalized {
|
||||
source_id: "slack:workspace-1".into(),
|
||||
source_kind: "chat".into(),
|
||||
chunks_written: 3,
|
||||
chunk_ids: vec!["chunk-1".into()],
|
||||
canonicalized_at: 1_700_000_000.0,
|
||||
body_preview: None,
|
||||
})
|
||||
.await;
|
||||
|
||||
tokio::task::yield_now().await;
|
||||
|
||||
let stages: Vec<String> = collector
|
||||
.events
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.filter_map(|event| match event {
|
||||
DomainEvent::MemorySyncStageChanged { stage, .. } => Some(stage.clone()),
|
||||
_ => None,
|
||||
})
|
||||
.collect();
|
||||
assert!(stages.contains(&"stored".to_string()));
|
||||
assert!(stages.contains(&"queued".to_string()));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn memory_ingestion_started_emits_ingesting_stage() {
|
||||
let _guard = test_mutex()
|
||||
.lock()
|
||||
.unwrap_or_else(|poisoned| poisoned.into_inner());
|
||||
init_global(event_bus::DEFAULT_CAPACITY);
|
||||
|
||||
let collector = StageCollector::default();
|
||||
let _subscription =
|
||||
subscribe_global(Arc::new(collector.clone())).expect("event bus initialized");
|
||||
|
||||
let bridge = MemorySyncStageBridge;
|
||||
bridge
|
||||
.handle(&DomainEvent::MemoryIngestionStarted {
|
||||
document_id: "doc-123".into(),
|
||||
title: "Vault Note".into(),
|
||||
namespace: "vault:v-1".into(),
|
||||
queue_depth: 2,
|
||||
})
|
||||
.await;
|
||||
|
||||
tokio::task::yield_now().await;
|
||||
|
||||
let ingesting = collector
|
||||
.events
|
||||
.lock()
|
||||
.unwrap()
|
||||
.iter()
|
||||
.find_map(|event| match event {
|
||||
DomainEvent::MemorySyncStageChanged {
|
||||
stage,
|
||||
provider,
|
||||
connection_id,
|
||||
detail,
|
||||
..
|
||||
} if stage == "ingesting" => {
|
||||
Some((provider.clone(), connection_id.clone(), detail.clone()))
|
||||
}
|
||||
_ => None,
|
||||
})
|
||||
.expect("ingesting stage should be emitted");
|
||||
|
||||
assert_eq!(ingesting.0.as_deref(), Some("vault:v-1"));
|
||||
assert_eq!(ingesting.1.as_deref(), Some("doc-123"));
|
||||
assert_eq!(ingesting.2.as_deref(), Some("queue_depth=2"));
|
||||
}
|
||||
}
|
||||
|
||||
@@ -424,7 +424,15 @@ fn empty_path_allowed() {
|
||||
|
||||
#[test]
|
||||
fn dotfile_in_workspace_allowed() {
|
||||
let p = default_policy();
|
||||
let workspace = tempfile::tempdir().expect("workspace tempdir");
|
||||
std::fs::write(workspace.path().join(".gitignore"), "target/\n").expect("write .gitignore");
|
||||
std::fs::write(workspace.path().join(".env"), "LOCAL_ONLY=1\n").expect("write .env");
|
||||
let p = SecurityPolicy {
|
||||
workspace_dir: workspace.path().to_path_buf(),
|
||||
workspace_only: true,
|
||||
forbidden_paths: vec![],
|
||||
..SecurityPolicy::default()
|
||||
};
|
||||
assert!(p.is_path_string_allowed(".gitignore"));
|
||||
assert!(p.is_path_string_allowed(".env"));
|
||||
}
|
||||
|
||||
@@ -0,0 +1,111 @@
|
||||
//! Integration coverage for on-disk memory artifacts.
|
||||
//!
|
||||
//! Uses a real Slack ingest to create raw/source artifacts, then stages a
|
||||
//! mocked summary record to verify the Obsidian-compatible vault layout and
|
||||
//! frontmatter contract without requiring a live summarizer model.
|
||||
|
||||
use tempfile::tempdir;
|
||||
|
||||
use chrono::{TimeZone, Utc};
|
||||
|
||||
use openhuman_core::openhuman::config::Config;
|
||||
use openhuman_core::openhuman::memory_queue::drain_until_idle;
|
||||
use openhuman_core::openhuman::memory_store::content::atomic::stage_summary;
|
||||
use openhuman_core::openhuman::memory_store::content::obsidian::ensure_obsidian_defaults;
|
||||
use openhuman_core::openhuman::memory_store::content::{SummaryComposeInput, SummaryTreeKind};
|
||||
use openhuman_core::openhuman::memory_sync::composio::providers::slack::ingest::ingest_page_into_memory_tree;
|
||||
use openhuman_core::openhuman::memory_sync::composio::providers::slack::SlackMessage;
|
||||
|
||||
fn make_config(workspace_dir: &std::path::Path) -> Config {
|
||||
let mut config = Config::default();
|
||||
config.workspace_dir = workspace_dir.to_path_buf();
|
||||
config
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn sync_raw_artifacts_and_mocked_summary_match_obsidian_contract() {
|
||||
let tmp = tempdir().expect("tempdir");
|
||||
let workspace_dir = tmp.path().join("workspace");
|
||||
std::fs::create_dir_all(&workspace_dir).expect("workspace dir");
|
||||
let config = make_config(&workspace_dir);
|
||||
|
||||
let ts = Utc.timestamp_opt(1_700_000_000, 0).single().unwrap();
|
||||
let msg = SlackMessage {
|
||||
channel_id: "C123".into(),
|
||||
channel_name: "engineering".into(),
|
||||
is_private: false,
|
||||
author: "alice".into(),
|
||||
author_id: "U123".into(),
|
||||
text: "Phoenix migration launch window is Friday at 22:00 UTC.".into(),
|
||||
timestamp: ts,
|
||||
ts_raw: "1700000000.000100".into(),
|
||||
thread_ts: None,
|
||||
permalink: Some("https://slack.example.test/archives/C123/p1700000000000100".into()),
|
||||
};
|
||||
ingest_page_into_memory_tree(&config, "alice", "conn-slack-1", &[msg])
|
||||
.await
|
||||
.expect("seed slack sync");
|
||||
drain_until_idle(&config).await.expect("drain sync jobs");
|
||||
|
||||
let content_root = config.memory_tree_content_root();
|
||||
let raw_file = content_root
|
||||
.join("raw")
|
||||
.join("slack-conn-slack-1")
|
||||
.join("chats")
|
||||
.join("1700000000000_1700000000.000100.md");
|
||||
let source_file = content_root
|
||||
.join("raw")
|
||||
.join("slack-conn-slack-1")
|
||||
.join("_source.md");
|
||||
assert!(raw_file.exists(), "expected raw markdown artifact");
|
||||
assert!(source_file.exists(), "expected source registry mirror");
|
||||
|
||||
let raw_body = std::fs::read_to_string(&raw_file).expect("read raw artifact");
|
||||
assert!(raw_body.contains("**Channel:** #engineering"));
|
||||
assert!(raw_body.contains("**Author:** alice"));
|
||||
assert!(raw_body.contains("Phoenix migration launch window"));
|
||||
|
||||
ensure_obsidian_defaults(&content_root).expect("stage obsidian defaults");
|
||||
|
||||
let sealed_at = Utc.with_ymd_and_hms(2026, 5, 24, 22, 0, 0).unwrap();
|
||||
let child_ids = vec!["chunk-1".to_string()];
|
||||
let child_basenames = vec![Some("1700000000000_1700000000.000100".to_string())];
|
||||
let staged = stage_summary(
|
||||
&content_root,
|
||||
&SummaryComposeInput {
|
||||
summary_id: "summary:1760000000000:L1-phoenix-window",
|
||||
tree_kind: SummaryTreeKind::Source,
|
||||
tree_id: "source:slack",
|
||||
tree_scope: "slack:conn-slack-1",
|
||||
level: 1,
|
||||
child_ids: &child_ids,
|
||||
child_basenames: Some(&child_basenames),
|
||||
child_count: 1,
|
||||
time_range_start: ts,
|
||||
time_range_end: ts,
|
||||
sealed_at,
|
||||
body: "Phoenix migration launch window confirmed for Friday 22:00 UTC.",
|
||||
},
|
||||
"slack-conn-slack-1",
|
||||
None,
|
||||
)
|
||||
.expect("stage mocked summary");
|
||||
|
||||
let summary_path = content_root.join(&staged.content_path);
|
||||
assert!(summary_path.exists(), "summary markdown should be written");
|
||||
let summary_body = std::fs::read_to_string(&summary_path).expect("read summary markdown");
|
||||
assert!(summary_body.contains("tree_kind: source"));
|
||||
assert!(summary_body.contains("tree_scope: \"slack:conn-slack-1\""));
|
||||
assert!(summary_body.contains("time_range_start: 2023-11-14T22:13:20+00:00"));
|
||||
assert!(summary_body.contains("time_range_end: 2023-11-14T22:13:20+00:00"));
|
||||
assert!(summary_body.contains("sealed_at: 2026-05-24T22:00:00+00:00"));
|
||||
assert!(summary_body.contains("[[1700000000000_1700000000.000100]]"));
|
||||
|
||||
let graph_json = content_root.join(".obsidian").join("graph.json");
|
||||
let types_json = content_root.join(".obsidian").join("types.json");
|
||||
assert!(graph_json.exists(), "graph defaults should exist");
|
||||
assert!(types_json.exists(), "type hints should exist");
|
||||
let types_body = std::fs::read_to_string(types_json).expect("read types.json");
|
||||
assert!(types_body.contains("\"time_range_start\": \"date\""));
|
||||
assert!(types_body.contains("\"sealed_at\": \"datetime\""));
|
||||
}
|
||||
@@ -0,0 +1,168 @@
|
||||
//! End-to-end coverage for vault sync lifecycle.
|
||||
//!
|
||||
//! Runs the public `vault.*` operations against a real temp workspace:
|
||||
//! create a vault, sync supported files, verify per-file ledger + memory
|
||||
//! metadata, then modify/delete/add files and sync again.
|
||||
|
||||
use std::path::Path;
|
||||
use std::sync::{Mutex, OnceLock};
|
||||
use std::time::Duration;
|
||||
|
||||
use tempfile::tempdir;
|
||||
|
||||
use openhuman_core::openhuman::config::Config;
|
||||
use openhuman_core::openhuman::memory::global as memory_global;
|
||||
use openhuman_core::openhuman::vault::ops;
|
||||
use openhuman_core::openhuman::vault::VaultSyncStatus;
|
||||
|
||||
static TEST_LOCK: OnceLock<Mutex<()>> = OnceLock::new();
|
||||
|
||||
fn test_lock() -> std::sync::MutexGuard<'static, ()> {
|
||||
TEST_LOCK
|
||||
.get_or_init(|| Mutex::new(()))
|
||||
.lock()
|
||||
.expect("test lock poisoned")
|
||||
}
|
||||
|
||||
fn make_config(workspace_dir: &Path) -> Config {
|
||||
let mut config = Config::default();
|
||||
config.workspace_dir = workspace_dir.to_path_buf();
|
||||
config
|
||||
}
|
||||
|
||||
async fn wait_for_sync(vault_id: &str) -> openhuman_core::openhuman::vault::VaultSyncState {
|
||||
for _ in 0..100 {
|
||||
let state = ops::vault_sync_status(vault_id)
|
||||
.await
|
||||
.expect("vault_sync_status")
|
||||
.value;
|
||||
if state.status != VaultSyncStatus::Running {
|
||||
return state;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(50)).await;
|
||||
}
|
||||
panic!("vault sync did not finish within polling window");
|
||||
}
|
||||
|
||||
fn documents_from_payload(payload: &serde_json::Value) -> Vec<serde_json::Value> {
|
||||
payload
|
||||
.get("documents")
|
||||
.and_then(serde_json::Value::as_array)
|
||||
.cloned()
|
||||
.unwrap_or_default()
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn vault_sync_roundtrip_updates_memory_and_ledger() {
|
||||
let _guard = test_lock();
|
||||
let tmp = tempdir().expect("tempdir");
|
||||
let workspace_dir = tmp.path().join("workspace");
|
||||
let vault_root = tmp.path().join("vault-root");
|
||||
std::fs::create_dir_all(&workspace_dir).expect("workspace dir");
|
||||
std::fs::create_dir_all(vault_root.join("notes")).expect("notes dir");
|
||||
std::fs::create_dir_all(vault_root.join("docs")).expect("docs dir");
|
||||
std::fs::create_dir_all(vault_root.join("node_modules")).expect("excluded dir");
|
||||
|
||||
std::fs::write(
|
||||
vault_root.join("notes").join("one.md"),
|
||||
"# One\n\nPhoenix migration checklist.\n",
|
||||
)
|
||||
.expect("write one.md");
|
||||
std::fs::write(
|
||||
vault_root.join("docs").join("two.json"),
|
||||
"{\"status\":\"green\",\"owner\":\"alice\"}\n",
|
||||
)
|
||||
.expect("write two.json");
|
||||
std::fs::write(vault_root.join("image.png"), b"not a real png").expect("write image.png");
|
||||
std::fs::write(
|
||||
vault_root.join("node_modules").join("skip.md"),
|
||||
"should be excluded",
|
||||
)
|
||||
.expect("write excluded file");
|
||||
|
||||
memory_global::init(workspace_dir.clone()).expect("init global memory client");
|
||||
let config = make_config(&workspace_dir);
|
||||
|
||||
let vault = ops::vault_create(
|
||||
&config,
|
||||
"Project Vault",
|
||||
vault_root.to_str().expect("vault root utf-8"),
|
||||
vec![],
|
||||
vec![],
|
||||
)
|
||||
.await
|
||||
.expect("vault_create")
|
||||
.value;
|
||||
|
||||
ops::vault_sync(&config, &vault.id)
|
||||
.await
|
||||
.expect("vault_sync first");
|
||||
let first = wait_for_sync(&vault.id).await;
|
||||
assert_eq!(first.status, VaultSyncStatus::Completed);
|
||||
assert_eq!(first.ingested, 2);
|
||||
assert_eq!(first.removed, 0);
|
||||
assert_eq!(first.failed, 0);
|
||||
assert_eq!(first.skipped_unsupported, 1);
|
||||
assert_eq!(first.scanned, 3);
|
||||
|
||||
let files = ops::vault_files(&config, &vault.id)
|
||||
.await
|
||||
.expect("vault_files after first sync")
|
||||
.value;
|
||||
assert_eq!(files.len(), 2);
|
||||
assert!(files.iter().any(|file| file.rel_path == "notes/one.md"));
|
||||
assert!(files.iter().any(|file| file.rel_path == "docs/two.json"));
|
||||
|
||||
let docs = memory_global::client()
|
||||
.expect("global memory client")
|
||||
.list_documents(Some(&vault.namespace))
|
||||
.await
|
||||
.expect("list vault documents");
|
||||
let docs = documents_from_payload(&docs);
|
||||
assert_eq!(docs.len(), 2);
|
||||
assert!(docs.iter().any(|doc| {
|
||||
doc.get("key").and_then(serde_json::Value::as_str) == Some("notes/one.md")
|
||||
&& doc.get("sourceType").and_then(serde_json::Value::as_str) == Some("vault")
|
||||
}));
|
||||
assert!(docs.iter().any(|doc| {
|
||||
doc.get("key").and_then(serde_json::Value::as_str) == Some("docs/two.json")
|
||||
}));
|
||||
|
||||
let note_ledger = files
|
||||
.iter()
|
||||
.find(|file| file.rel_path == "notes/one.md")
|
||||
.expect("note ledger entry");
|
||||
assert!(note_ledger.bytes > 0);
|
||||
assert_eq!(note_ledger.vault_id, vault.id);
|
||||
|
||||
std::fs::write(
|
||||
vault_root.join("notes").join("one.md"),
|
||||
"# One\n\nPhoenix migration checklist updated with rollback steps.\n",
|
||||
)
|
||||
.expect("rewrite one.md");
|
||||
std::fs::remove_file(vault_root.join("docs").join("two.json")).expect("remove two.json");
|
||||
std::fs::write(
|
||||
vault_root.join("docs").join("three.toml"),
|
||||
"status = \"ready\"\nowner = \"bob\"\n",
|
||||
)
|
||||
.expect("write three.toml");
|
||||
|
||||
ops::vault_sync(&config, &vault.id)
|
||||
.await
|
||||
.expect("vault_sync second");
|
||||
let second = wait_for_sync(&vault.id).await;
|
||||
assert_eq!(second.status, VaultSyncStatus::Completed);
|
||||
assert_eq!(second.ingested, 2);
|
||||
assert_eq!(second.removed, 1);
|
||||
assert_eq!(second.failed, 0);
|
||||
assert_eq!(second.skipped_unsupported, 1);
|
||||
|
||||
let files = ops::vault_files(&config, &vault.id)
|
||||
.await
|
||||
.expect("vault_files after second sync")
|
||||
.value;
|
||||
assert_eq!(files.len(), 2);
|
||||
assert!(files.iter().any(|file| file.rel_path == "notes/one.md"));
|
||||
assert!(files.iter().any(|file| file.rel_path == "docs/three.toml"));
|
||||
assert!(!files.iter().any(|file| file.rel_path == "docs/two.json"));
|
||||
}
|
||||
Reference in New Issue
Block a user