mock-upcloud 0.1.3

//! **The twenty-two behaviours, over a real socket.**
//!
//! Each test arms one behaviour and asserts the thing a client would actually
//! see: a status, an error code, a field's absence, a state that is not the one
//! that was asked for. They go over HTTP with the same blocking `reqwest` stack
//! the real plugin uses, because half of these behaviours (the transport reset,
//! the problem+json content type, the keep-alive pooling) do not exist above the
//! socket.
//!
//! The numbering matches the table in the crate header. A behaviour with no test
//! here is a behaviour that is not modelled, and the header says which those are.

use mock_upcloud::estate::{Estate, StorageKind};
use mock_upcloud::{Clock, Fault, Faults, Mock};
use serde_json::Value;

struct Rig {
    base: String,
    http: reqwest::blocking::Client,
    _rt: tokio::runtime::Runtime,
}

impl Rig {
    fn with(faults: Faults) -> Rig {
        // Virtual time: these tests are about WHAT is answered, not how long it
        // takes, and the one test that is about duration reads the clock.
        let estate = Estate::new(Clock::virtual_only(), faults, 1234);
        let rt = tokio::runtime::Builder::new_multi_thread().enable_all().build().unwrap();
        let port = rt.block_on(async {
            let (p, _h) = mock_upcloud::serve(Mock::new(estate), "127.0.0.1:0").await.unwrap();
            p
        });
        Rig {
            base: format!("http://127.0.0.1:{port}"),
            http: reqwest::blocking::Client::builder().build().unwrap(),
            _rt: rt,
        }
    }

    fn new() -> Rig {
        Rig::with(Faults::none())
    }

    fn get(&self, path: &str) -> (u16, Value) {
        let r = self
            .http
            .get(format!("{}{path}", self.base))
            .bearer_auth("ucat_mock")
            .send()
            .unwrap();
        let s = r.status().as_u16();
        let t = r.text().unwrap();
        (s, serde_json::from_str(&t).unwrap_or(Value::Null))
    }

    fn send(&self, method: reqwest::Method, path: &str, body: Value) -> (u16, Value) {
        let r = self
            .http
            .request(method, format!("{}{path}", self.base))
            .bearer_auth("ucat_mock")
            .json(&body)
            .send()
            .unwrap();
        let s = r.status().as_u16();
        let t = r.text().unwrap();
        (s, serde_json::from_str(&t).unwrap_or(Value::Null))
    }

    fn post(&self, path: &str, body: Value) -> (u16, Value) {
        self.send(reqwest::Method::POST, path, body)
    }
    fn put(&self, path: &str, body: Value) -> (u16, Value) {
        self.send(reqwest::Method::PUT, path, body)
    }
    fn delete(&self, path: &str) -> (u16, Value) {
        self.send(reqwest::Method::DELETE, path, Value::Null)
    }

    /// Create a server and poll it to `started`, as a client must.
    fn a_started_server(&self, title: &str) -> String {
        let (s, v) = self.post(
            "/1.3/server",
            serde_json::json!({"server": {
                "zone": "se-sto1", "title": title, "hostname": title, "plan": "2xCPU-4GB",
                "labels": {"label": [{"key": "site", "value": "gunnar.rs"}, {"key": "role", "value": "appliance"}]},
                "storage_devices": {"storage_device": [{"action": "create", "title": format!("{title}-boot"), "size": 20, "tier": "maxiops"}]}
            }}),
        );
        assert_eq!(s, 201, "{v}");
        let uuid = v["server"]["uuid"].as_str().unwrap().to_string();
        self.poll_until(&uuid, "started");
        uuid
    }

    fn poll_until(&self, uuid: &str, state: &str) -> u32 {
        for n in 1..=200 {
            let (_, v) = self.get(&format!("/1.3/server/{uuid}"));
            if v["server"]["state"] == state {
                return n;
            }
            if v["server"].is_null() {
                panic!("server {uuid} vanished while waiting for {state}");
            }
        }
        panic!("server {uuid} never reached {state}");
    }
}

// ── 1 ────────────────────────────────────────────────────────────────────────

/// **A firewall-rule read for a DELETED server answers 403
/// ERROR_AUTHENTICATION_FAILED, not 404.**
///
/// This is the one that stopped Terraform for an hour: it cannot tell this from
/// a revoked token, so it does not remove the resource from state — it halts.
/// The assertion is the `type` URL and the presence of a `correlation_id`,
/// because those two are what a client has to match on, and there is nothing
/// else in the body to match.
#[test]
fn b01_a_deleted_servers_firewall_rules_answer_403_authentication_failed() {
    let r = Rig::new();
    let uuid = r.a_started_server("front");
    // While it exists, the same call is a perfectly ordinary 200.
    let (s, _) = r.get(&format!("/1.3/server/{uuid}/firewall_rule"));
    assert_eq!(s, 200);

    assert_eq!(r.post(&format!("/1.3/server/{uuid}/stop"), serde_json::json!({"stop_server": {"stop_type": "hard", "timeout": "60"}})).0, 200);
    r.poll_until(&uuid, "stopped");
    assert_eq!(r.delete(&format!("/1.3/server/{uuid}?storages=1&backups=keep")).0, 204);
    for _ in 0..200 {
        let (s, _) = r.get(&format!("/1.3/server/{uuid}"));
        if s == 404 {
            break;
        }
    }

    let (s, v) = r.get(&format!("/1.3/server/{uuid}/firewall_rule"));
    assert_eq!(s, 403, "not 404 — that is the whole difficulty: {v}");
    assert_eq!(
        v["type"], "https://developers.upcloud.com/1.3/errors#ERROR_AUTHENTICATION_FAILED",
        "{v}"
    );
    assert!(v["correlation_id"].is_string(), "{v}");
    assert!(v["error"].is_null(), "it is problem+json, not the error envelope: {v}");
}

// ── 2 ────────────────────────────────────────────────────────────────────────

/// **`GET /1.3/price` fails at the transport level, not with a status.**
///
/// `reqwest` reports it as a request error with no status at all. monetize's
/// credential probe turned that into "the credential in UPCLOUD_TOKEN could not
/// be verified", which sent a person to look at a token that was fine.
#[test]
fn b02_the_price_list_drops_the_socket_rather_than_answering() {
    let f = Faults::none();
    f.arm(Fault::PriceTransportReset);
    let r = Rig::with(f);
    let e = r
        .http
        .get(format!("{}/1.3/price", r.base))
        .bearer_auth("ucat_mock")
        .send()
        .expect_err("the socket must close with nothing written");
    assert!(e.status().is_none(), "a transport failure has no status: {e:?}");
}

// ── 3 ────────────────────────────────────────────────────────────────────────

/// **412 out_of_stock on poweron, and it does not clear.** Verified against
/// Scaleway (fr-par-1 2026-09-09), not UpCloud; kept as a generic
/// cloud-provider fault.
#[test]
fn b03_poweron_is_refused_with_412_out_of_stock_and_stays_refused() {
    let r = Rig::new();
    let uuid = r.a_started_server("appliance");
    r.post(&format!("/1.3/server/{uuid}/stop"), serde_json::json!({"stop_server": {"stop_type": "hard", "timeout": "60"}}));
    r.poll_until(&uuid, "stopped");

    let (s, v) = r.post(&format!("/mock/fault/out-of-stock/arm"), Value::Null);
    assert_eq!(s, 200, "{v}");
    for attempt in 0..25 {
        let (s, v) = r.post(&format!("/1.3/server/{uuid}/start"), Value::Null);
        assert_eq!(s, 412, "attempt {attempt}: a retry loop must not outwait it: {v}");
        assert_eq!(v["error"]["error_code"], "out_of_stock", "{v}");
    }
    // And it is curable only by the operator, never by patience.
    r.post("/mock/fault/out-of-stock/disarm", Value::Null);
    assert_eq!(r.post(&format!("/1.3/server/{uuid}/start"), Value::Null).0, 200);
}

// ── 4 ────────────────────────────────────────────────────────────────────────

/// **The server LIST carries no attachments; the storage LIST carries labels.**
///
/// A caller that lists servers and reads `storage_devices` off the rows finds
/// nothing attached to anything, and concludes the account is empty of disks.
#[test]
fn b04_the_list_is_not_the_detail() {
    let r = Rig::new();
    let uuid = r.a_started_server("front");

    let (_, list) = r.get("/1.3/server");
    let row = &list["servers"]["server"][0];
    assert_eq!(row["uuid"], uuid.as_str());
    assert!(row["storage_devices"].is_null(), "the LIST must not carry attachments: {row}");
    assert!(row["ip_addresses"].is_null(), "nor addresses: {row}");

    let (_, detail) = r.get(&format!("/1.3/server/{uuid}"));
    assert!(
        detail["server"]["storage_devices"]["storage_device"][0]["storage"].is_string(),
        "the DETAIL is the only place they exist: {detail}"
    );

    // The storage list, by contrast, DOES carry labels — so a label search over
    // storages needs no second call and a label search over servers does.
    let (_, st) = r.get("/1.3/storage/private");
    let s0 = &st["storages"]["storage"][0];
    assert!(s0["labels"].is_array(), "{s0}");
}

// ── 5 ────────────────────────────────────────────────────────────────────────

/// **A revoked credential answers 0 servers and 403 on details. Never a clean
/// 401.** Verified against Scaleway (IAM change 2026-08-26), not UpCloud; kept
/// as a generic cloud-provider fault. A dead UpCloud token is behaviour 43.
///
/// The two halves together are the defect. A caller seeing the empty list
/// concludes there is nothing to clean up and reports success.
#[test]
fn b05_a_revoked_credential_is_an_empty_account_not_an_error() {
    let r = Rig::new();
    let uuid = r.a_started_server("appliance");
    assert_eq!(r.get("/1.3/server").1["servers"]["server"].as_array().unwrap().len(), 1);

    r.post("/mock/fault/revoked-credential/arm", Value::Null);

    let (s, v) = r.get("/1.3/server");
    assert_eq!(s, 200, "not 401 — that is the point");
    assert_eq!(v["servers"]["server"].as_array().unwrap().len(), 0, "{v}");

    let (s, v) = r.get(&format!("/1.3/server/{uuid}"));
    assert_eq!(s, 403, "{v}");
    assert!(v["correlation_id"].is_string(), "{v}");

    let (s, v) = r.get("/1.3/account");
    assert_eq!(s, 403, "{v}");
}

