haematite-cli 0.21.8

//! `branch.merge` through the gate: the merge action is the administrator's,
//! asked of Lys live, and a merge is refused by name when the branch it
//! merges has moved since its review or the branch it merges into has moved
//! since the fork.

use std::error::Error;
use std::num::NonZeroU32;
use std::sync::Arc;
use std::time::Duration;

use base64::Engine;
use base64::engine::general_purpose::URL_SAFE_NO_PAD;
use haematite::db::DatabaseConfig;
use haematite::store::NodeCacheBudget;
use lys_pass::deliberate::CheckAnswer;
use lys_pass::drafts::DraftRequest;
use lys_pass::{Error as LysError, Target};
use serde_json::{Value, json};

use super::collection_tests::pass;
use super::tests::{ISSUER, KID, role, signer};
use super::{Gate, LiveCheck};
use crate::serve::protocol::{BranchPage, Operation, ResponseFailure, ReviewPage, Version};
use crate::serve::service::Service;
use crate::serve::storage::Storage;

type TestResult = Result<(), Box<dyn Error>>;

const PAGE: NonZeroU32 = NonZeroU32::MIN.saturating_add(9);

/// Lys as the roles answer a merge: the administrator holds it outright,
/// everyone else holds nothing.
struct Roles;

impl LiveCheck for Roles {
    fn ask(&self, _token: &str, holder: &str, _target: &Target) -> Result<CheckAnswer, LysError> {
        let answer = if holder == "administrator" {
            json!({"allowed": true, "grant": "administrator-grant", "mode": "outright",
                "via": null, "refusal": null, "reason": null, "degraded": null})
        } else {
            json!({"allowed": false, "grant": null, "mode": null,
                "via": null, "refusal": null, "reason": null, "degraded": null})
        };
        serde_json::from_value(answer).map_err(|_| LysError::CannotAsk("answer fixture"))
    }

    fn draft(
        &self,
        _token: &str,
        _request: &DraftRequest,
        _judgment: &CheckAnswer,
    ) -> Result<String, LysError> {
        Err(LysError::CannotAsk("this double records no draft"))
    }
}

fn head(service: &Service, branch: &str) -> Result<Version, Box<dyn Error>> {
    let answer = service.dispatch(Operation::Branches(BranchPage {
        after: None,
        limit: PAGE,
    }))?;
    let row = answer["entries"]
        .as_array()
        .ok_or("branches has no entries")?
        .iter()
        .find(|row| row["branch"] == branch)
        .ok_or_else(|| format!("{branch} missing"))?;
    Ok(serde_json::from_value(row["version"].clone())?)
}

fn operation(
    service: &Service,
    branch: &str,
    request_id: &str,
    kind: &str,
    value: &Value,
) -> Result<Operation, Box<dyn Error>> {
    Ok(serde_json::from_value(json!({
        "method": "mutation.apply",
        "params": {"request_id": request_id, "branch": branch,
            "expected": head(service, branch)?, "change": {"kind": kind, "value": value}},
    }))?)
}

fn apply(service: &Service, branch: &str, kind: &str, value: &Value) -> TestResult {
    let request_id = format!("{kind} {value}");
    service.dispatch(operation(service, branch, &request_id, kind, value)?)?;
    Ok(())
}

fn put(service: &Service, branch: &str, id: &str) -> TestResult {
    apply(
        service,
        branch,
        "document.put",
        &json!({"collection": "patients", "document_id": id, "document": {"on": branch}}),
    )
}

fn fork(service: &Service, name: &str) -> TestResult {
    apply(
        service,
        "main",
        "branch.fork",
        &json!({"name": name, "from": head(service, "main")?}),
    )
}

/// The version of `branch` its review reads.
fn reviewed(service: &Service, branch: &str) -> Result<Version, Box<dyn Error>> {
    let answer = service.dispatch(Operation::Review(ReviewPage {
        branch: branch.to_owned(),
        after: None,
        limit: PAGE,
    }))?;
    Ok(serde_json::from_value(
        answer["selection"]["version"].clone(),
    )?)
}

fn merge(service: &Service, from: &str, reviewed: Version) -> Result<Operation, Box<dyn Error>> {
    operation(
        service,
        "main",
        &format!("merge {from}"),
        "branch.merge",
        &json!({"from": from, "reviewed": reviewed}),
    )
}

fn document(service: &Service, branch: &str, id: &str) -> Result<Option<Value>, Box<dyn Error>> {
    let read: Operation = serde_json::from_value(json!({
        "method": "documents.get",
        "params": {"selection": {"branch": branch, "version": null},
            "collection": "patients", "document_id": id},
    }))?;
    let answer = service.dispatch(read)?;
    Ok((!answer["document"].is_null()).then(|| answer["document"].clone()))
}

