mock-upcloud 0.1.3

//! **Behaviours 39–63: what the estate knew about UpCloud and the mock did not.**
//!
//! Each test here is one row of `BEHAVIOURS-LEDGER.md` that was missing from
//! the mock, or wrong in it. They are written against the HTTP face and the
//! `/mock/` door only, so the same file compiles against the mock as it was
//! before lane T13 — and there every one of them fails. That is the proof each
//! is a behaviour the mock gained, not a restatement of one it had.

use serde_json::{json, Value};

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

impl Rig {
    fn new() -> Rig {
        let estate = mock_upcloud::Estate::new(mock_upcloud::Clock::virtual_only(), mock_upcloud::Faults::none(), 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_upcloud::Mock::new(estate), "127.0.0.1:0").await.unwrap();
            p
        });
        Rig { base: format!("http://127.0.0.1:{port}"), http: reqwest::blocking::Client::new(), _rt: rt }
    }

    fn call(&self, method: reqwest::Method, path: &str, body: Option<Value>) -> (u16, Value) {
        let mut rq = self.http.request(method, format!("{}{path}", self.base)).bearer_auth("ucat_mock");
        if let Some(b) = body {
            rq = rq.json(&b);
        }
        let r = rq.send().unwrap();
        let s = r.status().as_u16();
        (s, serde_json::from_str(&r.text().unwrap()).unwrap_or(Value::Null))
    }
    fn get(&self, p: &str) -> (u16, Value) {
        self.call(reqwest::Method::GET, p, None)
    }
    fn post(&self, p: &str, b: Value) -> (u16, Value) {
        self.call(reqwest::Method::POST, p, Some(b))
    }
    fn put(&self, p: &str, b: Value) -> (u16, Value) {
        self.call(reqwest::Method::PUT, p, Some(b))
    }
    fn delete(&self, p: &str) -> (u16, Value) {
        self.call(reqwest::Method::DELETE, p, None)
    }
    fn arm(&self, fault: &str) {
        assert_eq!(self.post(&format!("/mock/fault/{fault}/arm"), Value::Null).0, 200, "no such fault: {fault}");
    }
    fn now(&self) -> u64 {
        self.get("/mock/seed").1["now_ms"].as_u64().unwrap()
    }
    fn state(&self, uuid: &str) -> String {
        self.get(&format!("/1.3/server/{uuid}")).1["server"]["state"].as_str().unwrap_or("").to_string()
    }
    fn sstate(&self, uuid: &str) -> String {
        self.get(&format!("/1.3/storage/{uuid}")).1["storage"]["state"].as_str().unwrap_or("").to_string()
    }
    fn until(&self, uuid: &str, want: &str) {
        for _ in 0..400 {
            if self.state(uuid) == want {
                return;
            }
        }
        panic!("server {uuid} never reached {want} (is {})", self.state(uuid));
    }
    fn until_storage(&self, uuid: &str, want: &str) {
        for _ in 0..400 {
            if self.sstate(uuid) == want {
                return;
            }
        }
        panic!("storage {uuid} never reached {want} (is {})", self.sstate(uuid));
    }
    fn server_with(&self, title: &str, extra: Value) -> String {
        let mut body = json!({"server": {
            "zone": "se-sto1", "title": title, "hostname": title, "plan": "2xCPU-4GB",
            "storage_devices": {"storage_device": [{"action": "create", "title": format!("{title}-boot"), "size": 20, "tier": "maxiops"}]}
        }});
        if let (Some(b), Some(x)) = (body["server"].as_object_mut(), extra.as_object()) {
            for (k, v) in x {
                b.insert(k.clone(), v.clone());
            }
        }
        let (s, v) = self.post("/1.3/server", body);
        assert_eq!(s, 201, "{v}");
        let u = v["server"]["uuid"].as_str().unwrap().to_string();
        self.until(&u, "started");
        u
    }
    fn server(&self, title: &str) -> String {
        self.server_with(title, json!({}))
    }
    fn stop(&self, uuid: &str) {
        self.post(&format!("/1.3/server/{uuid}/stop"), json!({"stop_server": {"stop_type": "hard"}}));
        self.until(uuid, "stopped");
    }
    fn volume(&self, title: &str, gib: u64, tier: &str) -> String {
        let (s, v) = self.post(
            "/1.3/storage",
            json!({"storage": {"size": gib, "tier": tier, "title": title, "zone": "se-sto1", "labels": []}}),
        );
        assert_eq!(s, 201, "{v}");
        let u = v["storage"]["uuid"].as_str().unwrap().to_string();
        self.until_storage(&u, "online");
        u
    }
    fn attach(&self, server: &str, storage: &str, address: Option<&str>) -> (u16, Value) {
        let mut d = json!({"storage_device": {"type": "disk", "storage": storage}});
        if let Some(a) = address {
            d["storage_device"]["address"] = json!(a);
        }
        self.post(&format!("/1.3/server/{server}/storage/attach"), d)
    }
    fn addresses(&self, server: &str) -> Vec<String> {
        let (_, v) = self.get(&format!("/1.3/server/{server}"));
        v["server"]["storage_devices"]["storage_device"]
            .as_array()
            .unwrap()
            .iter()
            .map(|d| d["address"].as_str().unwrap().to_string())
            .collect()
    }
    fn guest(&self, uuid: &str) -> String {
        let (_, est) = self.get("/mock/estate");
        let me = est["servers"].as_array().unwrap().iter().find(|s| s["uuid"] == uuid).unwrap().clone();
        me["guest"].as_str().unwrap().to_string()
    }
}

fn code(v: &Value) -> &str {
    v["error"]["error_code"].as_str().unwrap_or("")
}

// ── 39 ───────────────────────────────────────────────────────────────────────

/// An empty segment is a 404 at UpCloud, naming no path (gunnar `api.rs`).
#[test]
fn b39_a_doubled_slash_is_a_404_not_a_list() {
    let r = Rig::new();
    let (s, v) = r.get("/1.3//server");
    assert_eq!(s, 404, "{v}");
    assert_ne!(code(&v), "MOCK_UPCLOUD_NOT_IMPLEMENTED", "UpCloud's own 404, not the mock's: {v}");
    assert!(!v.to_string().contains("/1.3//server"), "and it names no path: {v}");
}

// ── 40 ───────────────────────────────────────────────────────────────────────