// ── 6, 7, 11 ─────────────────────────────────────────────────────────────────

/// **Creating a server takes 98–105 s, and the state before that is
/// `maintenance`.**
#[test]
fn b06_a_server_is_in_maintenance_for_ninety_eight_to_a_hundred_and_five_seconds() {
    let r = Rig::new();
    let (_, v) = r.post(
        "/1.3/server",
        serde_json::json!({"server": {"zone": "se-sto1", "title": "t", "hostname": "t", "plan": "2xCPU-4GB",
            "labels": {"label": []},
            "storage_devices": {"storage_device": [{"action": "create", "title": "t-boot", "size": 20, "tier": "maxiops"}]}}}),
    );
    assert_eq!(v["server"]["state"], "maintenance", "it is never born started: {v}");
    let uuid = v["server"]["uuid"].as_str().unwrap().to_string();
    let polls = r.poll_until(&uuid, "started");
    assert!(polls > 1, "a client that did not poll would have been wrong");

    let (_, seed) = r.get("/mock/seed");
    let elapsed = seed["now_ms"].as_u64().unwrap();
    assert!((98_000..=106_000).contains(&elapsed), "{elapsed} ms");
}

/// **Deleting an appliance that carries four member volumes takes more than
/// five minutes in `maintenance`; a front takes about a minute. And neither can
/// be deleted at all until it is STOPPED (behaviour 11).**
#[test]
fn b07_and_b11_a_delete_needs_a_stop_first_and_a_fat_appliance_is_slow() {
    let r = Rig::new();
    let uuid = r.a_started_server("appliance");

    // 11: a started server is refused, not queued.
    let (s, v) = r.delete(&format!("/1.3/server/{uuid}?storages=1&backups=keep"));
    assert_eq!(s, 409, "{v}");
    assert_eq!(v["error"]["error_code"], "SERVER_STATE_ILLEGAL", "{v}");

    r.post(&format!("/1.3/server/{uuid}/stop"), serde_json::json!({"stop_server": {"stop_type": "hard", "timeout": "60"}}));
    r.poll_until(&uuid, "stopped");

    // Four member volumes, attached while stopped.
    for i in 0..4 {
        let (s, sv) = r.post(
            "/1.3/storage",
            serde_json::json!({"storage": {"size": 100, "tier": "maxiops", "title": format!("member-{i}"), "zone": "se-sto1",
                "labels": [{"key": "volume", "value": format!("member-{i}")}]}}),
        );
        assert_eq!(s, 201, "{sv}");
        let su = sv["storage"]["uuid"].as_str().unwrap().to_string();
        // A storage is born in `maintenance` and an attach before `online` is
        // refused — the poll loop's reason to exist.
        loop {
            let (_, g) = r.get(&format!("/1.3/storage/{su}"));
            if g["storage"]["state"] == "online" {
                break;
            }
        }
        let (s, av) = r.post(
            &format!("/1.3/server/{uuid}/storage/attach"),
            serde_json::json!({"storage_device": {"type": "disk", "storage": su}}),
        );
        assert_eq!(s, 200, "{av}");
    }

    let before = r.get("/mock/seed").1["now_ms"].as_u64().unwrap();
    assert_eq!(r.delete(&format!("/1.3/server/{uuid}?storages=1&backups=keep")).0, 204);
    // 204 does NOT mean gone.
    let (_, v) = r.get(&format!("/1.3/server/{uuid}"));
    assert_eq!(v["server"]["state"], "maintenance", "a 204 is not a deletion: {v}");
    loop {
        let (s, _) = r.get(&format!("/1.3/server/{uuid}"));
        if s == 404 {
            break;
        }
    }
    let took = r.get("/mock/seed").1["now_ms"].as_u64().unwrap() - before;
    assert!(took > 300_000, "four member volumes is over five minutes, not {took} ms");
}

// ── 8 ────────────────────────────────────────────────────────────────────────

/// **The guest's own install is ~10 s, and nothing narrates it.**
#[test]
fn b08_the_install_is_ten_seconds_of_silence() {
    use mock_upcloud::kvm::{GuestEngine, GuestOutcome, GuestSpec, VirtualGuest};
    let out = VirtualGuest::default().boot(&GuestSpec::new("appliance", "/dev/null"));
    assert_eq!(out, GuestOutcome::Installed { ms: 10_011 });
    // There is no log to read. That is behaviour 20, expressed as a type: the
    // outcome carries no output field, so nothing can pretend to have watched.
}

// ── 9 ────────────────────────────────────────────────────────────────────────

/// **A stop-resize-start leaves a `Resize Backup` behind: detached, labelled
/// with the estate's own labels, and with NO "gunnar" in its title.**
///
/// The cleanup filtered by title prefix. This row is invisible to that filter
/// and bills maxiops forever.
#[test]
fn b09_a_resize_leaves_a_backup_no_title_filter_can_see() {
    let r = Rig::new();
    let uuid = r.a_started_server("gunnar-appliance");
    r.post(&format!("/1.3/server/{uuid}/stop"), serde_json::json!({"stop_server": {"stop_type": "soft", "timeout": "60"}}));
    r.poll_until(&uuid, "stopped");

    let before = r.get("/1.3/storage/private").1["storages"]["storage"].as_array().unwrap().len();
    let (s, v) = r.put(&format!("/1.3/server/{uuid}"), serde_json::json!({"server": {"plan": "4xCPU-8GB"}}));
    assert_eq!(s, 202, "{v}");
    r.poll_until(&uuid, "stopped");
    // The mint happens on arrival, so one more read after the transition.
    r.get("/1.3/server");

    let (_, list) = r.get("/1.3/storage/private");
    let rows = list["storages"]["storage"].as_array().unwrap();
    assert_eq!(rows.len(), before + 1, "exactly one new object: {list}");
    let backup = rows.iter().find(|s| s["type"] == "backup").expect("a backup row");

    assert!(
        !backup["title"].as_str().unwrap().to_lowercase().contains("gunnar"),
        "a title-prefix filter is BLIND to this row, and that is the defect: {backup}"
    );
    assert!(backup["title"].as_str().unwrap().starts_with("Resize Backup"), "{backup}");
    assert!(backup["origin"].is_string(), "it names an origin: {backup}");
    // It carries the estate's labels, so a LABEL search does find it — which is
    // the fix the cleanup needed and did not have.
    let labels = backup["labels"].as_array().unwrap();
    assert!(labels.iter().any(|l| l["key"] == "site"), "{backup}");
    // And it is attached to nothing, so no server delete will take it along.
    let (_, det) = r.get(&format!("/1.3/storage/{}", backup["uuid"].as_str().unwrap()));
    assert_eq!(det["storage"]["servers"]["server"].as_array().unwrap().len(), 0, "{det}");
}

// ── 34 ───────────────────────────────────────────────────────────────────────

/// **The filesystem resize is the door that actually leaks, and it hands the
/// leak back in its own reply.**
///
/// Behaviour 9 minted the backup on a plan change, which this estate never
/// does. What `cargo xtask grow` does to the twin is `PUT /storage/{uuid}` and
/// then `POST /storage/{uuid}/resize`, and the provider answers the second with
/// `resize_backup` — a whole storage object, its uuid included — which nobody
/// then deletes. MEASURED 2026-09-20: 44 GB, `Resize Backup`, origin = the
/// twin's data volume, and a sweep that filtered by title said "no orphans".
#[test]
fn b34_the_filesystem_resize_hands_back_the_backup_it_leaves_behind() {
    let r = Rig::new();
    let uuid = r.a_started_server("gunnar-twin");
    let (_, det) = r.get(&format!("/1.3/server/{uuid}"));
    let boot = det["server"]["storage_devices"]["storage_device"][0]["storage"].as_str().unwrap().to_string();

    // On a STARTED server it is refused by name — the estate stops the twin first.
    let (s, v) = r.post(&format!("/1.3/storage/{boot}/resize"), Value::Null);
    assert_eq!(s, 409, "{v}");
    assert_eq!(v["error"]["error_code"], "SERVER_STATE_ILLEGAL", "{v}");

    r.post(&format!("/1.3/server/{uuid}/stop"), serde_json::json!({"stop_server": {"stop_type": "soft", "timeout": "60"}}));
    r.poll_until(&uuid, "stopped");
    // Grow the volume first, as the estate does: the backup is then the GROWN size.
    assert_eq!(r.put(&format!("/1.3/storage/{boot}"), serde_json::json!({"storage": {"size": 44}})).0, 200);

    let before = r.get("/1.3/storage/private").1["storages"]["storage"].as_array().unwrap().len();
    let (s, v) = r.post(&format!("/1.3/storage/{boot}/resize"), Value::Null);
    assert_eq!(s, 200, "{v}");
    let backup = &v["resize_backup"];
    let bk = backup["uuid"].as_str().expect("the reply names the backup it took").to_string();
    assert_eq!(backup["type"], "backup", "{backup}");
    assert_eq!(backup["size"], 44, "the backup is the volume's size — the 44 GB that was measured: {backup}");
    assert_eq!(backup["origin"], boot, "and it names the volume it was taken from: {backup}");
    assert!(backup["title"].as_str().unwrap().starts_with("Resize Backup"), "{backup}");
    assert!(!backup["title"].as_str().unwrap().to_lowercase().contains("gunnar"), "provider-titled: {backup}");
    // The measurement: the labels ARE copied by default.
    assert!(backup["labels"].as_array().unwrap().iter().any(|l| l["key"] == "site"), "{backup}");

    // It is on the account, detached, beside the labelled set — and stays.
    let (_, list) = r.get("/1.3/storage/private");
    let rows = list["storages"]["storage"].as_array().unwrap();
    assert_eq!(rows.len(), before + 1, "{list}");
    let (_, det) = r.get(&format!("/1.3/storage/{bk}"));
    assert_eq!(det["storage"]["servers"]["server"].as_array().unwrap().len(), 0, "{det}");
    // The volume went through maintenance and comes back online, still 44.
    loop {
        let st = r.get(&format!("/1.3/storage/{boot}")).1["storage"].clone();
        if st["state"] == "online" {
            assert_eq!(st["size"], 44);
            break;
        }
    }
}

