//! End-to-end: CI against the real `forge` binary, driven by the real `forge-runner` library.
//!
//! The runner side uses `HostExecutor` (behind the `test-host-executor` feature), so jobs run as
//! plain `sh` on the test host. Everything else is the production path: runner registration over
//! the HTTP API, claims, the job's git fetch over smart HTTP with its job token, log shipping with
//! masking, artifact upload, and finish.
//!
//! Scenarios:
//! - a push run: secrets are masked in the log, artifacts download, a failure skips dependants;
//! - a fork pull request: it waits for approval, then runs without secrets;
//! - a live log: `log.txt` streams while a job runs and ends when the job finishes.

mod common;
use common::{BIN, Server, launch, pick_ports, write_config};
use forge_ci::proto::{JobAssignment, Outcome, RegisterRequest};
use forge_runner::client::ForgeClient;
use forge_runner::config::{Limits, RunnerConfig};
use forge_runner::executor::HostExecutor;
use forge_runner::runner::{RunOptions, Runner};
use std::collections::BTreeMap;
use std::fs;
use std::path::{Path, PathBuf};
use std::process::{Command, Output};
use std::sync::Arc;
use std::time::{Duration, Instant};
use tokio_util::sync::CancellationToken;

use reqwest::header::{LOCATION, SET_COOKIE};

const ALICE_PW: &str = "alice-password-123";
const BOB_PW: &str = "bob-password-4567";
const BOB_EMAIL: &str = "bob@users.noreply.127.0.0.1";
const REPO_PATH: &str = "/alice/r";
const SECRET_VALUE: &str = "supersecretvalue";
const LABELS: &[&str] = &["linux"];

/// The push-triggered workflow. Image strings are ignored by `HostExecutor`.
const CI_TOML: &str = r#"[on]
push = ["main"]
pull_request = true

[jobs.build]
image = "docker.io/library/alpine:3"
run = "echo hello; echo $MY_SECRET; mkdir -p out; echo data > out/a.txt"
secrets = ["MY_SECRET"]
artifacts = ["out/*.txt"]

[jobs.after]
image = "docker.io/library/alpine:3"
run = "echo after"
needs = ["build"]

[jobs.broken]
image = "docker.io/library/alpine:3"
run = "exit 3"

[jobs.skipme]
image = "docker.io/library/alpine:3"
run = "echo never"
needs = ["broken"]
"#;

/// The same workflow with the build line changed to print the secret in a form that stays
/// readable when the variable is unset. Bob's pull request uses it, so `set -u` does not abort.
const CI_TOML_FORK: &str = r#"[on]
push = ["main"]
pull_request = true

[jobs.build]
image = "docker.io/library/alpine:3"
run = "echo hello; echo \"value=[${MY_SECRET-}]\"; mkdir -p out; echo data > out/a.txt"
secrets = ["MY_SECRET"]
artifacts = ["out/*.txt"]

[jobs.after]
image = "docker.io/library/alpine:3"
run = "echo after"
needs = ["build"]

[jobs.broken]
image = "docker.io/library/alpine:3"
run = "exit 3"

[jobs.skipme]
image = "docker.io/library/alpine:3"
run = "echo never"
needs = ["broken"]
"#;

/// A parsed HTTP response. Only the parts the test needs are kept.
struct Resp {
    status: u16,
    location: Option<String>,
    set_cookies: Vec<String>,
    body: String,
}

struct Http {
    client: reqwest::Client,
    base: String,
}

impl Http {
    fn new(port: u16) -> Self {
        let client = reqwest::Client::builder()
            .redirect(reqwest::redirect::Policy::none())
            .timeout(Duration::from_secs(60))
            .build()
            .expect("http client");
        Self {
            client,
            base: format!("http://127.0.0.1:{port}"),
        }
    }

    async fn get(&self, path: &str, cookie: Option<&str>) -> Resp {
        self.send(self.client.get(format!("{}{path}", self.base)), cookie)
            .await
    }