/// A gated service on database `wards` whose main holds `patients` with `p0`.
fn service(directory: &std::path::Path) -> Result<Service, Box<dyn Error>> {
    let storage = Storage::create(
        DatabaseConfig {
            data_dir: directory.join("database"),
            shard_count: 2,
            distributed: None,
            executor_threads: Some(1),
            node_cache_budget: Some(NodeCacheBudget::Unlimited),
            retention: haematite::Retention::KeepAll,
        },
        NodeCacheBudget::Unlimited,
        Duration::from_secs(1),
    )?;
    let x = URL_SAFE_NO_PAD.encode(signer().verifying_key().as_bytes());
    let keys = json!({"keys": [{"kty": "OKP", "crv": "Ed25519", "kid": KID, "x": x, "alg": "EdDSA", "use": "sig"}]});
    let service = Service::new(storage, PAGE)?.with_lys(Gate::new(
        &keys.to_string(),
        ISSUER,
        "wards",
        Arc::new(Roles),
    )?);
    apply(
        &service,
        "main",
        "collection.put",
        &json!({"collection": "patients", "schema": null}),
    )?;
    put(&service, "main", "p0")?;
    Ok(service)
}

fn refused(answer: Result<Value, ResponseFailure>, code: &str) -> TestResult {
    let failure = answer
        .err()
        .ok_or_else(|| format!("{code}: the merge ran"))?;
    assert_eq!(failure.code, code, "{}", failure.message);
    Ok(())
}

#[test]
fn an_administrator_merges_a_reviewed_import_and_a_writer_is_refused() -> TestResult {
    let directory = tempfile::tempdir()?;
    let service = service(directory.path())?;
    fork(&service, "import")?;
    put(&service, "import", "a")?;
    put(&service, "import", "b")?;
    apply(
        &service,
        "import",
        "document.delete",
        &json!({"collection": "patients", "document_id": "p0"}),
    )?;
    let version = reviewed(&service, "import")?;
    let import = head(&service, "import")?;

    let writer = pass(
        "writer",
        &role(&["branch", "query", "read", "write"], "writer-grant"),
    );
    let refusal = service
        .authorize(Some(&writer), &merge(&service, "import", version)?, None)
        .err()
        .ok_or("a writer merged")?;
    assert_eq!(refusal.code, "permission_refused");
    let shape: Value = serde_json::from_str(&refusal.message)?;
    assert_eq!(shape["refused"]["needed"]["id"], "wards.main");
    assert_eq!(shape["refused"]["needed"]["action"], "merge");
    assert_eq!(shape["refused"]["request"], format!("{ISSUER}/#/requests"));
    assert_eq!(document(&service, "main", "a")?, None, "a refusal merged");

    let administrator = pass("administrator", &json!([]));
    let before = head(&service, "main")?;
    let request = merge(&service, "import", version)?;
    service.authorize(Some(&administrator), &request, None)?;
    let receipt = service.dispatch(request)?;
    assert_eq!(receipt["branch"], "main");
    for id in ["a", "b"] {
        let merged = document(&service, "main", id)?.ok_or("a merged record is missing")?;
        assert_eq!(merged["raw_json"], r#"{"on":"import"}"#, "{id}");
        assert_eq!(merged["revision"]["written_in"], "import", "{id}");
    }
    assert_eq!(
        document(&service, "main", "p0")?,
        None,
        "the removal merged"
    );
    assert_eq!(
        head(&service, "import")?,
        import,
        "the source is left as it is"
    );
    // The import's own fork origin stays with it: main is still no fork.
    refused(
        service.dispatch(Operation::Review(ReviewPage {
            branch: "main".to_owned(),
            after: None,
            limit: PAGE,
        })),
        "fork_provenance_missing",
    )?;
    // The same request again answers its receipt and merges nothing more.
    let again: Operation = serde_json::from_value(json!({
        "method": "mutation.apply",
        "params": {"request_id": "merge import", "branch": "main", "expected": before,
            "change": {"kind": "branch.merge", "value": {"from": "import", "reviewed": version}}},
    }))?;
    let after = head(&service, "main")?;
    assert_eq!(service.dispatch(again)?, receipt);
    assert_eq!(head(&service, "main")?, after);
    Ok(())
}

#[test]
fn a_branch_that_moved_after_its_review_or_onto_a_moved_target_is_refused() -> TestResult {
    let directory = tempfile::tempdir()?;
    let service = service(directory.path())?;
    fork(&service, "import")?;
    put(&service, "import", "a")?;
    let version = reviewed(&service, "import")?;
    put(&service, "import", "late")?;
    refused(
        service.dispatch(merge(&service, "import", version)?),
        "merge_source_moved",
    )?;
    assert_eq!(document(&service, "main", "a")?, None);

    fork(&service, "second")?;
    put(&service, "second", "c")?;
    let second = reviewed(&service, "second")?;
    put(&service, "main", "moved")?;
    refused(
        service.dispatch(merge(&service, "second", second)?),
        "merge_target_moved",
    )?;
    assert_eq!(document(&service, "main", "c")?, None);
    Ok(())
}