/// **The brief's hypothesis, by name: a backup carrying NEITHER a product title
/// NOR a label.** The default is the measurement (labels copied); armed by
/// name, the same door leaves an object that only its `origin` ties to the
/// estate. Never in seeded weather.
#[test]
fn b34b_unlabelled_is_a_hypothesis_and_arrives_only_when_asked_for() {
    let r = Rig::new();
    let uuid = r.a_started_server("gunnar-twin");
    let boot = r.get(&format!("/1.3/server/{uuid}")).1["server"]["storage_devices"]["storage_device"][0]["storage"]
        .as_str()
        .unwrap()
        .to_string();
    r.post(&format!("/1.3/server/{uuid}/stop"), serde_json::json!({"stop_server": {"stop_type": "soft", "timeout": "60"}}));
    r.poll_until(&uuid, "stopped");

    r.post("/mock/fault/resize-backup-unlabelled/arm", Value::Null);
    let (s, v) = r.post(&format!("/1.3/storage/{boot}/resize"), Value::Null);
    assert_eq!(s, 200, "{v}");
    let backup = &v["resize_backup"];
    assert!(backup["labels"].as_array().unwrap().is_empty(), "nothing on it but the provider's own words: {backup}");
    assert!(!backup["title"].as_str().unwrap().to_lowercase().contains("gunnar"), "{backup}");
    assert_eq!(backup["origin"], boot, "the origin is all that is left to recognise it by: {backup}");

    // And it is NOT weather: seeded faults never include it.
    let seeded = mock_upcloud::Faults::seeded(7);
    assert!(!seeded.is_armed(Fault::ResizeBackupUnlabelled));
    assert!(!seeded.is_armed(Fault::WithholdCreatedField), "the same rule as the other hypothesis");
}

// ── 10 ───────────────────────────────────────────────────────────────────────

/// **`created` IS sent, on the list AND the detail.**
///
/// This test used to assert the opposite, because the brief said so. MEASURED
/// against the live account 2026-09-20 — two volumes, both endpoints, the field
/// present every time — so the mock was stricter than the provider and was
/// manufacturing a verdict path the account cannot produce. The same mistake
/// this crate made with the virtio hot-plug attach, in the other direction.
///
/// What behaviour 9 actually turns on is the TITLE. The date was never the
/// missing half; the label was.
#[test]
fn b10_created_is_sent_on_both_the_list_and_the_detail() {
    let r = Rig::new();
    r.a_started_server("front");
    let (_, list) = r.get("/1.3/storage/private");
    let rows = list["storages"]["storage"].as_array().unwrap();
    assert!(!rows.is_empty());
    for row in rows {
        let c = row["created"].as_str().unwrap_or_else(|| panic!("the account sends this: {row}"));
        assert!(c.ends_with('Z') && c.contains('T'), "RFC 3339, as UpCloud sends it: {c}");
    }
    let uuid = rows[0]["uuid"].as_str().unwrap().to_string();
    let (_, detail) = r.get(&format!("/1.3/storage/{uuid}"));
    assert!(detail["storage"]["created"].is_string(), "and the DETAIL carries it too: {detail}");

    // The hypothesis is still reachable, by name, because the verdict paths that
    // would depend on an absent field are worth a regression test even though
    // the provider never produces one.
    r.post("/mock/fault/withhold-created-field/arm", Value::Null);
    let (_, list) = r.get("/1.3/storage/private");
    assert!(
        list["storages"]["storage"][0]["created"].is_null(),
        "armed by name, and only then: {list}"
    );
}

// ── 12 ───────────────────────────────────────────────────────────────────────

/// **Addresses move between lays.** A literal address in code went stale exactly
/// this way, three times.
#[test]
fn b12_the_addresses_are_not_the_same_on_the_next_lay() {
    let r = Rig::new();
    let a = r.a_started_server("appliance");
    let t = r.a_started_server("twin");
    let ip = |u: &str| {
        let (_, v) = r.get(&format!("/1.3/server/{u}"));
        let addrs = v["server"]["ip_addresses"]["ip_address"].as_array().unwrap().clone();
        let f = |access: &str| {
            addrs
                .iter()
                .find(|a| a["access"] == access)
                .unwrap()["address"]
                .as_str()
                .unwrap()
                .to_string()
        };
        (f("public"), f("utility"))
    };
    let lay1 = (ip(&a), ip(&t));
    assert_ne!(lay1.0, lay1.1, "two servers never share an address");

    r.post("/mock/relay", Value::Null);
    let a2 = r.a_started_server("appliance");
    let t2 = r.a_started_server("twin");
    let lay2 = (ip(&a2), ip(&t2));
    assert_ne!(lay1, lay2, "a re-lay that handed out the same addresses would hide the defect");
}

// ── 14 ───────────────────────────────────────────────────────────────────────

/// **The VNC port goes stale on every stop/start, and only a `remote_access`
/// no→yes toggle fixes what the API reports.**
#[test]
fn b14_the_reported_vnc_port_is_the_one_from_before_the_restart() {
    let r = Rig::new();
    let uuid = r.a_started_server("appliance");
    r.put(
        &format!("/1.3/server/{uuid}"),
        serde_json::json!({"server": {"remote_access_enabled": "yes", "remote_access_password": "hunter2"}}),
    );
    let port = |()| {
        r.get(&format!("/1.3/server/{uuid}"))
            .1["server"]["remote_access_port"]
            .as_str()
            .unwrap()
            .to_string()
    };
    let first = port(());

    r.post(&format!("/1.3/server/{uuid}/stop"), serde_json::json!({"stop_server": {"stop_type": "hard", "timeout": "60"}}));
    r.poll_until(&uuid, "stopped");
    r.post(&format!("/1.3/server/{uuid}/start"), Value::Null);
    r.poll_until(&uuid, "started");

    assert_eq!(port(()), first, "the API keeps reporting the OLD port — that is the defect");

    // The cure, and the only one: off, then on.
    r.put(&format!("/1.3/server/{uuid}"), serde_json::json!({"server": {"remote_access_enabled": "no"}}));
    // …with the pause every working tool has (behaviour 50).
    r.post("/mock/advance/2000", Value::Null);
    r.put(&format!("/1.3/server/{uuid}"), serde_json::json!({"server": {"remote_access_enabled": "yes"}}));
    assert_ne!(port(()), first, "after the toggle it reports the port the hypervisor is really on");
}

// ── 15 ───────────────────────────────────────────────────────────────────────

/// **SeaBIOS, and there is no knob.** A server reports `firmware: bios` and the
/// API offers no way to ask for anything else, which is why every gunnar ISO
/// must carry a BIOS boot path.
#[test]
fn b15_there_is_no_uefi_and_no_way_to_ask_for_one() {
    let r = Rig::new();
    let uuid = r.a_started_server("appliance");
    let (_, v) = r.get(&format!("/1.3/server/{uuid}"));
    assert_eq!(v["server"]["firmware"], "bios");
    // Asking for UEFI is not refused — it is IGNORED, which is worse, and is
    // what the real API does with a field it does not know.
    r.put(&format!("/1.3/server/{uuid}"), serde_json::json!({"server": {"firmware": "uefi"}}));
    let (_, v) = r.get(&format!("/1.3/server/{uuid}"));
    assert_eq!(v["server"]["firmware"], "bios", "the request was accepted and ignored");
}

// ── 16, 17 ───────────────────────────────────────────────────────────────────

/// **`cdrom/eject` works on a started server; a detach of the same device does
/// not — and the eject is what ends an installer loop.**
#[test]
fn b16_and_b17_eject_is_the_only_never_loop_primitive() {
    let f = Faults::none();
    f.arm(Fault::InstallerLoop);
    let r = Rig::with(f);
    let uuid = r.a_started_server("appliance");

    // Put an installer medium on, which needs the box stopped.
    r.post(&format!("/1.3/server/{uuid}/stop"), serde_json::json!({"stop_server": {"stop_type": "hard", "timeout": "60"}}));
    r.poll_until(&uuid, "stopped");
    let (_, iso) = r.post(
        "/1.3/storage",
        serde_json::json!({"storage": {"size": 2, "tier": "maxiops", "title": "gunnar-installer.iso", "zone": "se-sto1", "labels": []}}),
    );
    let iso_uuid = iso["storage"]["uuid"].as_str().unwrap().to_string();
    loop {
        if r.get(&format!("/1.3/storage/{iso_uuid}")).1["storage"]["state"] == "online" {
            break;
        }
    }
    assert_eq!(
        r.post(&format!("/1.3/server/{uuid}/storage/attach"), serde_json::json!({"storage_device": {"type": "cdrom", "storage": iso_uuid}})).0,
        200
    );
    r.put(&format!("/1.3/server/{uuid}"), serde_json::json!({"server": {"boot_order": "cdrom"}}));
    r.post(&format!("/1.3/server/{uuid}/start"), Value::Null);
    r.poll_until(&uuid, "started");

    // 17: the guest is looping, and a SOFT stop of a wedged installer never
    // completes — which is why the procedure uses a hard stop after the media go on.
    let (s, v) = r.post(&format!("/1.3/server/{uuid}/stop"), serde_json::json!({"stop_server": {"stop_type": "soft", "timeout": "60"}}));
    assert_eq!(s, 409, "a wedged installer answers no ACPI event: {v}");

    // 16a: a detach of the ide device on a STARTED server is refused by name.
    let (s, v) = r.post(&format!("/1.3/server/{uuid}/storage/detach"), serde_json::json!({"storage_device": {"address": "ide:0:0"}}));
    assert_eq!(s, 409, "{v}");
    assert_eq!(v["error"]["error_code"], "IDE_HOTPLUG_UNSUPPORTED", "{v}");

    // 16b: the eject works, on the same started server, and ends the loop.
    let (s, v) = r.post(&format!("/1.3/server/{uuid}/cdrom/eject"), Value::Null);
    assert_eq!(s, 200, "{v}");
    let (_, est) = r.get("/mock/estate");
    let me = est["servers"].as_array().unwrap().iter().find(|s| s["uuid"] == uuid.as_str()).unwrap();
    assert_eq!(me["guest"], "Installed", "the loop ended: {me}");
}