#[test]
fn b40_a_read_can_answer_502_and_a_write_still_works() {
    let r = Rig::new();
    r.arm("read-bad-gateway");
    assert_eq!(r.get("/1.3/server").0, 502);
    let (s, _) = r.post("/1.3/storage", json!({"storage": {"size": 1, "tier": "maxiops", "title": "x", "zone": "se-sto1"}}));
    assert_eq!(s, 201, "only READS wobble this way");
}

// ── 41 ───────────────────────────────────────────────────────────────────────

#[test]
fn b41_a_server_put_answers_202() {
    let r = Rig::new();
    let u = r.server("front");
    let (s, v) = r.put(&format!("/1.3/server/{u}"), json!({"server": {"boot_order": "disk"}}));
    assert_eq!(s, 202, "MEASURED 2026-09-19: {v}");
}

// ── 42 ───────────────────────────────────────────────────────────────────────

#[test]
fn b42_a_chunked_upload_is_411() {
    let r = Rig::new();
    let (_, v) = r.post("/1.3/storage", json!({"storage": {"size": 1, "tier": "maxiops", "title": "iso", "zone": "se-sto1"}}));
    let u = v["storage"]["uuid"].as_str().unwrap().to_string();
    r.until_storage(&u, "online");
    let (_, sess) = r.post(&format!("/1.3/storage/{u}/import"), json!({"storage_import": {"source": "direct_upload"}}));
    let url = sess["storage_import"]["direct_upload_url"].as_str().unwrap().to_string();
    // A reader of unknown length: reqwest sends it chunked.
    let body = reqwest::blocking::Body::new(std::io::Cursor::new(vec![0u8; 4096]));
    let resp = r.http.put(&url).body(body).send().unwrap();
    assert_eq!(resp.status().as_u16(), 411);
}

// ── 43 ───────────────────────────────────────────────────────────────────────

/// A DEAD UpCloud token is a clean 401 everywhere (MEASURED). The "0 rows +
/// 403" shape is behaviour 5, whose only dated measurement is Scaleway's.
#[test]
fn b43_a_dead_token_is_a_clean_401() {
    let r = Rig::new();
    r.arm("dead-token");
    let (a, v) = r.get("/1.3/account");
    assert_eq!((a, code(&v)), (401, "AUTHENTICATION_FAILED"), "{v}");
    assert_eq!(r.get("/1.3/server").0, 401, "the LIST too — not 200 with zero rows");
}

// ── 44 ───────────────────────────────────────────────────────────────────────

#[test]
fn b44_the_maxiops_quota_is_ten_thousand_two_hundred_and_forty_gib() {
    let r = Rig::new();
    let (s, v) = r.post("/1.3/storage", json!({"storage": {"size": 10_241, "tier": "maxiops", "title": "big", "zone": "se-sto1"}}));
    assert_eq!((s, code(&v)), (403, "MAXIOPS_STORAGE_LIMIT_REACHED"), "{v}");
    let (s, _) = r.post("/1.3/storage", json!({"storage": {"size": 10_241, "tier": "hdd", "title": "big", "zone": "se-sto1"}}));
    assert_eq!(s, 201, "only the MaxIOPS quota was measured");
}

// ── 45 ───────────────────────────────────────────────────────────────────────

#[test]
fn b45_a_disk_boot_is_started_in_ten_seconds() {
    let r = Rig::new();
    let u = r.server("front");
    r.stop(&u);
    let t0 = r.now();
    assert_eq!(r.post(&format!("/1.3/server/{u}/start"), Value::Null).0, 200);
    r.until(&u, "started");
    let took = r.now() - t0;
    assert!((9_000..=11_000).contains(&took), "MEASURED ~10 s; the mock said 20: {took} ms");
}

// ── 46 ───────────────────────────────────────────────────────────────────────

/// **An installer start reads `started` BRIEFLY, then `maintenance` for the
/// pass, then `stopped`.** MEASURED 2026-09-21 08:42 on the live re-image
/// (private-gunnar-ops `.reimage/reinstall-receipt.json`, `done: true`): step
/// `start` 377 421 ms → `started` read at 381 091 ms → `maintenance` at
/// 381 174 ms → `stopped` at 530 945 ms. The earlier "never `started`"
/// (2026-09-19) was polled every 10 s from t+10 s and could not see a window of
/// under 100 ms; its `maintenance` … `stopped` half stands. The pass is 130–146 s
/// from the start, whatever the guest's own 2–5 s.
#[test]
fn b46_an_install_pass_reads_started_briefly_then_maintenance_then_stopped() {
    let r = Rig::new();
    let u = r.server("appliance");
    r.stop(&u);
    let iso = r.volume("gunnar-installer.iso", 1, "maxiops");
    let (s, v) = r.post(&format!("/1.3/server/{u}/storage/attach"), json!({"storage_device": {"type": "cdrom", "storage": iso}}));
    assert_eq!(s, 200, "{v}");
    r.put(&format!("/1.3/server/{u}"), json!({"server": {"boot_order": "cdrom,disk"}}));
    let t0 = r.now();
    assert_eq!(r.post(&format!("/1.3/server/{u}/start"), Value::Null).0, 200);
    // (state, first seen, last seen), in virtual ms after the start. A read
    // settles first and THEN moves virtual time toward the next deadline, so
    // the clock after the read that first shows a state can be well past the
    // moment it began; the LAST read that still showed the previous state is
    // the honest bound on when it ended.
    let mut seen: Vec<(String, u64, u64)> = vec![];
    for _ in 0..400 {
        let st = r.state(&u);
        let now = r.now() - t0;
        match seen.last_mut() {
            Some((s, _, last)) if *s == st => *last = now,
            _ => seen.push((st.clone(), now, now)),
        }
        if st == "stopped" {
            break;
        }
    }
    let states: Vec<&str> = seen.iter().map(|(s, _, _)| s.as_str()).collect();
    assert_eq!(states, ["maintenance", "started", "maintenance", "stopped"], "{seen:?}");
    // `started` at ~3.6 s, and only for a window the receipt bounds at < 100 ms.
    let (first, last) = (seen[1].1, seen[1].2);
    assert!((3_600..=3_700).contains(&first), "the brief `started` began at {first} ms: {seen:?}");
    assert!(last <= 3_700 && last - first <= 100, "`started` was still read at {last} ms: {seen:?}");
    let took = seen[3].1;
    assert!((130_000..=146_000).contains(&took), "the provider's pass, not the guest's 2–5 s: {took} ms");
    assert_eq!(r.guest(&u), "Installed");
}

// ── 47 ───────────────────────────────────────────────────────────────────────