    async fn get_bearer(&self, path: &str, token: &str) -> Resp {
        let req = self
            .client
            .get(format!("{}{path}", self.base))
            .bearer_auth(token);
        self.send(req, None).await
    }

    async fn post_form(&self, path: &str, form: &[(&str, &str)], cookie: Option<&str>) -> Resp {
        self.send(
            self.client.post(format!("{}{path}", self.base)).form(form),
            cookie,
        )
        .await
    }

    async fn send(&self, req: reqwest::RequestBuilder, cookie: Option<&str>) -> Resp {
        let req = match cookie {
            Some(cookie) => req.header(reqwest::header::COOKIE, cookie),
            None => req,
        };
        let res = req.send().await.expect("http request");
        let status = res.status().as_u16();
        let location = res
            .headers()
            .get(LOCATION)
            .and_then(|v| v.to_str().ok())
            .map(str::to_owned);
        let set_cookies = res
            .headers()
            .get_all(SET_COOKIE)
            .iter()
            .filter_map(|v| v.to_str().ok().map(str::to_owned))
            .collect();
        let body = res.text().await.unwrap_or_default();
        Resp {
            status,
            location,
            set_cookies,
            body,
        }
    }
}

/// Returns `name=value` for the cookie whose name ends with `suffix`. Cleared cookies are ignored.
fn cookie_named(resp: &Resp, suffix: &str) -> Option<String> {
    resp.set_cookies.iter().find_map(|set| {
        let pair = set.split(';').next()?;
        let (name, value) = pair.split_once('=')?;
        (name.ends_with(suffix) && !value.is_empty()).then(|| format!("{name}={value}"))
    })
}

/// The value of the first `<input name="...">` on a page.
fn input_value(body: &str, name: &str) -> String {
    let doc = scraper::Html::parse_document(body);
    let selector = scraper::Selector::parse(&format!("input[name=\"{name}\"]")).expect("selector");
    doc.select(&selector)
        .next()
        .and_then(|el| el.value().attr("value"))
        .unwrap_or_else(|| panic!("no input named {name} on page"))
        .to_owned()
}

/// The text of the first element matching `selector`.
fn first_text(body: &str, selector: &str) -> String {
    let doc = scraper::Html::parse_document(body);
    let selector = scraper::Selector::parse(selector).expect("selector");
    doc.select(&selector)
        .next()
        .unwrap_or_else(|| panic!("no element matches {selector:?}"))
        .text()
        .collect::<String>()
        .trim()
        .to_owned()
}

fn run_admin(config: &Path, username: &str, extra: &[&str], password: &str) {
    let out = Command::new(BIN)
        .arg("--config")
        .arg(config)
        .args(["admin", "create-user", username])
        .args(extra)
        .env("FORGE_PASSWORD", password)
        .output()
        .expect("run admin create-user");
    ok(&out, &format!("create-user {username}"));
}

fn ok(out: &Output, what: &str) {
    assert!(
        out.status.success(),
        "{what} failed:\n{}",
        String::from_utf8_lossy(&out.stderr)
    );
}

/// Runs git in `cwd` with `env` applied on top of the inherited environment. Returns stdout.
fn git_env(cwd: &Path, args: &[&str], env: &[(&str, &str)]) -> String {
    let mut cmd = Command::new("git");
    cmd.current_dir(cwd)
        .args(["-c", "credential.helper="])
        .args(args)
        .env("GIT_TERMINAL_PROMPT", "0");
    for (key, value) in env {
        cmd.env(key, value);
    }
    let out = cmd.output().expect("run git");
    assert!(
        out.status.success(),
        "git {args:?} failed:\n{}",
        String::from_utf8_lossy(&out.stderr)
    );
    String::from_utf8_lossy(&out.stdout).trim().to_owned()
}