// ── 22 ───────────────────────────────────────────────────────────────────────

/// **A volume never shrinks.** Refused by name, not clamped: a clamp would let a
/// caller believe it had shrunk something, and a growth is billed forever.
#[test]
fn b22_a_volume_grows_and_never_shrinks() {
    let r = Rig::new();
    let (_, v) = r.post(
        "/1.3/storage",
        serde_json::json!({"storage": {"size": 100, "tier": "maxiops", "title": "data", "zone": "se-sto1", "labels": []}}),
    );
    let u = v["storage"]["uuid"].as_str().unwrap().to_string();
    loop {
        if r.get(&format!("/1.3/storage/{u}")).1["storage"]["state"] == "online" {
            break;
        }
    }
    assert_eq!(r.put(&format!("/1.3/storage/{u}"), serde_json::json!({"storage": {"size": 112}})).0, 200);
    let (s, v) = r.put(&format!("/1.3/storage/{u}"), serde_json::json!({"storage": {"size": 100}}));
    assert_eq!(s, 400, "{v}");
    // Behaviour 54: the MEASURED code and sentence (it was STORAGE_INVALID_SIZE here).
    assert_eq!(v["error"]["error_code"], "SIZE_INVALID", "{v}");
    assert!(
        v["error"]["error_message"].as_str().unwrap().contains("must be greater than the old size"),
        "refused BY NAME, so nobody believes a shrink happened: {v}"
    );
    // …and a SAME-size retry is refused the same way (MEASURED).
    let (s, v) = r.put(&format!("/1.3/storage/{u}"), serde_json::json!({"storage": {"size": 112}}));
    assert_eq!((s, v["error"]["error_code"].as_str()), (400, Some("SIZE_INVALID")), "{v}");
    assert_eq!(r.get(&format!("/1.3/storage/{u}")).1["storage"]["size"], 112);
}

// ── the mock's own promises ──────────────────────────────────────────────────

/// An unimplemented path must never read as a working one.
#[test]
fn an_unimplemented_path_is_loud_and_names_itself() {
    let r = Rig::new();
    let (s, v) = r.get("/1.3/load_balancer");
    assert_eq!(s, 404);
    assert_eq!(v["error"]["error_code"], "MOCK_UPCLOUD_NOT_IMPLEMENTED");
    assert!(v["error"]["error_message"].as_str().unwrap().contains("/1.3/load_balancer"), "{v}");
}

/// No credential at all IS a clean 401. A revoked one is not (behaviour 5), and
/// the difference between those two is the whole confusion.
#[test]
fn no_credential_is_the_only_clean_401() {
    let r = Rig::new();
    let resp = r.http.get(format!("{}/1.3/server", r.base)).send().unwrap();
    assert_eq!(resp.status().as_u16(), 401);
}

/// ★ **A PUT that names the SAME plan is not a plan change.**
///
/// MEASURED with `--log`: the provider grows a disk by sending the whole server
/// object, plan included and unchanged. Reading that as a resize put the machine
/// into `maintenance`, refused the filesystem resize that followed, and minted a
/// `Resize Backup` for a resize that never happened — the mock manufacturing
/// both a broken growth path and the leak it exists to reproduce.
#[test]
fn a_put_that_names_the_same_plan_changes_nothing() {
    let r = Rig::new();
    let uuid = r.a_started_server("front");
    r.post(&format!("/1.3/server/{uuid}/stop"), serde_json::json!({"stop_server": {"stop_type": "hard"}}));
    for _ in 0..40 {
        let (_, v) = r.get(&format!("/1.3/server/{uuid}"));
        if v["server"]["state"] == "stopped" {
            break;
        }
    }
    let (_, before) = r.get(&format!("/1.3/server/{uuid}"));
    assert_eq!(before["server"]["state"], "stopped", "the stop never completed: {}", before["server"]["state"]);
    let plan = before["server"]["plan"].as_str().expect("a plan").to_string();
    let backups_before = backup_count(&r);

    let (code, _) = r.put(&format!("/1.3/server/{uuid}"), serde_json::json!({"server": {"plan": plan}}));
    assert_eq!(code, 202, "behaviour 41: a server PUT is 202, MEASURED");

    let (_, after) = r.get(&format!("/1.3/server/{uuid}"));
    assert_eq!(after["server"]["state"], "stopped", "a no-op PUT moved the machine into {}", after["server"]["state"]);
    assert_eq!(backup_count(&r), backups_before, "a no-op PUT minted a Resize Backup");
}

fn backup_count(r: &Rig) -> usize {
    let (_, v) = r.get("/1.3/storage/private");
    v["storages"]["storage"]
        .as_array()
        .map(|a| a.iter().filter(|s| s["type"] == "backup").count())
        .unwrap_or(0)
}

/// ★ **A grow that no read reports is a write that did not happen.**
///
/// MEASURED 2026-09-21: terraform grew a system disk 20 → 25 GB, the PUT
/// succeeded, and the provider refused its own apply — `.template[0].size: was
/// 25, but now 20` — because it reads the size off the SERVER's
/// `storage_devices`, which this mock left at the size the device had when it
/// was attached.
#[test]
fn a_grown_volume_is_grown_everywhere_it_is_reported() {
    let r = Rig::new();
    let uuid = r.a_started_server("front");
    let (_, v) = r.get(&format!("/1.3/server/{uuid}"));
    let dev = v["server"]["storage_devices"]["storage_device"][0].clone();
    let disk = dev["storage"].as_str().expect("a boot disk").to_string();
    let was = dev["storage_size"].as_u64().expect("a number");
    // terraform stops the server before it grows a template (the measured run
    // did); a grow under a RUNNING server is refused (behaviour 56).
    r.post(&format!("/1.3/server/{uuid}/stop"), serde_json::json!({"stop_server": {"stop_type": "hard"}}));
    r.poll_until(&uuid, "stopped");

    let (code, _) = r.put(&format!("/1.3/storage/{disk}"), serde_json::json!({"storage": {"size": was + 5}}));
    assert_eq!(code, 200);

    let (_, s) = r.get(&format!("/1.3/storage/{disk}"));
    assert_eq!(s["storage"]["size"].as_u64(), Some(was + 5), "the volume itself");
    let (_, v2) = r.get(&format!("/1.3/server/{uuid}"));
    let after = v2["server"]["storage_devices"]["storage_device"][0]["storage_size"].as_u64();
    assert_eq!(after, Some(was + 5), "the DEVICE still reports the old size: {}", v2["server"]["storage_devices"]);
}

/// The numbers UpCloud stringifies must be strings here too, or a client that
/// cannot read `"10"` passes the mock and fails the provider.
#[test]
fn the_inconsistent_stringification_is_reproduced() {
    let r = Rig::new();
    let uuid = r.a_started_server("front");
    let (_, v) = r.get(&format!("/1.3/server/{uuid}"));
    let s = &v["server"];
    assert!(s["core_number"].is_string(), "core_number is a STRING: {s}");
    assert!(s["memory_amount"].is_string(), "{s}");
    // …but NOT a device's `storage_size`. That was this crate's own invention
    // and `UpCloudLtd/upcloud` 5.44.1 refuses it — `cannot unmarshal string
    // into Go struct field …storage_size of type int` — so every terraform
    // server create against the mock died on a shape the account cannot send.
    // See the comment at the render site.
    assert!(
        s["storage_devices"]["storage_device"][0]["storage_size"].is_number(),
        "a device's storage_size is a NUMBER — the official provider's SDK declares it int: {s}"
    );
    let (_, st) = r.get("/1.3/storage/private");
    assert!(st["storages"]["storage"][0]["size"].is_number(), "but a storage's size is a NUMBER");
}

/// A bare `GET /1.3/storage` lists the public templates; `/storage/private`
/// does not. A cleanup that used the former would walk thousands of rows that
/// are not the account's.
#[test]
fn the_public_templates_are_in_the_bare_storage_list() {
    let r = Rig::new();
    let all = r.get("/1.3/storage").1["storages"]["storage"].as_array().unwrap().len();
    let mine = r.get("/1.3/storage/private").1["storages"]["storage"].as_array().unwrap().len();
    assert!(all > mine, "{all} vs {mine}");
}

/// The estate is a state machine, not a script: the same seed produces the same
/// uuids, the same addresses and the same durations, so a failing storm run is
/// replayable from its seed alone.
#[test]
fn a_seed_reproduces_the_estate() {
    let one = Estate::new(Clock::virtual_only(), Faults::seeded(77), 77);
    let two = Estate::new(Clock::virtual_only(), Faults::seeded(77), 77);
    let names = |e: &Estate| {
        e.all_storages()
            .filter(|s| s.kind == StorageKind::Template)
            .map(|s| s.uuid.clone())
            .collect::<Vec<_>>()
    };
    assert_eq!(names(&one), names(&two));
}

// ── 23, 24, 25, 26 — the import, the sync, and the fix that may not be one ──

/// The measured ISO: 43 485 184 bytes.
fn an_iso() -> Vec<u8> {
    // The bytes are not the real ISO's, but the COUNT is, and the digests the
    // mock answers are the real digests of whatever is sent. A test that sent
    // ten bytes would not exercise the upload duration, which is per-MiB.
    let mut v = vec![0u8; 43_485_184];
    for (i, b) in v.iter_mut().enumerate() {
        *b = (i % 251) as u8;
    }
    v
}