#[test]
fn b47_stop_has_no_timeout_action() {
    let r = Rig::new();
    let u = r.server("front");
    let (s, v) = r.post(
        &format!("/1.3/server/{u}/stop"),
        json!({"stop_server": {"stop_type": "soft", "timeout": "60", "timeout_action": "destroy"}}),
    );
    assert_eq!(s, 400, "MEASURED: that field belongs to restart: {v}");
    assert_eq!(r.state(&u), "started", "and nothing was stopped");
}

// ── 48 ───────────────────────────────────────────────────────────────────────

/// terraform 5.44.1 starts the server straight after a plan-change PUT. The
/// mock used to be in `maintenance` then, let the start overwrite the pending
/// transition, and lose the Resize Backup. A start during a real
/// `maintenance` is refused instead.
#[test]
fn b48_a_plan_change_is_done_when_the_put_answers_and_a_start_in_maintenance_is_refused() {
    let r = Rig::new();
    let u = r.server("appliance");
    r.post(&format!("/1.3/server/{u}/stop"), json!({"stop_server": {"stop_type": "hard"}}));
    let (s, v) = r.post(&format!("/1.3/server/{u}/start"), Value::Null);
    assert_eq!((s, code(&v)), (409, "SERVER_STATE_ILLEGAL"), "a start mid-stop is refused: {v}");
    r.until(&u, "stopped");

    let backups = |r: &Rig| {
        r.get("/1.3/storage/private").1["storages"]["storage"].as_array().unwrap().iter().filter(|s| s["type"] == "backup").count()
    };
    let before = backups(&r);
    let (s, _) = r.put(&format!("/1.3/server/{u}"), json!({"server": {"plan": "4xCPU-8GB"}}));
    assert_eq!(s, 202);
    assert_eq!(r.state(&u), "stopped", "nothing to wait for");
    assert_eq!(r.post(&format!("/1.3/server/{u}/start"), Value::Null).0, 200, "the provider's immediate start");
    assert_eq!(backups(&r), before + 1, "and the Resize Backup is not lost");
}

// ── 49 ───────────────────────────────────────────────────────────────────────

/// `boot_order` is honoured at the next HYPERVISOR start, not a guest reboot:
/// a PUT does not end an install loop (MEASURED 2026-09-14). Only eject does.
#[test]
fn b49_a_boot_order_put_does_not_end_a_loop() {
    let r = Rig::new();
    r.arm("installer-loop");
    let u = r.server("appliance");
    r.stop(&u);
    let iso = r.volume("gunnar-installer.iso", 1, "maxiops");
    r.post(&format!("/1.3/server/{u}/storage/attach"), json!({"storage_device": {"type": "cdrom", "storage": iso}}));
    r.put(&format!("/1.3/server/{u}"), json!({"server": {"boot_order": "cdrom"}}));
    r.post(&format!("/1.3/server/{u}/start"), Value::Null);
    r.until(&u, "started");
    assert!(r.guest(&u).starts_with("Looping"));
    r.put(&format!("/1.3/server/{u}"), json!({"server": {"boot_order": "disk"}}));
    r.get("/1.3/server");
    assert!(r.guest(&u).starts_with("Looping"), "still looping: {}", r.guest(&u));
    assert_eq!(r.post(&format!("/1.3/server/{u}/cdrom/eject"), Value::Null).0, 200);
    assert_eq!(r.guest(&u), "Installed");
}

// ── 50 ───────────────────────────────────────────────────────────────────────

#[test]
fn b50_a_toggle_without_the_pause_gives_the_old_console_back() {
    let r = Rig::new();
    let u = r.server("appliance");
    r.put(&format!("/1.3/server/{u}"), json!({"server": {"remote_access_enabled": "yes"}}));
    let port = |r: &Rig| r.get(&format!("/1.3/server/{u}")).1["server"]["remote_access_port"].as_str().unwrap().to_string();
    let first = port(&r);
    r.stop(&u);
    r.post(&format!("/1.3/server/{u}/start"), Value::Null);
    r.until(&u, "started");
    assert_eq!(port(&r), first, "stale (behaviour 14)");

    // No pause: the old endpoint comes back.
    r.put(&format!("/1.3/server/{u}"), json!({"server": {"remote_access_enabled": "no"}}));
    r.put(&format!("/1.3/server/{u}"), json!({"server": {"remote_access_enabled": "yes"}}));
    assert_eq!(port(&r), first, "MEASURED: without the pause the old port comes back");

    // With it: the cure.
    r.put(&format!("/1.3/server/{u}"), json!({"server": {"remote_access_enabled": "no"}}));
    r.post("/mock/advance/2000", Value::Null);
    r.put(&format!("/1.3/server/{u}"), json!({"server": {"remote_access_enabled": "yes"}}));
    assert_ne!(port(&r), first);
}

// ── 51 ───────────────────────────────────────────────────────────────────────

#[test]
fn b51_the_console_toggle_is_409_in_maintenance() {
    let r = Rig::new();
    let u = r.server("appliance");
    r.post(&format!("/1.3/server/{u}/stop"), json!({"stop_server": {"stop_type": "hard"}}));
    assert_eq!(r.state(&u), "maintenance");
    let (s, v) = r.put(&format!("/1.3/server/{u}"), json!({"server": {"remote_access_enabled": "yes"}}));
    assert_eq!(s, 409, "MEASURED: retry after maintenance: {v}");
}

// ── 52 ───────────────────────────────────────────────────────────────────────

#[test]
fn b52_an_import_ends_syncing_then_maintenance_then_online() {
    let r = Rig::new();
    let (_, v) = r.post("/1.3/storage", json!({"storage": {"size": 1, "tier": "maxiops", "title": "iso", "zone": "se-sto1"}}));
    let u = v["storage"]["uuid"].as_str().unwrap().to_string();
    r.until_storage(&u, "online");
    let (_, sess) = r.post(&format!("/1.3/storage/{u}/import"), json!({"storage_import": {"source": "direct_upload"}}));
    let url = sess["storage_import"]["direct_upload_url"].as_str().unwrap().to_string();
    assert_eq!(r.http.put(&url).body(vec![0u8; 4096]).send().unwrap().status().as_u16(), 200);
    let mut seen: Vec<String> = vec![];
    for _ in 0..400 {
        let st = r.sstate(&u);
        if seen.last() != Some(&st) {
            seen.push(st.clone());
        }
        if st == "online" {
            break;
        }
    }
    assert_eq!(seen, ["maintenance", "syncing", "maintenance", "online"], "MEASURED sequence");
}

// ── 54 ───────────────────────────────────────────────────────────────────────

