//! The `w2-loopback-harness` acceptance rows — the harness proved
//! live: a full in-process connect → authenticate (fixture key through
//! `MemIdentityProvider`) → authenticated-session observation through
//! the wave-2 stack, plus the observation shapes (auth failure,
//! open-failure reason, host-key verdicts) the later waves' contract
//! tests speak through the same helpers.
//!
//! Every row drives the crate through the harness (`common::harness`)
//! — the recorded transport (tokio duplex pair, no TCP port) and the
//! recorded client posture (fixture keys, `MemHostKeyPolicy`, the
//! `Limits` keepalive/wall-clock posture) are the harness's own
//! decisions, exercised here.
//!
//! The rows are contract stamps: each names the pinning ADR § (the
//! greppable `Contract stamp:` inventory applies).

mod common;

use std::sync::Arc;
use std::time::Duration;

use alkcall::core::auth::IdentityProvider;
use common::harness::{
    LOOPBACK_HOST, LOOPBACK_PORT, authenticate_publickey, client_config, drive, remaining_methods,
    serve_config,
};
use common::{MemHostKeyPolicy, MemIdentityProvider, fixture, identity};
use russh::{Error, MethodSet};

use alkssh::{Limits, SshAdapter};

fn registered_provider(fingerprints: &[&str]) -> Arc<MemIdentityProvider> {
    let provider = Arc::new(MemIdentityProvider::new());
    for (index, registered) in fingerprints.iter().enumerate() {
        provider.register(registered, identity(&format!("user-{index}"), &["ssh"]));
    }
    provider
}

fn permissive_policy() -> Arc<MemHostKeyPolicy> {
    Arc::new(MemHostKeyPolicy::with_default(true))
}

/// - *Contract stamp:* ADR-001 (the mount's per-stream edge
///   termination, run to completion over the harness's recorded
///   in-process transport) + ADR-004 decision 1 (the fixture key's
///   fingerprint accepts through `MemIdentityProvider`) + ADR-005
///   decision 6 (the host key set loads decrypt-at-construction from
///   `MemHostKeySource` — the wave-1 doubles live through the wave-2
///   stack).
#[tokio::test]
async fn in_process_pair_connects_and_authenticates_end_to_end() {
    let provider = registered_provider(&[fixture("alpha").fingerprint]);
    let adapter = SshAdapter::new(serve_config(
        provider as Arc<dyn IdentityProvider>,
        &[fixture("bravo")],
        Limits::default(),
    ))
    .await
    .expect("adapter constructs");

    let host_key_policy = permissive_policy();
    let mut session = drive(adapter, host_key_policy, Limits::default())
        .await
        .expect("the in-process pair connects");

    let result = authenticate_publickey(&mut session.client, "anyone", fixture("alpha")).await;
    assert!(
        result.success(),
        "the fixture key authenticates: {result:?}"
    );

    let decisions = session.host_key_policy.decisions();
    assert_eq!(
        decisions.len(),
        1,
        "the host-key decision rode the MemHostKeyPolicy double: {decisions:?}"
    );
    assert!(decisions[0].accepted, "the permissive posture accepted");
    assert_eq!(decisions[0].presented.host, LOOPBACK_HOST);
    assert_eq!(decisions[0].presented.port, LOOPBACK_PORT);
    assert_eq!(
        decisions[0].presented.fingerprint,
        fixture("bravo").fingerprint,
        "the presented host key is the adapter's fixture set, fingerprinted on the crate's path"
    );

    session.unmount().await;
}

/// - *Contract stamp:* ADR-001 (the authenticated session is live and
///   serving over the mount) + `mapping.md` §Decline matrix (the
///   fail-closed decline is observable through the harness as a
///   ChannelOpenFailure reason — the *specific* reason is
///   `w2-dispatch-decline-core`'s row, which landed it; in builds with
///   a session-serving feature this row never sees a decline, so it
///   is pinned to the featureless builds it speaks about).
#[cfg(not(any(feature = "session", feature = "git")))]
#[tokio::test]
async fn authenticated_session_serves_and_declines_with_an_observed_reason() {
    use common::harness::open_session_channel;
    let provider = registered_provider(&[fixture("alpha").fingerprint]);
    let adapter = SshAdapter::new(serve_config(
        provider as Arc<dyn IdentityProvider>,
        &[fixture("bravo")],
        Limits::default(),
    ))
    .await
    .expect("adapter constructs");

    let mut session = drive(adapter, permissive_policy(), Limits::default())
        .await
        .expect("the in-process pair connects");

    let result = authenticate_publickey(&mut session.client, "anyone", fixture("alpha")).await;
    assert!(result.success(), "authenticated: {result:?}");

    let opened = open_session_channel(&session.client).await;
    assert!(
        matches!(opened, Err(Error::ChannelOpenFailure(_))),
        "the authenticated session answers a channel open with a decline reason: {opened:?}"
    );

    session.unmount().await;
}