/// Open a session, PUT the bytes, and hand back (import json, storage uuid).
fn upload_a_medium(r: &Rig, title: &str) -> (Value, String) {
    let (s, v) = r.post(
        "/1.3/storage",
        serde_json::json!({"storage": {"size": 1, "tier": "maxiops", "title": title, "zone": "se-sto1", "labels": []}}),
    );
    assert_eq!(s, 201, "{v}");
    let u = v["storage"]["uuid"].as_str().unwrap().to_string();
    loop {
        if r.get(&format!("/1.3/storage/{u}")).1["storage"]["state"] == "online" {
            break;
        }
    }
    let (s, sess) = r.post(
        &format!("/1.3/storage/{u}/import"),
        serde_json::json!({"storage_import": {"source": "direct_upload"}}),
    );
    assert_eq!(s, 201, "{sess}");
    let url = sess["storage_import"]["direct_upload_url"].as_str().unwrap().to_string();
    // FOLLOW the url the provider gave, rather than building one: that is what
    // a caller does, and it is the half a hand-built URL never tests.
    let resp = r.http.put(&url).body(an_iso()).send().unwrap();
    assert_eq!(resp.status().as_u16(), 200);
    // The uploader's reply is UN-enveloped (behaviour 67 / ledger L63), and the
    // callers here read it bare.
    let body: Value = serde_json::from_str(&resp.text().unwrap()).unwrap();
    assert!(body.get("storage_import").is_none() && body.get("written_bytes").is_some(), "{body}");
    let im = body;
    (im, u)
}

/// **The import and the storage are two different clocks.**
///
/// The upload finishes in five seconds with every byte accounted for and both
/// checksums present — and then the volume sits in `syncing` for a hundred and
/// something seconds more, while nothing is happening on either side. A poller
/// that watched the IMPORT would call the medium ready twenty times too early.
#[test]
fn b23_the_import_says_completed_while_the_storage_is_still_syncing() {
    let r = Rig::new();
    let (im, u) = upload_a_medium(&r, "gunnar-installer.iso");
    assert_eq!(im["read_bytes"], 43_485_184u64);
    assert_eq!(im["written_bytes"], 43_485_184u64);

    let t0 = r.get("/mock/seed").1["now_ms"].as_u64().unwrap();
    // First: the import completes and the storage enters `syncing`.
    let (mut import_done_at, mut online_at) = (None, None);
    let mut saw_syncing_while_completed = false;
    for _ in 0..500 {
        let now = r.get("/mock/seed").1["now_ms"].as_u64().unwrap();
        let (_, v) = r.get(&format!("/1.3/storage/{u}"));
        let st = v["storage"]["state"].as_str().unwrap().to_string();
        let ist = v["storage"]["storage_import"]["state"].as_str().unwrap_or("").to_string();
        if ist == "completed" && import_done_at.is_none() {
            import_done_at = Some(now);
        }
        if ist == "completed" && st == "syncing" {
            saw_syncing_while_completed = true;
        }
        if st == "online" {
            online_at = Some(now);
            break;
        }
    }
    let import_done_at = import_done_at.expect("the import must complete");
    let online_at = online_at.expect("the storage must eventually come online");
    assert!(
        saw_syncing_while_completed,
        "the whole point: the import says completed while the storage says syncing"
    );

    let upload_took = import_done_at - t0;
    let sync_took = online_at - import_done_at;
    assert!((4_000..=6_500).contains(&upload_took), "the upload is five seconds: {upload_took} ms");
    assert!(
        (95_000..=140_000).contains(&sync_took),
        "and then a hundred-and-something seconds of nothing: {sync_took} ms"
    );
    assert!(sync_took > upload_took * 15, "the WAIT is the cost, not the transfer");
}

/// **The checksums are the real ones.** The ladder compares `sha256sum` with
/// the local file's, so a mock that made one up would make the verification
/// pass without ever having run.
#[test]
fn b24_the_import_carries_the_real_digests_of_the_real_bytes() {
    let r = Rig::new();
    let (im, _) = upload_a_medium(&r, "iso");
    let bytes = an_iso();
    assert_eq!(
        im["sha256sum"].as_str().unwrap(),
        mock_upcloud::digest::sha256_hex(&bytes)
    );
    assert_eq!(
        im["md5sum"].as_str().unwrap(),
        mock_upcloud::digest::md5_hex(&bytes)
    );
    // Changed bytes, changed digest. Without this the test above proves only
    // that two copies of the same function agree.
    let mut other = bytes.clone();
    other[0] ^= 0xff;
    assert_ne!(mock_upcloud::digest::sha256_hex(&other), mock_upcloud::digest::sha256_hex(&bytes));
}

/// **The re-image pays the sync TWICE**, because the same medium goes on once
/// as a CD-ROM and once as a virtio disk — `korp-installer` probes virtio and
/// nothing else. Measured here as the cost the fix would have to beat.
#[test]
fn b25_the_same_medium_is_uploaded_twice_and_waits_twice() {
    let r = Rig::new();
    let t0 = r.get("/mock/seed").1["now_ms"].as_u64().unwrap();
    for title in ["gunnar-installer.iso (cdrom)", "gunnar-installer.iso (virtio)"] {
        let (_, u) = upload_a_medium(&r, title);
        loop {
            if r.get(&format!("/1.3/storage/{u}")).1["storage"]["state"] == "online" {
                break;
            }
        }
    }
    let took = r.get("/mock/seed").1["now_ms"].as_u64().unwrap() - t0;
    assert!(took > 200_000, "two media is over three minutes of waiting: {took} ms");
}

/// **Behaviour 26, corrected to the measurement (behaviour 53).**
///
/// This test used to prove the pessimistic DEFAULT — a clone syncs 100–130 s
/// like an import — on the grounds that the real answer was "NOT MEASURED". It
/// was measured, on 2026-09-20, in gunnar `deploy/upcloud/tests/clone_probe.rs`:
/// the call 728 ms, then `maintenance` → `online` in 47 s, NO `syncing`. The
/// old guess is still reachable, by its own name, and the old name that
/// promised the optimistic guess no longer parses — loudly.
#[test]
fn b26_a_clone_skips_the_sync_and_goes_online_in_forty_seven_seconds() {
    let r = Rig::new();
    let (im, first) = upload_a_medium(&r, "gunnar-installer.iso (cdrom)");
    loop {
        if r.get(&format!("/1.3/storage/{first}")).1["storage"]["state"] == "online" {
            break;
        }
    }
    let sha = im["sha256sum"].as_str().unwrap().to_string();

    let clone_of = |title: &str| -> (u64, bool) {
        let t0 = r.get("/mock/seed").1["now_ms"].as_u64().unwrap();
        let (s, v) = r.post(
            &format!("/1.3/storage/{first}/clone"),
            serde_json::json!({"storage": {"title": title, "tier": "maxiops"}}),
        );
        assert_eq!(s, 201, "{v}");
        let c = v["storage"]["uuid"].as_str().unwrap().to_string();
        let mut saw_syncing = false;
        loop {
            let (_, g) = r.get(&format!("/1.3/storage/{c}"));
            if g["storage"]["state"] == "syncing" {
                saw_syncing = true;
            }
            if g["storage"]["state"] == "online" {
                assert_eq!(g["storage"]["storage_import"]["sha256sum"].as_str().unwrap(), sha, "same bytes");
                assert_eq!(g["storage"]["origin"].as_str().unwrap(), first);
                break;
            }
        }
        (r.get("/mock/seed").1["now_ms"].as_u64().unwrap() - t0, saw_syncing)
    };

    let (took, synced) = clone_of("gunnar-installer.iso (virtio)");
    assert!(!synced, "MEASURED: a clone never enters `syncing`");
    assert!((40_000..=60_000).contains(&took), "MEASURED: online in ~47 s, not {took} ms");

    // The old guess, by its own name only.
    assert_eq!(r.post("/mock/fault/clone-skips-sync/arm", Value::Null).0, 404, "the old name is gone, loudly");
    assert_eq!(r.post("/mock/fault/clone-syncs-like-import/arm", Value::Null).0, 200);
    let (took2, synced2) = clone_of("third");
    assert!(synced2 && took2 > 95_000, "the OLD guess: syncs like an import ({took2} ms)");
}

/// A sync that outlasts the caller's 1200 s budget. A timeout that has never
/// fired is a timeout nobody has read the handler of.
#[test]
fn a_sync_can_exceed_the_callers_budget() {
    let f = Faults::none();
    f.arm(Fault::SyncExceedsBudget);
    let r = Rig::with(f);
    let (_, u) = upload_a_medium(&r, "iso");
    let t0 = r.get("/mock/seed").1["now_ms"].as_u64().unwrap();
    loop {
        if r.get(&format!("/1.3/storage/{u}")).1["storage"]["state"] == "online" {
            break;
        }
    }
    let took = r.get("/mock/seed").1["now_ms"].as_u64().unwrap() - t0;
    assert!(took > 1_200_000, "past the 1200 s budget: {took} ms");
}

/// An import that fails after every byte arrived: the object goes `failed` with
/// a code and a message, and the STORAGE lands in `error` — which a poll loop
/// looking only for `online` will wait its whole budget for.
#[test]
fn an_import_can_fail_after_the_bytes_arrived() {
    let f = Faults::none();
    f.arm(Fault::ImportFailed);
    let r = Rig::with(f);
    let (_, u) = upload_a_medium(&r, "iso");
    let mut state = String::new();
    for _ in 0..200 {
        let (_, v) = r.get(&format!("/1.3/storage/{u}"));
        state = v["storage"]["state"].as_str().unwrap().to_string();
        if state == "error" {
            assert_eq!(v["storage"]["storage_import"]["state"], "failed", "{v}");
            assert_eq!(v["storage"]["storage_import"]["error_code"], "IMPORT_FAILED", "{v}");
            assert!(v["storage"]["storage_import"]["error_message"].is_string(), "{v}");
            // Every byte arrived. The failure is AFTER the transfer, which is
            // why a caller that checks read_bytes is still wrong.
            assert_eq!(v["storage"]["storage_import"]["read_bytes"], 43_485_184u64, "{v}");
            return;
        }
    }
    panic!("the storage never reached error; it is {state}");
}

// ── 19, 18, 27, 28 — the clock, and the six reds it causes ──────────────────