fn git(cwd: &Path, args: &[&str]) -> String {
    git_env(cwd, args, &[])
}

/// Writes `file` (creating its directory), commits it with `who` as author and committer, and
/// returns the new HEAD.
fn commit(
    repo: &Path,
    file: &str,
    contents: &str,
    message: &str,
    who: &str,
    email: &str,
) -> String {
    let path = repo.join(file);
    if let Some(parent) = path.parent() {
        fs::create_dir_all(parent).expect("create dir");
    }
    fs::write(&path, contents).expect("write file");
    git(repo, &["add", file]);
    git_env(
        repo,
        &["-c", "commit.gpgsign=false", "commit", "-q", "-m", message],
        &[
            ("GIT_AUTHOR_NAME", who),
            ("GIT_AUTHOR_EMAIL", email),
            ("GIT_COMMITTER_NAME", who),
            ("GIT_COMMITTER_EMAIL", email),
        ],
    );
    git(repo, &["rev-parse", "HEAD"])
}

/// Clones `url` into `into` (a path that does not exist yet) and returns it.
fn clone(url: &str, into: &Path) -> PathBuf {
    let parent = into.parent().expect("clone target has a parent");
    git(
        parent,
        &["clone", "-q", url, into.to_str().expect("utf-8 path")],
    );
    into.to_path_buf()
}

/// The HTTPS remote for `owner/repo`, authenticated with a token.
fn remote(port: u16, user: &str, token: &str, owner: &str, repo: &str) -> String {
    format!("http://{user}:{token}@127.0.0.1:{port}/{owner}/{repo}.git")
}

/// The CSRF token shown on any settings page of the signed-in user.
async fn csrf_from(http: &Http, path: &str, session: &str) -> String {
    let page = http.get(path, Some(session)).await;
    assert_eq!(page.status, 200, "GET {path}");
    input_value(&page.body, "csrf")
}

/// Logs in through the real `/login` form and returns the `__Host-forge_session` cookie header.
async fn login(http: &Http, username: &str, password: &str) -> String {
    let page = http.get("/login", None).await;
    assert_eq!(page.status, 200, "GET /login");
    let precsrf_cookie = cookie_named(&page, "forge_precsrf").expect("precsrf cookie");
    let precsrf = input_value(&page.body, "precsrf");
    let res = http
        .post_form(
            "/login",
            &[
                ("username", username),
                ("password", password),
                ("next", "/"),
                ("precsrf", &precsrf),
            ],
            Some(&precsrf_cookie),
        )
        .await;
    assert_eq!(res.status, 303, "login as {username}: {}", res.body);
    cookie_named(&res, "forge_session").expect("session cookie after login")
}

/// Creates an access token through `/settings/tokens` and returns the token shown once.
async fn create_token(http: &Http, session: &str, name: &str) -> String {
    let csrf = csrf_from(http, "/settings/tokens", session).await;
    let res = http
        .post_form(
            "/settings/tokens",
            &[("csrf", &csrf), ("name", name), ("expires", "30")],
            Some(session),
        )
        .await;
    assert_eq!(res.status, 200, "create token: {}", res.body);
    let token = first_text(&res.body, "pre.code code");
    assert!(token.starts_with("forge_"), "unexpected token shape");
    token
}

/// Creates a public repo `name` owned by alice through `/new`.
async fn create_repo(http: &Http, session: &str, name: &str) {
    let page = http.get("/new", Some(session)).await;
    assert_eq!(page.status, 200, "GET /new");
    let csrf = input_value(&page.body, "csrf");
    let res = http
        .post_form(
            "/new",
            &[
                ("csrf", &csrf),
                ("name", name),
                ("description", ""),
                ("visibility", "public"),
                ("default_branch", "main"),
            ],
            Some(session),
        )
        .await;
    assert_eq!(res.status, 303, "create repo {name}: {}", res.body);
    assert_eq!(
        res.location.as_deref(),
        Some(format!("/alice/{name}").as_str())
    );
}