#[test]
fn b54_same_size_is_size_invalid_too() {
    let r = Rig::new();
    let u = r.volume("data", 10, "maxiops");
    let (s, v) = r.put(&format!("/1.3/storage/{u}"), json!({"storage": {"size": 10}}));
    assert_eq!((s, code(&v)), (400, "SIZE_INVALID"), "{v}");
}

// ── 55 ───────────────────────────────────────────────────────────────────────

#[test]
fn b55_a_grow_is_maintenance_for_thirty_seven_seconds() {
    let r = Rig::new();
    let u = r.volume("data", 1, "maxiops");
    let t0 = r.now();
    assert_eq!(r.put(&format!("/1.3/storage/{u}"), json!({"storage": {"size": 2}})).0, 200);
    assert_eq!(r.sstate(&u), "maintenance", "not instant");
    r.until_storage(&u, "online");
    let took = r.now() - t0;
    assert!((36_000..=38_000).contains(&took), "MEASURED 37 s: {took} ms");
}

// ── 56 ───────────────────────────────────────────────────────────────────────

#[test]
fn b56_a_grow_under_a_running_server_is_refused() {
    let r = Rig::new();
    let u = r.server("appliance");
    let vol = r.volume("member", 10, "maxiops");
    assert_eq!(r.attach(&u, &vol, None).0, 200);
    let (s, v) = r.put(&format!("/1.3/storage/{vol}"), json!({"storage": {"size": 20}}));
    assert_eq!((s, code(&v)), (409, "STORAGE_ATTACHED"), "{v}");
    r.stop(&u);
    assert_eq!(r.put(&format!("/1.3/storage/{vol}"), json!({"storage": {"size": 20}})).0, 200, "stopped: allowed");
    r.until_storage(&vol, "online");
    r.arm("resize-requires-detach");
    let (s, v) = r.put(&format!("/1.3/storage/{vol}"), json!({"storage": {"size": 30}}));
    assert_eq!((s, code(&v)), (409, "STORAGE_ATTACHED"), "the stricter docs page: {v}");
}

// ── 57 ───────────────────────────────────────────────────────────────────────

#[test]
fn b57_a_storage_is_attached_once() {
    let r = Rig::new();
    let a = r.server("a");
    let b = r.server("b");
    let vol = r.volume("shared", 10, "maxiops");
    assert_eq!(r.attach(&a, &vol, None).0, 200);
    let (s, v) = r.attach(&b, &vol, None);
    assert_eq!((s, code(&v)), (409, "STORAGE_ATTACHED"), "{v}");
}

// ── 58 ───────────────────────────────────────────────────────────────────────

/// A requested address is honoured, `virtio` takes the first FREE slot, and
/// the guest names disks in slot order, contiguously (vda, vdb, …).
#[test]
fn b58_addresses_are_honoured_and_the_guest_names_follow_the_slots() {
    let r = Rig::new();
    let u = r.server("appliance");
    let v1 = r.volume("d1", 10, "maxiops");
    let v3 = r.volume("d3", 10, "maxiops");
    let vx = r.volume("dx", 10, "maxiops");
    assert_eq!(r.attach(&u, &v3, Some("virtio:3")).1["server"].is_object(), true);
    assert!(r.addresses(&u).contains(&"virtio:3".to_string()), "{:?}", r.addresses(&u));
    assert_eq!(r.attach(&u, &v1, Some("virtio")).0, 200);
    assert!(r.addresses(&u).contains(&"virtio:1".to_string()), "the first FREE slot: {:?}", r.addresses(&u));
    let (s, _) = r.attach(&u, &vx, Some("virtio:3"));
    assert_eq!(s, 409, "a taken address is refused");
    let (_, d) = r.get(&format!("/mock/disks/{u}"));
    let names: Vec<(String, String)> = d["disks"]
        .as_array()
        .unwrap()
        .iter()
        .map(|x| (x["address"].as_str().unwrap().into(), x["name"].as_str().unwrap().into()))
        .collect();
    assert_eq!(
        names,
        vec![
            ("virtio:0".to_string(), "vda".to_string()),
            ("virtio:1".to_string(), "vdb".to_string()),
            ("virtio:3".to_string(), "vdc".to_string()),
        ]
    );
}

// ── 59 ───────────────────────────────────────────────────────────────────────

#[test]
fn b59_a_storage_mid_operation_or_a_clones_source_cannot_be_deleted() {
    let r = Rig::new();
    let (_, v) = r.post("/1.3/storage", json!({"storage": {"size": 1, "tier": "maxiops", "title": "young", "zone": "se-sto1"}}));
    let young = v["storage"]["uuid"].as_str().unwrap().to_string();
    let (s, v) = r.delete(&format!("/1.3/storage/{young}"));
    assert_eq!((s, code(&v)), (409, "STORAGE_STATE_ILLEGAL"), "{v}");

    let src = r.volume("src", 1, "maxiops");
    let (s, _) = r.post(&format!("/1.3/storage/{src}/clone"), json!({"storage": {"title": "c", "tier": "maxiops"}}));
    assert_eq!(s, 201);
    let (s, v) = r.delete(&format!("/1.3/storage/{src}"));
    assert_eq!((s, code(&v)), (409, "STORAGE_STATE_ILLEGAL"), "the clone is still being made: {v}");
}

// ── 60 ───────────────────────────────────────────────────────────────────────

/// Hot-plug is the GUEST's. A kernel that can do it: a virtio detach on a
/// running server succeeds. One that cannot: `511 HOTPLUG_FAILED`, both ways
/// (MEASURED 2026-09-14). The mock used to answer an invented `409
/// VIRTIO_HOTPLUG_UNSUPPORTED` for every guest.
#[test]
fn b60_hot_plug_is_the_guests_and_its_failure_is_511() {
    let r = Rig::new();
    let u = r.server("front");
    let vol = r.volume("m", 10, "maxiops");
    assert_eq!(r.attach(&u, &vol, None).0, 200);
    let (s, v) = r.post(&format!("/1.3/server/{u}/storage/detach"), json!({"storage_device": {"address": "virtio:1"}}));
    assert_eq!(s, 200, "a hot-plug kernel acks the eject: {v}");

    r.arm("guest-kernel-lacks-hotplug");
    let old = r.server("appliance-old-kernel");
    let (s, v) = r.attach(&old, &vol, None);
    assert_eq!((s, code(&v)), (511, "HOTPLUG_FAILED"), "{v}");
    r.stop(&old);
    assert_eq!(r.attach(&old, &vol, None).0, 200, "stopped: no hot-plug needed");
    r.post(&format!("/1.3/server/{old}/start"), Value::Null);
    r.until(&old, "started");
    let (s, v) = r.post(&format!("/1.3/server/{old}/storage/detach"), json!({"storage_device": {"address": "virtio:1"}}));
    assert_eq!((s, code(&v)), (511, "HOTPLUG_FAILED"), "MEASURED: {v}");
}

