OpenFUT/openfut-utas-host/tests/host_test.rs

//! Integration tests for the FIFA 17 UTAS migration host: `/club` served from a
//! fake Core through the real adapter, faithful Python passthrough against a mock
//! upstream, route-classification safety, negatives, and an end-to-end socket run.

use std::collections::HashMap;
use std::io::{BufRead, BufReader, Read, Write};
use std::net::{TcpListener, TcpStream};
use std::sync::atomic::{AtomicUsize, Ordering};
use std::sync::Arc;

use openfut_adapter_fifa17::fut::catalog::Fifa17CardCatalog;
use openfut_adapter_fifa17::fut::club_response::{CoreOwnedItem, ItemIdentityResolver};
use openfut_adapter_fifa17::fut::entities::Fifa17Entities;
use openfut_identity::JsonIdentityStore;
use openfut_utas_host::account_store::AccountStore;
use openfut_utas_host::{
    classify, handle_club, handle_put_squad, handle_squad_active, handle_squad_list,
    handle_user_mass_info, read_request, ClubDeps, CoreAccess, CoreError, CoreExtState,
    CoreKitAssignments, CorePage, CoreReplaceRequest, CoreReplaceResult, CoreSquadRead,
    CoreSquadSlot, Fifa17IdentityResolver, HttpCoreClient, PassClient, Route, Server, SquadDeps,
};
use parking_lot::Mutex;
use serde_json::Value;

// ── Fakes ────────────────────────────────────────────────────────────────────

/// (method, target, body) recorded by the mock upstream.
type Recorded = Arc<Mutex<Vec<(String, String, Vec<u8>)>>>;

/// A recorded squad replacement (the fields a host test asserts on).
#[derive(Clone)]
struct StoredReplace {
    formation: Option<String>,
    slots: Vec<(String, i64, bool, bool)>, // owned_card_id, index, is_captain, is_on_bench
    ext_payload: String,
    chemistry: Option<i64>,
}

#[derive(Default)]
struct FakeCore {
    items: Vec<CoreOwnedItem>,
    total: i64,
    calls: AtomicUsize,
    read_calls: AtomicUsize,
    replace_calls: AtomicUsize,
    last_params: Mutex<Vec<(String, String)>>,
    /// The stored squad + extension returned by `read_squad_ext`. A successful
    /// `replace_squad` overwrites it (Fresh), enabling coupled read-after-write.
    squad: Mutex<Option<CoreSquadRead>>,
    /// The ownership-backed manager assignment, exactly as Core persists it: a
    /// full squad replacement writes it (or clears it with `None`).
    manager: Mutex<Option<String>>,
    replaced: Mutex<Vec<StoredReplace>>,
    panic_if_called: bool,
    return_err: bool,
}

impl FakeCore {
    fn new(items: Vec<CoreOwnedItem>, total: i64) -> Self {
        FakeCore {
            items,
            total,
            ..Default::default()
        }
    }
    fn forbidden() -> Self {
        FakeCore {
            panic_if_called: true,
            ..Default::default()
        }
    }
    fn erroring() -> Self {
        FakeCore {
            return_err: true,
            ..Default::default()
        }
    }
    /// Preset the stored squad read (for read-path tests without a prior PUT).
    fn with_squad(self, read: CoreSquadRead) -> Self {
        *self.squad.lock() = Some(read);
        self
    }
    fn calls(&self) -> usize {
        self.calls.load(Ordering::SeqCst)
    }
    fn read_calls(&self) -> usize {
        self.read_calls.load(Ordering::SeqCst)
    }
    fn replace_calls(&self) -> usize {
        self.replace_calls.load(Ordering::SeqCst)
    }
    fn replaced(&self) -> Vec<StoredReplace> {
        self.replaced.lock().clone()
    }
    fn last(&self) -> Vec<(String, String)> {
        self.last_params.lock().clone()
    }
    fn manager(&self) -> Option<String> {
        self.manager.lock().clone()
    }
}

impl CoreAccess for FakeCore {
    fn query_owned(&self, params: &[(&str, String)]) -> Result<CorePage, CoreError> {
        assert!(
            !self.panic_if_called,
            "Core must NOT be called on this path"
        );
        self.calls.fetch_add(1, Ordering::SeqCst);
        if self.return_err {
            return Err(CoreError::Status(500));
        }
        *self.last_params.lock() = params
            .iter()
            .map(|(k, v)| (k.to_string(), v.clone()))
            .collect();
        Ok(CorePage {
            items: self.items.clone(),
            total: self.total,
        })
    }

    fn read_squad_ext(&self, _namespace: &str) -> Result<CoreSquadRead, CoreError> {
        assert!(
            !self.panic_if_called,
            "Core must NOT be called on this path"
        );
        self.read_calls.fetch_add(1, Ordering::SeqCst);
        if self.return_err {
            return Err(CoreError::Status(500));
        }
        self.squad.lock().clone().ok_or(CoreError::Status(404))
    }

    fn replace_squad(&self, req: &CoreReplaceRequest) -> Result<CoreReplaceResult, CoreError> {
        assert!(
            !self.panic_if_called,
            "Core must NOT be called on this path"
        );
        self.replace_calls.fetch_add(1, Ordering::SeqCst);
        if self.return_err {
            return Err(CoreError::Status(500));
        }
        self.replaced.lock().push(StoredReplace {
            formation: req.formation.clone(),
            slots: req
                .slots
                .iter()
                .map(|s| {
                    (
                        s.owned_card_id.clone(),
                        s.index,
                        s.is_captain,
                        s.is_on_bench,
                    )
                })
                .collect(),
            ext_payload: req.ext_payload.clone(),
            chemistry: req.client_reported.chemistry,
        });
        // Reflect the committed state so a subsequent read is Fresh (deterministic
        // fingerprint by construction — the fake owns both sides).
        let slots = req
            .slots
            .iter()
            .map(|s| CoreSquadSlot {
                owned_card_id: s.owned_card_id.clone(),
                index: s.index,
                is_captain: s.is_captain,
                is_on_bench: s.is_on_bench,
            })
            .collect();
        let fingerprint = format!(
            "fp-{}-{}",
            req.formation.clone().unwrap_or_default(),
            req.slots.len()
        );
        *self.squad.lock() = Some(CoreSquadRead {
            name: req.name.clone().unwrap_or_default(),
            formation: req.formation.clone().unwrap_or_default(),
            slots,
            ext: CoreExtState::Fresh {
                schema_version: req.ext_schema_version,
                payload: req.ext_payload.clone(),
            },
        });
        Ok(CoreReplaceResult {
            squad_id: "sq-1".into(),
            canonical_fingerprint: fingerprint,
            slots_written: req.slots.len(),
        })
    }

    fn get_squad_manager(&self) -> Result<Option<String>, CoreError> {
        if self.return_err {
            return Err(CoreError::Status(500));
        }
        Ok(self.manager.lock().clone())
    }

    fn set_squad_manager(&self, owned_card_id: Option<&str>) -> Result<(), CoreError> {
        if self.return_err {
            return Err(CoreError::Status(500));
        }
        *self.manager.lock() = owned_card_id.map(str::to_string);
        Ok(())
    }
}

fn entities() -> Fifa17Entities {
    Fifa17Entities::from_maps(
        HashMap::from([(13, "Premier League".to_string())]),
        HashMap::from([(52, "Argentina".to_string())]),
        HashMap::from([(5, "Chelsea".to_string())]),
    )
}

fn item(
    owned: &str,
    card: &str,
    rating: u8,
    pos: &str,
    nation: &str,
    league: &str,
    club: &str,
) -> CoreOwnedItem {
    CoreOwnedItem {
        owned_card_id: owned.into(),
        card_id: card.into(),
        rating,
        position: pos.into(),
        nation: nation.into(),
        league: league.into(),
        club: club.into(),
        attributes: [90, 88, 70, 85, 40, 78],
    }
}

/// A mock Python upstream: records each request, replies 200 + `X-From-Python`.
fn spawn_mock_python() -> (String, Recorded) {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    let rec = Arc::new(Mutex::new(Vec::new()));
    let rec2 = rec.clone();
    std::thread::spawn(move || {
        for stream in listener.incoming() {
            let mut s = match stream {
                Ok(s) => s,
                Err(_) => continue,
            };
            let mut r = BufReader::new(s.try_clone().unwrap());
            if let Ok(Some(req)) = read_request(&mut r) {
                rec2.lock()
                    .push((req.method.clone(), req.target.clone(), req.body.clone()));
                let body = br#"{"python":true}"#;
                let head = format!(
                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nX-From-Python: 1\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
                    body.len()
                );
                let _ = s.write_all(head.as_bytes());
                let _ = s.write_all(body);
            }
        }
    });
    (format!("http://{addr}"), rec)
}