/// Adds or replaces a repo secret through the settings form.
async fn set_secret(http: &Http, session: &str, name: &str, value: &str) {
    let csrf = csrf_from(http, &format!("{REPO_PATH}/settings/secrets"), session).await;
    let res = http
        .post_form(
            &format!("{REPO_PATH}/settings/secrets"),
            &[("csrf", &csrf), ("name", name), ("value", value)],
            Some(session),
        )
        .await;
    assert_eq!(res.status, 303, "set secret {name}: {}", res.body);
}

/// A running server with its own temp data dir. Field order matters: the server is dropped (and
/// killed) before the temp dir is removed.
struct Booted {
    _server: Server,
    _root: tempfile::TempDir,
    dir: PathBuf,
    http: Http,
    base_url: String,
    port: u16,
}

/// The server's log, for assertion messages.
fn server_log(b: &Booted) -> String {
    fs::read_to_string(b.dir.join("serve.log")).unwrap_or_default()
}

/// Creates alice (admin) and bob, then boots the server on fresh ports.
async fn boot() -> Booted {
    // Keep the path short: the internal unix socket path is limited to 108 bytes.
    let root = tempfile::Builder::new()
        .prefix("ci2e")
        .tempdir_in("/tmp")
        .expect("temp dir");
    let dir = root.path().to_path_buf();
    let ports = pick_ports();
    let config = write_config(&dir, ports);

    run_admin(&config, "alice", &["--admin"], ALICE_PW);
    run_admin(&config, "bob", &[], BOB_PW);
    let (server, ports) = launch(&dir, ports).await;
    Booted {
        _server: server,
        _root: root,
        dir,
        http: Http::new(ports.http),
        base_url: format!("http://127.0.0.1:{}", ports.http),
        port: ports.http,
    }
}

/// A registered runner and the secret it claims with.
struct Worker {
    runner: Runner,
    secret: String,
    labels: Vec<String>,
}

impl Worker {
    /// One claim. `None` means the forge had nothing for this runner (after its long poll).
    async fn claim(&self) -> Option<JobAssignment> {
        self.runner
            .client()
            .claim(&self.secret, &self.labels)
            .await
            .expect("claim")
    }

    /// Claims and runs jobs until the forge has nothing left. Returns how many ran.
    async fn drain(&self) -> usize {
        let mut ran = 0;
        while let Some(job) = self.claim().await {
            self.runner.process(job, CancellationToken::new()).await;
            ran += 1;
        }
        ran
    }
}

/// Registers a runner through the admin page and the runner API. The admin is `admin_session`.
async fn register_runner(b: &Booted, admin_session: &str, name: &str) -> Worker {
    let page = b.http.get("/admin/runners", Some(admin_session)).await;
    assert_eq!(page.status, 200, "GET /admin/runners");
    let csrf = input_value(&page.body, "csrf");
    let res = b
        .http
        .post_form(
            "/admin/runners/tokens",
            &[("csrf", &csrf)],
            Some(admin_session),
        )
        .await;
    assert_eq!(res.status, 200, "create registration token: {}", res.body);
    let token = first_text(&res.body, "pre code");

    let client = ForgeClient::new(&b.base_url).expect("runner client");
    let registered = client
        .register(&RegisterRequest {
            token,
            name: name.to_owned(),
            labels: LABELS.iter().map(|l| (*l).to_owned()).collect(),
        })
        .await
        .expect("register runner");
    let config = RunnerConfig {
        url: b.base_url.clone(),
        runner_id: registered.runner_id,
        runner_secret: registered.runner_secret.clone(),
        name: name.to_owned(),
        labels: LABELS.iter().map(|l| (*l).to_owned()).collect(),
        work_dir: b.dir.join("runner-work"),
        concurrency: 1,
        limits: Limits::default(),
        executor: forge_runner::config::ExecutorKind::default(),
        images: forge_runner::config::ImagesConfig::default(),
        sandbox: forge_runner::config::SandboxConfig::default(),
        registries: Vec::new(),
    };
    let runner =
        Runner::new(config, Arc::new(HostExecutor), RunOptions::default()).expect("runner");
    Worker {
        runner,
        secret: registered.runner_secret,
        labels: LABELS.iter().map(|l| (*l).to_owned()).collect(),
    }
}