/// - *Contract stamp:* ADR-004 decision 1 (`None` → reject — the
///   failure shape rides the harness with the advertisement intact)
///   + ADR-004 decision 4 (the remaining-methods set is exactly
///   `{publickey}`).
#[tokio::test]
async fn auth_failure_shape_and_remaining_advertisement_are_observed() {
    let provider = registered_provider(&[fixture("alpha").fingerprint]);
    let adapter = SshAdapter::new(serve_config(
        provider as Arc<dyn IdentityProvider>,
        &[fixture("bravo")],
        Limits::default(),
    ))
    .await
    .expect("adapter constructs");

    let mut session = drive(adapter, permissive_policy(), Limits::default())
        .await
        .expect("the in-process pair connects");

    let result = authenticate_publickey(&mut session.client, "anyone", fixture("delta")).await;
    assert!(!result.success(), "the unregistered key must fail");
    let remaining =
        remaining_methods(&result).expect("the failure shape carries the remaining-methods set");
    assert_eq!(
        remaining,
        MethodSet::from(&[russh::MethodKind::PublicKey][..]),
        "the publickey-only advertisement survives the decline"
    );

    session.unmount().await;
}

/// - *Contract stamp:* `auth.md` §Host-key policy (the crate is
///   policy-blind; the fail-closed verdict is the policy's, and the
///   reject disposition rides russh's hook as a hard, clean decline).
#[tokio::test]
async fn fail_closed_host_key_policy_verdict_disconnects_the_dial() {
    let provider = registered_provider(&[]);
    let adapter = SshAdapter::new(serve_config(
        provider as Arc<dyn IdentityProvider>,
        &[fixture("bravo")],
        Limits::default(),
    ))
    .await
    .expect("adapter constructs");

    let rejecting_policy = Arc::new(MemHostKeyPolicy::with_default(false));
    let outcome = drive(adapter, rejecting_policy, Limits::default()).await;

    let err = outcome
        .err()
        .expect("a rejecting policy never connects (the drive errs with the decline)");
    assert!(
        matches!(err, Error::UnknownKey),
        "the reject disposition is the clean dial decline: {err:?}"
    );
}

/// - *Contract stamp:* `auth.md` §Host-key policy case 1 (the
///   trust-anchor shape: an exact `(host, fingerprint)` pin accepts
///   with a fail-closed default for everything else) — the policy
///   double's rule vocabulary driven over the wire.
#[tokio::test]
async fn pinned_policy_rule_serves_the_trust_anchor_case() {
    let provider = registered_provider(&[fixture("alpha").fingerprint]);
    let adapter = SshAdapter::new(serve_config(
        provider as Arc<dyn IdentityProvider>,
        &[fixture("bravo")],
        Limits::default(),
    ))
    .await
    .expect("adapter constructs");

    let pinned_policy = Arc::new(MemHostKeyPolicy::with_default(false));
    pinned_policy.pin(LOOPBACK_HOST, fixture("bravo").fingerprint);

    let mut session = drive(adapter, pinned_policy, Limits::default())
        .await
        .expect("the pin accepts the presented host key");
    assert_eq!(
        session.host_key_policy.decisions().len(),
        1,
        "one decision, from the pin"
    );

    let result = authenticate_publickey(&mut session.client, "anyone", fixture("alpha")).await;
    assert!(result.success(), "the trust-anchored session authenticates");

    session.unmount().await;
}

/// - *Contract stamp:* `overview.md` §Bounded-resources policy (the
///   client direction's `Limits` postures map into the harness
///   client config verbatim: keepalive interval/max, wall clock
///   always set).
#[test]
fn client_config_maps_the_limits_postures() {
    let limits = Limits {
        keepalive_interval: Some(Duration::from_secs(7)),
        keepalive_max: 11,
        inactivity_timeout: Duration::from_secs(1_337),
        ..Limits::default()
    };
    let config = client_config(limits);

    assert_eq!(config.keepalive_interval, Some(Duration::from_secs(7)));
    assert_eq!(config.keepalive_max, 11);
    assert_eq!(config.inactivity_timeout, Some(Duration::from_secs(1_337)));

    let default_config = client_config(Limits::default());
    assert_eq!(
        default_config.keepalive_interval,
        alkssh::Limits::default().keepalive_interval,
        "the default posture rides by default"
    );
    assert_eq!(
        default_config.inactivity_timeout,
        Some(alkssh::Limits::default().inactivity_timeout)
    );
}