/// **The appliance's clock is one timezone offset ahead and the front's is
/// not**, over the wire, from a hypervisor that correctly reports `UTC`.
///
/// That last clause is the finding. `GET /1.3/server/{uuid}` answers
/// `"timezone": "UTC"` for both boxes, truthfully, and one of them is still two
/// hours out — because the skew is not the hypervisor's, it is the guest's
/// userland failing to establish what the RTC holds. A mock that skewed its own
/// clock would let a fix that subtracts two hours pass, which is not a fix.
#[test]
fn b19_the_appliance_misreads_the_rtc_and_the_front_does_not() {
    let f = Faults::none();
    f.arm(Fault::GuestReadsRtcAsLocalTime);
    let r = Rig::with(f);
    let appliance = r.a_started_server("appliance");
    // Disarm, and the NEXT server created is a front: same hypervisor, same
    // zone, minutes later, and correct.
    r.post("/mock/fault/guest-reads-rtc-as-local-time/disarm", Value::Null);
    let front = r.a_started_server("front");

    for u in [&appliance, &front] {
        let (_, v) = r.get(&format!("/1.3/server/{u}"));
        assert_eq!(v["server"]["timezone"], "UTC", "the hypervisor tells the truth: {v}");
    }

    let (_, a) = r.get(&format!("/mock/clock/{appliance}"));
    let (_, f2) = r.get(&format!("/mock/clock/{front}"));
    assert_eq!(a["guest_clock_skew_ms"], 7_200_000i64, "measured +7 198 668 ms; the mechanism is exactly 2 h: {a}");
    assert_eq!(f2["guest_clock_skew_ms"], 0i64, "same hypervisor, same zone, correct: {f2}");
    assert_eq!(a["hypervisor_rtc"], "UTC");
    assert_eq!(a["guest_reads_rtc_as"], "LocalTime");
    assert_eq!(f2["guest_reads_rtc_as"], "Utc");
}

/// **No UDP reply ever comes back**, so the box cannot fix itself with NTP.
#[test]
fn b18_no_udp_reply_comes_back_so_ntp_is_useless() {
    let r = Rig::new();
    let u = r.a_started_server("appliance");
    let (_, v) = r.get(&format!("/mock/clock/{u}"));
    assert_eq!(v["udp_reply_arrives"], false, "default ON: it is the provider's normal, not an exception");
}

/// **One wrong clock is six red rows**, in the order of causation, and the last
/// one (`monetize-poll: tenants=0`) is the one you notice and the wrong one to
/// chase.
#[test]
fn b27_one_wrong_clock_is_six_reds_in_causal_order() {
    let f = Faults::none();
    f.arm(Fault::GuestReadsRtcAsLocalTime);
    let r = Rig::with(f);
    let u = r.a_started_server("appliance");
    let (_, v) = r.get(&format!("/mock/clock/{u}"));
    let rows: Vec<String> = v["red"]
        .as_array()
        .unwrap()
        .iter()
        .map(|x| x["row"].as_str().unwrap().to_string())
        .collect();
    assert_eq!(
        rows,
        vec!["skew-refusal", "no-quorum", "clock-unchanged", "console-key", "banner", "monetize-poll"],
        "{v}"
    );
    let console = v["red"].as_array().unwrap().iter().find(|x| x["row"] == "console-key").unwrap();
    assert!(
        console["why"].as_str().unwrap().contains("the window is ±300000 ms"),
        "the measured sentence, verbatim: {console}"
    );
    // A correct box has no reds at all, or the cascade would be weather.
    r.post("/mock/fault/guest-reads-rtc-as-local-time/disarm", Value::Null);
    let front = r.a_started_server("front");
    let (_, v) = r.get(&format!("/mock/clock/{front}"));
    assert_eq!(v["red"].as_array().unwrap().len(), 0, "{v}");
}

// ── 29, 30 — the two asymmetries, which is why both looked like something else

/// Find two started servers whose utility addresses are in DIFFERENT /22s.
/// The address pool spans both and is shuffled per lay, so which pair straddles
/// varies — and that is the point of behaviour 12, so the test looks rather than
/// assumes.
fn a_straddling_pair(r: &Rig, first: &str, rest: &[&str]) -> (String, String) {
    let a = r.a_started_server(first);
    let net_of = |u: &str| {
        let (_, v) = r.get(&format!("/mock/dhcp/{u}"));
        let ip = v["address"].as_str().unwrap().to_string();
        (ip.clone(), mock_upcloud::net::net_of(&ip).unwrap().to_string())
    };
    let (_, a_net) = net_of(&a);
    for t in rest {
        let b = r.a_started_server(t);
        let (_, b_net) = net_of(&b);
        if b_net != a_net {
            return (a, b);
        }
    }
    panic!("the utility pool must span two /22s or behaviour 29 is unreachable");
}

/// **The afternoon, in one test.**
///
/// The appliance's utility NIC gets its address by DHCP. The front is on a
/// different /22. The route between them arrives ONLY as classless static
/// routes — option 121 — and the offer carries no default gateway at all. A
/// guest that ignores the option comes up with a good address, answers
/// everything sent to it, and cannot reach the front. Its clock sync dies
/// outbound while every inbound probe says it is healthy, which reads exactly
/// like a two-hour clock bug and is not one.
#[test]
fn b29_a_guest_that_ignores_option_121_dies_outbound_and_looks_healthy_inbound() {
    let f = Faults::none();
    f.arm(Fault::GuestIgnoresDhcpOption121);
    let r = Rig::with(f);
    // The appliance is created while the fault is armed; everything after it is
    // an ordinary guest.
    let appliance = r.a_started_server("appliance");
    r.post("/mock/fault/guest-ignores-dhcp-option-121/disarm", Value::Null);

    let (_, offer) = r.get(&format!("/mock/dhcp/{appliance}"));
    assert!(offer["router"].is_null(), "no default gateway: that is why option 121 is load-bearing: {offer}");
    assert_eq!(offer["option_121"].as_array().unwrap().len(), 1, "the offer IS complete: {offer}");
    assert_eq!(offer["guest_dhcp_client"], "IgnoresOption121", "…and the guest is what drops it: {offer}");

    // A front on the other /22.
    let a_ip = offer["address"].as_str().unwrap().to_string();
    let a_net = mock_upcloud::net::net_of(&a_ip).unwrap().to_string();
    let mut front = None;
    for t in ["front-a", "front-b", "front-c", "front-d", "front-e"] {
        let u = r.a_started_server(t);
        let (_, o) = r.get(&format!("/mock/dhcp/{u}"));
        let ip = o["address"].as_str().unwrap().to_string();
        if mock_upcloud::net::net_of(&ip).unwrap().to_string() != a_net {
            front = Some((u, ip));
            break;
        }
    }
    let (front_uuid, front_ip) = front.expect("the pool spans two /22s");

    // OUTBOUND: nowhere to go.
    let (_, out) = r.get(&format!("/mock/reach/{appliance}?dest={front_ip}&port=443"));
    assert_eq!(out["ok"], false, "{out}");
    assert_eq!(out["kind"], "no-route-outbound", "{out}");
    assert!(out["why"].as_str().unwrap().contains("did not install it"), "{out}");
    // INBOUND: perfectly healthy, which is the whole diagnostic difficulty.
    assert_eq!(out["inbound_ok"], true, "every health check says this box is up: {out}");

    // The front, on the same estate, with a client that reads the option:
    // reaches the appliance without trouble.
    let (_, back) = r.get(&format!("/mock/reach/{front_uuid}?dest={a_ip}&port=443"));
    assert_eq!(back["ok"], true, "{back}");
}

/// **Hairpin NAT does not exist.** The front DNATs `:2222` to the appliance.
/// From outside it works. From the front itself, to its own public address, it
/// is `connection refused` — locally-generated traffic never traverses
/// `prerouting`. A healthy forge was diagnosed as broken on exactly this.
#[test]
fn b30_the_box_holding_the_dnat_cannot_use_it_on_itself() {
    let r = Rig::new();
    let (front, appliance) = a_straddling_pair(&r, "front", &["appliance-a", "appliance-b", "appliance-c", "appliance-d"]);
    let (_, av) = r.get(&format!("/1.3/server/{appliance}"));
    let a_util = av["server"]["ip_addresses"]["ip_address"]
        .as_array()
        .unwrap()
        .iter()
        .find(|x| x["access"] == "utility")
        .unwrap()["address"]
        .as_str()
        .unwrap()
        .to_string();
    let (_, fv) = r.get(&format!("/1.3/server/{front}"));
    let f_pub = fv["server"]["ip_addresses"]["ip_address"]
        .as_array()
        .unwrap()
        .iter()
        .find(|x| x["access"] == "public")
        .unwrap()["address"]
        .as_str()
        .unwrap()
        .to_string();

    assert_eq!(r.post(&format!("/mock/dnat/{front}/2222/{a_util}"), Value::Null).0, 200);

    // The front, to its own public address: refused.
    let (_, own) = r.get(&format!("/mock/reach/{front}?dest={f_pub}&port=2222"));
    assert_eq!(own["kind"], "no-hairpin", "{own}");
    assert!(own["why"].as_str().unwrap().contains("connection refused"), "the kernel's own words: {own}");
    assert!(own["why"].as_str().unwrap().contains("prerouting"), "and the reason: {own}");

    // Anyone else, to the same address and port: through. Same rule, same
    // instant, opposite answer.
    let (_, outside) = r.get(&format!("/mock/reach/{appliance}?dest={f_pub}&port=2222"));
    assert_eq!(outside["ok"], true, "{outside}");
}

// ── 31 — the console is a HOST and a port ───────────────────────────────────