/// A unique temp path for a per-server identity store (integration tests run in
/// the same process; each server gets its own store file).
fn unique_store_path() -> std::path::PathBuf {
    static N: AtomicUsize = AtomicUsize::new(0);
    let n = N.fetch_add(1, Ordering::SeqCst);
    std::env::temp_dir().join(format!("ofut-host-it-{}-{n}.json", std::process::id()))
}

fn build_server(
    core: Arc<FakeCore>,
    upstream: &str,
    assets_map: Option<HashMap<String, u32>>,
) -> Server {
    // The map (card id → asset id) becomes a real identity catalog; an absent
    // map is an empty catalog (every item dropped — the honest no-mapping case).
    let cards = assets_map.unwrap_or_default();
    let entries: Vec<String> = cards
        .iter()
        .map(|(id, asset)| format!("\"{id}\":{{\"asset_id\":{asset}}}"))
        .collect();
    let doc = format!(
        "{{\"schema_version\":1,\"game\":\"fifa17\",\"cards\":{{{}}}}}",
        entries.join(",")
    );
    let catalog = Fifa17CardCatalog::from_json_str(&doc).unwrap();
    let store = JsonIdentityStore::open(unique_store_path()).unwrap();
    let resolver = Arc::new(Fifa17IdentityResolver::new(catalog, Arc::new(store)));
    Server::new(
        core,
        Arc::new(entities()),
        resolver,
        Arc::new(PassClient::new(upstream)),
        33_068_179,
    )
}

/// A real identity resolver over a `card_id -> asset_id` map (same construction
/// `build_server` uses), for tests that call `handle_club` directly.
fn resolver_for(cards: &[(&str, u32)]) -> Arc<Fifa17IdentityResolver> {
    let entries: Vec<String> = cards
        .iter()
        .map(|(id, asset)| format!("\"{id}\":{{\"asset_id\":{asset}}}"))
        .collect();
    let doc = format!(
        "{{\"schema_version\":1,\"game\":\"fifa17\",\"cards\":{{{}}}}}",
        entries.join(",")
    );
    let catalog = Fifa17CardCatalog::from_json_str(&doc).unwrap();
    let store = JsonIdentityStore::open(unique_store_path()).unwrap();
    Arc::new(Fifa17IdentityResolver::new(catalog, Arc::new(store)))
}

/// A card with an ACTIVE listing has LEFT the club: `/club` must not show it, and
/// pagination must run over the CLUB-VISIBLE set (never Core's unfiltered page,
/// which would hand back short pages).
#[test]
fn club_excludes_listed_items_and_paginates_the_visible_set() {
    let core = Arc::new(FakeCore::new(
        vec![
            item(
                "oc1",
                "card_a",
                86,
                "CDM",
                "Argentina",
                "Premier League",
                "Chelsea",
            ),
            item(
                "oc2",
                "card_b",
                85,
                "ST",
                "Argentina",
                "Premier League",
                "Chelsea",
            ),
            item(
                "oc3",
                "card_c",
                84,
                "CB",
                "Argentina",
                "Premier League",
                "Chelsea",
            ),
        ],
        3,
    ));
    let resolver = resolver_for(&[("card_a", 20801), ("card_b", 20802), ("card_c", 20803)]);
    let ents = entities();

    // oc2 has an active transfer-market listing.
    let hidden: std::collections::HashSet<String> = ["oc2".to_string()].into_iter().collect();
    let active_kits = CoreKitAssignments::default();
    let deps = ClubDeps {
        core: core.as_ref(),
        entities: &ents,
        assets: resolver.as_ref(),
        hidden: &hidden,
        active_kits: &active_kits,
    };
    let (resp, log) = handle_club("", &deps);
    let v: Value = serde_json::from_slice(&resp.body).unwrap();
    let assets: Vec<i64> = v["itemData"]
        .as_array()
        .unwrap()
        .iter()
        .map(|i| i["assetId"].as_i64().unwrap())
        .collect();
    assert_eq!(
        assets,
        vec![20801, 20803],
        "the transfer-pile card is not in the club"
    );
    assert_eq!(log.total, 2, "total is the club-visible count");

    // A page of 2 over a 2-item visible set is FULL — not short because a hidden
    // item consumed a slot.
    let (resp2, log2) = handle_club("count=2", &deps);
    let v2: Value = serde_json::from_slice(&resp2.body).unwrap();
    assert_eq!(
        v2["itemData"].as_array().unwrap().len(),
        2,
        "full-width page from the visible set"
    );
    assert_eq!(log2.total, 2);

    // Nothing hidden → all three player items remain visible.
    let none: std::collections::HashSet<String> = std::collections::HashSet::new();
    let deps_all = ClubDeps {
        core: core.as_ref(),
        entities: &ents,
        assets: resolver.as_ref(),
        hidden: &none,
        active_kits: &active_kits,
    };
    let (resp3, log3) = handle_club("", &deps_all);
    let v3: Value = serde_json::from_slice(&resp3.body).unwrap();
    assert_eq!(v3["itemData"].as_array().unwrap().len(), 3);
    assert_eq!(log3.total, 3);
}

#[test]
fn club_projects_only_owned_kits_with_active_designations() {
    let core = Arc::new(FakeCore::new(
        vec![
            item(
                "player",
                "card_player",
                90,
                "ST",
                "Argentina",
                "Premier League",
                "Chelsea",
            ),
            item("home", "kit_home", 0, "", "", "", ""),
            item("away", "kit_away", 0, "", "", "", ""),
        ],
        3,
    ));
    let catalog = Fifa17CardCatalog::from_json_str(
        r#"{"schema_version":1,"game":"fifa17","cards":{
            "card_player":{"asset_id":20801},
            "kit_home":{"asset_id":6300006,"kind":"kit","subtype":9,
                "card_asset_id":35,"team_id":21,"rareflag":0},
            "kit_away":{"asset_id":6400003,"kind":"kit","subtype":9,
                "card_asset_id":35,"team_id":21,"rareflag":0}
        }}"#,
    )
    .unwrap();
    let resolver = Arc::new(Fifa17IdentityResolver::new(
        catalog,
        Arc::new(JsonIdentityStore::open(unique_store_path()).unwrap()),
    ));
    let ents = entities();
    let hidden = std::collections::HashSet::new();
    let active_kits = CoreKitAssignments {
        home_owned_card_id: Some("home".into()),
        away_owned_card_id: Some("away".into()),
    };
    let deps = ClubDeps {
        core: core.as_ref(),
        entities: &ents,
        assets: resolver.as_ref(),
        hidden: &hidden,
        active_kits: &active_kits,
    };

    let (kit_response, kit_log) = handle_club("type=kit", &deps);
    let kits: Value = serde_json::from_slice(&kit_response.body).unwrap();
    assert_eq!(kit_log.total, 2);
    assert_eq!(kits["itemData"][0]["itemState"], "activeHomeKit");
    assert_eq!(kits["itemData"][1]["itemState"], "activeAwayKit");
    assert_eq!(kits["itemData"][0]["cardassetid"], 35);
    assert_eq!(kits["itemData"][0]["teamid"], 21);

    let (player_response, player_log) = handle_club("type=player", &deps);
    let players: Value = serde_json::from_slice(&player_response.body).unwrap();
    assert_eq!(player_log.total, 1);
    assert_eq!(players["itemData"].as_array().unwrap().len(), 1);
    assert_eq!(players["itemData"][0]["itemType"], "player");
}

// ── /club served from Core ─────────────────────────────────────────────────

