//! 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, /// The reviews to ask the forge for. Not limited to agents: a reviewer is /// a forge account. pub requests: Vec, /// Recipients that are not agents of this swarm, so got nothing. pub skipped: Vec, 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 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(); 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()); } Op::Clear => { cleared.insert(action.agent.clone()); } } } 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(), is_pull: false, assignees: assignees.iter().map(|a| (*a).to_owned()).collect(), } } fn agents(a: &Routed) -> Vec<(&str, Op)> { a.actions .iter() .map(|x| (x.agent.as_str(), x.op.op)) .collect() } /// 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()]), }; let r = route("pull_request_sync", &pull, &before, &flows, |_| false); assert_eq!( r.requests, [ReviewRequest { flow: "head-classifier".to_owned(), node: "request".to_owned(), repo: "o/r".to_owned(), number: 9, reviewer: "review-classifier".to_owned(), }] ); assert!(r.actions.is_empty() && r.skipped.is_empty()); assert_eq!(r.next, before); // The source gates: another PR event requests nothing. for other in ["pull_request", "pull_request_assign", "issues"] { let r = route(other, &pull, &before, &flows, |_| true); assert!(r.requests.is_empty(), "{other}"); } // 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()); } #[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()]), }; 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 { 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()); } }