#![forbid(unsafe_code)]
use std::collections::BTreeMap;
use std::error::Error;
use std::num::{NonZeroU32, NonZeroUsize};
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use haematite_cli::serve::protocol::{
Change, CollectionWrite, DocumentWrite, Fork, Metric, Mutation, Operation, Selection,
VectorDelete, VectorPage, VectorRead, VectorSearch, VectorWrite, Version,
};
use haematite_cli::serve::service::Service;
use haematite_cli::serve::storage::{Receipt, Storage, reads};
use serde_json::{Value, json, value::RawValue};
type TestResult = Result<(), Box<dyn Error>>;
struct Fixture {
service: Service,
directory: tempfile::TempDir,
versions: BTreeMap<String, Version>,
requests: u64,
}
impl Fixture {
fn new() -> Result<Self, Box<dyn Error>> {
let directory = tempfile::tempdir()?;
let storage = Storage::create(
DatabaseConfig {
data_dir: directory.path().join("database"),
shard_count: 2,
distributed: None,
executor_threads: Some(2),
node_cache_budget: Some(NodeCacheBudget::Unlimited),
retention: haematite::Retention::KeepAll,
},
NodeCacheBudget::Unlimited,
long_wait_interval(),
)?;
let version = storage.capture(&selection("main", None))?.version;
let mut fixture = Self {
directory,
service: Service::new(storage, NonZeroU32::new(8).ok_or("zero page limit")?)?,
versions: BTreeMap::from([("main".to_owned(), version)]),
requests: 0,
};
fixture.apply(
"main",
Change::Collection(CollectionWrite {
collection: "records".to_owned(),
schema: None,
}),
)?;
Ok(fixture)
}
fn version(&self, branch: &str) -> Result<Version, Box<dyn Error>> {
self.versions
.get(branch)
.copied()
.ok_or_else(|| "missing fixture branch".into())
}
fn apply(&mut self, branch: &str, change: Change) -> Result<Receipt, Box<dyn Error>> {
self.requests += 1;
let reply = self.service.dispatch(Operation::Mutate(Mutation {
request_id: format!("request-{}", self.requests),
branch: branch.to_owned(),
expected: self.version(branch)?,
change,
}))?;
let receipt: Receipt = serde_json::from_value(reply)?;
self.versions
.insert(receipt.branch.clone(), receipt.version);
Ok(receipt)
}
fn document(&mut self, branch: &str, id: &str) -> TestResult {
self.apply(
branch,
Change::Put(DocumentWrite {
collection: "records".to_owned(),
document_id: id.to_owned(),
document: RawValue::from_string("{}".to_owned())?,
}),
)?;
Ok(())
}
fn vector(&mut self, branch: &str, id: &str, values: Vec<f32>) -> TestResult {
self.apply(branch, vector(id, values))?;
Ok(())
}
fn get(&self, branch: &str, id: &str) -> Result<Value, Box<dyn Error>> {
Ok(self.service.dispatch(Operation::VectorGet(VectorRead {
selection: selection(branch, None),
collection: "records".to_owned(),
document_id: id.to_owned(),
field: "embedding".to_owned(),
}))?)
}
fn refuse(&mut self, change: Change, code: &str) -> TestResult {
let before = self.version("main")?;
let error = self
.apply("main", change)
.err()
.ok_or("invalid vector accepted")?;
let failure = error
.downcast_ref::<haematite_cli::serve::protocol::ResponseFailure>()
.ok_or("unexpected error type")?;
assert_eq!(failure.code, code);
assert_eq!(self.version("main")?, before);
let branches = self.service.dispatch(Operation::Branches(
haematite_cli::serve::protocol::BranchPage {
after: None,
limit: NonZeroU32::MIN,
},
))?;
assert_eq!(
serde_json::from_value::<Version>(branches["entries"][0]["version"].clone())?,
before
);
Ok(())
}
}
fn selection(branch: &str, version: Option<Version>) -> Selection {
Selection {
branch: branch.to_owned(),
version,
}
}
fn vector(id: &str, values: Vec<f32>) -> Change {
Change::VectorPut(VectorWrite {
collection: "records".to_owned(),
document_id: id.to_owned(),
field: "embedding".to_owned(),
vector: values,
})
}
fn page(version: Option<Version>, after: Option<&str>, limit: NonZeroU32) -> Operation {
Operation::VectorList(VectorPage {
selection: selection("main", version),
collection: "records".to_owned(),
field: "embedding".to_owned(),
after: after.map(str::to_owned),
limit,
})
}
fn search(query: Vec<f32>, metric: Metric, k: NonZeroU32) -> Operation {
Operation::VectorSearch(VectorSearch {
selection: selection("main", None),
collection: "records".to_owned(),
field: "embedding".to_owned(),
query,
metric,
k,
})
}
#[test]
fn vectors_put_get_delete_and_restart_preserve_members_and_dimension() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.document("main", "a")?;
assert!(fixture.get("main", "a")?["vector"].is_null());
let written = fixture
.apply("main", vector("a", vec![1.5, -2.25, 0.0]))?
.version;
assert_eq!(
fixture.get("main", "a")?["vector"],
json!([1.5, -2.25, 0.0])
);
let data_dir = fixture.directory.path().join("database");
let Fixture {
directory,
service,
versions,
requests,
} = fixture;
drop(service);
let storage = Storage::open(&data_dir, NodeCacheBudget::Unlimited, long_wait_interval())?;
let at = storage.capture(&selection("main", Some(written)))?;
let reader = storage.reader()?;
assert_eq!(
reads::get_bytes(&reader, &at, b"w\0records\0embedding")?,
Some(b"3".to_vec())
);
assert_eq!(
reads::get_bytes(&reader, &at, b"v\0records\0embedding\0a")?,
Some(vec![0, 0, 192, 63, 0, 0, 16, 192, 0, 0, 0, 0])
);
drop(reader);
let mut fixture = Fixture {
directory,
service: Service::new(storage, NonZeroU32::new(8).ok_or("zero page limit")?)?,
versions,
requests,
};
assert_eq!(
fixture.get("main", "a")?["vector"],
json!([1.5, -2.25, 0.0])
);
fixture.apply(
"main",
Change::VectorDelete(VectorDelete {
collection: "records".to_owned(),
document_id: "a".to_owned(),
field: "embedding".to_owned(),
}),
)?;
assert!(fixture.get("main", "a")?["vector"].is_null());
let historical = fixture.service.dispatch(Operation::VectorGet(VectorRead {
selection: selection("main", Some(written)),
collection: "records".to_owned(),
document_id: "a".to_owned(),
field: "embedding".to_owned(),
}))?;
assert_eq!(historical["vector"], json!([1.5, -2.25, 0.0]));
fixture.refuse(vector("a", vec![1.0]), "vector_dimension")?;
Ok(())
}
#[test]
fn vector_mutations_refuse_empty_non_finite_missing_document_and_wrong_dimension() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.document("main", "a")?;
fixture.refuse(vector("a", Vec::new()), "vector_empty")?;
for member in [f32::NAN, f32::INFINITY, f32::NEG_INFINITY] {
fixture.refuse(vector("a", vec![member]), "vector_non_finite")?;
}
fixture.refuse(vector("missing", vec![1.0]), "document_missing")?;
fixture.vector("main", "a", vec![1.0, 2.0])?;
fixture.refuse(vector("a", vec![3.0]), "vector_dimension")?;
assert_eq!(fixture.get("main", "a")?["vector"], json!([1.0, 2.0]));
for (collection, field, document_id) in [
("", "embedding", "a"),
("records", "bad\0field", "a"),
("records", "embedding", ""),
] {
fixture.refuse(
Change::VectorPut(VectorWrite {
collection: collection.to_owned(),
field: field.to_owned(),
document_id: document_id.to_owned(),
vector: vec![1.0, 2.0],
}),
"invalid_identifier",
)?;
fixture.refuse(
Change::VectorDelete(VectorDelete {
collection: collection.to_owned(),
field: field.to_owned(),
document_id: document_id.to_owned(),
}),
"invalid_identifier",
)?;
}
Ok(())
}
#[test]
fn vector_pages_require_and_keep_the_captured_version() -> TestResult {
let mut fixture = Fixture::new()?;
for id in ["a", "b", "bb", "c"] {
fixture.document("main", id)?;
}
for id in ["c", "a", "b"] {
fixture.vector("main", id, vec![1.0, 2.0])?;
}
let first = fixture
.service
.dispatch(page(None, None, NonZeroU32::MIN))?;
assert_eq!(
first["vectors"],
json!([{"document_id":"a", "vector":[1.0,2.0]}])
);
assert_eq!(first["next_after"], "a");
let captured: Version = serde_json::from_value(first["selection"]["version"].clone())?;
fixture.vector("main", "bb", vec![3.0, 4.0])?;
let refused = fixture
.service
.dispatch(page(None, Some("a"), NonZeroU32::MIN))
.err()
.ok_or("uncaptured continuation accepted")?;
assert_eq!(refused.code, "version_required");
let second = fixture
.service
.dispatch(page(Some(captured), Some("a"), NonZeroU32::MIN))?;
assert_eq!(second["vectors"][0]["document_id"], "b");
assert_eq!(second["next_after"], "b");
let last = fixture
.service
.dispatch(page(Some(captured), Some("b"), NonZeroU32::MIN))?;
assert_eq!(last["vectors"][0]["document_id"], "c");
assert!(last["next_after"].is_null());
assert_eq!(last["selection"]["version"], first["selection"]["version"]);
let fresh =
fixture
.service
.dispatch(page(None, None, NonZeroU32::new(8).ok_or("zero page")?))?;
assert_eq!(
fresh["vectors"].as_array().ok_or("vectors absent")?.len(),
4
);
let too_large = fixture
.service
.dispatch(page(None, None, NonZeroU32::new(9).ok_or("zero page")?))
.err()
.ok_or("oversize page accepted")?;
assert_eq!(too_large.code, "page_limit");
Ok(())
}
#[test]
fn exact_metrics_score_every_vector_and_break_ties_by_document_id() -> TestResult {
let mut fixture = Fixture::new()?;
for (id, values) in [
("c", vec![3.0, 0.0]),
("b", vec![0.0, 4.0]),
("a", vec![3.0, 0.0]),
("d", vec![-3.0, 0.0]),
("e", vec![0.0, 0.0]),
] {
fixture.document("main", id)?;
fixture.vector("main", id, values)?;
}
let limit = NonZeroU32::new(5).ok_or("zero search limit")?;
for (metric, expected) in [
(
Metric::Dot,
vec![("a", 3.0), ("c", 3.0), ("b", 0.0), ("e", 0.0), ("d", -3.0)],
),
(
Metric::Cosine,
vec![("a", 1.0), ("c", 1.0), ("b", 0.0), ("e", 0.0), ("d", -1.0)],
),
(
Metric::L2,
vec![
("e", 1.0),
("a", 2.0),
("c", 2.0),
("d", 4.0),
("b", 17.0_f64.sqrt()),
],
),
] {
let result = fixture
.service
.dispatch(search(vec![1.0, 0.0], metric, limit))?;
assert_eq!(result["scanned"], 5);
let hits = result["hits"].as_array().ok_or("hits absent")?;
assert_eq!(hits.len(), expected.len());
for (hit, (id, expected_score)) in hits.iter().zip(expected) {
assert_eq!(hit["document_id"], id);
let score = hit["score"].as_f64().ok_or("score absent")?;
assert!((score - expected_score).abs() < 1e-12);
}
}
let first = fixture
.service
.dispatch(search(vec![1.0, 0.0], Metric::Dot, NonZeroU32::MIN))?;
assert_eq!(first["scanned"], 5);
assert_eq!(first["hits"], json!([{"document_id":"a", "score":3.0}]));
let zero = fixture
.service
.dispatch(search(vec![0.0, 0.0], Metric::Cosine, limit))?;
assert_eq!(
zero["hits"]
.as_array()
.ok_or("hits absent")?
.iter()
.map(|hit| hit["document_id"].as_str().ok_or("id absent"))
.collect::<Result<Vec<_>, _>>()?,
vec!["a", "b", "c", "d", "e"]
);
for hit in zero["hits"].as_array().ok_or("hits absent")? {
assert!(hit["score"].as_f64().ok_or("score absent")?.abs() < f64::EPSILON);
}
Ok(())
}
#[test]
fn vector_search_refuses_bad_query_or_limit_and_empty_field_reports_zero() -> TestResult {
let mut fixture = Fixture::new()?;
let empty = fixture
.service
.dispatch(search(vec![1.0, 2.0], Metric::Dot, NonZeroU32::MIN))?;
assert_eq!(empty["hits"], json!([]));
assert_eq!(empty["scanned"], 0);
fixture.document("main", "a")?;
fixture.vector("main", "a", vec![1.0, 2.0])?;
for (query, k, code) in [
(Vec::new(), NonZeroU32::MIN, "vector_empty"),
(vec![1.0], NonZeroU32::MIN, "vector_dimension"),
(vec![f32::NAN, 0.0], NonZeroU32::MIN, "vector_non_finite"),
(
vec![f32::INFINITY, 0.0],
NonZeroU32::MIN,
"vector_non_finite",
),
(
vec![f32::NEG_INFINITY, 0.0],
NonZeroU32::MIN,
"vector_non_finite",
),
(
vec![1.0, 2.0],
NonZeroU32::new(9).ok_or("zero limit")?,
"page_limit",
),
] {
let refused = fixture
.service
.dispatch(search(query, Metric::Dot, k))
.err()
.ok_or("invalid search accepted")?;
assert_eq!(refused.code, code);
}
Ok(())
}
#[test]
fn vectors_on_a_fork_are_invisible_on_the_source_branch() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.document("main", "a")?;
let source = fixture.version("main")?;
fixture.apply(
"main",
Change::Fork(Fork {
name: "experiment".to_owned(),
from: source,
}),
)?;
fixture.vector("experiment", "a", vec![5.0, 6.0])?;
assert!(fixture.get("main", "a")?["vector"].is_null());
assert_eq!(fixture.get("experiment", "a")?["vector"], json!([5.0, 6.0]));
fixture.vector("main", "a", vec![7.0])?;
assert_eq!(fixture.get("main", "a")?["vector"], json!([7.0]));
assert_eq!(fixture.get("experiment", "a")?["vector"], json!([5.0, 6.0]));
Ok(())
}
#[test]
fn vector_pages_refuse_a_first_entry_above_the_response_byte_budget() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.document("main", "a")?;
fixture.vector("main", "a", vec![1.0; 128])?;
let refused = fixture
.service
.dispatch_bounded(
page(None, None, NonZeroU32::MIN),
NonZeroUsize::new(64).ok_or("zero byte limit")?,
)
.err()
.ok_or("oversize vector entry accepted")?;
assert_eq!(refused.code, "response_limit");
let reply = fixture
.service
.dispatch(page(None, None, NonZeroU32::MIN))?;
assert_eq!(
reply["vectors"][0]["vector"]
.as_array()
.ok_or("vector absent")?
.len(),
128
);
Ok(())
}
fn declaration() -> Value {
json!({
"kind": "dense", "source": "body",
"model": {"name": "local-model", "revision": "v1", "manifest_sha256": "a".repeat(64)},
"dimensions": 2, "distance": "cosine"
})
}
fn declare_change(declaration: Value) -> Result<Change, serde_json::Error> {
let value = serde_json::Map::from_iter([
("collection".to_owned(), json!("records")),
("field".to_owned(), json!("embedding")),
("declaration".to_owned(), declaration),
]);
serde_json::from_value(Value::Object(serde_json::Map::from_iter([
("kind".to_owned(), json!("vector.declare")),
("value".to_owned(), Value::Object(value)),
])))
}
fn field_listing(fixture: &Fixture) -> Result<Value, Box<dyn Error>> {
Ok(fixture.service.dispatch(Operation::VectorFields(
haematite_cli::serve::protocol::StoreRead {
selection: selection("main", None),
},
))?)
}
fn refuse_declaration_change(fixture: &mut Fixture, change: Change, code: &str) -> TestResult {
let before = field_listing(fixture)?;
fixture.requests += 1;
let failure = fixture
.service
.dispatch(Operation::Mutate(Mutation {
request_id: format!("request-{}", fixture.requests),
branch: "main".to_owned(),
expected: fixture.version("main")?,
change,
}))
.err()
.ok_or("invalid declaration mutation accepted")?;
let answer = serde_json::to_value(failure)?;
assert_eq!(answer["code"], code);
assert_eq!(answer["outcome"], "not_applied");
assert_eq!(field_listing(fixture)?, before);
Ok(())
}
#[test]
fn declared_field_listing_keeps_supplied_reader_and_has_no_false_count() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.document("main", "a")?;
fixture.apply(
"main",
Change::VectorPut(VectorWrite {
collection: "records".to_owned(),
field: "supplied".to_owned(),
document_id: "a".to_owned(),
vector: vec![1.0, 2.0],
}),
)?;
fixture.apply("main", declare_change(declaration())?)?;
let answer = field_listing(&fixture)?;
assert_eq!(
answer["fields"],
json!([{
"collection":"records", "field":"supplied", "dimensions":2, "count":1
}])
);
let old_fields: Vec<Value> = serde_json::from_value(answer["fields"].clone())?;
for field in old_fields {
let collection: String = serde_json::from_value(field["collection"].clone())?;
let field_name: String = serde_json::from_value(field["field"].clone())?;
let dimensions: usize = serde_json::from_value(field["dimensions"].clone())?;
let count: u64 = serde_json::from_value(field["count"].clone())?;
assert_eq!(
(collection.as_str(), field_name.as_str(), dimensions, count),
("records", "supplied", 2, 1)
);
}
let mut embedded = declaration();
embedded["collection"] = json!("records");
embedded["field"] = json!("embedding");
embedded["status"] = json!("embed_not_configured");
assert_eq!(answer["embedded_fields"], json!([embedded]));
assert!(answer["embedded_fields"][0].get("count").is_none());
Ok(())
}
#[test]
fn declared_field_allows_record_delete_and_preserves_its_declaration() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.document("main", "a")?;
fixture.apply("main", declare_change(declaration())?)?;
let before = field_listing(&fixture)?;
fixture.apply(
"main",
Change::Delete(haematite_cli::serve::protocol::DocumentDelete {
collection: "records".to_owned(),
document_id: "a".to_owned(),
}),
)?;
let document = fixture.service.dispatch(Operation::Get(
haematite_cli::serve::protocol::DocumentRead {
selection: selection("main", None),
collection: "records".to_owned(),
document_id: "a".to_owned(),
},
))?;
assert!(document["document"].is_null());
assert_eq!(
field_listing(&fixture)?["embedded_fields"],
before["embedded_fields"]
);
Ok(())
}
#[test]
fn declared_field_refuses_handmade_vector_without_a_write() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.document("main", "a")?;
fixture.apply("main", declare_change(declaration())?)?;
refuse_declaration_change(
&mut fixture,
vector("a", vec![1.0, 2.0]),
"vector_field_embedded",
)?;
Ok(())
}
#[test]
fn supplied_field_even_when_emptied_cannot_be_declared_embedded() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.document("main", "a")?;
fixture.vector("main", "a", vec![1.0, 2.0])?;
refuse_declaration_change(
&mut fixture,
declare_change(declaration())?,
"vector_field_supplied",
)?;
fixture.apply(
"main",
Change::VectorDelete(VectorDelete {
collection: "records".to_owned(),
field: "embedding".to_owned(),
document_id: "a".to_owned(),
}),
)?;
refuse_declaration_change(
&mut fixture,
declare_change(declaration())?,
"vector_field_supplied",
)?;
Ok(())
}
#[test]
fn declared_field_refuses_every_different_declaration() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.apply("main", declare_change(declaration())?)?;
let mut replacements = Vec::new();
for (member, value) in [
("name", "other"),
("revision", "v2"),
("manifest_sha256", &"b".repeat(64)),
] {
let mut replacement = declaration();
replacement["model"][member] = json!(value);
replacements.push(replacement);
}
for (member, value) in [
("source", json!("title")),
("dimensions", json!(3)),
("distance", json!("dot")),
] {
let mut replacement = declaration();
replacement[member] = value;
replacements.push(replacement);
}
for replacement in replacements {
refuse_declaration_change(
&mut fixture,
declare_change(replacement)?,
"embed_model_change_not_available",
)?;
}
Ok(())
}
#[test]
fn declare_replay_keeps_original_receipt_and_new_request_matches_collection_put() -> TestResult {
let mut fixture = Fixture::new()?;
let expected = fixture.version("main")?;
let request = || -> Result<Mutation, Box<dyn Error>> {
Ok(Mutation {
request_id: "declaration-replay".to_owned(),
branch: "main".to_owned(),
expected,
change: declare_change(declaration())?,
})
};
let first = fixture.service.dispatch(Operation::Mutate(request()?))?;
let receipt: Receipt = serde_json::from_value(first.clone())?;
fixture.versions.insert("main".to_owned(), receipt.version);
fixture.document("main", "a")?;
let before = field_listing(&fixture)?;
assert_eq!(
fixture.service.dispatch(Operation::Mutate(request()?))?,
first
);
assert_eq!(field_listing(&fixture)?, before);
let declared = fixture.apply("main", declare_change(declaration())?)?;
assert_eq!(declared.version.sequence, fixture.version("main")?.sequence);
let collection = fixture.apply(
"main",
Change::Collection(CollectionWrite {
collection: "records".to_owned(),
schema: None,
}),
)?;
let repeated_collection = fixture.apply(
"main",
Change::Collection(CollectionWrite {
collection: "records".to_owned(),
schema: None,
}),
)?;
let before_version: Version = serde_json::from_value(before["selection"]["version"].clone())?;
assert_eq!(declared.version.sequence - before_version.sequence, 1);
assert_eq!(
repeated_collection.version.sequence - collection.version.sequence,
1
);
assert_eq!(
field_listing(&fixture)?["embedded_fields"],
before["embedded_fields"]
);
Ok(())
}
#[test]
fn declared_vector_get_list_and_search_refuse_unserved_companion() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.document("main", "a")?;
fixture.apply("main", declare_change(declaration())?)?;
for operation in [
Operation::VectorGet(VectorRead {
selection: selection("main", None),
collection: "records".to_owned(),
field: "embedding".to_owned(),
document_id: "a".to_owned(),
}),
page(None, None, NonZeroU32::MIN),
search(vec![1.0, 2.0], Metric::Cosine, NonZeroU32::MIN),
] {
let failure = fixture
.service
.dispatch(operation)
.err()
.ok_or("embedded read answered empty")?;
assert_eq!(failure.code, "embed_companion_not_served");
}
Ok(())
}
#[test]
fn declaration_boundary_refuses_invalid_names_manifest_and_missing_collection() -> TestResult {
let mut fixture = Fixture::new()?;
for (member, value) in [("source", json!("")), ("source", json!("bad\0field"))] {
let mut invalid = declaration();
invalid[member] = value;
refuse_declaration_change(&mut fixture, declare_change(invalid)?, "invalid_identifier")?;
}
for member in ["name", "revision"] {
let mut invalid = declaration();
invalid["model"][member] = json!("");
refuse_declaration_change(&mut fixture, declare_change(invalid)?, "invalid_identifier")?;
}
for hash in ["abc".to_owned(), "A".repeat(64), "g".repeat(64)] {
let mut invalid = declaration();
invalid["model"]["manifest_sha256"] = json!(hash);
refuse_declaration_change(
&mut fixture,
declare_change(invalid)?,
"embed_manifest_invalid",
)?;
}
for (collection, field, code) in [
("missing", "embedding", "collection_missing"),
("", "embedding", "invalid_identifier"),
("records", "bad\0field", "invalid_identifier"),
] {
let change = serde_json::from_value(json!({"kind":"vector.declare", "value": {
"collection":collection, "field":field, "declaration":declaration()
}}))?;
refuse_declaration_change(&mut fixture, change, code)?;
}
let mut zero = declaration();
zero["dimensions"] = json!(0);
assert!(declare_change(zero).is_err());
Ok(())
}
#[test]
fn declared_field_branch_review_renders_canonical_declaration() -> TestResult {
let mut fixture = Fixture::new()?;
fixture.apply(
"main",
Change::Fork(Fork {
name: "feature".to_owned(),
from: fixture.version("main")?,
}),
)?;
fixture.apply("feature", declare_change(declaration())?)?;
let mut lines = Vec::new();
let mut after = None;
loop {
let review = fixture.service.dispatch(Operation::Review(
haematite_cli::serve::protocol::ReviewPage {
branch: "feature".to_owned(),
after,
limit: NonZeroU32::new(8).ok_or("zero page")?,
},
))?;
lines.extend(
review["diff"]
.as_array()
.ok_or("no review diff")?
.iter()
.cloned(),
);
after = review["next_after"].as_str().map(str::to_owned);
if after.is_none() {
break;
}
}
let line = lines
.iter()
.find(|line| line["what"] == "vector field embedding")
.ok_or("no declaration review line")?;
assert_eq!(line["base"], "");
let rendered: Value =
serde_json::from_str(line["here"].as_str().ok_or("no declaration text")?)?;
assert_eq!(rendered, declaration());
Ok(())
}
const fn long_wait_interval() -> std::time::Duration {
const LONG_WAIT_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
LONG_WAIT_INTERVAL
}