/// **The console endpoint is a pair, and it must be re-read as a pair.**
///
/// MEASURED: the API said `remote_access_enabled=false` and the banner was
/// dead; two PUTs, one field each, nothing stopped or rebooted; and the console
/// came back as `se-sto1.vnc.upcloud.com:60031` — a ZONE host, not the server's
/// own address, and a five-figure port.
///
/// Whether the HOST changes between provisionings is not something this estate
/// has measured, so the mock does not claim it does. What it does claim, and
/// what this asserts, is that the pair is the provider's to hand out: a client
/// that constructs the host itself, or re-reads only the port after the cure,
/// is dialling somewhere of its own invention.
#[test]
fn b31_the_console_is_a_zone_host_and_a_port_and_both_are_the_providers() {
    let r = Rig::new();
    let u = r.a_started_server("appliance");

    // Remote access off: the API reports NEITHER half. A stale endpoint
    // answered confidently is what makes a dead console look alive.
    let (_, v) = r.get(&format!("/1.3/server/{u}"));
    assert_eq!(v["server"]["remote_access_enabled"], "no");
    assert!(v["server"]["remote_access_host"].is_null(), "no stale host: {v}");
    assert!(v["server"]["remote_access_port"].is_null(), "no stale port: {v}");

    // Two PUTs, one field each, nothing stopped or rebooted.
    r.put(&format!("/1.3/server/{u}"), serde_json::json!({"server": {"remote_access_enabled": "yes"}}));
    r.put(&format!("/1.3/server/{u}"), serde_json::json!({"server": {"remote_access_password": "hunter2"}}));
    let (_, v) = r.get(&format!("/1.3/server/{u}"));
    let host = v["server"]["remote_access_host"].as_str().unwrap().to_string();
    let port: u16 = v["server"]["remote_access_port"].as_str().unwrap().parse().unwrap();
    assert_eq!(host, "se-sto1.vnc.mock.invalid", "the ZONE's console (RFC 2606 in the mock), not the server's own address: {v}");
    assert!(!host.contains(&u[..8]), "a client that builds the host from the uuid is wrong: {host}");
    assert!((60_000..61_000).contains(&port), "five figures, as measured: {port}");

    // Behaviour 14 still holds, and now over the PAIR: a stop/start leaves both
    // halves stale, and the toggle reconciles both.
    r.post(&format!("/1.3/server/{u}/stop"), serde_json::json!({"stop_server": {"stop_type": "hard", "timeout": "60"}}));
    r.poll_until(&u, "stopped");
    r.post(&format!("/1.3/server/{u}/start"), Value::Null);
    r.poll_until(&u, "started");
    let (_, v) = r.get(&format!("/1.3/server/{u}"));
    assert_eq!(v["server"]["remote_access_port"].as_str().unwrap().parse::<u16>().unwrap(), port, "stale, as before");
    r.put(&format!("/1.3/server/{u}"), serde_json::json!({"server": {"remote_access_enabled": "no"}}));
    r.post("/mock/advance/2000", Value::Null);
    r.put(&format!("/1.3/server/{u}"), serde_json::json!({"server": {"remote_access_enabled": "yes"}}));
    let (_, v) = r.get(&format!("/1.3/server/{u}"));
    assert_ne!(
        v["server"]["remote_access_port"].as_str().unwrap().parse::<u16>().unwrap(),
        port,
        "the cure re-provisions the endpoint: {v}"
    );
    assert!(v["server"]["remote_access_host"].is_string(), "and hands back a host with it: {v}");
}

// ── 32 — two observers, two orders of magnitude ─────────────────────────────

/// **The installer is not slow; the provider is.**
///
/// MEASURED across three re-images: `INSTALL-OK` at 2 001, 2 751 and 4 752 ms
/// on the UART, against a ladder `install-time` of 167 s and 253 s wall. Both
/// numbers are right and they measure different things — the guest's own
/// install, and everything the provider does around it (create, media sync,
/// firmware, boot order, DHCP).
///
/// A bar set at 60 s against the ladder's number and named "install" reads RED
/// forever and tempts somebody to move the bar. The mock reports both, named by
/// observer, so the thing to argue about is which number the bar is for.
#[test]
fn b32_the_guests_install_and_the_ladders_wall_are_two_orders_of_magnitude_apart() {
    let r = Rig::new();
    let u = r.a_started_server("appliance");
    let (_, v) = r.get(&format!("/mock/reimage/{u}"));
    let uart = v["guest_uart_ms"].as_u64().unwrap();
    let wall = v["ladder_wall_ms"].as_u64().unwrap();
    assert!((2_001..=4_752).contains(&uart), "the guest's own clock: {v}");
    assert!((110_000..=253_000).contains(&wall), "the ladder's wall clock: {v}");
    assert!(wall > uart * 20, "two orders of magnitude, and it is the provider's: {v}");
    assert!(v["observers"]["guest_uart_ms"].is_string(), "each number names who measured it: {v}");
    assert!(v["observers"]["ladder_wall_ms"].is_string(), "{v}");
    // A 60-second bar against the wall number is unpassable, and that is a fact
    // about the provider rather than about the installer.
    assert!(wall > 60_000, "a 60 s bar reads RED forever against this observer: {wall}");
}

// ── 33 — a new host key every re-image, and the three paths that vet it ─────

/// **A re-image mints a new SSH host key, every time**, so
/// `REMOTE HOST IDENTIFICATION HAS CHANGED` is expected and cannot be alarming
/// on its own. What separates a re-imaged machine from a hijacked name is that
/// the SAME key answers on the name, on the front's public address, and on the
/// appliance's own public address bypassing the DNAT. Three paths, one key.
#[test]
fn b33_three_paths_one_key_and_a_reimage_changes_it() {
    let r = Rig::new();
    let u = r.a_started_server("appliance");
    let (s, v) = r.get(&format!("/mock/hostkey/{u}"));
    assert_eq!(s, 200, "{v}");
    assert_eq!(v["agree"], true, "{v}");
    let first = v["paths"]["name"].as_str().unwrap().to_string();
    assert_eq!(v["paths"]["front"].as_str().unwrap(), first);
    assert_eq!(v["paths"]["direct"].as_str().unwrap(), first);
    assert!(first.starts_with("SHA256:"), "{first}");

    // Boot it again — a fresh install, a fresh key. This is the line every
    // bring-up trips over, and it is not an attack.
    r.post(&format!("/1.3/server/{u}/stop"), serde_json::json!({"stop_server": {"stop_type": "hard", "timeout": "60"}}));
    r.poll_until(&u, "stopped");
    r.post(&format!("/1.3/server/{u}/start"), Value::Null);
    r.poll_until(&u, "started");
    let (_, v) = r.get(&format!("/mock/hostkey/{u}"));
    assert_ne!(v["paths"]["direct"].as_str().unwrap(), first, "a re-image mints a new key: {v}");
    assert_eq!(v["agree"], true, "…and all three paths still agree, so it is the same machine: {v}");
}

/// **And this is what the three paths are FOR.** With a name answering for a
/// machine that is not behind the DNAT, the `name` path disagrees with the
/// other two — and a verifier that looked at one path would have accepted it,
/// because a changed key is exactly what a legitimate re-image produces.
#[test]
fn b33b_a_hijacked_name_is_caught_only_because_three_paths_are_checked() {
    let f = Faults::none();
    f.arm(Fault::HijackedName);
    let r = Rig::with(f);
    let u = r.a_started_server("appliance");
    let (_, v) = r.get(&format!("/mock/hostkey/{u}"));
    assert_eq!(v["agree"], false, "{v}");
    assert_ne!(v["paths"]["name"], v["paths"]["direct"], "{v}");
    assert_eq!(v["paths"]["front"], v["paths"]["direct"], "only the NAME is lying: {v}");
    assert!(v["verdict"].as_str().unwrap().contains("not behind the DNAT"), "{v}");
}

// ── 35, 36 — and the shape that needs no fault ──────────────────────────────

/// **The firewall of a server that READS FINE answers the same 403 as a dead
/// one's.** Behaviour 1's body, byte for byte, beside a 200 on the server's
/// own detail: a credential without the firewall permission. The assertion is
/// that the two replies are indistinguishable at the firewall endpoint — same
/// `type`, a `correlation_id` in both — so a caller that decides off this
/// endpoint alone cannot tell "forbidden" from "gone".
#[test]
fn b35_a_live_servers_firewall_answers_403_under_a_scoped_credential() {
    let f = Faults::none();
    f.arm(Fault::FirewallForbidden);
    let r = Rig::with(f);
    let uuid = r.a_started_server("front");

    let (s, v) = r.get(&format!("/1.3/server/{uuid}"));
    assert_eq!(s, 200, "the server itself reads fine: {v}");
    assert_eq!(v["server"]["uuid"], uuid.as_str());
    let (s, fw) = r.get(&format!("/1.3/server/{uuid}/firewall_rule"));
    assert_eq!(s, 403, "{fw}");
    assert!(fw["type"].as_str().unwrap().ends_with("ERROR_AUTHENTICATION_FAILED"), "{fw}");
    assert!(fw["correlation_id"].is_string(), "{fw}");

    // And the list still carries it: nothing about the account says "gone".
    let (_, list) = r.get("/1.3/server");
    let listed: Vec<&str> = list["servers"]["server"].as_array().unwrap().iter().filter_map(|s| s["uuid"].as_str()).collect();
    assert_eq!(listed, vec![uuid.as_str()]);

    // The same body a DEAD server's firewall answers (behaviour 1), so the
    // only thing that separates 34 from 1 is the server's own detail.
    let (s, dead) = r.get("/1.3/server/00000000-0000-4000-8000-000000000000/firewall_rule");
    assert_eq!(s, 403);
    assert_eq!(dead["type"], fw["type"]);
    assert_eq!(dead["title"], fw["title"]);
}

/// **The detail says 404 for a uuid the list carries.** The account
/// contradicting itself; same `SERVER_NOT_FOUND` envelope as a genuine 404, so
/// only a caller holding the list beside the detail can name it.
#[test]
fn b36_the_detail_answers_404_for_a_server_the_list_shows() {
    let r = Rig::new();
    let uuid = r.a_started_server("appliance");
    // Armed AFTER the server is up: the create's own poll reads the detail,
    // and a contradiction that fired during it would read as "vanished".
    r.post("/mock/fault/detail-404-for-listed-server/arm", Value::Null);

    let (_, list) = r.get("/1.3/server");
    let listed: Vec<&str> = list["servers"]["server"].as_array().unwrap().iter().filter_map(|s| s["uuid"].as_str()).collect();
    assert_eq!(listed, vec![uuid.as_str()], "the list shows it");

    let (s, v) = r.get(&format!("/1.3/server/{uuid}"));
    assert_eq!(s, 404, "{v}");
    assert_eq!(v["error"]["error_code"], "SERVER_NOT_FOUND", "{v}");

    // A uuid that was never created answers the identical envelope: the fault
    // adds no tell of its own, which is what makes it worth having.
    let (s, never) = r.get("/1.3/server/00000000-0000-4000-8000-000000000000");
    assert_eq!(s, 404);
    assert_eq!(never["error"]["error_code"], v["error"]["error_code"]);
}

