johnandersen777.bsky.social/knotless

Switch repository

Loading…

//! An agent keeping pulls the merge queue turned away moving, on the
//! reference PDS: one that conflicts with its base only where both sides
//! added lines it brings up to date and queues again, one where they
//! changed the same lines it tells its author of, once, and any other it
//! leaves alone.

#[path = "../../knotless/tests/common/mod.rs"]
mod common;

use common::{Sandbox, net, wait_until};
use knotless::forge::Target;
use knotless::queue::Ejected;
use knotless::url::RepoUrl;
use knotless_agent::{Job, Options};
use serde_json::Value;
use std::time::Duration;

/// An appview over the reference PDS, following its firehose.
fn appview(rt: &tokio::runtime::Runtime, s: &Sandbox) -> String {
    let listener = rt
        .block_on(tokio::net::TcpListener::bind("127.0.0.1:0"))
        .unwrap();
    let base = format!("http://localhost:{}", listener.local_addr().unwrap().port());
    let index = knotless_serve::index::Index::open(&s.path("index.db")).unwrap();
    let app = knotless_serve::App::build(common::ctx(s), Duration::ZERO, None, Some(index));
    let router = knotless_serve::router(app.clone());
    rt.spawn(async move { axum::serve(listener, router).await.unwrap() });
    rt.spawn(knotless_serve::firehose::watch(
        app,
        net().pds.replace("http://", "ws://"),
    ));
    base
}

#[track_caller]
fn ok(s: &Sandbox, args: &[&str]) -> String {
    let (code, out) = s.cli(args, b"");
    assert_eq!(code, 0, "{args:?}: {out}");
    out
}

/// The comments `who` wrote on `pull`, oldest first.
fn said(who: &Sandbox, pull: &str) -> Vec<String> {
    let mut said: Vec<(String, Value)> = who.records("dev.knotless.comment");
    said.retain(|(_, c)| c["subject"] == pull);
    said.sort_by(|a, b| a.0.cmp(&b.0));
    let bodies = said
        .into_iter()
        .map(|(_, c)| c["body"].as_str().unwrap().to_string());
    bodies.collect()
}