// ── 61 ───────────────────────────────────────────────────────────────────────

/// UpCloud gives IPv6 when asked; "no IPv6" is the estate's law, and a mock
/// that answered IPv4 regardless would hide a declaration breaking it.
#[test]
fn b61_an_interface_that_asks_for_ipv6_gets_ipv6() {
    let r = Rig::new();
    let u = r.server_with(
        "v6",
        json!({"networking": {"interfaces": {"interface": [
            {"index": 1, "type": "public", "ip_addresses": {"ip_address": [{"family": "IPv6"}]}},
            {"index": 2, "type": "utility", "ip_addresses": {"ip_address": [{"family": "IPv4"}]}}
        ]}}}),
    );
    let (_, v) = r.get(&format!("/1.3/server/{u}"));
    let ifs = v["server"]["networking"]["interfaces"]["interface"].as_array().unwrap();
    assert_eq!(ifs[0]["ip_addresses"]["ip_address"][0]["family"], "IPv6", "{v}");
    assert!(ifs[0]["ip_addresses"]["ip_address"][0]["address"].as_str().unwrap().contains(':'));
    assert_eq!(ifs[1]["ip_addresses"]["ip_address"][0]["family"], "IPv4");
}

// ── 62 ───────────────────────────────────────────────────────────────────────

#[test]
fn b62_the_firewall_is_evaluated() {
    let r = Rig::new();
    let fw = r.server_with("fw", json!({"firewall": "on"}));
    let open = r.server_with("open", json!({"firewall": "off"}));
    let (s, _) = r.put(
        &format!("/1.3/server/{fw}/firewall_rule"),
        json!({"firewall_rules": {"firewall_rule": [
            {"direction": "in", "action": "accept", "family": "IPv4", "protocol": "tcp", "destination_port_start": "22", "destination_port_end": "22"},
            {"direction": "in", "action": "accept", "family": "IPv4", "protocol": "tcp",
             "source_address_start": "10.13.8.99", "source_address_end": "10.13.8.99",
             "destination_port_start": "50051", "destination_port_end": "50051"},
            {"direction": "in", "action": "drop", "family": "IPv4"}
        ]}}),
    );
    assert_eq!(s, 200);
    let inb = |to: &str, from: &str, port: u16| {
        r.get(&format!("/mock/inbound/{to}?from={from}&proto=tcp&port={port}")).1["why"].as_str().unwrap().to_string()
    };
    assert_eq!(inb(&fw, "1.2.3.4", 22), "ok");
    assert!(inb(&fw, "1.2.3.4", 80).contains("dropped"), "silence, not a refusal");
    assert_eq!(inb(&fw, "10.13.8.99", 50051), "ok");
    assert!(inb(&fw, "10.13.7.210", 50051).contains("dropped"), "the utility network is filtered too");
    assert_eq!(inb(&open, "1.2.3.4", 80), "ok", "firewall off: wide open");

    let udp = |on: &str| r.get(&format!("/mock/udp-reply/{on}?from=192.0.2.9&port=53")).1["arrives"].as_bool().unwrap();
    assert!(!udp(&fw), "the catch-all drop eats the DNS reply (MEASURED: +notcp times out)");
    assert!(udp(&open), "no rules: the reply arrives");
}

// ── 63 ───────────────────────────────────────────────────────────────────────

/// The mock records whom it heard, by digest, so a verb can prove the
/// terraform provider spoke to THIS mock before it lets `apply` run.
#[test]
fn b63_the_mock_can_say_whether_it_heard_a_token() {
    let r = Rig::new();
    let sha = |t: &str| mock_upcloud::digest::sha256_hex(t.as_bytes());
    assert_eq!(r.get(&format!("/mock/heard?token_sha256={}", sha("ucat_mock"))).1["account_calls"], 0);
    r.get("/1.3/account");
    r.get("/1.3/account");
    let (_, h) = r.get(&format!("/mock/heard?token_sha256={}", sha("ucat_mock")));
    assert_eq!(h["account_calls"], 2, "{h}");
    assert_eq!(r.get(&format!("/mock/heard?token_sha256={}", sha("ucat_other"))).1["requests"], 0);
}

// ── 64 ───────────────────────────────────────────────────────────────────────

/// A server PUT changes `hostname` and `title` in place. The mock dropped both,
/// and terraform 5.44.1 then refused its own apply ("inconsistent result").
#[test]
fn b64_a_server_put_renames_in_place() {
    let r = Rig::new();
    let u = r.server("front");
    assert_eq!(r.put(&format!("/1.3/server/{u}"), json!({"server": {"hostname": "front2", "title": "front 2"}})).0, 202);
    let (_, v) = r.get(&format!("/1.3/server/{u}"));
    assert_eq!((v["server"]["hostname"].as_str(), v["server"]["title"].as_str()), (Some("front2"), Some("front 2")), "{v}");
}

// ── 65 ───────────────────────────────────────────────────────────────────────