#[test]
fn club_route_maps_query_and_shapes_core_items() {
    let core = Arc::new(FakeCore::new(
        vec![item(
            "oc1",
            "card_ch_1",
            86,
            "CDM",
            "Argentina",
            "Premier League",
            "Chelsea",
        )],
        1,
    ));
    let server = build_server(
        core.clone(),
        "http://127.0.0.1:1", // passthrough must not be used
        Some(HashMap::from([("card_ch_1".to_string(), 20801u32)])),
    );

    let resp = server.handle(
        "GET",
        "/ut/game/fifa17/club?year=2017&type=player&count=11&level=gold&nation=52&league=13&team=5&sort=desc&start=0",
        &[],
        b"",
    );
    assert_eq!(resp.status, 200);
    let body: Value = serde_json::from_slice(&resp.body).unwrap();
    let it = &body["itemData"][0];
    assert_eq!(it["resourceId"], 20801, "real asset id from the resolver");
    assert_eq!(it["rating"], 86);
    assert_eq!(it["preferredPosition"], "CDM");
    assert_eq!(it["leagueId"], 13);
    assert_eq!(it["teamid"], 5);
    assert_eq!(it["nation"], 52);

    // Core was queried once with SEMANTIC params (ids resolved to names), and the
    // FIFA UI window (start/count) became semantic offset/limit.
    assert_eq!(core.calls(), 1);
    let p = core.last();
    assert!(
        p.contains(&("quality".into(), "gold".into())),
        "level=gold -> quality {p:?}"
    );
    assert!(
        p.contains(&("league".into(), "Premier League".into())),
        "league id 13 -> name {p:?}"
    );
    assert!(
        p.contains(&("club".into(), "Chelsea".into())),
        "team id 5 -> club name {p:?}"
    );
    // The host must NOT ask Core for a window. Kind and transfer-pile membership
    // are host-side concepts Core cannot express, so the visible set only exists
    // after local filtering and shaping; a Core-side window would paginate the
    // unfiltered set and hand the client short pages.
    assert!(
        !p.iter().any(|(k, _)| k == "offset" || k == "limit"),
        "pagination must not be delegated to Core {p:?}"
    );
    // No raw FIFA id reached Core.
    for (_, v) in &p {
        if v == "13" || v == "5" || v == "52" {
            panic!("raw FIFA id leaked into Core params: {p:?}");
        }
    }
}

/// The client's `start`/`count` window is applied by the host, over the set that
/// survives kind + transfer-pile filtering. A window past the end is an empty
/// page, not the first item.
#[test]
fn club_window_is_applied_locally_after_filtering() {
    let core = Arc::new(FakeCore::new(
        vec![item(
            "oc1",
            "card_ch_1",
            86,
            "CDM",
            "Argentina",
            "Premier League",
            "Chelsea",
        )],
        1,
    ));
    let server = build_server(
        core.clone(),
        "http://127.0.0.1:1",
        Some(HashMap::from([("card_ch_1".to_string(), 20801u32)])),
    );

    let in_window = server.handle(
        "GET",
        "/ut/game/fifa17/club?type=player&start=0&count=11",
        &[],
        b"",
    );
    let body: Value = serde_json::from_slice(&in_window.body).unwrap();
    assert_eq!(body["itemData"].as_array().unwrap().len(), 1);

    let past_end = server.handle(
        "GET",
        "/ut/game/fifa17/club?type=player&start=10&count=11",
        &[],
        b"",
    );
    assert_eq!(
        past_end.status, 200,
        "a window past the end is still a page"
    );
    let body: Value = serde_json::from_slice(&past_end.body).unwrap();
    assert!(
        body["itemData"].as_array().unwrap().is_empty(),
        "offset past the visible set yields an empty page, not the first item"
    );
}

#[test]
fn club_route_with_empty_asset_map_drops_items_not_fakes_them() {
    // The current production reality: no card→asset mapping → empty itemData.
    let core = Arc::new(FakeCore::new(
        vec![item(
            "oc1",
            "card_pl_001",
            84,
            "ST",
            "England",
            "Premier League",
            "Northgate United",
        )],
        1,
    ));
    let server = build_server(core.clone(), "http://127.0.0.1:1", None);
    let resp = server.handle("GET", "/ut/game/fifa17/club?level=any", &[], b"");
    assert_eq!(resp.status, 200);
    let body: Value = serde_json::from_slice(&resp.body).unwrap();
    assert_eq!(
        body["itemData"].as_array().unwrap().len(),
        0,
        "no fabricated ids"
    );
    assert_eq!(
        core.calls(),
        1,
        "Core still queried; drop happens at shaping"
    );
}

#[test]
fn club_route_unknown_id_returns_empty_and_never_calls_core() {
    let core = Arc::new(FakeCore::forbidden());
    let server = build_server(core.clone(), "http://127.0.0.1:1", None);
    // league 9999 is not in the entity map → hard MapError → empty, no Core call.
    let resp = server.handle("GET", "/ut/game/fifa17/club?league=9999", &[], b"");
    assert_eq!(resp.status, 200);
    let body: Value = serde_json::from_slice(&resp.body).unwrap();
    assert_eq!(body["itemData"].as_array().unwrap().len(), 0);
    assert_eq!(core.calls(), 0, "unknown id must short-circuit before Core");
}

// ── Passthrough to Python ────────────────────────────────────────────────────