/// One commit status as the status API reports it.
#[derive(Debug, Clone)]
struct Status {
    state: String,
    description: String,
    target_url: String,
}

/// The statuses on `sha` of alice/r, keyed by context.
async fn statuses(http: &Http, token: &str, sha: &str) -> BTreeMap<String, Status> {
    let res = http
        .get_bearer(
            &format!("/api/v1/repos/alice/r/commits/{sha}/statuses"),
            token,
        )
        .await;
    assert_eq!(res.status, 200, "statuses for {sha}: {}", res.body);
    let json: serde_json::Value = serde_json::from_str(&res.body).expect("statuses json");
    json["statuses"]
        .as_array()
        .expect("statuses array")
        .iter()
        .map(|s| {
            let field = |k: &str| s[k].as_str().unwrap_or_default().to_owned();
            (
                field("context"),
                Status {
                    state: field("state"),
                    description: field("description"),
                    target_url: field("target_url"),
                },
            )
        })
        .collect()
}

/// Polls the statuses until every context in `contexts` is present, then returns them.
async fn wait_statuses(
    http: &Http,
    token: &str,
    sha: &str,
    contexts: &[&str],
) -> BTreeMap<String, Status> {
    let deadline = Instant::now() + Duration::from_secs(20);
    loop {
        let found = statuses(http, token, sha).await;
        if contexts.iter().all(|c| found.contains_key(*c)) {
            return found;
        }
        assert!(
            Instant::now() < deadline,
            "statuses never showed {contexts:?}; got {found:?}"
        );
        tokio::time::sleep(Duration::from_millis(100)).await;
    }
}

/// The job id at the end of a status `target_url` (`.../ci/jobs/<id>`).
fn job_id_of(status: &Status) -> String {
    status
        .target_url
        .rsplit('/')
        .next()
        .filter(|id| !id.is_empty())
        .unwrap_or_else(|| panic!("no job id in {:?}", status.target_url))
        .to_owned()
}

/// Pushes `.forge/ci.toml` (with `contents`) to `main` as alice. Returns the pushed commit.
fn push_ci_config(work: &Path, contents: &str) -> String {
    let sha = commit(
        work,
        ".forge/ci.toml",
        contents,
        "add ci",
        "alice",
        "alice@example.test",
    );
    git(work, &["push", "-q", "origin", "HEAD:refs/heads/main"]);
    sha
}