/// **`cdrom/load` re-images with a DIFFERENT medium** (L110, documented, not
/// measured): only into an empty tray, and the new medium is what the device
/// row — and so the next hypervisor start — carries.
#[test]
fn b65_cdrom_load_puts_a_new_medium_into_an_empty_tray() {
    let r = Rig::new();
    let first = r.volume("medium-a", 1, "maxiops");
    let second = r.volume("medium-b", 1, "maxiops");
    let u = r.server("appliance");
    let cdrom = |r: &Rig| -> String {
        let v = r.get(&format!("/1.3/server/{u}")).1;
        v["server"]["storage_devices"]["storage_device"]
            .as_array()
            .unwrap()
            .iter()
            .find(|d| d["type"] == "cdrom")
            .map(|d| d["storage"].as_str().unwrap_or("").to_string())
            .unwrap_or_else(|| "<no cdrom device>".into())
    };
    let load = |r: &Rig, st: &str| r.post(&format!("/1.3/server/{u}/cdrom/load"), json!({"storage_device": {"storage": st}}));

    // No cdrom device yet: refused by name.
    let (s, v) = load(&r, &first);
    assert_eq!((s, v["error"]["error_code"].as_str()), (404, Some("CDROM_DEVICE_NOT_FOUND")), "{v}");

    r.stop(&u);
    assert_eq!(r.post(&format!("/1.3/server/{u}/storage/attach"), json!({"storage_device": {"type": "cdrom", "storage": first}})).0, 200);
    r.put(&format!("/1.3/server/{u}"), json!({"server": {"boot_order": "cdrom,disk"}}));
    assert_eq!(cdrom(&r), first);

    // A full tray is refused; the medium in it is untouched.
    let (s, v) = load(&r, &second);
    assert_eq!((s, v["error"]["error_code"].as_str()), (409, Some("CDROM_DEVICE_IN_USE")), "{v}");
    assert_eq!(cdrom(&r), first);

    // Missing or unknown storage: refused by name.
    let (s, v) = load(&r, "");
    assert_eq!((s, v["error"]["error_code"].as_str()), (400, Some("STORAGE_MISSING")), "{v}");
    assert_eq!(r.post(&format!("/1.3/server/{u}/cdrom/eject"), Value::Null).0, 200);
    let (s, v) = load(&r, "00000000-0000-0000-0000-000000000000");
    assert_eq!((s, v["error"]["error_code"].as_str()), (404, Some("STORAGE_NOT_FOUND")), "{v}");

    // Eject, then load the OTHER medium: it is what the tray now holds, on a
    // started server too (like eject, L52).
    assert_eq!(r.post(&format!("/1.3/server/{u}/start"), Value::Null).0, 200);
    let (s, v) = load(&r, &second);
    assert_eq!(s, 200, "{v}");
    assert_eq!(cdrom(&r), second);
}

// ── 66 ───────────────────────────────────────────────────────────────────────

/// An import straight after `POST /storage`, with no wait, is accepted — the
/// way gunnar's `start_medium` has done it on the live account since
/// 2026-09-07 (receipt 2026-09-21: two media, no 409). A storage mid-grow is
/// still refused: only the create's own settling is let through.
#[test]
fn b66_an_import_right_after_the_create_is_accepted() {
    let r = Rig::new();
    let (s, v) = r.post("/1.3/storage", json!({"storage": {"size": 1, "tier": "maxiops", "title": "seed", "zone": "se-sto1"}}));
    assert_eq!(s, 201, "{v}");
    let u = v["storage"]["uuid"].as_str().unwrap().to_string();
    let (s, im) = r.post(&format!("/1.3/storage/{u}/import"), json!({"storage_import": {"source": "direct_upload"}}));
    assert_eq!(s, 201, "no wait between create and import, as the live runs do: {im}");
    let url = im["storage_import"]["direct_upload_url"].as_str().unwrap().to_string();
    assert_eq!(r.http.put(&url).body(vec![0u8; 4096]).send().unwrap().status().as_u16(), 200);
    r.until_storage(&u, "online");

    let g = r.volume("growing", 1, "maxiops");
    r.put(&format!("/1.3/storage/{g}"), json!({"storage": {"size": 2}}));
    let (s, v) = r.post(&format!("/1.3/storage/{g}/import"), json!({"storage_import": {"source": "direct_upload"}}));
    assert_eq!((s, code(&v)), (409, "STORAGE_STATE_ILLEGAL"), "a grow's maintenance is not a create's: {v}");
}

/// **The direct-upload `PUT` answers the import object BARE** — `written_bytes`,
/// `sha256sum` and the rest at the top level, not under `storage_import`
/// (behaviour 67). MEASURED by the live re-image of 2026-09-21 08:42:
/// `gunnar-upcloud` reads `written_bytes` at the top level and its receipt says
/// `done: true`. The wrapped shape the mock answered stopped the SAME procedure,
/// run through its ported client against this mock, with "the upload reply
/// carries no `written_bytes`" (lane T14's proof run).
#[test]
fn b67_the_upload_put_answers_the_import_bare() {
    let r = Rig::new();
    let (_, v) = r.post("/1.3/storage", json!({"storage": {"size": 1, "tier": "maxiops", "title": "seed", "zone": "se-sto1"}}));
    let u = v["storage"]["uuid"].as_str().unwrap().to_string();
    let (_, im) = r.post(&format!("/1.3/storage/{u}/import"), json!({"storage_import": {"source": "direct_upload"}}));
    let url = im["storage_import"]["direct_upload_url"].as_str().unwrap().to_string();
    let resp = r.http.put(&url).body(vec![7u8; 4096]).send().unwrap();
    assert_eq!(resp.status().as_u16(), 200);
    let got: Value = serde_json::from_str(&resp.text().unwrap()).unwrap();
    assert_eq!(got["written_bytes"], 4096u64, "{got}");
    assert_eq!(got["sha256sum"].as_str().unwrap(), mock_upcloud::digest::sha256_hex(&[7u8; 4096]), "{got}");
    assert!(got.get("storage_import").is_none(), "not wrapped: {got}");
}

// ── 12 / 31: nothing the mock reports is a real host ──────────────────────────

/// Is `ip` loopback, RFC 5737 TEST-NET or RFC 1918?
fn allowed_ip(ip: std::net::Ipv4Addr) -> bool {
    let o = ip.octets();
    ip.is_loopback()
        || ip.is_private()
        || (o[0], o[1], o[2]) == (192, 0, 2)
        || (o[0], o[1], o[2]) == (198, 51, 100)
        || (o[0], o[1], o[2]) == (203, 0, 113)
}

/// A string that names a host must name one nobody can reach by accident.
fn check_host(host: &str) -> Result<(), String> {
    let h = host.trim_end_matches('.').to_ascii_lowercase();
    if let Ok(ip) = h.parse::<std::net::Ipv4Addr>() {
        return if allowed_ip(ip) { Ok(()) } else { Err(format!("routable IPv4 {h}")) };
    }
    if h == "localhost" || h.ends_with(".invalid") {
        return Ok(());
    }
    Err(format!("real-looking host {h}"))
}

