use std::num::NonZeroU32;
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use serde_json::{Value, json, value::RawValue};
use super::Collection;
use crate::serve::protocol::{
BranchPage, Change, CollectionPage, CollectionWrite, DocumentImport, DocumentRead,
DocumentWrite, ImportedDocument, Mutation, Operation, ReceiptQuery, Selection, Version,
};
use crate::serve::service::Service;
use crate::serve::storage::{Receipt, Storage};
type TestResult = Result<(), Box<dyn std::error::Error>>;
const LIMIT: NonZeroU32 = NonZeroU32::MIN.saturating_add(9);
fn selection() -> Selection {
Selection {
branch: "main".to_owned(),
version: None,
}
}
fn with_service(test: impl FnOnce(&Service) -> TestResult) -> TestResult {
let directory = tempfile::tempdir()?;
let storage = Storage::create(
DatabaseConfig {
data_dir: directory.path().join("database"),
shard_count: 1,
distributed: None,
executor_threads: Some(1),
node_cache_budget: Some(NodeCacheBudget::Unlimited),
retention: haematite::Retention::KeepAll,
},
NodeCacheBudget::Unlimited,
fixture_long_wait_interval(),
)?;
let service = Service::new(storage, LIMIT)?;
test(&service)
}
fn head(service: &Service) -> Result<Version, Box<dyn std::error::Error>> {
let answer = service.dispatch(Operation::Branches(BranchPage {
after: None,
limit: LIMIT,
}))?;
let row = answer["entries"]
.as_array()
.ok_or("branches has no entries")?
.iter()
.find(|row| row["branch"] == "main")
.ok_or("main branch missing")?;
Ok(serde_json::from_value(row["version"].clone())?)
}
fn mutation(
service: &Service,
id: &str,
change: Change,
) -> Result<Mutation, Box<dyn std::error::Error>> {
Ok(Mutation {
request_id: id.to_owned(),
branch: "main".to_owned(),
expected: head(service)?,
change,
})
}
fn apply(
service: &Service,
id: &str,
change: Change,
) -> Result<Receipt, Box<dyn std::error::Error>> {
Ok(serde_json::from_value(service.dispatch(
Operation::Mutate(mutation(service, id, change)?),
)?)?)
}
fn shape() -> Value {
json!({
"type": "object",
"properties": {
"payload": {"type": "string", "contentMediaType": "application/json"}
},
"required": ["payload"]
})
}
fn declare(service: &Service, id: &str, schema: Option<Value>) -> TestResult {
apply(
service,
id,
Change::Collection(CollectionWrite {
collection: "records".to_owned(),
schema,
}),
)?;
Ok(())
}
fn put(id: &str, raw: &str) -> Result<Change, serde_json::Error> {
Ok(Change::Put(DocumentWrite {
collection: "records".to_owned(),
document_id: id.to_owned(),
document: RawValue::from_string(raw.to_owned())?,
}))
}
fn imported(id: &str, raw: &str) -> Result<ImportedDocument, serde_json::Error> {
Ok(ImportedDocument {
document_id: id.to_owned(),
document: RawValue::from_string(raw.to_owned())?,
})
}
fn get(service: &Service, id: &str) -> Result<Value, Box<dyn std::error::Error>> {
Ok(service.dispatch(Operation::Get(DocumentRead {
selection: selection(),
collection: "records".to_owned(),
document_id: id.to_owned(),
}))?)
}
fn receipt_absent(service: &Service, id: &str) -> TestResult {
assert!(
service
.dispatch(Operation::Receipt(ReceiptQuery {
branch: "main".to_owned(),
request_id: id.to_owned(),
}))?
.is_null()
);
Ok(())
}
#[test]
fn json_text_declaration_reads_back_in_schema_json() -> TestResult {
with_service(|service| {
let schema = shape();
declare(service, "declare", Some(schema.clone()))?;
let answer = service.dispatch(Operation::Collections(CollectionPage {
selection: selection(),
after: None,
limit: LIMIT,
}))?;
let entries = answer["entries"]
.as_array()
.ok_or("collections has no entries")?;
assert_eq!(entries.len(), 1);
let raw = entries[0]["schema_json"]
.as_str()
.ok_or("schema_json missing")?;
assert_eq!(serde_json::from_str::<Value>(raw)?, schema);
Ok(())
})
}
#[test]
fn valid_json_text_keeps_the_submitted_document_bytes() -> TestResult {
with_service(|service| {
declare(service, "declare", Some(shape()))?;
let raw = r#"{ "payload" : "{ \"items\" : [1,true,null] }", "untouched": [ 1, 2 ] }"#;
apply(service, "put", put("one", raw)?)?;
let answer = get(service, "one")?;
assert_eq!(answer["document"]["raw_json"].as_str(), Some(raw));
let stored: Value = serde_json::from_str(
answer["document"]["raw_json"]
.as_str()
.ok_or("raw missing")?,
)?;
assert!(stored["payload"].is_string());
Ok(())
})
}
#[test]
fn invalid_json_text_put_leaves_version_receipt_and_document_unchanged() -> TestResult {
with_service(|service| {
declare(service, "declare", Some(shape()))?;
apply(service, "seed", put("one", r#"{"payload":"null"}"#)?)?;
let before = head(service)?;
let document = get(service, "one")?;
receipt_absent(service, "bad")?;
let refused = service
.dispatch(Operation::Mutate(mutation(
service,
"bad",
put("one", r#"{"payload":"{not json"}"#)?,
)?))
.err()
.ok_or("invalid JSON text put was accepted")?;
assert_eq!(refused.code, "json_text_invalid");
assert_eq!(refused.path.as_deref(), Some("/payload"));
assert!(refused.message.contains("not JSON"));
assert_eq!(serde_json::to_value(&refused)?["outcome"], "not_applied");
assert_eq!(head(service)?, before);
assert_eq!(get(service, "one")?, document);
receipt_absent(service, "bad")
})
}
#[test]
fn an_invalid_import_row_refuses_every_row_without_a_receipt() -> TestResult {
with_service(|service| {
declare(service, "declare", Some(shape()))?;
apply(service, "seed", put("kept", r#"{"payload":"true"}"#)?)?;
let before = head(service)?;
let kept = get(service, "kept")?;
let change = Change::PutMany(DocumentImport {
collection: "records".to_owned(),
documents: vec![
imported("good", r#"{"payload":"[]"}"#)?,
imported("bad", r#"{"payload":"{not json"}"#)?,
],
});
let refused = service
.dispatch(Operation::Mutate(mutation(service, "import", change)?))
.err()
.ok_or("invalid JSON text import was accepted")?;
assert_eq!(refused.code, "json_text_invalid");
assert_eq!(
refused.path.as_deref(),
Some("/documents/1/document/payload")
);
assert!(refused.message.contains("not JSON"));
assert_eq!(serde_json::to_value(&refused)?["outcome"], "not_applied");
assert_eq!(head(service)?, before);
assert!(get(service, "good")?["document"].is_null());
assert!(get(service, "bad")?["document"].is_null());
assert_eq!(get(service, "kept")?, kept);
receipt_absent(service, "import")
})
}
#[test]
fn declaring_json_text_does_not_scan_or_change_an_existing_row() -> TestResult {
with_service(|service| {
declare(service, "untyped", None)?;
let raw = r#"{"payload":"{not json"}"#;
apply(service, "old", put("one", raw)?)?;
let before = get(service, "one")?["document"].clone();
declare(service, "typed", Some(shape()))?;
let after = get(service, "one")?;
assert_eq!(after["document"], before);
assert_eq!(after["document"]["raw_json"].as_str(), Some(raw));
Ok(())
})
}
#[test]
fn json_content_media_type_requires_exactly_type_string() -> TestResult {
for schema in [
json!({"type":"object", "contentMediaType":"application/json"}),
json!({"contentMediaType":"application/json"}),
json!({"type":["string"], "contentMediaType":"application/json"}),
] {
let refusal = Collection::compile(&schema)
.err()
.ok_or("invalid keyword location accepted")?;
assert_eq!(refusal.path, "/contentMediaType");
assert!(refusal.reason.contains("type string"));
}
Ok(())
}
#[test]
fn unsupported_media_type_is_named_and_content_schema_stays_refused() -> TestResult {
for schema in [
json!({"type":"string", "contentMediaType":"text/plain"}),
json!({"type":"string", "contentMediaType":true}),
json!({"type":"string", "contentSchema":{"type":"object"}}),
] {
let refusal = Collection::compile(&schema)
.err()
.ok_or("unsupported keyword value accepted")?;
assert!(refusal.reason.contains("unsupported"));
}
Ok(())
}
#[test]
fn json_text_accepts_each_json_kind_and_refuses_trailing_values() -> TestResult {
let schema =
Collection::compile(&json!({"type":"string","contentMediaType":"application/json"}))?;
for text in [
"null",
"true",
"18446744073709551616001",
r#""text""#,
"{}",
"[1]",
" \n [] \t",
] {
schema.validate(&json!(text))?;
}
for text in ["", "null true", "{} trailing", "{not json"] {
let refusal = schema
.validate(&json!(text))
.err()
.ok_or("invalid JSON text accepted")?;
assert!(refusal.reason.contains("not JSON"));
assert_eq!(refusal.path, "");
}
Ok(())
}
#[test]
fn nested_array_json_text_names_the_escaped_document_pointer() -> TestResult {
let schema = Collection::compile(&json!({
"type":"object",
"properties":{"a/b~c":{"type":"array","items":{"type":"string","contentMediaType":"application/json"}}}
}))?;
schema.validate(&json!({"a/b~c":["null","[]"]}))?;
let refusal = schema
.validate(&json!({"a/b~c":["null","{not json"]}))
.err()
.ok_or("invalid array member accepted")?;
assert_eq!(refusal.path, "/a~1b~0c/1");
assert!(refusal.reason.contains("not JSON"));
Ok(())
}
#[test]
fn additional_property_json_text_is_checked_without_rechecking_named_fields() -> TestResult {
let schema = Collection::compile(&json!({
"type":"object",
"properties":{"number":{"type":"integer"}},
"additionalProperties":{"type":"string","contentMediaType":"application/json"}
}))?;
schema.validate(&json!({"number":1,"extra":"true"}))?;
let refusal = schema
.validate(&json!({"number":1,"extra":"{not json"}))
.err()
.ok_or("invalid extra property accepted")?;
assert_eq!(refusal.path, "/extra");
assert!(refusal.reason.contains("not JSON"));
Ok(())
}
#[test]
fn missing_optional_json_text_and_wrong_field_type_keep_schema_semantics() -> TestResult {
let schema = Collection::compile(&json!({
"type":"object",
"properties":{"payload":{"type":"string","contentMediaType":"application/json"}}
}))?;
schema.validate(&json!({}))?;
let refusal = schema
.validate(&json!({"payload":{}}))
.err()
.ok_or("wrong field type accepted")?;
assert_eq!(refusal.path, "/payload");
assert!(!refusal.reason.contains("not JSON"));
Ok(())
}
#[test]
fn inner_json_uses_the_outer_parser_recursion_bound() -> TestResult {
let text = format!("{}0{}", "[".repeat(256), "]".repeat(256));
let raw = serde_json::to_string(&text)?;
let outer: Value = serde_json::from_str(&raw)?;
let schema =
Collection::compile(&json!({"type":"string","contentMediaType":"application/json"}))?;
let refusal = schema
.validate(&outer)
.err()
.ok_or("deep inner JSON accepted")?;
assert!(refusal.reason.contains("not JSON"));
assert!(refusal.reason.contains("recursion limit"));
Ok(())
}
#[test]
fn existing_schema_errors_keep_their_code_on_put_and_import() -> TestResult {
with_service(|service| {
declare(
service,
"declare",
Some(json!({
"type":"object",
"properties":{"payload":{"type":"integer","minimum":0,"maximum":10,"enum":[0,2,4]}},
"required":["payload"]
})),
)?;
let before = head(service)?;
for (index, raw) in [
r#"{"payload":"wrong"}"#,
r#"{"payload":-1}"#,
r#"{"payload":12}"#,
r#"{"payload":3}"#,
r"{}",
]
.into_iter()
.enumerate()
{
for change in [
put("one", raw)?,
Change::PutMany(DocumentImport {
collection: "records".to_owned(),
documents: vec![imported("one", raw)?],
}),
] {
let id = format!("bad-{index}");
let refused = service
.dispatch(Operation::Mutate(mutation(service, &id, change)?))
.err()
.ok_or("ordinary schema error accepted")?;
assert_eq!(refused.code, "schema_refused");
assert_eq!(head(service)?, before);
assert!(get(service, "one")?["document"].is_null());
receipt_absent(service, &id)?;
}
}
Ok(())
})
}
#[test]
fn declaration_errors_keep_the_schema_refusal_code() -> TestResult {
with_service(|service| {
let before = head(service)?;
for schema in [
json!({"type":"string","contentMediaType":"text/plain"}),
json!({"type":"object","contentMediaType":"application/json"}),
json!({"type":"string","contentSchema":{"type":"object"}}),
] {
let change = Change::Collection(CollectionWrite {
collection: "records".to_owned(),
schema: Some(schema),
});
let refusal = service
.dispatch(Operation::Mutate(mutation(service, "bad-shape", change)?))
.err()
.ok_or("invalid declaration was accepted")?;
assert_eq!(refusal.code, "schema_refused");
assert_eq!(head(service)?, before);
receipt_absent(service, "bad-shape")?;
}
Ok(())
})
}
const fn fixture_long_wait_interval() -> std::time::Duration {
std::time::Duration::from_secs(1)
}