#[tokio::test]
async fn push_run_masks_secrets_keeps_artifacts_and_skips_dependants() {
    let b = boot().await;
    let alice = login(&b.http, "alice", ALICE_PW).await;
    let token = create_token(&b.http, &alice, "ci-token").await;
    create_repo(&b.http, &alice, "r").await;
    set_secret(&b.http, &alice, "MY_SECRET", SECRET_VALUE).await;
    let worker = register_runner(&b, &alice, "runner-a").await;

    let work = clone(
        &remote(b.port, "alice", &token, "alice", "r"),
        &b.dir.join("alice-work"),
    );
    let sha = push_ci_config(&work, CI_TOML);

    let found = wait_statuses(
        &b.http,
        &token,
        &sha,
        &["ci/build", "ci/after", "ci/broken", "ci/skipme"],
    )
    .await;
    assert_eq!(found["ci/build"].state, "pending", "build is queued");

    let ran = worker.drain().await;
    assert_eq!(ran, 3, "build, after and broken run; skipme never does");

    let found = statuses(&b.http, &token, &sha).await;
    assert_eq!(found["ci/build"].state, "success");
    assert_eq!(found["ci/after"].state, "success");
    assert_eq!(found["ci/broken"].state, "failure");
    assert_eq!(found["ci/skipme"].state, "error");
    assert_eq!(found["ci/skipme"].description, "skipped");

    // The build log: the secret is masked, never printed.
    let build_id = job_id_of(&found["ci/build"]);
    let log = b
        .http
        .get(&format!("{REPO_PATH}/ci/jobs/{build_id}/log.txt"), None)
        .await;
    assert_eq!(log.status, 200, "GET log.txt");
    assert!(log.body.contains("hello"), "log has the echo: {}", log.body);
    assert!(log.body.contains("***"), "secret is masked: {}", log.body);
    assert!(
        !log.body.contains(SECRET_VALUE),
        "secret leaked into the log: {}",
        log.body
    );

    // The artifact comes back with the right bytes. `out/a.txt` is stored as `out_a.txt`.
    let artifact = b
        .http
        .get(
            &format!("{REPO_PATH}/ci/jobs/{build_id}/artifacts/out_a.txt"),
            Some(&alice),
        )
        .await;
    assert_eq!(artifact.status, 200, "GET artifact: {}", server_log(&b));
    assert_eq!(artifact.body, "data\n", "artifact bytes");

    // The run page lists every job.
    let run = b
        .http
        .get(&format!("{REPO_PATH}/ci/runs/1"), Some(&alice))
        .await;
    assert_eq!(run.status, 200, "GET run page");
    for job in ["build", "after", "broken", "skipme"] {
        assert!(run.body.contains(job), "run page lists {job}");
    }
}