/// Walk every string in a reply. IPv4 literals and URL hosts are checked
/// everywhere; host-shaped words only under a key that names a host, so a
/// title like `gunnar-installer.iso` is not mistaken for one. The ONE
/// exception is RFC 7807's `type`, which UpCloud itself returns as a
/// documentation URI (behaviour 1): an identifier, not a host to dial.
fn walk(path: &str, key: &str, v: &Value, bad: &mut Vec<String>) {
    match v {
        Value::Object(m) => m.iter().for_each(|(k, x)| walk(&format!("{path}.{k}"), k, x, bad)),
        Value::Array(a) => a.iter().enumerate().for_each(|(i, x)| walk(&format!("{path}[{i}]"), key, x, bad)),
        Value::String(s) => {
            if key == "type" && s.starts_with("https://developers.upcloud.com/1.3/errors#") {
                return;
            }
            let verdict = if let Ok(ip) = s.parse::<std::net::Ipv4Addr>() {
                check_host(&ip.to_string())
            } else if let Some(rest) = s.split_once("://").map(|(_, r)| r) {
                let authority = rest.split('/').next().unwrap_or("");
                let host = authority.rsplit_once(':').map(|(h, _)| h).unwrap_or(authority);
                check_host(host)
            } else if ["host", "url", "fqdn", "domain", "endpoint"].iter().any(|w| key.to_ascii_lowercase().contains(w))
                && s.contains('.')
                && !s.contains(' ')
            {
                let host = s.rsplit_once(':').filter(|(_, p)| p.parse::<u16>().is_ok()).map(|(h, _)| h).unwrap_or(s);
                check_host(host)
            } else {
                Ok(())
            };
            if let Err(why) = verdict {
                bad.push(format!("{path} = {s:?}: {why}"));
            }
        }
        _ => {}
    }
}

/// **No host or address the mock reports is outside loopback, TEST-NET,
/// RFC 1918 or `.invalid`.** The pool once held REAL UpCloud addresses (one a
/// stranger's host with port 22 open) and the console was
/// `<zone>.vnc.upcloud.com`: a mock step that dialled what it was told went out
/// onto the internet. Driven over the API a caller uses, through every state
/// that changes an address: create, remote access on, stop/start (a fresh
/// console), a medium upload, a relay, plus the mock's own door.
#[test]
fn b12_nothing_the_mock_reports_is_a_real_host() {
    let r = Rig::new();
    let a = r.server("appliance");
    let f = r.server("front");
    r.put(&format!("/1.3/server/{a}"), json!({"server": {"remote_access_enabled": "yes", "remote_access_type": "vnc", "remote_access_password": "abcd1234"}}));
    r.stop(&a);
    r.post(&format!("/1.3/server/{a}/start"), Value::Null);
    r.until(&a, "started");
    let iso = r.volume("gunnar-installer.iso", 1, "maxiops");
    let (_, sess) = r.post(&format!("/1.3/storage/{iso}/import"), json!({"storage_import": {"source": "direct_upload"}}));
    r.post("/mock/relay", Value::Null);
    let b = r.server("after-relay");

    let mut paths: Vec<String> = [
        "/1.3/server", "/1.3/storage", "/1.3/storage/private", "/1.3/account", "/1.3/zone", "/1.3/price", "/1.3/plan",
        "/mock/estate",
    ]
    .iter()
    .map(|s| s.to_string())
    .collect();
    for u in [&a, &f, &b] {
        paths.push(format!("/1.3/server/{u}"));
        paths.push(format!("/1.3/server/{u}/firewall_rule"));
        paths.push(format!("/mock/dhcp/{u}"));
    }
    paths.push(format!("/1.3/storage/{iso}"));
    paths.push(format!("/1.3/storage/{iso}/import"));

    let mut bad = Vec::new();
    walk("import-reply", "", &sess, &mut bad);
    for p in &paths {
        let (_, v) = r.get(p);
        walk(p, "", &v, &mut bad);
    }
    assert!(bad.is_empty(), "the mock reported real hosts:\n{}", bad.join("\n"));
    // The console field was really there to check, and it is RFC 2606.
    let host = r.get(&format!("/1.3/server/{a}")).1["server"]["remote_access_host"].as_str().unwrap_or("").to_string();
    assert!(host.ends_with(".vnc.mock.invalid"), "console host {host:?}");

    // The checker itself can say no: yesterday's pool and console fail it.
    let mut caught = Vec::new();
    walk("t", "", &json!({"ip_address": {"address": "94.237.30.222"}, "remote_access_host": "se-sto1.vnc.upcloud.com",
                         "direct_upload_url": "https://se-sto1.img.upcloud.com/uploader/session/x"}), &mut caught);
    assert_eq!(caught.len(), 3, "{caught:?}");
}

/// **Behaviour 68 / L112: the mock accepts BOTH spellings of a storage size
/// and of a stop timeout** — the verified one (size as a number, timeout as a
/// string) and the unverified one — because a mock that refused the unverified
/// half would be claiming a measurement nobody has.
#[test]
fn b68_both_spellings_are_accepted_because_only_one_of_each_is_verified() {
    let r = Rig::new();
    let a = r.volume("sized-by-number", 10, "maxiops");
    let (s, v) = r.put(&format!("/1.3/storage/{a}"), json!({"storage": {"size": 20}}));
    assert_eq!(s, 200, "the VERIFIED spelling: {v}");
    let b = r.volume("sized-by-string", 10, "maxiops");
    let (s, v) = r.put(&format!("/1.3/storage/{b}"), json!({"storage": {"size": "20"}}));
    assert_eq!(s, 200, "the unverified spelling is not refused either: {v}");
}

// ── L111 ─────────────────────────────────────────────────────────────────────

/// **A seed takes its import straight after the create**, as the live tool
/// does (gunnar deploy/upcloud/src/plan.rs:538-542, no wait between them) and
/// as the live re-image of 2026-09-21 08:42 did twice. A storage that is busy
/// with something ELSE (an import already open) is still refused.
#[test]
fn b111_an_import_is_accepted_straight_after_the_create() {
    let r = Rig::new();
    let (s, v) = r.post(
        "/1.3/storage",
        json!({"storage": {"size": 1, "tier": "maxiops", "title": "seed", "zone": "se-sto1", "labels": []}}),
    );
    assert_eq!(s, 201, "{v}");
    let u = v["storage"]["uuid"].as_str().unwrap().to_string();
    assert_eq!(v["storage"]["state"], "maintenance", "a new storage is still being created");
    let (s, v) = r.post(&format!("/1.3/storage/{u}/import"), json!({"storage_import": {"source": "direct_upload"}}));
    assert_eq!(s, 201, "the import straight after the create: {v}");
    let url = v["storage_import"]["direct_upload_url"].as_str().unwrap().to_string();
    let up = r.http.put(&url).body(vec![0u8; 4096]).send().unwrap();
    assert_eq!(up.status().as_u16(), 200);
    // Busy with the import now: a second session is refused.
    let (s, v) = r.post(&format!("/1.3/storage/{u}/import"), json!({"storage_import": {"source": "direct_upload"}}));
    assert_eq!((s, code(&v)), (409, "STORAGE_STATE_ILLEGAL"), "{v}");
}