#[test]
fn an_agent_keeps_turned_away_pulls_moving() {
    let rt = tokio::runtime::Runtime::new().unwrap();
    let base = appview(&rt, &Sandbox::new());
    let mut owner = Sandbox::new();
    let handle = owner.account.handle.clone();
    owner.set("KNOTLESS_APPVIEW", &base);
    owner.set("KNOTLESS_IDENTIFIER", &handle);
    let url = owner.url("kept");
    let work = owner.work("w", 1);
    owner.commit_file(&work, "notes", "one\n");
    owner.git(&work, &["push", "-q", &url, "main"]);
    // A runner named, and the queue on for main, so its pulls may be queued.
    ok(&owner, &["runner", "add", &url, &handle]);
    ok(
        &owner,
        &["branch", "protect", &url, "main", "--merges", "--queue"],
    );
    // `branch` as main is now, with `file` holding `text`, pushed.
    let write = |branch: &str, from: &str, file: &str, text: &str| {
        owner.git(&work, &["fetch", "-q", &url, from]);
        owner.git(&work, &["checkout", "-q", "-B", branch, "FETCH_HEAD"]);
        let sha = owner.commit_file(&work, file, text);
        owner.git(&work, &["push", "-q", "-f", &url, branch]);
        sha
    };
    let ctx = common::ctx(&owner);
    let repo = RepoUrl::parse(&url).unwrap();
    // Turns `pull` away from the queue at `head`, as a runner does.
    let eject = |pull: &str, head: &str| {
        let t = rt.block_on(Target::open(&ctx, &repo)).unwrap();
        let ejected = Ejected {
            reason: Some("it conflicts in notes"),
            head: Some(head),
            checks: &[],
        };
        rt.block_on(knotless::queue::eject(&ctx, &t, pull, ejected))
            .unwrap();
    };
    let pull = |branch: &str, title: &str, body: &str| {
        let pull = ok(
            &owner,
            &[
                "pull", "create", &url, "main", &url, branch, title, "-m", body,
            ],
        );
        wait_until(|| ok(&owner, &["pull", "list", &url]).contains(&pull));
        pull
    };
    let shown = |pull: &str| ok(&owner, &["pull", "show", &url, pull]);

    // Pulls cut from main, then main moved on.
    let added = write("added", "main", "notes", "one\nadded\n");
    let changed = write("changed", "main", "notes", "uno\n");
    let apart = write("apart", "main", "apart", "apart\n");
    let off = write("off", "main", "notes", "one\noff\n");
    let moved = write("moved", "main", "notes", "one\nmoved\n");
    let gone = write("gone", "main", "notes", "one\ngone\n");
    // A branch of another of the owner's repositories: not the keeper's
    // to push to.
    let other = owner.url("elsewhere");
    let away = write("away", "main", "notes", "one\naway\n");
    owner.git(&work, &["push", "-q", &other, "away:main", "away"]);
    let from_elsewhere = || {
        let args = [
            "pull",
            "create",
            &url,
            "main",
            &other,
            "away",
            "Away",
            "-m",
            "From elsewhere",
        ];
        let pull = ok(&owner, &args);
        wait_until(|| ok(&owner, &["pull", "list", &url]).contains(&pull));
        pull
    };
    let pulls = [
        (pull("added", "Added", "More notes"), added),
        (pull("changed", "Changed", "Other notes"), changed),
        (pull("apart", "Apart", "Elsewhere"), apart),
        (pull("off", "Off", "Mine to fix\n\nKeeper: off"), off),
        (pull("moved", "Moved", "Pushed to since"), moved),
        (pull("gone", "Gone", "Its branch deleted"), gone),
        (from_elsewhere(), away),
    ];
    write("main", "main", "notes", "one\nmain\n");
    for (pull, head) in &pulls {
        eject(pull, head);
    }
    // Pushed to since it was turned away, it is its author's again.
    write("moved", "moved", "notes", "one\nmoved, again\n");
    ok(&owner, &["branch", "delete", &url, "gone"]);
    for (pull, _) in &pulls {
        wait_until(|| shown(pull).contains("ejected"));
    }
    let added_at = pulls[0].1.clone();
    let [added, changed, apart, off, moved, gone, away] = pulls.map(|(p, _)| p);

    let opts = Options {
        job: Job::Keep,
        command: String::new(),
        all: false,
        rounds: 3,
        timeout: Duration::from_secs(3),
        jetstream: "ws://127.0.0.1:9".into(),
        sweep: Duration::from_secs(600),
        settle: Duration::ZERO,
        once: true,
        repos: vec![format!("{handle}/kept")],
        owners: Vec::new(),
        attempts: 2,
    };
    let once = |opts: &Options| {
        rt.block_on(knotless_agent::run(opts.clone(), common::ctx(&owner)))
            .unwrap()
    };
    once(&opts);

    // Lines both sides added: brought up to date, main's first, and queued.
    owner.git(&work, &["fetch", "-q", &url, "added"]);
    let notes = owner.git(&work, &["show", "FETCH_HEAD:notes"]);
    assert_eq!(notes, "one\nmain\nadded");
    let told = said(&owner, &added);
    assert_eq!(told.len(), 1, "{told:?}");
    assert!(
        told[0].starts_with("Merged main into this pull"),
        "{}",
        told[0]
    );
    assert!(
        told[0].contains("Both sides only added lines in notes"),
        "{}",
        told[0]
    );
    wait_until(|| shown(&added).contains("merge queue: queued"));
    // The same line changed: its author is told where, and it stays.
    let told = said(&owner, &changed);
    assert_eq!(told.len(), 1, "{told:?}");
    assert!(
        told[0].contains("conflicts with main in notes"),
        "{}",
        told[0]
    );
    owner.git(&work, &["fetch", "-q", &url, "changed"]);
    assert_eq!(owner.git(&work, &["show", "FETCH_HEAD:notes"]), "uno");
    // Turned away for another reason, asked to be left, pushed to, its
    // branch gone or elsewhere: nothing.
    for pull in [&apart, &off, &moved, &gone, &away] {
        assert!(said(&owner, pull).is_empty(), "{pull}");
    }

    // Read again, it says nothing twice.
    wait_until(|| shown(&changed).matches("conflicts with main").count() == 1);
    once(&opts);
    assert_eq!(said(&owner, &changed).len(), 1);
    assert_eq!(said(&owner, &added).len(), 1);

    // Left turned away at the head it had before it was brought up to
    // date, as when queueing it failed: queued again, nothing more said.
    // With the queue off for main it is left alone, until it is on again.
    eject(&added, &added_at);
    wait_until(|| shown(&added).contains("merge queue: ejected"));
    ok(&owner, &["branch", "protect", &url, "main", "--undo"]);
    once(&opts);
    assert!(shown(&added).contains("merge queue: ejected"));
    assert_eq!(said(&owner, &added).len(), 1);
    ok(
        &owner,
        &["branch", "protect", &url, "main", "--merges", "--queue"],
    );
    once(&opts);
    wait_until(|| shown(&added).contains("merge queue: queued"));
    assert_eq!(said(&owner, &added).len(), 1);

    // Turned away again once main moved on, it is brought up to date
    // again, but no more often than it may be.
    owner.git(&work, &["fetch", "-q", &url, "added"]);
    let head = owner.git(&work, &["rev-parse", "FETCH_HEAD"]);
    write("main", "main", "notes", "one\nmain\nmain again\n");
    eject(&added, &head);
    wait_until(|| shown(&added).contains("merge queue: ejected"));
    let once_more = Options {
        attempts: 1,
        ..opts.clone()
    };
    once(&once_more);
    assert_eq!(said(&owner, &added).len(), 1);
    once(&opts);
    let told = said(&owner, &added);
    assert_eq!(told.len(), 2, "{told:?}");
    owner.git(&work, &["fetch", "-q", &url, "added"]);
    let notes = owner.git(&work, &["show", "FETCH_HEAD:notes"]);
    assert_eq!(notes, "one\nmain\nmain again\nadded");
    // Its author was told of the head that conflicts once, though main
    // moved on.
    assert_eq!(said(&owner, &changed).len(), 1);
}