#[tokio::test]
async fn fork_pull_waits_for_approval_and_runs_without_secrets() {
    let b = boot().await;
    let alice = login(&b.http, "alice", ALICE_PW).await;
    let token = create_token(&b.http, &alice, "ci-token").await;
    create_repo(&b.http, &alice, "r").await;
    set_secret(&b.http, &alice, "MY_SECRET", SECRET_VALUE).await;
    let worker = register_runner(&b, &alice, "runner-a").await;

    // Base branch with the workflow, run to completion (run 1).
    let work = clone(
        &remote(b.port, "alice", &token, "alice", "r"),
        &b.dir.join("alice-work"),
    );
    let base_sha = push_ci_config(&work, CI_TOML);
    wait_statuses(&b.http, &token, &base_sha, &["ci/build"]).await;
    worker.drain().await;

    // Bob forks and changes the workflow on a branch.
    let bob = login(&b.http, "bob", BOB_PW).await;
    let bob_token = create_token(&b.http, &bob, "bob-token").await;
    let csrf = csrf_from(&b.http, "/settings/tokens", &bob).await;
    let fork = b
        .http
        .post_form(&format!("{REPO_PATH}/fork"), &[("csrf", &csrf)], Some(&bob))
        .await;
    assert_eq!(fork.status, 303, "fork: {}\n{}", fork.body, server_log(&b));

    let bob_work = clone(
        &remote(b.port, "bob", &bob_token, "bob", "r"),
        &b.dir.join("bob-work"),
    );
    git(&bob_work, &["checkout", "-q", "-b", "bob-feature"]);
    let head = commit(
        &bob_work,
        ".forge/ci.toml",
        CI_TOML_FORK,
        "print the secret",
        "bob",
        BOB_EMAIL,
    );
    git(
        &bob_work,
        &["push", "-q", "origin", "HEAD:refs/heads/bob-feature"],
    );

    // Bob opens the pull request. The run waits for the owner.
    let compare = b
        .http
        .get(
            &format!("{REPO_PATH}/compare/main...bob:bob-feature"),
            Some(&bob),
        )
        .await;
    assert_eq!(compare.status, 200, "GET compare");
    let csrf = csrf_from(&b.http, "/settings/tokens", &bob).await;
    let open = b
        .http
        .post_form(
            &format!("{REPO_PATH}/pulls"),
            &[
                ("csrf", &csrf),
                ("title", "Bob change"),
                ("body", ""),
                ("base", "main"),
                ("head", "bob:bob-feature"),
            ],
            Some(&bob),
        )
        .await;
    assert_eq!(open.status, 303, "open pull: {}", open.body);

    let found = wait_statuses(&b.http, &token, &head, &["ci/build"]).await;
    assert_eq!(found["ci/build"].state, "pending");
    assert_eq!(found["ci/build"].description, "Waiting for approval");

    let run_page = b
        .http
        .get(&format!("{REPO_PATH}/ci/runs/2"), Some(&alice))
        .await;
    assert_eq!(run_page.status, 200, "GET run 2");
    assert!(
        run_page.body.contains("Approve and run"),
        "run 2 waits for approval: {}",
        run_page.body
    );

    // The owner approves, and the job runs with no secrets.
    let approve = b
        .http
        .post_form(
            &format!("{REPO_PATH}/ci/runs/2/approve"),
            &[("csrf", &input_value(&run_page.body, "csrf"))],
            Some(&alice),
        )
        .await;
    assert_eq!(approve.status, 303, "approve run 2: {}", approve.body);
    worker.drain().await;

    let found = statuses(&b.http, &token, &head).await;
    assert_eq!(found["ci/build"].state, "success", "fork build runs");
    assert_eq!(found["ci/skipme"].state, "error");

    let build_id = job_id_of(&found["ci/build"]);
    let log = b
        .http
        .get(&format!("{REPO_PATH}/ci/jobs/{build_id}/log.txt"), None)
        .await;
    assert_eq!(log.status, 200, "GET log.txt");
    assert!(
        log.body.contains("value=[]"),
        "the secret is empty for a fork run: {}",
        log.body
    );
    assert!(!log.body.contains("***"), "no value to mask: {}", log.body);
    assert!(!log.body.contains(SECRET_VALUE), "secret leaked");
}

#[tokio::test]
async fn live_log_streams_until_the_job_finishes() {
    let b = boot().await;
    let alice = login(&b.http, "alice", ALICE_PW).await;
    let token = create_token(&b.http, &alice, "ci-token").await;
    create_repo(&b.http, &alice, "r").await;
    let worker = register_runner(&b, &alice, "runner-a").await;

    let slow = r#"[on]
push = ["main"]

[jobs.slow]
image = "docker.io/library/alpine:3"
run = "echo started; sleep 5; echo done"
"#;
    let work = clone(
        &remote(b.port, "alice", &token, "alice", "r"),
        &b.dir.join("alice-work"),
    );
    let sha = push_ci_config(&work, slow);
    let found = wait_statuses(&b.http, &token, &sha, &["ci/slow"]).await;
    let job_id = job_id_of(&found["ci/slow"]);

    let job = worker.claim().await.expect("the slow job is claimed");
    assert_eq!(job.job_id, job_id, "claimed the job the status points at");
    let runner = worker.runner.clone();
    let handle = tokio::spawn(async move { runner.process(job, CancellationToken::new()).await });

    // Read the stream chunk by chunk. The first line appears while the job is still sleeping.
    let url = format!("{}{REPO_PATH}/ci/jobs/{job_id}/log.txt", b.http.base);
    let mut res = b.http.client.get(url).send().await.expect("GET log.txt");
    assert_eq!(res.status().as_u16(), 200, "GET log.txt");
    let mut body = String::new();
    let mut seen_started = false;
    let read = async {
        while let Some(chunk) = res.chunk().await.expect("log chunk") {
            body.push_str(&String::from_utf8_lossy(&chunk));
            if !seen_started && body.contains("started") {
                seen_started = true;
                assert!(
                    !handle.is_finished(),
                    "the first line arrived only after the job had finished"
                );
            }
        }
    };
    tokio::time::timeout(Duration::from_secs(90), read)
        .await
        .expect("the log stream ends after the job finishes");

    assert!(seen_started, "the live log showed the first line: {body}");
    assert!(
        body.contains("done"),
        "the stream ends with the last line: {body}"
    );
    let outcome = handle.await.expect("job task");
    assert_eq!(outcome, Outcome::Success);
    let found = statuses(&b.http, &token, &sha).await;
    assert_eq!(found["ci/slow"].state, "success");
}