// ── 69 ───────────────────────────────────────────────────────────────────────

/// The guest as it is NOW (`/mock/estate?as_is`): the plain view runs the
/// estate to quiet, and a looping medium never is.
fn guest_now(r: &Rig, u: &str) -> String {
    let (_, est) = r.get("/mock/estate?as_is");
    let me = est["servers"].as_array().unwrap().iter().find(|s| s["uuid"] == u).unwrap().clone();
    me["guest"].as_str().unwrap().to_string()
}

/// A server with a rebooting medium in its tray, CD first, and the start taken.
fn rebooting_installer_start(r: &Rig) -> (String, u64) {
    r.arm("installer-reboots");
    let u = r.server("appliance");
    r.stop(&u);
    let iso = r.volume("korp-installer-0.1.3.iso", 1, "maxiops");
    let (s, v) = r.post(&format!("/1.3/server/{u}/storage/attach"), json!({"storage_device": {"type": "cdrom", "storage": iso}}));
    assert_eq!(s, 200, "{v}");
    r.put(&format!("/1.3/server/{u}"), json!({"server": {"boot_order": "cdrom,disk"}}));
    let t0 = r.now();
    assert_eq!(r.post(&format!("/1.3/server/{u}/start"), Value::Null).0, 200);
    (u, t0)
}

/// The waiter `gunnar-upcloud` runs WITHOUT `--medium-powers-off`
/// (plan.rs:691-693 → `wait_for_state(.., "started", ..)` under the
/// `stopping` budget, 300 s): it wants `started`, and anything else is waiting.
/// Returns the virtual ms it took, or panics as the tool would time out.
fn tool_waits_for_started(r: &Rig, u: &str, t0: u64) -> u64 {
    loop {
        // A read settles at the clock it finds and only THEN moves virtual
        // time on, so the state it returns is the state at the clock read
        // BEFORE it — the moment a 2 s poller would have asked.
        let waited = r.now() - t0;
        let st = r.state(u);
        assert!(waited <= 300_000, "the tool times out after 300 s; last seen {st:?}");
        assert_ne!(st, "stopped", "a rebooting medium never powers the box off");
        if st == "started" {
            return waited;
        }
    }
}

/// Read, as a poller does, until the guest is no longer `from`: every read in
/// between must say `started`. Returns the virtual ms (before the read) at
/// which the change was first seen. Bounded: virtual time stands still once
/// nothing is scheduled, so a loop on the clock alone would never end.
fn started_until_guest_leaves(r: &Rig, u: &str, t0: u64, from: &str) -> u64 {
    for _ in 0..2000 {
        let at = r.now() - t0;
        assert_eq!(r.state(u), "started", "the pass reads `started`, at t+{at} ms");
        if guest_now(r, u) != from {
            return at;
        }
    }
    panic!("the guest never left {from}");
}

/// **Behaviour 69, ledger L113 — the REBOOTING medium** (no
/// `install_then=poweroff`; korp-installer < 0.1.4; the 2026-09-14 loop). Its
/// installer start reads `started` — which is what the tool's plain wait wants
/// (plan.rs:691-693), so it passes — and stays `started` for the whole
/// 900–1100 s pass (plan.rs: every measured pass ≥ 900 s; 1000–1100 s live
/// 2026-09-08); the eject works on the started box mid-pass (MEASURED
/// 2026-09-14, L52); at the end the guest reboots, and with the tray empty it
/// boots the disk.
#[test]
fn b69_a_rebooting_medium_reads_started_for_its_pass_and_then_boots_the_disk() {
    let r = Rig::new();
    let (u, t0) = rebooting_installer_start(&r);
    let took = tool_waits_for_started(&r, &u, t0);
    assert!(took <= 10_000, "the hypervisor start, not the pass: {took} ms");
    assert_eq!(guest_now(&r, &u), "Installing");

    // Mid-pass, as the tool's loop does it: eject on the STARTED box.
    assert_eq!(r.state(&u), "started");
    let mid = r.now() - t0;
    assert!(mid < 900_000, "still inside the pass at t+{mid} ms");
    let (s, v) = r.post(&format!("/1.3/server/{u}/cdrom/eject"), Value::Null);
    assert_eq!(s, 200, "the eject works on a started box mid-pass (L52): {v}");
    assert_eq!(guest_now(&r, &u), "Installing", "the pass is not over");

    // The rest of the pass reads `started`; the reboot finds no medium and the
    // disk boots, 900–1100 s after the start.
    let rebooted = started_until_guest_leaves(&r, &u, t0, "Installing");
    assert!((900_000..=1_110_000).contains(&rebooted), "the pass lasted {rebooted} ms");
    assert_eq!(guest_now(&r, &u), "Installed", "no medium in the tray: the reboot boots the disk");
    assert_eq!(r.state(&u), "started");
}

/// **The same medium left in, CD first: the reboot installs again** (17) —
/// and the API says `started` the whole time, so nothing outside the guest
/// can see the loop. Round after round, a new host key each time (33); only
/// an eject from outside ends it.
#[test]
fn b69_left_in_and_first_the_rebooting_medium_installs_again() {
    let r = Rig::new();
    let (u, t0) = rebooting_installer_start(&r);
    tool_waits_for_started(&r, &u, t0);
    let key0 = r.get(&format!("/mock/hostkey/{u}?as_is")).1.to_string();
    let first = started_until_guest_leaves(&r, &u, t0, "Installing");
    assert!((900_000..=1_110_000).contains(&first), "{first}");
    assert_eq!(guest_now(&r, &u), "Looping { rounds: 1 }", "the CD was first and loaded: the installer ran again");
    let second = started_until_guest_leaves(&r, &u, t0, "Looping { rounds: 1 }");
    assert!(second - first >= 900_000, "a whole pass again: {first} -> {second}");
    assert_eq!(guest_now(&r, &u), "Looping { rounds: 2 }");
    let key2 = r.get(&format!("/mock/hostkey/{u}?as_is")).1.to_string();
    assert_ne!(key2, key0, "every pass mints a new host key");
    // Only an eject from outside ends it; the next reboot boots the disk.
    assert_eq!(r.post(&format!("/1.3/server/{u}/cdrom/eject"), Value::Null).0, 200);
    started_until_guest_leaves(&r, &u, t0, "Looping { rounds: 2 }");
    assert_eq!(guest_now(&r, &u), "Installed");
}