#[test]
fn passthrough_forwards_verbatim_and_never_calls_core() {
    let (upstream, rec) = spawn_mock_python();
    let core = Arc::new(FakeCore::forbidden()); // proves /club-only for Core
    let server = build_server(core, &upstream, None);

    let resp = server.handle(
        "POST",
        "/ut/game/fifa17/purchased/items",
        &[("X-UT-SID".into(), "sess".into())],
        br#"{"itemData":[{"id":1}]}"#,
    );
    // upstream response returned faithfully
    assert_eq!(resp.status, 200);
    assert!(
        resp.headers
            .iter()
            .any(|(k, v)| k.eq_ignore_ascii_case("x-from-python") && v == "1"),
        "upstream headers preserved: {:?}",
        resp.headers
    );
    let body: Value = serde_json::from_slice(&resp.body).unwrap();
    assert_eq!(body["python"], true);

    // upstream received the exact method, target and body
    std::thread::sleep(std::time::Duration::from_millis(50));
    let got = rec.lock().clone();
    assert_eq!(got.len(), 1);
    assert_eq!(got[0].0, "POST");
    assert_eq!(got[0].1, "/ut/game/fifa17/purchased/items");
    assert_eq!(got[0].2, br#"{"itemData":[{"id":1}]}"#);
}

#[test]
fn club_rename_is_rust_owned_persistent_and_never_calls_python_or_core() {
    assert_eq!(classify("PUT", "/ut/game/fifa17/club"), Route::ClubRename);
    assert_eq!(
        classify("PUT", "/ut/game/fifa17/user/club"),
        Route::ClubRename
    );
    assert_eq!(
        classify("POST", "/ut/game/fifa17/user/club"),
        Route::ClubRename
    );
    assert_eq!(
        classify("POST", "/ut/game/fifa17/squad/0"),
        Route::Passthrough
    );

    let (upstream, rec) = spawn_mock_python();
    let core = Arc::new(FakeCore::forbidden());
    let account_path = unique_store_path();
    let account = Arc::new(AccountStore::open(&account_path));
    let server = build_server(core.clone(), &upstream, None).with_account(account.clone());
    let resp = server.handle(
        "PUT",
        "/ut/game/fifa17/user/club",
        &[],
        br#"{"clubName":"Real FUT","clubAbbr":"RF"}"#,
    );
    assert_eq!(resp.status, 200);
    assert_eq!(
        serde_json::from_slice::<Value>(&resp.body).unwrap(),
        json!({})
    );
    assert_eq!(account.club().name, "Real FUT");
    assert_eq!(account.club().abbr, "RF");
    assert_eq!(AccountStore::open(&account_path).club(), account.club());
    assert_eq!(core.calls(), 0);
    assert_eq!(rec.lock().len(), 0, "rename never reached Python");
    let _ = std::fs::remove_file(account_path);
}

// ── End-to-end over a socket (read_request + write_response + keep-alive) ─────

#[test]
fn end_to_end_socket_serves_club_and_passthrough() {
    let (upstream, rec) = spawn_mock_python();
    let core = Arc::new(FakeCore::new(
        vec![item(
            "oc1",
            "card_ch_1",
            86,
            "CDM",
            "Argentina",
            "Premier League",
            "Chelsea",
        )],
        1,
    ));
    let server = build_server(
        core,
        &upstream,
        Some(HashMap::from([("card_ch_1".to_string(), 20801u32)])),
    );
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    std::thread::spawn(move || server.serve_listener(listener));

    // /club → Rust/Core
    let club = raw_get(
        &addr.to_string(),
        "/ut/game/fifa17/club?level=gold&league=13",
    );
    assert!(club.contains("200 OK"), "club status: {club}");
    assert!(
        club.contains("\"resourceId\":20801") || club.contains("\"resourceId\": 20801"),
        "club body: {club}"
    );

    // passthrough → Python
    let pt = raw_get(&addr.to_string(), "/ut/game/fifa17/tradePile?x=1");
    assert!(pt.contains("200 OK"));
    assert!(
        pt.to_lowercase().contains("x-from-python"),
        "upstream header relayed (case-insensitive): {pt}"
    );

    std::thread::sleep(std::time::Duration::from_millis(50));
    let got = rec.lock().clone();
    assert!(
        got.iter()
            .any(|(m, t, _)| m == "GET" && t == "/ut/game/fifa17/tradePile?x=1"),
        "python saw the passthrough with query intact: {got:?}"
    );
    assert!(
        !got.iter().any(|(_, t, _)| t.contains("/club")),
        "python must NOT have seen the /club request: {got:?}"
    );
}

/// Minimal raw HTTP/1.1 GET (Connection: close) returning the whole response text.
fn raw_get(addr: &str, target: &str) -> String {
    let mut s = TcpStream::connect(addr).unwrap();
    let req = format!("GET {target} HTTP/1.1\r\nHost: fifa\r\nConnection: close\r\n\r\n");
    s.write_all(req.as_bytes()).unwrap();
    let mut buf = String::new();
    s.read_to_string(&mut buf).unwrap();
    buf
}

#[test]
fn club_core_error_returns_empty_and_never_falls_back_to_python() {
    // A Core failure on /club must degrade to an empty page, NOT retry on Python
    // (which could double-apply a mutation on other routes; the rule is absolute).
    let (upstream, rec) = spawn_mock_python();
    let core = Arc::new(FakeCore::erroring());
    let server = build_server(
        core.clone(),
        &upstream,
        Some(HashMap::from([("c".into(), 1u32)])),
    );
    let resp = server.handle("GET", "/ut/game/fifa17/club?level=any", &[], b"");
    assert_eq!(resp.status, 200);
    let body: Value = serde_json::from_slice(&resp.body).unwrap();
    assert_eq!(body["itemData"].as_array().unwrap().len(), 0);
    assert_eq!(core.calls(), 1, "Core was attempted once");
    std::thread::sleep(std::time::Duration::from_millis(50));
    assert!(
        rec.lock().is_empty(),
        "Python must NOT be contacted on a /club core error"
    );
}

#[test]
fn host_does_not_refilter_core_results() {
    // Filtering is Core's job. The host must emit whatever Core returns; if it
    // re-applied the filter it would drop items Core already vetted.
    let core = Arc::new(FakeCore::new(
        vec![
            item(
                "oc1",
                "card_a",
                90,
                "ST",
                "Argentina",
                "Premier League",
                "Chelsea",
            ),
            item(
                "oc2",
                "card_b",
                60,
                "GK",
                "England",
                "Premier League",
                "Chelsea",
            ),
        ],
        2,
    ));
    let server = build_server(
        core,
        "http://127.0.0.1:1",
        Some(HashMap::from([
            ("card_a".into(), 20801u32),
            ("card_b".into(), 158023u32),
        ])),
    );
    // Query says gold; Core (faked) returns both regardless. Host must emit BOTH.
    let resp = server.handle("GET", "/ut/game/fifa17/club?level=gold", &[], b"");
    let body: Value = serde_json::from_slice(&resp.body).unwrap();
    assert_eq!(
        body["itemData"].as_array().unwrap().len(),
        2,
        "host must not second-guess Core's filtering"
    );
}

// ── End-to-end through the REAL resolver + Core over HTTP ────────────────────

/// A mock OpenFUT Core: records each request's header block and replies with a
/// fixed `/collection` body. Lets us prove the host talks to Core over real HTTP
/// (game header + query) and composes the real identity resolver.
fn spawn_mock_core(collection: Value) -> (String, Recorded) {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    let rec: Recorded = Arc::new(Mutex::new(Vec::new()));
    let rec2 = rec.clone();
    let body = collection.to_string();
    std::thread::spawn(move || {
        for stream in listener.incoming() {
            let mut stream = stream.unwrap();
            let mut reader = BufReader::new(stream.try_clone().unwrap());
            let mut headers = String::new();
            loop {
                let mut line = String::new();
                if reader.read_line(&mut line).unwrap() == 0 || line == "\r\n" {
                    break;
                }
                headers.push_str(&line);
            }
            rec2.lock().push(("GET".into(), headers, vec![]));
            let resp = format!(
                "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{}",
                body.len(),
                body
            );
            stream.write_all(resp.as_bytes()).unwrap();
        }
    });
    (format!("http://{addr}"), rec)
}

#[test]
fn club_end_to_end_through_real_resolver_and_sends_game_header() {
    // Core returns one asset-mapped item (fifa17_20801) and one synthetic item
    // with no catalog identity.
    let collection = serde_json::json!({
        "collection": [
            {"owned_card_id":"oc-mapped","effective_position":"ST","card":{"id":"fifa17_20801","overall":94,"position":"ST","nation":"Argentina","league":"Premier League","club":"Chelsea","pace":90,"shooting":92,"passing":81,"dribbling":91,"defending":33,"physical":80}},
            {"owned_card_id":"oc-synth","effective_position":"CB","card":{"id":"card_synth_1","overall":70,"position":"CB","nation":"England","league":"Premier League","club":"Chelsea","pace":60,"shooting":40,"passing":55,"dribbling":50,"defending":72,"physical":75}}
        ],
        "total": 2
    });
    let (core_url, rec) = spawn_mock_core(collection);

    let catalog = Fifa17CardCatalog::from_json_str(
        "{\"schema_version\":1,\"game\":\"fifa17\",\"cards\":{\"fifa17_20801\":{\"asset_id\":20801}}}",
    )
    .unwrap();
    let store = JsonIdentityStore::open(unique_store_path()).unwrap();
    let resolver = Arc::new(Fifa17IdentityResolver::new(catalog, Arc::new(store)));
    let server = Server::new(
        Arc::new(HttpCoreClient::new(core_url, "fifa17")),
        Arc::new(entities()),
        resolver,
        Arc::new(PassClient::new("http://127.0.0.1:1")),
        33_068_179,
    );

    let resp = server.handle("GET", "/ut/game/fifa17/club?level=gold", &[], b"");
    assert_eq!(resp.status, 200);
    let body: Value = serde_json::from_slice(&resp.body).unwrap();
    let items = body["itemData"].as_array().unwrap();
    assert_eq!(
        items.len(),
        1,
        "mapped item renders; synthetic (no catalog id) dropped"
    );
    assert_eq!(
        items[0]["resourceId"], 20801,
        "real asset id from the catalog"
    );
    assert!(
        items[0]["id"].as_i64().unwrap() >= 100_000_001,
        "wire id minted by the store, not a placeholder"
    );

    // The game header reached Core so it can select the fifa17 (all-mapped) profile.
    std::thread::sleep(std::time::Duration::from_millis(30));
    let got = rec.lock().clone();
    assert!(
        got.iter()
            .any(|(_, h, _)| h.to_lowercase().contains("x-openfut-game: fifa17")),
        "X-OpenFUT-Game: fifa17 sent to Core: {got:?}"
    );
}

// ── Squad host: routing, PUT pipeline, authorization, coupled read-after-write ─

use serde_json::json;

/// Build the production resolver over a card→asset catalog and PRE-ALLOCATE a
/// wire id for each owned item (as `/club` would before FIFA ever references
/// them). Returns the resolver + the wire id assigned to each owned_card_id.
fn resolver_with_wires(
    items: &[CoreOwnedItem],
    assets: &[(&str, u32)],
) -> (Fifa17IdentityResolver, HashMap<String, i64>) {
    let entries: Vec<String> = assets
        .iter()
        .map(|(id, a)| format!("\"{id}\":{{\"asset_id\":{a}}}"))
        .collect();
    let doc = format!(
        "{{\"schema_version\":1,\"game\":\"fifa17\",\"cards\":{{{}}}}}",
        entries.join(",")
    );
    let catalog = Fifa17CardCatalog::from_json_str(&doc).unwrap();
    let store = JsonIdentityStore::open(unique_store_path()).unwrap();
    let resolver = Fifa17IdentityResolver::new(catalog, Arc::new(store));
    let mut wires = HashMap::new();
    for it in items {
        if let Some(id) = resolver.resolve(it) {
            wires.insert(it.owned_card_id.clone(), id.item_id as i64);
        }
    }
    (resolver, wires)
}

/// A FIFA squad-save body. `players` = (index, wire id, kit number).
///
/// `manager` is explicit because a squad save carries an ownership-backed
/// manager assignment: a ref the resolver cannot map to an owned instance is an
/// unresolved wire id and the whole save is refused (you cannot manage with a
/// card you do not own). Tests that are not about the manager pass `None`.
fn put_body(
    formation: &str,
    captain: i64,
    players: &[(i64, i64, i64)],
    custom: &str,
    manager: Option<i64>,
) -> Vec<u8> {
    let ps: Vec<Value> = players
        .iter()
        .map(|(i, w, k)| json!({"index": i, "itemData": {"id": w, "dream": false}, "kitNumber": k}))
        .collect();
    let mgr: Vec<Value> = manager
        .into_iter()
        .map(|id| json!({"id": id, "dream": false}))
        .collect();
    json!({
        "id": 0,
        "formation": formation,
        "squadName": "OpenFUT",
        "squadType": "REGULAR_SQUAD",
        "chemistry": 52,
        "rating": 90,
        "starRating": 90,
        "captain": captain,
        "custom": custom,
        "manager": mgr,
        "players": ps,
        "kicktakers": [{"index": 0, "id": captain, "dream": false}],
    })
    .to_string()
    .into_bytes()
}

fn gk() -> CoreOwnedItem {
    item(
        "oc-a",
        "card_a",
        87,
        "GK",
        "Argentina",
        "Premier League",
        "Chelsea",
    )
}
fn st() -> CoreOwnedItem {
    item(
        "oc-b",
        "card_b",
        90,
        "ST",
        "Argentina",
        "Premier League",
        "Chelsea",
    )
}
const ASSETS: &[(&str, u32)] = &[("card_a", 20801), ("card_b", 158023)];

/// A mock Python returning a fixed userMassInfo JSON (userInfo/settings/... +
/// a Python-authored squad the overlay must replace).
fn spawn_mock_python_json(body: Value) -> (String, Recorded) {
    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
    let addr = listener.local_addr().unwrap();
    let rec = Arc::new(Mutex::new(Vec::new()));
    let rec2 = rec.clone();
    let text = body.to_string();
    std::thread::spawn(move || {
        for stream in listener.incoming() {
            let mut s = match stream {
                Ok(s) => s,
                Err(_) => continue,
            };
            let mut r = BufReader::new(s.try_clone().unwrap());
            if let Ok(Some(req)) = read_request(&mut r) {
                rec2.lock()
                    .push((req.method.clone(), req.target.clone(), req.body.clone()));
                let head = format!(
                    "HTTP/1.1 200 OK\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n",
                    text.len()
                );
                let _ = s.write_all(head.as_bytes());
                let _ = s.write_all(text.as_bytes());
            }
        }
    });
    (format!("http://{addr}"), rec)
}

// ── Classification ──────────────────────────────────────────────────────────

#[test]
fn classify_squad_and_usermassinfo_routes() {
    assert_eq!(
        classify("PUT", "/ut/game/fifa17/squad/0"),
        Route::SquadReplace
    );
    assert_eq!(
        classify("PUT", "/ut/game/fifa17/squad/3"),
        Route::SquadReplace
    );
    assert_eq!(
        classify("GET", "/ut/game/fifa17/squad/list"),
        Route::SquadList
    );
    assert_eq!(
        classify("GET", "/ut/game/fifa17/userMassInfo"),
        Route::UserMassInfo
    );
    // GET /squad/active and every numeric GET are the one Core-backed current
    // squad, matching the oracle's single-squad response regardless of URL id.
    assert_eq!(
        classify("GET", "/ut/game/fifa17/squad/active"),
        Route::SquadActive
    );
    assert_eq!(
        classify("PUT", "/ut/game/fifa17/squad/active"),
        Route::Passthrough
    );
    assert_eq!(
        classify("GET", "/ut/game/fifa17/squad/0"),
        Route::SquadActive
    );
    assert_eq!(
        classify("GET", "/ut/game/fifa17/squad/5"),
        Route::SquadActive
    );
    assert_eq!(
        classify("PUT", "/ut/game/fifa17/squad/list"),
        Route::Passthrough
    );
    assert_eq!(classify("GET", "/ut/game/fifa17/club"), Route::Club);
}

#[test]
fn numbered_squad_get_returns_current_core_squad_without_python() {
    let items = vec![gk(), st()];
    let (resolver, wires) = resolver_with_wires(&items, ASSETS);
    let core = Arc::new(FakeCore::new(items, 2));
    let (python, recorded) = spawn_mock_python();
    let server = Server::new(
        core,
        Arc::new(entities()),
        Arc::new(resolver),
        Arc::new(PassClient::new(&python)),
        33_068_179,
    );
    let put = server.handle(
        "PUT",
        "/ut/game/fifa17/squad/0",
        &[],
        &put_body(
            "f442",
            wires["oc-a"],
            &[(0, wires["oc-a"], 1), (1, wires["oc-b"], 9)],
            "[1,2,3]",
            None,
        ),
    );
    assert_eq!(put.status, 200);

    let active = server.handle("GET", "/ut/game/fifa17/squad/active", &[], b"");
    let numbered = server.handle("GET", "/ut/game/fifa17/squad/5", &[], b"");
    assert_eq!(numbered.status, 200);
    assert_eq!(numbered.body, active.body);
    assert_eq!(recorded.lock().len(), 0, "numeric GET never reached Python");
}

// ── PUT pipeline ────────────────────────────────────────────────────────────

#[test]
fn put_full_replacement_commits_canonical_and_extension_and_acks_id0() {
    let items = vec![gk(), st()];
    let (resolver, w) = resolver_with_wires(&items, ASSETS);
    let core = FakeCore::new(items.clone(), 2);
    let ent = entities();
    let deps = SquadDeps {
        core: &core,
        resolver: &resolver,
        entities: &ent,
    };
    let body = put_body(
        "f442",
        w["oc-a"],
        &[(0, w["oc-a"], 1), (1, w["oc-b"], 9)],
        "[1,2,3]",
        None,
    );

    let (resp, log) = handle_put_squad(&body, &deps);
    assert_eq!(resp.status, 200, "{log:?}");
    assert_eq!(resp.body, br#"{"id":0}"#, "exact save ack");

    let recs = core.replaced();
    assert_eq!(recs.len(), 1);
    let r = &recs[0];
    assert_eq!(r.formation.as_deref(), Some("f442"));
    // Canonical slots carry Core owned ids, never FIFA wire integers.
    let mut owned: Vec<&str> = r.slots.iter().map(|(o, ..)| o.as_str()).collect();
    owned.sort();
    assert_eq!(owned, vec!["oc-a", "oc-b"]);
    assert!(r.slots.iter().all(|(o, ..)| o.starts_with("oc-")));
    // Captain is the semantic owned item (oc-a), flagged in canonical.
    assert!(r.slots.iter().any(|(o, _, cap, _)| o == "oc-a" && *cap));
    // Extension payload round-trips the opaque custom + kit-by-owned-id.
    assert!(r.ext_payload.contains("[1,2,3]"), "custom in ext payload");
    assert_eq!(r.chemistry, Some(52), "client-reported shadow carried");
}

/// The squad's manager is an ownership-backed assignment, not an opaque wire
/// echo: a save assigns the owned instance behind the ref, and a later save
/// without a manager CLEARS it (a full replacement replaces the manager too).
#[test]
fn put_assigns_the_owned_manager_and_a_later_save_clears_it() {
    let items = vec![gk(), st()];
    let (resolver, w) = resolver_with_wires(&items, ASSETS);
    let core = FakeCore::new(items.clone(), 2);
    let ent = entities();
    let deps = SquadDeps {
        core: &core,
        resolver: &resolver,
        entities: &ent,
    };

    let with_manager = put_body(
        "f442",
        w["oc-a"],
        &[(0, w["oc-a"], 1)],
        "[]",
        Some(w["oc-b"]),
    );
    let (resp, log) = handle_put_squad(&with_manager, &deps);
    assert_eq!(resp.status, 200, "{log:?}");
    assert_eq!(
        core.manager().as_deref(),
        Some("oc-b"),
        "manager persisted as the Core owned id, never the wire id"
    );

    let without_manager = put_body("f442", w["oc-a"], &[(0, w["oc-a"], 1)], "[]", None);
    let (resp, log) = handle_put_squad(&without_manager, &deps);
    assert_eq!(resp.status, 200, "{log:?}");
    assert_eq!(
        core.manager(),
        None,
        "a full replacement without a manager clears the assignment"
    );
}

/// The REAL client always sends a manager ref, and on a real profile it does not
/// resolve: production's own save points at 100000427, which is absent from
/// production's `/club/staff`, and the client accepts that squad back unchanged.
/// Refusing the save would therefore break EVERY squad save, so an unresolvable
/// manager ref commits the squad with NO manager assignment.
#[test]
fn put_saves_the_squad_when_the_manager_ref_does_not_resolve() {
    let items = vec![gk(), st()];
    let (resolver, w) = resolver_with_wires(&items, ASSETS);
    let core = FakeCore::new(items.clone(), 2);
    let ent = entities();
    let deps = SquadDeps {
        core: &core,
        resolver: &resolver,
        entities: &ent,
    };

    let body = put_body(
        "f442",
        w["oc-a"],
        &[(0, w["oc-a"], 1)],
        "[]",
        Some(100_000_427),
    );
    let (resp, log) = handle_put_squad(&body, &deps);
    assert_eq!(resp.status, 200, "{log:?}");
    assert_eq!(resp.body, br#"{"id":0}"#);
    assert_eq!(core.replace_calls(), 1, "the squad itself is committed");
    assert_eq!(
        core.manager(),
        None,
        "no ownership-backed manager is invented from a ref we cannot map"
    );

    // An occupied PLAYER slot that does not resolve is the opposite case: it
    // would silently lose an owned card, so it still refuses the whole save.
    let bad_player = put_body(
        "f442",
        w["oc-a"],
        &[(0, w["oc-a"], 1), (1, 999_999_999, 9)],
        "[]",
        None,
    );
    let (resp, log) = handle_put_squad(&bad_player, &deps);
    assert_eq!(resp.status, 400);
    assert_eq!(log.outcome, "unresolved_wire_ids");
    assert_eq!(core.replace_calls(), 1, "nothing further committed");
}

#[test]
fn put_rejects_wire_id_owned_by_another_profile_core_unchanged() {
    // oc-b resolves (identity) but is NOT in the active club's owned set.
    let items = vec![gk(), st()];
    let (resolver, w) = resolver_with_wires(&items, ASSETS);
    let core = FakeCore::new(vec![gk()], 1); // only oc-a owned
    let ent = entities();
    let deps = SquadDeps {
        core: &core,
        resolver: &resolver,
        entities: &ent,
    };
    let body = put_body(
        "f442",
        w["oc-a"],
        &[(0, w["oc-a"], 1), (1, w["oc-b"], 9)],
        "[]",
        None,
    );

    let (resp, log) = handle_put_squad(&body, &deps);
    assert_eq!(resp.status, 403, "resolvable != authorized");
    assert_eq!(log.outcome, "unauthorized_item");
    assert_eq!(core.replace_calls(), 0, "Core squad left unchanged");
    assert!(core.replaced().is_empty());
}

#[test]
fn put_rejects_unknown_wire_id() {
    let items = vec![gk(), st()];
    let (resolver, w) = resolver_with_wires(&items, ASSETS);
    let core = FakeCore::new(items.clone(), 2);
    let ent = entities();
    let deps = SquadDeps {
        core: &core,
        resolver: &resolver,
        entities: &ent,
    };
    // 999_999_999 was never allocated → unresolvable.
    let body = put_body(
        "f442",
        w["oc-a"],
        &[(0, w["oc-a"], 1), (1, 999_999_999, 9)],
        "[]",
        None,
    );
    let (resp, log) = handle_put_squad(&body, &deps);
    assert_eq!(resp.status, 400);
    assert_eq!(log.outcome, "unresolved_wire_ids");
    assert_eq!(core.replace_calls(), 0);
}

#[test]
fn put_rejects_duplicate_owned_item() {
    let items = vec![gk(), st()];
    let (resolver, w) = resolver_with_wires(&items, ASSETS);
    let core = FakeCore::new(items.clone(), 2);
    let ent = entities();
    let deps = SquadDeps {
        core: &core,
        resolver: &resolver,
        entities: &ent,
    };
    let body = put_body(
        "f442",
        w["oc-a"],
        &[(0, w["oc-a"], 1), (1, w["oc-a"], 9)],
        "[]",
        None,
    );
    let (resp, log) = handle_put_squad(&body, &deps);
    assert_eq!(resp.status, 400);
    assert_eq!(log.outcome, "duplicate_owned_item");
    assert_eq!(core.replace_calls(), 0);
}

#[test]
fn put_core_failure_returns_error_never_python() {
    // handle_put_squad has no PassClient at all — a Core failure cannot fall back.
    let items = vec![gk(), st()];
    let (resolver, w) = resolver_with_wires(&items, ASSETS);
    let core = FakeCore::erroring();
    let ent = entities();
    let deps = SquadDeps {
        core: &core,
        resolver: &resolver,
        entities: &ent,
    };
    let body = put_body("f442", w["oc-a"], &[(0, w["oc-a"], 1)], "[]", None);
    let (resp, log) = handle_put_squad(&body, &deps);
    assert_eq!(resp.status, 502);
    assert_eq!(log.outcome, "core_error");
}

#[test]
fn repeated_identical_put_is_idempotent() {
    let items = vec![gk(), st()];
    let (resolver, w) = resolver_with_wires(&items, ASSETS);
    let core = FakeCore::new(items.clone(), 2);
    let ent = entities();
    let deps = SquadDeps {
        core: &core,
        resolver: &resolver,
        entities: &ent,
    };
    let body = put_body(
        "f442",
        w["oc-a"],
        &[(0, w["oc-a"], 1), (1, w["oc-b"], 9)],
        "[1,2,3]",
        None,
    );
    let (r1, _) = handle_put_squad(&body, &deps);
    let (r2, _) = handle_put_squad(&body, &deps);
    assert_eq!(r1.body, br#"{"id":0}"#);
    assert_eq!(r2.body, br#"{"id":0}"#);
    let recs = core.replaced();
    assert_eq!(recs.len(), 2);
    assert_eq!(recs[0].slots, recs[1].slots, "canonical converges");
    assert_eq!(recs[0].formation, recs[1].formation);
    assert_eq!(
        recs[0].ext_payload, recs[1].ext_payload,
        "extension converges"
    );
}

// ── Coupled read-after-write ─────────────────────────────────────────────────

/// PUT, then both reads (list + userMassInfo overlay) reflect the same committed
/// squad, projected via the shared identity path (real resourceIds + wire ids).
#[test]
fn coupled_read_after_write_list_and_usermassinfo_agree() {
    let items = vec![gk(), st()];
    let (resolver, w) = resolver_with_wires(&items, ASSETS);
    let core = FakeCore::new(items.clone(), 2);
    let ent = entities();
    let deps = SquadDeps {
        core: &core,
        resolver: &resolver,
        entities: &ent,
    };
    let body = put_body(
        "f442",
        w["oc-a"],
        &[(0, w["oc-a"], 1), (1, w["oc-b"], 9)],
        "[1,2,3]",
        None,
    );
    let (put, _) = handle_put_squad(&body, &deps);
    assert_eq!(put.status, 200);

    // GET /squad/list
    let (list_resp, list_log) = handle_squad_list(&deps);
    assert_eq!(list_log.outcome, "ok");
    let list: Value = serde_json::from_slice(&list_resp.body).unwrap();
    let entry = &list["squad"][0];
    assert_eq!(entry["formation"], "f442");
    assert_eq!(entry["id"], 0);
    assert_eq!(entry["squadType"], "REGULAR_SQUAD");

    // userMassInfo overlay against a mock Python
    let py_body = json!({
        "userInfo": {"personaId": 42, "clubName": "OpenFUT"},
        "settings": {"a": 1},
        "userData": {"b": 2},
        "pileSizeClientData": {"c": 3},
        "squad": {"id": 0, "personaId": 42, "players": [{"index": 0, "itemData": {"id": 777}, "kitNumber": 99}]},
    });
    let (py_url, _rec) = spawn_mock_python_json(py_body);
    let pass = PassClient::new(&py_url);
    let (umi_resp, umi_log) = handle_user_mass_info(
        "GET",
        "/ut/game/fifa17/userMassInfo",
        &[],
        b"",
        &deps,
        &pass,
    );
    assert_eq!(umi_log.outcome, "ok");
    let umi: Value = serde_json::from_slice(&umi_resp.body).unwrap();

    // Unrelated fields preserved verbatim.
    assert_eq!(umi["userInfo"]["personaId"], 42);
    assert_eq!(umi["settings"]["a"], 1);
    assert_eq!(umi["userData"]["b"], 2);
    assert_eq!(umi["pileSizeClientData"]["c"], 3);

    // .squad replaced by the Rust projection: persona preserved, envelope added.
    let sq = &umi["squad"];
    assert_eq!(sq["personaId"], 42, "persona preserved from Python");
    assert_eq!(sq["formation"], "f442");
    assert_eq!(
        sq["captain"], w["oc-a"],
        "captain is the wire id, not resourceId"
    );
    let occ: Vec<&Value> = sq["players"]
        .as_array()
        .unwrap()
        .iter()
        .filter(|p| p["itemData"]["id"].as_i64().unwrap() != 0)
        .collect();
    assert_eq!(occ.len(), 2, "the two committed players, not Python's 777");
    // DERIVED identity: wire id + real resourceId from the production catalog.
    let p0 = occ.iter().find(|p| p["index"] == 0).unwrap();
    assert_eq!(p0["itemData"]["id"], w["oc-a"]);
    assert_eq!(p0["itemData"]["resourceId"], 20801);
    assert_eq!(p0["kitNumber"], 1);
    let p1 = occ.iter().find(|p| p["index"] == 1).unwrap();
    assert_eq!(p1["itemData"]["resourceId"], 158023);
    assert_eq!(p1["kitNumber"], 9);
    assert!(
        umi["squad"]["players"]
            .as_array()
            .unwrap()
            .iter()
            .all(|p| p["itemData"]["id"] != 777),
        "no Python squad content survives"
    );

    // Content-Length matches the rewritten body.
    let cl = umi_resp
        .headers
        .iter()
        .find(|(k, _)| k.eq_ignore_ascii_case("content-length"))
        .map(|(_, v)| v.parse::<usize>().unwrap());
    assert_eq!(
        cl,
        Some(umi_resp.body.len()),
        "Content-Length fixed after overlay"
    );
}

/// `GET /squad/active` serves the Core-backed squad object at top level, stamped
/// with the host-configured persona (NOT proxied from Python), byte-consistent
/// with the committed squad.
#[test]
fn squad_active_serves_core_backed_object_with_configured_persona() {
    let items = vec![gk(), st()];
    let (resolver, w) = resolver_with_wires(&items, ASSETS);
    let core = FakeCore::new(items.clone(), 2);
    let ent = entities();
    let deps = SquadDeps {
        core: &core,
        resolver: &resolver,
        entities: &ent,
    };
    // Commit a squad so Core has a fresh canonical squad + extension.
    let body = put_body(
        "f442",
        w["oc-a"],
        &[(0, w["oc-a"], 1), (1, w["oc-b"], 9)],
        "[1,2,3]",
        None,
    );
    let (put, _) = handle_put_squad(&body, &deps);
    assert_eq!(put.status, 200);

    const PERSONA: i64 = 33_068_179;
    let (resp, log) = handle_squad_active(&deps, PERSONA);
    assert_eq!(log.outcome, "ok");
    assert_eq!(resp.status, 200);
    let sq: Value = serde_json::from_slice(&resp.body).unwrap();

    // Top-level squad object (not wrapped), persona from host config not Python.
    assert_eq!(sq["id"], 0);
    assert_eq!(sq["personaId"], PERSONA, "persona from host config");
    assert_eq!(sq["formation"], "f442");
    assert_eq!(sq["captain"], w["oc-a"], "captain is the wire id");
    assert_eq!(sq["changed"], 0);
    assert!(sq["actives"].as_array().unwrap().is_empty());
    let occ: Vec<&Value> = sq["players"]
        .as_array()
        .unwrap()
        .iter()
        .filter(|p| p["itemData"]["id"].as_i64().unwrap() != 0)
        .collect();
    assert_eq!(occ.len(), 2, "the two committed players");
    let p0 = occ.iter().find(|p| p["index"] == 0).unwrap();
    assert_eq!(p0["itemData"]["id"], w["oc-a"]);
    assert_eq!(p0["itemData"]["resourceId"], 20801);
}

#[test]
fn read_path_is_bounded_no_per_slot_lookup() {
    let items = vec![gk(), st()];
    let (resolver, w) = resolver_with_wires(&items, ASSETS);
    let core = FakeCore::new(items.clone(), 2);
    let ent = entities();
    let deps = SquadDeps {
        core: &core,
        resolver: &resolver,
        entities: &ent,
    };
    let body = put_body(
        "f442",
        w["oc-a"],
        &[(0, w["oc-a"], 1), (1, w["oc-b"], 9)],
        "[]",
        None,
    );
    handle_put_squad(&body, &deps);
    let reads_before = core.read_calls();
    let owned_before = core.calls();
    let (_r, log) = handle_squad_list(&deps);
    assert_eq!(log.outcome, "ok");
    assert_eq!(
        core.read_calls() - reads_before,
        1,
        "exactly one squad read"
    );
    assert_eq!(
        core.calls() - owned_before,
        1,
        "exactly one batch owned fetch (no per-slot)"
    );
}

// ── Stale / Missing integrity ────────────────────────────────────────────────

fn read_with_ext(ext: CoreExtState) -> CoreSquadRead {
    CoreSquadRead {
        name: "OpenFUT".into(),
        formation: "f442".into(),
        slots: vec![CoreSquadSlot {
            owned_card_id: "oc-a".into(),
            index: 0,
            is_captain: true,
            is_on_bench: false,
        }],
        ext,
    }
}

#[test]
fn stale_extension_is_not_applied_on_reads() {
    let items = vec![gk()];
    let (resolver, _w) = resolver_with_wires(&items, ASSETS);
    let core = FakeCore::new(items.clone(), 1).with_squad(read_with_ext(CoreExtState::Stale {
        schema_version: 1,
        payload: "{}".into(),
    }));
    let ent = entities();
    let deps = SquadDeps {
        core: &core,
        resolver: &resolver,
        entities: &ent,
    };
    let (resp, log) = handle_squad_list(&deps);
    assert_eq!(log.outcome, "stale_integrity");
    let v: Value = serde_json::from_slice(&resp.body).unwrap();
    assert_eq!(
        v["squad"].as_array().unwrap().len(),
        0,
        "stale never projected"
    );
}

#[test]
fn missing_extension_is_explicit_on_reads() {
    let items = vec![gk()];
    let (resolver, _w) = resolver_with_wires(&items, ASSETS);
    let core = FakeCore::new(items.clone(), 1).with_squad(read_with_ext(CoreExtState::Missing));
    let ent = entities();
    let deps = SquadDeps {
        core: &core,
        resolver: &resolver,
        entities: &ent,
    };
    let (resp, log) = handle_squad_list(&deps);
    assert_eq!(log.outcome, "missing_integrity");
    let v: Value = serde_json::from_slice(&resp.body).unwrap();
    assert_eq!(
        v["squad"].as_array().unwrap().len(),
        0,
        "missing never fabricated"
    );
}

#[test]
fn usermassinfo_never_serves_python_squad_on_integrity_failure() {
    let items = vec![gk()];
    let (resolver, _w) = resolver_with_wires(&items, ASSETS);
    let core = FakeCore::new(items.clone(), 1).with_squad(read_with_ext(CoreExtState::Missing));
    let ent = entities();
    let deps = SquadDeps {
        core: &core,
        resolver: &resolver,
        entities: &ent,
    };
    let py_body = json!({
        "userInfo": {"personaId": 7},
        "squad": {"id": 0, "players": [{"index": 0, "itemData": {"id": 777}, "kitNumber": 5}]},
    });
    let (py_url, _rec) = spawn_mock_python_json(py_body);
    let pass = PassClient::new(&py_url);
    let (resp, log) = handle_user_mass_info(
        "GET",
        "/ut/game/fifa17/userMassInfo",
        &[],
        b"",
        &deps,
        &pass,
    );
    assert_eq!(log.outcome, "missing_integrity");
    let v: Value = serde_json::from_slice(&resp.body).unwrap();
    assert_eq!(
        v["userInfo"]["personaId"], 7,
        "unrelated fields still Python"
    );
    // Rust owns squad: Python's squad (id 777) must NOT survive.
    assert!(v["squad"]["players"].as_array().unwrap().is_empty());
    assert_eq!(
        v["squad"]["personaId"], 7,
        "persona preserved, squad emptied"
    );
}

#[test]
fn unrelated_route_still_reaches_python() {
    let core = Arc::new(FakeCore::forbidden());
    let (py_url, rec) = spawn_mock_python();
    let server = build_server(core, &py_url, None);
    let resp = server.handle("POST", "/ut/game/fifa17/packs/purchase", &[], b"{}");
    assert_eq!(resp.status, 200);
    assert!(resp
        .headers
        .iter()
        .any(|(k, _)| k.eq_ignore_ascii_case("x-from-python")));
    assert_eq!(rec.lock().len(), 1, "reached Python");
}

#[test]
fn duplicate_definition_instances_stay_distinct_through_host() {
    // Two owned copies of ONE definition (card_a): distinct owned ids and wire
    // ids, one shared resourceId — must not collapse anywhere in the host.
    let a = item(
        "oc-a",
        "card_a",
        87,
        "GK",
        "Argentina",
        "Premier League",
        "Chelsea",
    );
    let c = item(
        "oc-c",
        "card_a",
        87,
        "ST",
        "Argentina",
        "Premier League",
        "Chelsea",
    );
    let items = vec![a, c];
    let (resolver, w) = resolver_with_wires(&items, &[("card_a", 20801)]);
    assert_ne!(w["oc-a"], w["oc-c"], "two copies get distinct wire ids");
    let core = FakeCore::new(items.clone(), 2);
    let ent = entities();
    let deps = SquadDeps {
        core: &core,
        resolver: &resolver,
        entities: &ent,
    };
    let body = put_body(
        "f442",
        w["oc-a"],
        &[(0, w["oc-a"], 1), (1, w["oc-c"], 7)],
        "[]",
        None,
    );
    let (put, _) = handle_put_squad(&body, &deps);
    assert_eq!(put.status, 200);

    let py = json!({"userInfo": {"personaId": 1}, "squad": {"players": []}});
    let (url, _r) = spawn_mock_python_json(py);
    let pass = PassClient::new(&url);
    let (umi, _) = handle_user_mass_info(
        "GET",
        "/ut/game/fifa17/userMassInfo",
        &[],
        b"",
        &deps,
        &pass,
    );
    let v: Value = serde_json::from_slice(&umi.body).unwrap();
    let occ: Vec<&Value> = v["squad"]["players"]
        .as_array()
        .unwrap()
        .iter()
        .filter(|p| p["itemData"]["id"].as_i64().unwrap() != 0)
        .collect();
    assert_eq!(occ.len(), 2);
    assert_ne!(
        occ[0]["itemData"]["id"], occ[1]["itemData"]["id"],
        "two copies of one definition keep distinct wire ids"
    );
    for p in &occ {
        assert_eq!(p["itemData"]["resourceId"], 20801, "shared asset id");
    }
}

// ── Non-economy routes owned by Rust (no Python fallback, no Core) ──────────────

/// The migrated non-economy static routes are served entirely by Rust: exact
/// oracle-matching bodies, never proxied to Python, never touching Core. The
/// security-question route fails closed (400) for an unknown session in Rust —
/// again without any Python fallback. This is the per-domain no-fallback proof.
#[test]
fn non_economy_routes_rust_owned_no_python_no_core() {
    let core = Arc::new(FakeCore::forbidden()); // any Core call panics
    let (py_url, rec) = spawn_mock_python();
    let server = build_server(core, &py_url, None);

    let cases: &[(&str, &str, serde_json::Value)] = &[
        (
            "GET",
            "/ut/game/fifa17/user/accountinfo",
            serde_json::json!({}),
        ),
        (
            "GET",
            "/ut/game/fifa17/settings",
            serde_json::json!({ "configs": [] }),
        ),
        (
            "GET",
            "/ut/game/fifa17/leaderboards/options",
            serde_json::json!({}),
        ),
        ("PUT", "/ut/game/fifa17/match/reset", serde_json::json!({})),
        (
            "GET",
            "/ut/game/fifa17/club/stats/staff",
            serde_json::json!({}),
        ),
    ];
    for (m, p, want) in cases {
        let resp = server.handle(m, p, &[], b"");
        assert_eq!(resp.status, 200, "{m} {p} status");
        let body: Value = serde_json::from_slice(&resp.body).unwrap();
        assert_eq!(&body, want, "{m} {p} body");
    }

    // Security-question with an unknown session fails closed in Rust (never proxied).
    let resp = server.handle(
        "GET",
        "/ut/game/fifa17/phishing/trusteddevice?deviceId=6236375476659cd0f6c780e728774b71",
        &[("X-UT-SID".into(), "unknown-sid".into())],
        b"",
    );
    assert_eq!(resp.status, 400);
    let body: Value = serde_json::from_slice(&resp.body).unwrap();
    assert_eq!(body, serde_json::json!({ "reason": "invalid_session" }));

    // None of the migrated routes reached the Python upstream.
    assert_eq!(
        rec.lock().len(),
        0,
        "no Python fallback for migrated non-economy routes"
    );
}

/// `GET /hub` is served from Rust: `clubPlayers` counts owned PLAYER cards in
/// Core, auction/tradePile counts come from the durable market store (0 with no
/// economy wired), and Python is never consulted.
#[test]
fn hub_counts_players_from_core_no_python() {
    let items = vec![
        item("oc1", "card_a", 84, "ST", "Brazil", "La Liga", "Barcelona"),
        item("oc2", "card_b", 80, "CM", "Spain", "La Liga", "Real Madrid"),
        item("oc3", "card_c", 77, "CB", "France", "Ligue 1", "PSG"),
    ];
    let core = Arc::new(FakeCore::new(items, 3));
    let (py_url, rec) = spawn_mock_python();
    let server = build_server(core, &py_url, None);

    let resp = server.handle("GET", "/ut/game/fifa17/hub", &[], b"");
    assert_eq!(resp.status, 200);
    let body: Value = serde_json::from_slice(&resp.body).unwrap();
    assert_eq!(body["clubPlayers"], 3, "counts owned player cards");
    assert_eq!(body["auctionCount"], 0, "no economy wired => 0 listings");
    assert_eq!(
        body["tradePile"],
        serde_json::json!({ "count": 0, "selling": 0, "sold": 0 })
    );
    assert_eq!(rec.lock().len(), 0, "hub never reaches Python");
}

/// `GET /club/stats/year` is served from Rust with Core-accurate player tier
/// counts (contextId 1 global bucket), never reaching Python.
#[test]
fn club_stats_year_counts_tiers_from_core_no_python() {
    let items = vec![
        item("oc1", "card_a", 90, "ST", "Brazil", "La Liga", "Barcelona"), // gold
        item("oc2", "card_b", 70, "CM", "Spain", "La Liga", "Real Madrid"), // silver
        item("oc3", "card_c", 60, "CB", "France", "Ligue 1", "PSG"),       // bronze
    ];
    let core = Arc::new(FakeCore::new(items, 3));
    let (py_url, rec) = spawn_mock_python();
    let server = build_server(core, &py_url, None);

    let resp = server.handle("GET", "/ut/game/fifa17/club/stats/year", &[], b"");
    assert_eq!(resp.status, 200);
    let body: Value = serde_json::from_slice(&resp.body).unwrap();
    let g: std::collections::HashMap<String, i64> = body["stat"]
        .as_array()
        .unwrap()
        .iter()
        .filter(|r| r["contextId"] == 1)
        .map(|r| {
            (
                r["type"].as_str().unwrap().to_string(),
                r["typeValue"].as_i64().unwrap(),
            )
        })
        .collect();
    assert_eq!(g["players"], 3);
    assert_eq!(g["playersGold"], 1);
    assert_eq!(g["playersSilver"], 1);
    assert_eq!(g["playersBronze"], 1);
    assert_eq!(g["consumables"], 0);
    assert_eq!(g["staff"], 0);
    assert_eq!(rec.lock().len(), 0, "club/stats never reaches Python");
}