#[tokio::test]
async fn ci_job_token_clones_the_repos_wiki_read_only() {
    let b = boot().await;
    let alice = login(&b.http, "alice", ALICE_PW).await;
    let token = create_token(&b.http, &alice, "ci-token").await;
    create_repo(&b.http, &alice, "r").await;
    let worker = register_runner(&b, &alice, "runner-a").await;

    // The wiki gets its first page through the web form.
    let csrf = input_value(
        &b.http
            .get(&format!("{REPO_PATH}/wiki/-/new"), Some(&alice))
            .await
            .body,
        "csrf",
    );
    let saved = b
        .http
        .post_form(
            &format!("{REPO_PATH}/wiki/-/new"),
            &[
                ("csrf", &csrf),
                ("name", "Home"),
                ("base", ""),
                ("content", "# Welcome from the wiki\n"),
                ("message", "first page"),
            ],
            Some(&alice),
        )
        .await;
    assert_eq!(saved.status, 303, "create Home: {}", saved.body);

    // A slow job keeps running, so its token is valid while the test clones.
    let slow = r#"[on]
push = ["main"]

[jobs.slow]
image = "docker.io/library/alpine:3"
run = "echo started; sleep 5; echo done"
"#;
    let work = clone(
        &remote(b.port, "alice", &token, "alice", "r"),
        &b.dir.join("alice-work"),
    );
    let sha = push_ci_config(&work, slow);
    let found = wait_statuses(&b.http, &token, &sha, &["ci/slow"]).await;
    let job = worker.claim().await.expect("the slow job is claimed");
    assert_eq!(job.job_id, job_id_of(&found["ci/slow"]));

    // The job reads the wiki with its own credential: `ci-job` and `<job id>:<job token>`.
    let wiki_url = format!(
        "http://ci-job:{}:{}@127.0.0.1:{}/alice/r.wiki.git",
        job.job_id, job.job_token, b.port
    );
    let job_wiki = clone(&wiki_url, &b.dir.join("job-wiki"));
    let home = fs::read_to_string(job_wiki.join("Home.md")).expect("Home.md in the job's clone");
    assert!(home.contains("Welcome from the wiki"), "{home}");

    // Pushing to the wiki is refused for a job token.
    fs::write(job_wiki.join("Home.md"), "# changed by a job\n").expect("write");
    git(&job_wiki, &["add", "Home.md"]);
    git_env(
        &job_wiki,
        &[
            "-c",
            "commit.gpgsign=false",
            "commit",
            "-q",
            "-m",
            "job edit",
        ],
        &[
            ("GIT_AUTHOR_NAME", "job"),
            ("GIT_AUTHOR_EMAIL", "job@example.invalid"),
            ("GIT_COMMITTER_NAME", "job"),
            ("GIT_COMMITTER_EMAIL", "job@example.invalid"),
        ],
    );
    let push = Command::new("git")
        .current_dir(&job_wiki)
        .args([
            "-c",
            "credential.helper=",
            "push",
            "-q",
            "origin",
            "HEAD:main",
        ])
        .env("GIT_TERMINAL_PROMPT", "0")
        .output()
        .expect("run git");
    assert!(
        !push.status.success(),
        "a CI job token must not push to the wiki"
    );
}