529 lines
18 KiB
Rust
529 lines
18 KiB
Rust
//! One event through every flow: the ops to queue and the issue's next
|
|
//! snapshot. No I/O, so the whole decision is testable.
|
|
|
|
use std::collections::BTreeSet;
|
|
|
|
use swarm_wire::router_todo::Op;
|
|
|
|
use crate::event::{Delivery, Snapshot, fields, is_assign};
|
|
use swarm_wire::notify_flows::Flow;
|
|
|
|
use crate::flow::{Action, ReviewRequest, evaluate, requests};
|
|
|
|
/// What one event comes to.
|
|
#[derive(Debug, PartialEq, Eq)]
|
|
pub struct Routed {
|
|
/// The ops to queue, one per agent that exists.
|
|
pub actions: Vec<Action>,
|
|
/// The reviews to ask the forge for. Not limited to agents: a reviewer is
|
|
/// a forge account.
|
|
pub requests: Vec<ReviewRequest>,
|
|
/// Recipients that are not agents of this swarm, so got nothing.
|
|
pub skipped: Vec<String>,
|
|
pub next: Snapshot,
|
|
}
|
|
|
|
/// Run `flows` over one delivery of `event_type` on an issue whose snapshot is
|
|
/// `snapshot`. `is_agent` says whether a recipient is an agent of this swarm.
|
|
pub fn route(
|
|
event_type: &str,
|
|
delivery: &Delivery,
|
|
snapshot: &Snapshot,
|
|
flows: &[Flow],
|
|
is_agent: impl Fn(&str) -> bool,
|
|
) -> Routed {
|
|
let is_run = event_type.starts_with("workflow_run_");
|
|
// Forgejo cancels the runs of a PR when it is pushed to, and an old run can
|
|
// finish after the push cleared the PR's CI todo. Such a run began at a
|
|
// commit the PR has since moved past, so it tells nobody anything.
|
|
if is_run
|
|
&& !delivery.head.is_empty()
|
|
&& snapshot.head.as_deref().is_some_and(|h| h != delivery.head)
|
|
{
|
|
return Routed {
|
|
actions: Vec::new(),
|
|
requests: Vec::new(),
|
|
skipped: Vec::new(),
|
|
next: snapshot.clone(),
|
|
};
|
|
}
|
|
let fields = fields(event_type, delivery, snapshot);
|
|
let mut actions = Vec::new();
|
|
let mut skipped = Vec::new();
|
|
let mut review_requests = Vec::new();
|
|
for flow in flows {
|
|
review_requests.extend(requests(flow, event_type, &fields));
|
|
}
|
|
for action in flows.iter().flat_map(|f| evaluate(f, event_type, &fields)) {
|
|
if is_agent(&action.agent) {
|
|
actions.push(action);
|
|
} else if !skipped.contains(&action.agent) {
|
|
skipped.push(action.agent);
|
|
}
|
|
}
|
|
let mut next = snapshot.clone();
|
|
// A run's delivery carries the PR as it was when the run began, so only
|
|
// the other deliveries say where the PR's head is now.
|
|
if !is_run && delivery.is_pull && !delivery.head.is_empty() {
|
|
next.head = Some(delivery.head.clone());
|
|
}
|
|
if is_assign(event_type) {
|
|
// A skipped assignee is left out of the snapshot, so the next assign
|
|
// event counts it as added again: an agent assigned before it is in
|
|
// the roster gets its todo once it is.
|
|
let mut assignees = delivery.assignees.clone();
|
|
assignees.retain(|a| !skipped.contains(a));
|
|
next.assignees = Some(assignees);
|
|
}
|
|
let mut cleared = BTreeSet::new();
|
|
for action in &actions {
|
|
match action.op.op {
|
|
Op::Upsert => {
|
|
next.holders.insert(action.agent.clone());
|
|
}
|
|
// Only a prefix clear can take every todo the agent held for the
|
|
// issue. A clear of one key leaves it a holder: it may still hold
|
|
// the assignment, which the `closed` flow must find.
|
|
Op::Clear if action.op.prefix => {
|
|
cleared.insert(action.agent.clone());
|
|
}
|
|
Op::Clear => {}
|
|
}
|
|
}
|
|
next.holders.retain(|h| !cleared.contains(h));
|
|
Routed {
|
|
actions,
|
|
requests: review_requests,
|
|
skipped,
|
|
next,
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
use swarm_wire::notify_flows::{Condition, Edge, Node, NodeKind, builtins};
|
|
|
|
fn delivery(action: &str, assignees: &[&str]) -> Delivery {
|
|
Delivery {
|
|
action: action.to_owned(),
|
|
repo: "o/r".to_owned(),
|
|
repo_id: 1,
|
|
number: 9,
|
|
title: "t".to_owned(),
|
|
url: "u".to_owned(),
|
|
sender: "mara".to_owned(),
|
|
author: "mara".to_owned(),
|
|
head: String::new(),
|
|
is_pull: false,
|
|
assignees: assignees.iter().map(|a| (*a).to_owned()).collect(),
|
|
draft: false,
|
|
undrafted: false,
|
|
}
|
|
}
|
|
|
|
fn agents(a: &Routed) -> Vec<(&str, Op)> {
|
|
a.actions
|
|
.iter()
|
|
.map(|x| (x.agent.as_str(), x.op.op))
|
|
.collect()
|
|
}
|
|
|
|
fn pull_by(author: &str, action: &str) -> Delivery {
|
|
Delivery {
|
|
is_pull: true,
|
|
author: author.to_owned(),
|
|
..delivery(action, &[])
|
|
}
|
|
}
|
|
|
|
/// A failed run tells the PR's author, a cancelled one rewrites that todo,
|
|
/// and a push clears it. The author keeps holding the PR, so a later close
|
|
/// still reaches them.
|
|
#[test]
|
|
fn a_failed_run_tells_the_author_and_a_push_clears_it() {
|
|
let flows = builtins();
|
|
let is_agent = |a: &str| a == "atlas";
|
|
|
|
let failed = route(
|
|
"workflow_run_failure",
|
|
&pull_by("atlas", "failure"),
|
|
&Snapshot::default(),
|
|
&flows,
|
|
is_agent,
|
|
);
|
|
assert_eq!(agents(&failed), [("atlas", Op::Upsert)]);
|
|
assert_eq!(failed.actions[0].op.key, "forge:o/r#9:ci");
|
|
assert!(
|
|
failed.actions[0]
|
|
.op
|
|
.summary
|
|
.starts_with("CI failed on PR o/r#9")
|
|
);
|
|
assert_eq!(failed.next.holders, BTreeSet::from(["atlas".to_owned()]));
|
|
|
|
let cancelled = route(
|
|
"workflow_run_cancelled",
|
|
&pull_by("atlas", "cancelled"),
|
|
&failed.next,
|
|
&flows,
|
|
is_agent,
|
|
);
|
|
assert_eq!(agents(&cancelled), [("atlas", Op::Upsert)]);
|
|
assert_eq!(cancelled.actions[0].op.key, "forge:o/r#9:ci");
|
|
assert!(
|
|
cancelled.actions[0]
|
|
.op
|
|
.summary
|
|
.starts_with("CI was cancelled on PR")
|
|
);
|
|
|
|
let pushed = route(
|
|
"pull_request_sync",
|
|
&pull_by("atlas", "synchronized"),
|
|
&cancelled.next,
|
|
&flows,
|
|
is_agent,
|
|
);
|
|
let ci_clear: Vec<_> = pushed
|
|
.actions
|
|
.iter()
|
|
.filter(|a| a.op.op == Op::Clear)
|
|
.collect();
|
|
assert_eq!(ci_clear.len(), 1);
|
|
assert_eq!(ci_clear[0].op.key, "forge:o/r#9:ci");
|
|
assert!(
|
|
!ci_clear[0].op.prefix,
|
|
"only the CI todo goes, not the assignment"
|
|
);
|
|
assert_eq!(pushed.next.holders, BTreeSet::from(["atlas".to_owned()]));
|
|
}
|
|
|
|
fn at_head(mut d: Delivery, head: &str) -> Delivery {
|
|
head.clone_into(&mut d.head);
|
|
d
|
|
}
|
|
|
|
/// The push that clears the CI todo also cancels the runs before it, and
|
|
/// one of them can report in after the clear. It began at the old head, so
|
|
/// it must not bring the todo back.
|
|
#[test]
|
|
fn a_run_from_before_the_latest_push_tells_nobody() {
|
|
let flows = builtins();
|
|
let is_agent = |a: &str| a == "atlas";
|
|
let run = |event: &str, head: &str, before: &Snapshot| {
|
|
// The router names a run's action after its outcome.
|
|
let outcome = event.trim_start_matches("workflow_run_");
|
|
route(
|
|
event,
|
|
&at_head(pull_by("atlas", outcome), head),
|
|
before,
|
|
&flows,
|
|
is_agent,
|
|
)
|
|
};
|
|
|
|
// Before the router has seen the PR's head it can't tell, so it tells.
|
|
let unknown = run("workflow_run_cancelled", "old", &Snapshot::default());
|
|
assert_eq!(agents(&unknown), [("atlas", Op::Upsert)]);
|
|
assert_eq!(unknown.next.head, None, "a run says nothing about the head");
|
|
|
|
let pushed = route(
|
|
"pull_request_sync",
|
|
&at_head(pull_by("atlas", "synchronized"), "new"),
|
|
&unknown.next,
|
|
&flows,
|
|
is_agent,
|
|
);
|
|
assert_eq!(pushed.next.head.as_deref(), Some("new"));
|
|
|
|
let late = run("workflow_run_cancelled", "old", &pushed.next);
|
|
assert!(late.actions.is_empty() && late.skipped.is_empty());
|
|
assert_eq!(late.next, pushed.next);
|
|
|
|
let current = run("workflow_run_failure", "new", &pushed.next);
|
|
assert_eq!(agents(¤t), [("atlas", Op::Upsert)]);
|
|
}
|
|
|
|
#[test]
|
|
fn a_failed_run_on_a_pull_request_by_a_non_agent_queues_nothing() {
|
|
let got = route(
|
|
"workflow_run_failure",
|
|
&pull_by("mara", "failure"),
|
|
&Snapshot::default(),
|
|
&builtins(),
|
|
|a| a == "atlas",
|
|
);
|
|
assert!(got.actions.is_empty());
|
|
assert_eq!(got.skipped, ["mara"]);
|
|
}
|
|
|
|
/// Assign, reassign, close: the three built-ins in the order an issue
|
|
/// meets them, carrying the snapshot from one event to the next.
|
|
#[test]
|
|
fn an_issue_assigned_reassigned_and_closed_queues_and_clears_its_todos() {
|
|
let flows = builtins();
|
|
let is_agent = |a: &str| a != "mara";
|
|
|
|
let one = route(
|
|
"issue_assign",
|
|
&delivery("assigned", &["atlas", "mara"]),
|
|
&Snapshot::default(),
|
|
&flows,
|
|
is_agent,
|
|
);
|
|
assert_eq!(agents(&one), [("atlas", Op::Upsert)]);
|
|
assert_eq!(one.skipped, ["mara"]);
|
|
assert_eq!(one.next.holders, BTreeSet::from(["atlas".to_owned()]));
|
|
|
|
let two = route(
|
|
"issue_assign",
|
|
&delivery("unassigned", &["mara"]),
|
|
&one.next,
|
|
&flows,
|
|
is_agent,
|
|
);
|
|
assert_eq!(agents(&two), [("atlas", Op::Upsert)]);
|
|
assert!(two.actions[0].op.summary.contains("not assigned"));
|
|
|
|
let three = route(
|
|
"issue_assign",
|
|
&delivery("assigned", &["mara", "iris"]),
|
|
&two.next,
|
|
&flows,
|
|
is_agent,
|
|
);
|
|
assert_eq!(agents(&three), [("iris", Op::Upsert)]);
|
|
|
|
let closed = route(
|
|
"issues",
|
|
&delivery("closed", &["mara", "iris"]),
|
|
&three.next,
|
|
&flows,
|
|
is_agent,
|
|
);
|
|
// Per agent the clear comes before the closed todo that its prefix would take.
|
|
let mut got = agents(&closed);
|
|
got.sort_by_key(|(a, _)| *a);
|
|
assert_eq!(
|
|
got,
|
|
[
|
|
("atlas", Op::Clear),
|
|
("atlas", Op::Upsert),
|
|
("iris", Op::Clear),
|
|
("iris", Op::Upsert)
|
|
]
|
|
);
|
|
assert!(closed.next.holders.is_empty());
|
|
// A close leaves the assignee snapshot as the last assign set it.
|
|
assert_eq!(closed.next.assignees, three.next.assignees);
|
|
}
|
|
|
|
/// An agent assigned before it reports status gets nothing then, and its
|
|
/// todo on the next assign event, once it is in the roster.
|
|
#[test]
|
|
fn an_assignee_not_yet_in_the_roster_gets_its_todo_on_the_next_assign() {
|
|
let flows = builtins();
|
|
let early = route(
|
|
"issue_assign",
|
|
&delivery("assigned", &["nova"]),
|
|
&Snapshot::default(),
|
|
&flows,
|
|
|_| false,
|
|
);
|
|
assert!(early.actions.is_empty());
|
|
assert_eq!(early.skipped, ["nova"]);
|
|
assert_eq!(early.next.assignees, Some(BTreeSet::new()));
|
|
|
|
let later = route(
|
|
"issue_assign",
|
|
&delivery("assigned", &["nova", "atlas"]),
|
|
&early.next,
|
|
&flows,
|
|
|_| true,
|
|
);
|
|
let mut got = agents(&later);
|
|
got.sort_unstable_by_key(|(a, _)| *a);
|
|
assert_eq!(got, [("atlas", Op::Upsert), ("nova", Op::Upsert)]);
|
|
assert_eq!(
|
|
later.next.assignees,
|
|
Some(BTreeSet::from(["atlas".to_owned(), "nova".to_owned()]))
|
|
);
|
|
}
|
|
|
|
/// A push to a PR makes one request for the classifier's review, whether
|
|
/// or not the classifier is an agent, and touches nothing else.
|
|
#[test]
|
|
fn a_push_to_a_pull_request_requests_the_classifiers_review_once() {
|
|
let flows = builtins();
|
|
let pull = Delivery {
|
|
is_pull: true,
|
|
..delivery("synchronized", &[])
|
|
};
|
|
let before = Snapshot {
|
|
assignees: Some(BTreeSet::from(["atlas".to_owned()])),
|
|
holders: BTreeSet::from(["atlas".to_owned()]),
|
|
..Snapshot::default()
|
|
};
|
|
let r = route("pull_request_sync", &pull, &before, &flows, |_| false);
|
|
assert_eq!(
|
|
r.requests,
|
|
[ReviewRequest {
|
|
flow: "head-classifier".to_owned(),
|
|
node: "request-synchronized".to_owned(),
|
|
repo: "o/r".to_owned(),
|
|
number: 9,
|
|
reviewer: "review-classifier".to_owned(),
|
|
}]
|
|
);
|
|
// The push also clears the author's CI todo, but the author is no
|
|
// agent here, so that goes nowhere.
|
|
assert!(r.actions.is_empty());
|
|
assert_eq!(r.skipped, ["mara"]);
|
|
assert_eq!(r.next, before);
|
|
|
|
// The source gates: another PR event requests nothing.
|
|
for other in ["pull_request_assign", "issues"] {
|
|
let r = route(other, &pull, &before, &flows, |_| true);
|
|
assert!(r.requests.is_empty(), "{other}");
|
|
}
|
|
// Opening a PR is one `pull_request` event with action `opened`.
|
|
let opened = Delivery {
|
|
is_pull: true,
|
|
..delivery("opened", &[])
|
|
};
|
|
let r = route("pull_request", &opened, &before, &flows, |_| false);
|
|
assert_eq!(r.requests.len(), 1);
|
|
assert_eq!(r.requests[0].node, "request-opened");
|
|
assert!(r.actions.is_empty() && r.skipped.is_empty());
|
|
// Only a PR can be asked to review.
|
|
let issue = delivery("synchronized", &[]);
|
|
let r = route("pull_request_sync", &issue, &before, &flows, |_| true);
|
|
assert!(r.requests.is_empty());
|
|
}
|
|
|
|
/// A push to a draft PR requests nothing; the title edit that takes it out
|
|
/// of draft requests once.
|
|
#[test]
|
|
fn a_draft_gets_no_request_until_it_leaves_draft() {
|
|
let flows = builtins();
|
|
let before = Snapshot::default();
|
|
let draft = Delivery {
|
|
is_pull: true,
|
|
draft: true,
|
|
..delivery("synchronized", &[])
|
|
};
|
|
let r = route("pull_request_sync", &draft, &before, &flows, |_| false);
|
|
assert!(r.requests.is_empty());
|
|
let ready = Delivery {
|
|
is_pull: true,
|
|
undrafted: true,
|
|
..delivery("edited", &[])
|
|
};
|
|
let r = route("pull_request", &ready, &before, &flows, |_| false);
|
|
assert_eq!(r.requests.len(), 1);
|
|
assert_eq!(r.requests[0].node, "request-undrafted");
|
|
}
|
|
|
|
#[test]
|
|
fn an_event_no_flow_matches_changes_nothing() {
|
|
let before = Snapshot {
|
|
assignees: Some(BTreeSet::from(["atlas".to_owned()])),
|
|
holders: BTreeSet::from(["atlas".to_owned()]),
|
|
..Snapshot::default()
|
|
};
|
|
let r = route(
|
|
"issue_comment",
|
|
&delivery("created", &[]),
|
|
&before,
|
|
&builtins(),
|
|
|_| true,
|
|
);
|
|
assert!(r.actions.is_empty());
|
|
assert_eq!(r.next, before);
|
|
}
|
|
|
|
/// The body Forgejo sends for a submitted review: a pull request payload
|
|
/// whose `sender` is the reviewer.
|
|
fn review_body(reviewer: &str) -> Vec<u8> {
|
|
serde_json::to_vec(&serde_json::json!({
|
|
"action": "reviewed",
|
|
"number": 9,
|
|
"pull_request": {
|
|
"number": 9,
|
|
"title": "t",
|
|
"html_url": "u",
|
|
"assignees": [],
|
|
},
|
|
"repository": { "id": 1, "full_name": "o/r" },
|
|
"sender": { "login": reviewer },
|
|
"review": { "type": "pull_request_review_approved", "content": "" },
|
|
}))
|
|
.expect("encodes")
|
|
}
|
|
|
|
/// "A review by the classifier on a PR → request argus's review", as a
|
|
/// user builds it: valid, routed from the delivery body, and gated on the
|
|
/// reviewer.
|
|
#[test]
|
|
fn a_review_by_the_classifier_requests_argus() {
|
|
let id = |s: &str| s.to_owned();
|
|
let flow = Flow {
|
|
id: id("classifier-then-argus"),
|
|
name: id("classifier-then-argus"),
|
|
enabled: true,
|
|
nodes: vec![
|
|
Node {
|
|
id: id("source"),
|
|
kind: NodeKind::Source {
|
|
events: vec![id("pull_request_review_approved")],
|
|
},
|
|
},
|
|
Node {
|
|
id: id("match"),
|
|
kind: NodeKind::Match {
|
|
when: vec![Condition {
|
|
field: id("sender"),
|
|
equals: id("review-classifier"),
|
|
}],
|
|
},
|
|
},
|
|
Node {
|
|
id: id("request"),
|
|
kind: NodeKind::RequestReviewer {
|
|
reviewer: id("argus"),
|
|
},
|
|
},
|
|
],
|
|
edges: vec![
|
|
Edge {
|
|
from: id("source"),
|
|
to: id("match"),
|
|
},
|
|
Edge {
|
|
from: id("match"),
|
|
to: id("request"),
|
|
},
|
|
],
|
|
};
|
|
assert_eq!(flow.validate(), Ok(()));
|
|
let flows = [flow];
|
|
let routed = |event: &str, reviewer: &str| {
|
|
let d = Delivery::parse(&review_body(reviewer)).expect("parses");
|
|
route(event, &d, &Snapshot::default(), &flows, |_| true).requests
|
|
};
|
|
assert_eq!(
|
|
routed("pull_request_review_approved", "review-classifier"),
|
|
[ReviewRequest {
|
|
flow: "classifier-then-argus".to_owned(),
|
|
node: "request".to_owned(),
|
|
repo: "o/r".to_owned(),
|
|
number: 9,
|
|
reviewer: "argus".to_owned(),
|
|
}]
|
|
);
|
|
// Another reviewer, and another review event, request nothing.
|
|
assert!(routed("pull_request_review_approved", "mara").is_empty());
|
|
assert!(routed("pull_request_review_rejected", "review-classifier").is_empty());
|
|
}
|
|
}
|