/// **The 2026-09-20 shape needs no fault.** A state that names a uuid the
/// account never created (a stale state after a re-lay) reads 404 on the
/// server and 403 on its firewall, while the list shows the SAME TITLES alive
/// under other uuids. That pair is exactly what terraform saw, and a plan off
/// it would create a second gunnar-front beside the live one.
#[test]
fn b01b_a_stale_uuid_reads_404_and_its_firewall_403_while_the_title_is_alive_under_another_uuid() {
    let r = Rig::new();
    let live = r.a_started_server("gunnar-front");
    let stale = "0031e940-3f7d-4116-bc74-b8131f6d07a3";
    assert_ne!(live, stale);

    let (s, v) = r.get(&format!("/1.3/server/{stale}"));
    assert_eq!(s, 404, "{v}");
    assert_eq!(v["error"]["error_code"], "SERVER_NOT_FOUND");
    let (s, fw) = r.get(&format!("/1.3/server/{stale}/firewall_rule"));
    assert_eq!(s, 403, "{fw}");
    assert!(fw["correlation_id"].is_string(), "{fw}");

    let (_, list) = r.get("/1.3/server");
    let rows = list["servers"]["server"].as_array().unwrap();
    assert_eq!(rows.len(), 1);
    assert_eq!(rows[0]["title"], "gunnar-front");
    assert_eq!(rows[0]["uuid"], live.as_str(), "alive, and NOT under the state's uuid");
}

// ── 37 ───────────────────────────────────────────────────────────────────────

/// **A sold-out zone refuses the CREATE, not only the poweron — and the mock
/// can report both answers at the same door.**
///
/// Scaleway's `fr-par-1` (verified against Scaleway, not UpCloud; kept as a
/// generic cloud-provider fault) answered `412 out_of_stock` to `poweron` from 2026-09-09 and had
/// not cleared by 2026-09-20. What is out of stock is CAPACITY FOR A PLAN IN A
/// ZONE, and a create asks for exactly that, first. The mock served only the
/// poweron door until now, so a caller that met the outage at the create could
/// not be driven against it at all.
///
/// The control is the point: the SAME `POST /1.3/server` is a `201` with the
/// fault disarmed and a `412` with it armed, and the account is untouched in
/// the second case. A test that only asserted the 412 would pass against a mock
/// that refused every create.
#[test]
fn b37_a_sold_out_zone_refuses_the_create_and_the_same_call_succeeds_when_it_is_not() {
    let body = serde_json::json!({"server": {
        "zone": "se-sto1", "title": "gunnar-front", "hostname": "gunnar-front", "plan": "2xCPU-4GB",
        "storage_devices": {"storage_device": [{"action": "create", "title": "boot", "size": 20, "tier": "maxiops"}]}
    }});

    // CONTROL — disarmed. The create is an ordinary 201 and the account holds it.
    let r = Rig::new();
    let (s, v) = r.post("/1.3/server", body.clone());
    assert_eq!(s, 201, "with the zone in stock this is a create like any other: {v}");
    assert!(v["server"]["uuid"].is_string(), "{v}");
    assert_eq!(r.get("/1.3/server").1["servers"]["server"].as_array().unwrap().len(), 1);

    // ARMED — the same call, the same body, refused before anything is minted.
    let r = Rig::new();
    r.post("/mock/fault/out-of-stock/arm", Value::Null);
    let (s, v) = r.post("/1.3/server", body.clone());
    assert_eq!(s, 412, "the create needs the capacity too: {v}");
    assert_eq!(v["error"]["error_code"], "out_of_stock", "{v}");

    // And it minted NOTHING: no half a server, no orphan boot disk to pay for.
    assert_eq!(r.get("/1.3/server").1["servers"]["server"].as_array().unwrap().len(), 0);
    assert_eq!(r.get("/1.3/storage/private").1["storages"]["storage"].as_array().unwrap().len(), 0);

    // Sticky, at this door as at the other: a retry loop cannot outwait it.
    for attempt in 0..25 {
        let (s, v) = r.post("/1.3/server", body.clone());
        assert_eq!(s, 412, "attempt {attempt}: {v}");
    }

    // One fault, both doors: disarm it and the create goes through, and so does
    // the poweron that follows it.
    r.post("/mock/fault/out-of-stock/disarm", Value::Null);
    let (s, v) = r.post("/1.3/server", body);
    assert_eq!(s, 201, "{v}");
    let uuid = v["server"]["uuid"].as_str().unwrap().to_string();
    r.poll_until(&uuid, "started");
}

// ── 38 ───────────────────────────────────────────────────────────────────────

/// **A detach that answers 200 and leaves the volume attached — and the
/// instrument can report BOTH outcomes at the same door.**
///
/// This test is written as a CONTROL and a case on purpose, and the control is
/// the half that does the work. With the fault DISARMED the detach 200s and the
/// device is GONE from `GET /1.3/server/{uuid}`. With it ARMED the detach 200s
/// — the same status, the same reply shape — and the device is STILL THERE.
/// One assertion about the armed case alone would pass against a mock that had
/// no removal path at all, which is exactly the way a green gets printed by an
/// instrument that cannot say anything else.
#[test]
fn b38_a_detach_can_answer_success_and_leave_the_volume_attached() {
    // ── CONTROL: disarmed. The detach really detaches. ──────────────────────
    let r = Rig::new();
    let uuid = r.a_started_server("gunnar-twin");
    let member = attach_a_member(&r, &uuid, "twin-data");
    let before = devices(&r, &uuid);
    assert!(before.contains(&member.clone()), "the member is on the server to start with: {before:?}");

    r.post(&format!("/1.3/server/{uuid}/stop"), serde_json::json!({"stop_server": {"stop_type": "soft", "timeout": "60"}}));
    r.poll_until(&uuid, "stopped");
    let (s, v) = r.post(
        &format!("/1.3/server/{uuid}/storage/detach"),
        serde_json::json!({"storage_device": {"address": "virtio:1"}}),
    );
    assert_eq!(s, 200, "{v}");
    let after = devices(&r, &uuid);
    assert!(!after.contains(&member), "disarmed, the read-back shows it GONE: {after:?}");
    assert_eq!(after.len(), before.len() - 1);

    // ── THE CASE: armed. The same 200, and the read-back disagrees with it. ──
    let r = Rig::new();
    let uuid = r.a_started_server("gunnar-twin");
    let member = attach_a_member(&r, &uuid, "twin-data");
    r.post(&format!("/1.3/server/{uuid}/stop"), serde_json::json!({"stop_server": {"stop_type": "soft", "timeout": "60"}}));
    r.poll_until(&uuid, "stopped");
    r.post("/mock/fault/detach-says-success/arm", Value::Null);

    let (s, v) = r.post(
        &format!("/1.3/server/{uuid}/storage/detach"),
        serde_json::json!({"storage_device": {"address": "virtio:1"}}),
    );
    assert_eq!(s, 200, "the lie is a clean success, not an error: {v}");
    assert!(v["server"]["uuid"].is_string(), "and it carries the server object, like the real one: {v}");
    let after = devices(&r, &uuid);
    assert!(after.contains(&member), "armed, the volume is STILL ATTACHED: {after:?}");

    // The refusals above it still hold — this is "the detach that would have
    // worked lied", not "detach is broken".
    let (s, v) = r.post(
        &format!("/1.3/server/{uuid}/storage/detach"),
        serde_json::json!({"storage_device": {"address": "virtio:9"}}),
    );
    assert_eq!(s, 404, "an address that is not there is still a 404: {v}");
    assert_eq!(v["error"]["error_code"], "STORAGE_DEVICE_NOT_FOUND", "{v}");

    // Not sticky: disarm it and the very next detach is honest again, which is
    // the read-back-and-retry path the fault exists to exercise.
    r.post("/mock/fault/detach-says-success/disarm", Value::Null);
    let (s, v) = r.post(
        &format!("/1.3/server/{uuid}/storage/detach"),
        serde_json::json!({"storage_device": {"address": "virtio:1"}}),
    );
    assert_eq!(s, 200, "{v}");
    assert!(!devices(&r, &uuid).contains(&member), "and this time it is gone");

    // It is never weather: a write that lies about itself is asked for by name.
    assert!(!mock_upcloud::Faults::seeded(7).is_armed(Fault::DetachSaysSuccessButStaysAttached));
}

/// Create a member volume and hot-plug it onto a started server, which is what
/// a growth does. Returns its uuid.
fn attach_a_member(r: &Rig, server: &str, title: &str) -> String {
    let (s, v) = r.post(
        "/1.3/storage",
        serde_json::json!({"storage": {"title": title, "size": 10, "tier": "maxiops", "zone": "se-sto1"}}),
    );
    assert_eq!(s, 201, "{v}");
    let uuid = v["storage"]["uuid"].as_str().unwrap().to_string();
    for _ in 0..200 {
        if r.get(&format!("/1.3/storage/{uuid}")).1["storage"]["state"] == "online" {
            break;
        }
    }
    let (s, v) = r.post(
        &format!("/1.3/server/{server}/storage/attach"),
        serde_json::json!({"storage_device": {"storage": uuid, "type": "disk"}}),
    );
    assert_eq!(s, 200, "{v}");
    uuid
}

/// The storage uuids `GET /1.3/server/{uuid}` says are attached. The read-back
/// that tells a real detach from one that only said so.
fn devices(r: &Rig, server: &str) -> Vec<String> {
    r.get(&format!("/1.3/server/{server}")).1["server"]["storage_devices"]["storage_device"]
        .as_array()
        .unwrap()
        .iter()
        .filter_map(|d| d["storage"].as_str().map(str::to_string))
        .collect()
}