470 lines
18 KiB
Rust
470 lines
18 KiB
Rust
//! A forge delivery as the fields a flow reads, and the per-issue snapshot
|
|
//! that fills in what the delivery leaves out.
|
|
//!
|
|
//! Forgejo's `issue_assign` and `pull_request_assign` deliveries carry the
|
|
//! assignees *after* the change and not which one changed. [`Snapshot`] holds
|
|
//! the assignees the router last saw on an assign delivery, so the added and
|
|
//! removed ones are the difference. With no snapshot yet, every current
|
|
//! assignee counts as added and none as removed: an unassign is noticed only
|
|
//! on an issue the router has seen an assign delivery for.
|
|
|
|
use std::collections::BTreeSet;
|
|
|
|
use serde::{Deserialize, Serialize};
|
|
use serde_json::Value as Json;
|
|
|
|
use crate::flow::{Fields, Value};
|
|
|
|
/// The parts of an `issues` or `pull_request` family delivery the router
|
|
/// uses.
|
|
#[derive(Clone, Debug, PartialEq, Eq)]
|
|
pub struct Delivery {
|
|
pub action: String,
|
|
pub repo: String,
|
|
pub repo_id: i64,
|
|
pub number: i64,
|
|
pub title: String,
|
|
pub url: String,
|
|
pub sender: String,
|
|
/// The account that opened the issue or PR.
|
|
pub author: String,
|
|
/// The commit at the PR's head when the delivery was made; empty for an
|
|
/// issue. A CI run's delivery carries the PR as it was when the run began.
|
|
pub head: String,
|
|
pub is_pull: bool,
|
|
pub assignees: BTreeSet<String>,
|
|
/// The PR is a draft: the forge's `pull_request.draft`, true for a title
|
|
/// that starts with one of its work-in-progress prefixes.
|
|
pub draft: bool,
|
|
/// The delivery is a title edit that took a PR out of draft: `edited`, no
|
|
/// longer a draft, and the old title in `changes.title.from` starts with a
|
|
/// work-in-progress prefix.
|
|
pub undrafted: bool,
|
|
}
|
|
|
|
impl Delivery {
|
|
/// Read a delivery body, or `None` when it names no issue or PR.
|
|
pub fn parse(body: &[u8]) -> Option<Self> {
|
|
let json: Json = serde_json::from_slice(body).ok()?;
|
|
let (item, is_pull) = match json.get("pull_request") {
|
|
Some(pr) if pr.is_object() => (pr, true),
|
|
_ => (json.get("issue")?, false),
|
|
};
|
|
let text = |v: &Json, k: &str| {
|
|
v.get(k)
|
|
.and_then(Json::as_str)
|
|
.unwrap_or_default()
|
|
.to_owned()
|
|
};
|
|
let repository = json.get("repository")?;
|
|
let assignees = item
|
|
.get("assignees")
|
|
.and_then(Json::as_array)
|
|
.map(|a| {
|
|
a.iter()
|
|
.filter_map(|u| u.get("login").and_then(Json::as_str))
|
|
.map(str::to_owned)
|
|
.collect()
|
|
})
|
|
.unwrap_or_default();
|
|
let action = text(&json, "action");
|
|
let draft = is_pull && item.get("draft").and_then(Json::as_bool).unwrap_or(false);
|
|
let undrafted = is_pull
|
|
&& action == "edited"
|
|
&& !draft
|
|
&& json
|
|
.pointer("/changes/title/from")
|
|
.and_then(Json::as_str)
|
|
.is_some_and(is_wip_title);
|
|
Some(Self {
|
|
action,
|
|
repo: text(repository, "full_name"),
|
|
repo_id: repository.get("id")?.as_i64()?,
|
|
number: item.get("number")?.as_i64()?,
|
|
title: text(item, "title"),
|
|
url: text(item, "html_url"),
|
|
sender: json
|
|
.get("sender")
|
|
.map(|s| text(s, "login"))
|
|
.unwrap_or_default(),
|
|
author: item
|
|
.get("user")
|
|
.map(|u| text(u, "login"))
|
|
.unwrap_or_default(),
|
|
head: item
|
|
.pointer("/head/sha")
|
|
.and_then(Json::as_str)
|
|
.unwrap_or_default()
|
|
.to_owned(),
|
|
is_pull,
|
|
assignees,
|
|
draft,
|
|
undrafted,
|
|
})
|
|
}
|
|
|
|
/// Read the delivery of Forgejo event type `event_type`. A `workflow_run_*`
|
|
/// delivery carries the webhook that started the run as a JSON string in
|
|
/// `run.event_payload`; for a run on a PR that is the PR's own delivery, so
|
|
/// the PR is read from it, `action` is the run's outcome (`failure`,
|
|
/// `cancelled`) and `sender` is the account that triggered the run. A run
|
|
/// on anything else (a push to a branch with no PR) names no PR and gives
|
|
/// `None`.
|
|
pub fn from_event(event_type: &str, body: &[u8]) -> Option<Self> {
|
|
let Some(outcome) = event_type.strip_prefix("workflow_run_") else {
|
|
return Self::parse(body);
|
|
};
|
|
let json: Json = serde_json::from_slice(body).ok()?;
|
|
let run = json.get("run")?;
|
|
let mut delivery = Self::parse(run.get("event_payload")?.as_str()?.as_bytes())?;
|
|
if !delivery.is_pull {
|
|
return None;
|
|
}
|
|
outcome.clone_into(&mut delivery.action);
|
|
if let Some(trigger) = run.pointer("/trigger_user/login").and_then(Json::as_str) {
|
|
trigger.clone_into(&mut delivery.sender);
|
|
}
|
|
Some(delivery)
|
|
}
|
|
|
|
/// The state bucket key for this delivery's issue: `<repo id>.<number>`,
|
|
/// numeric so it is always a valid key whatever the repo is named.
|
|
pub fn state_key(&self) -> String {
|
|
format!("{}.{}", self.repo_id, self.number)
|
|
}
|
|
}
|
|
|
|
/// The prefixes the forge treats as work-in-progress (its default
|
|
/// `WORK_IN_PROGRESS_PREFIXES`), compared without regard to case.
|
|
const WIP_PREFIXES: [&str; 2] = ["WIP:", "[WIP]"];
|
|
|
|
/// Whether `title` starts with a work-in-progress prefix. A title edit's
|
|
/// payload carries the old title and no flag for it, so the prefix is the only
|
|
/// way to tell what the PR was before.
|
|
fn is_wip_title(title: &str) -> bool {
|
|
let title = title.trim_start();
|
|
WIP_PREFIXES.iter().any(|p| {
|
|
title
|
|
.get(..p.len())
|
|
.is_some_and(|head| head.eq_ignore_ascii_case(p))
|
|
})
|
|
}
|
|
|
|
/// What the router keeps per issue in its state bucket.
|
|
#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)]
|
|
pub struct Snapshot {
|
|
/// The assignees after the last assign delivery seen, less those that
|
|
/// were no agent of the roster then, or `None` before the first.
|
|
#[serde(default)]
|
|
pub assignees: Option<BTreeSet<String>>,
|
|
/// The agents holding a router todo for this issue.
|
|
#[serde(default)]
|
|
pub holders: BTreeSet<String>,
|
|
/// The commit at the head of the PR as of the last delivery about the PR
|
|
/// that was no CI run. A run that began at another commit is superseded
|
|
/// and tells nobody anything.
|
|
#[serde(default)]
|
|
pub head: Option<String>,
|
|
}
|
|
|
|
/// Whether `event_type` is an assign delivery, whose assignee list is the
|
|
/// set after the change.
|
|
pub fn is_assign(event_type: &str) -> bool {
|
|
matches!(event_type, "issue_assign" | "pull_request_assign")
|
|
}
|
|
|
|
/// The fields a flow reads for `delivery`, given the issue's `snapshot`.
|
|
pub fn fields(event_type: &str, delivery: &Delivery, snapshot: &Snapshot) -> Fields {
|
|
let (added, removed) = if is_assign(event_type) {
|
|
match &snapshot.assignees {
|
|
Some(before) => (
|
|
delivery.assignees.difference(before).cloned().collect(),
|
|
before.difference(&delivery.assignees).cloned().collect(),
|
|
),
|
|
None => (delivery.assignees.iter().cloned().collect(), Vec::new()),
|
|
}
|
|
} else {
|
|
(Vec::new(), Vec::new())
|
|
};
|
|
let text = |s: &str| Value::Text(s.to_owned());
|
|
Fields::from([
|
|
("event", text(event_type)),
|
|
("action", text(&delivery.action)),
|
|
("kind", text(if delivery.is_pull { "PR" } else { "issue" })),
|
|
("repo", text(&delivery.repo)),
|
|
("number", Value::Text(delivery.number.to_string())),
|
|
("title", text(&delivery.title)),
|
|
("url", text(&delivery.url)),
|
|
("sender", text(&delivery.sender)),
|
|
("author", text(&delivery.author)),
|
|
("draft", text(bool_text(delivery.draft))),
|
|
("undrafted", text(bool_text(delivery.undrafted))),
|
|
(
|
|
"assignees",
|
|
Value::List(delivery.assignees.iter().cloned().collect()),
|
|
),
|
|
("assignees_added", Value::List(added)),
|
|
("assignees_removed", Value::List(removed)),
|
|
(
|
|
"todo_holders",
|
|
Value::List(snapshot.holders.iter().cloned().collect()),
|
|
),
|
|
])
|
|
}
|
|
|
|
const fn bool_text(b: bool) -> &'static str {
|
|
if b { "true" } else { "false" }
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
fn issue_assign(action: &str, assignees: &[&str]) -> Vec<u8> {
|
|
let users: Vec<Json> = assignees
|
|
.iter()
|
|
.map(|a| serde_json::json!({ "login": a }))
|
|
.collect();
|
|
serde_json::to_vec(&serde_json::json!({
|
|
"action": action,
|
|
"number": 7,
|
|
"issue": {
|
|
"number": 7,
|
|
"title": "router",
|
|
"html_url": "https://forge/issue-7",
|
|
"assignees": users,
|
|
},
|
|
"repository": { "id": 42, "full_name": "hyperhive/hyperhive" },
|
|
"sender": { "login": "mara" },
|
|
}))
|
|
.expect("encodes")
|
|
}
|
|
|
|
fn list(fields: &Fields, name: &str) -> Vec<String> {
|
|
match fields.get(name) {
|
|
Some(Value::List(v)) => v.clone(),
|
|
other => panic!("{name}: {other:?}"),
|
|
}
|
|
}
|
|
|
|
/// The names a stored flow may read are the ones `fields` produces.
|
|
#[test]
|
|
fn the_fields_are_the_ones_a_stored_flow_may_name() {
|
|
let d = Delivery::parse(&issue_assign("assigned", &["atlas"])).expect("parses");
|
|
for event in ["issues", "issue_assign"] {
|
|
let mut got: Vec<&str> = fields(event, &d, &Snapshot::default())
|
|
.keys()
|
|
.copied()
|
|
.collect();
|
|
let mut want = swarm_wire::notify_flows::FIELDS
|
|
.iter()
|
|
.map(|(n, _)| *n)
|
|
.collect::<Vec<_>>();
|
|
got.sort_unstable();
|
|
want.sort_unstable();
|
|
assert_eq!(got, want, "{event}");
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn an_issue_delivery_reads_back_its_fields() {
|
|
let d = Delivery::parse(&issue_assign("assigned", &["atlas", "iris"])).expect("parses");
|
|
assert_eq!(d.repo, "hyperhive/hyperhive");
|
|
assert_eq!(d.number, 7);
|
|
assert!(!d.is_pull);
|
|
assert_eq!(d.state_key(), "42.7");
|
|
assert_eq!(
|
|
d.assignees,
|
|
BTreeSet::from(["atlas".to_owned(), "iris".to_owned()])
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn a_pull_request_delivery_reads_the_pull_request_object() {
|
|
let body = serde_json::to_vec(&serde_json::json!({
|
|
"action": "closed",
|
|
"number": 7,
|
|
"pull_request": { "number": 7, "title": "t", "html_url": "u", "assignees": null },
|
|
"repository": { "id": 1, "full_name": "o/r" },
|
|
"sender": { "login": "s" },
|
|
}))
|
|
.expect("encodes");
|
|
let d = Delivery::parse(&body).expect("parses");
|
|
assert!(d.is_pull);
|
|
assert_eq!(d.number, 7);
|
|
assert!(d.assignees.is_empty());
|
|
assert_eq!(Delivery::parse(b"{\"ref\":\"main\"}"), None);
|
|
}
|
|
|
|
fn pull_body(action: &str, title: &str, draft: bool, was: Option<&str>) -> Vec<u8> {
|
|
let mut body = serde_json::json!({
|
|
"action": action,
|
|
"number": 7,
|
|
"pull_request": { "number": 7, "title": title, "html_url": "u", "draft": draft },
|
|
"repository": { "id": 1, "full_name": "o/r" },
|
|
"sender": { "login": "s" },
|
|
});
|
|
if let Some(was) = was {
|
|
body["changes"] = serde_json::json!({ "title": { "from": was } });
|
|
}
|
|
serde_json::to_vec(&body).expect("encodes")
|
|
}
|
|
|
|
/// A `workflow_run_*` delivery: the run, holding the webhook that started
|
|
/// it as a JSON string.
|
|
fn run_body(started_by: &Json) -> Vec<u8> {
|
|
serde_json::to_vec(&serde_json::json!({
|
|
"action": "failure",
|
|
"run": {
|
|
"id": 5,
|
|
"trigger_user": { "login": "atlas" },
|
|
"event_payload": serde_json::to_string(started_by).expect("encodes"),
|
|
},
|
|
}))
|
|
.expect("encodes")
|
|
}
|
|
|
|
fn pull_payload(author: &str) -> Json {
|
|
serde_json::json!({
|
|
"action": "synchronized",
|
|
"number": 7,
|
|
"pull_request": {
|
|
"number": 7, "title": "t", "html_url": "u", "user": { "login": author },
|
|
"head": { "sha": "abc123" },
|
|
},
|
|
"repository": { "id": 1, "full_name": "o/r" },
|
|
"sender": { "login": "pusher" },
|
|
})
|
|
}
|
|
|
|
#[test]
|
|
fn a_run_on_a_pull_request_reads_as_that_pull_request_with_the_runs_outcome() {
|
|
let body = run_body(&pull_payload("iris"));
|
|
for (event, outcome) in [
|
|
("workflow_run_failure", "failure"),
|
|
("workflow_run_cancelled", "cancelled"),
|
|
] {
|
|
let d = Delivery::from_event(event, &body).expect("a PR run");
|
|
assert!(d.is_pull);
|
|
assert_eq!((d.repo.as_str(), d.number), ("o/r", 7));
|
|
assert_eq!(d.action, outcome);
|
|
assert_eq!(d.author, "iris");
|
|
assert_eq!(d.head, "abc123", "the commit the run began at");
|
|
assert_eq!(d.sender, "atlas", "the account that triggered the run");
|
|
assert_eq!(d.state_key(), "1.7");
|
|
let f = fields(event, &d, &Snapshot::default());
|
|
assert_eq!(f.get("author"), Some(&Value::Text("iris".to_owned())));
|
|
assert_eq!(f.get("kind"), Some(&Value::Text("PR".to_owned())));
|
|
}
|
|
}
|
|
|
|
#[test]
|
|
fn a_run_that_names_no_pull_request_is_ignored() {
|
|
let push = serde_json::json!({
|
|
"ref": "refs/heads/main",
|
|
"repository": { "id": 1, "full_name": "o/r" },
|
|
});
|
|
assert_eq!(
|
|
Delivery::from_event("workflow_run_failure", &run_body(&push)),
|
|
None
|
|
);
|
|
let issue: Json =
|
|
serde_json::from_slice(&issue_assign("assigned", &["atlas"])).expect("json");
|
|
assert_eq!(
|
|
Delivery::from_event("workflow_run_failure", &run_body(&issue)),
|
|
None
|
|
);
|
|
assert_eq!(
|
|
Delivery::from_event("workflow_run_failure", b"not json"),
|
|
None
|
|
);
|
|
assert_eq!(
|
|
Delivery::from_event("workflow_run_failure", b"{\"run\":{}}"),
|
|
None
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn other_events_read_the_body_directly() {
|
|
let body = pull_body("opened", "t", false, None);
|
|
assert_eq!(
|
|
Delivery::from_event("pull_request", &body),
|
|
Delivery::parse(&body)
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn the_forges_draft_flag_marks_a_draft() {
|
|
let d = Delivery::parse(&pull_body("synchronized", "WIP: t", true, None)).expect("parses");
|
|
assert!(d.draft && !d.undrafted);
|
|
let d = Delivery::parse(&pull_body("synchronized", "t", false, None)).expect("parses");
|
|
assert!(!d.draft && !d.undrafted);
|
|
let f = fields("pull_request_sync", &d, &Snapshot::default());
|
|
assert_eq!(f.get("draft"), Some(&Value::Text("false".to_owned())));
|
|
}
|
|
|
|
/// Only an edit that drops a WIP prefix from the title and leaves the PR
|
|
/// not a draft is an un-draft.
|
|
#[test]
|
|
fn a_title_edit_that_drops_the_wip_prefix_is_an_undraft() {
|
|
for was in ["WIP: t", "[WIP] t", "wip: t", " [wip] t"] {
|
|
let d = Delivery::parse(&pull_body("edited", "t", false, Some(was))).expect("parses");
|
|
assert!(d.undrafted, "{was}");
|
|
}
|
|
for (action, title, draft, was) in [
|
|
("edited", "t", false, Some("old title")),
|
|
("edited", "t", false, None),
|
|
("edited", "WIP: t", true, Some("WIP: u")),
|
|
("opened", "t", false, Some("WIP: t")),
|
|
] {
|
|
let d = Delivery::parse(&pull_body(action, title, draft, was)).expect("parses");
|
|
assert!(!d.undrafted, "{action} {title} {was:?}");
|
|
}
|
|
let f = fields(
|
|
"pull_request",
|
|
&Delivery::parse(&pull_body("edited", "t", false, Some("WIP: t"))).expect("parses"),
|
|
&Snapshot::default(),
|
|
);
|
|
assert_eq!(f.get("undrafted"), Some(&Value::Text("true".to_owned())));
|
|
}
|
|
|
|
#[test]
|
|
fn the_snapshot_turns_the_after_list_into_added_and_removed() {
|
|
let d = Delivery::parse(&issue_assign("unassigned", &["iris"])).expect("parses");
|
|
let before = Snapshot {
|
|
assignees: Some(BTreeSet::from(["atlas".to_owned(), "iris".to_owned()])),
|
|
holders: BTreeSet::new(),
|
|
..Snapshot::default()
|
|
};
|
|
let f = fields("issue_assign", &d, &before);
|
|
assert_eq!(list(&f, "assignees_removed"), ["atlas"]);
|
|
assert!(list(&f, "assignees_added").is_empty());
|
|
|
|
let d = Delivery::parse(&issue_assign("assigned", &["atlas", "iris"])).expect("parses");
|
|
let before = Snapshot {
|
|
assignees: Some(BTreeSet::from(["iris".to_owned()])),
|
|
holders: BTreeSet::new(),
|
|
..Snapshot::default()
|
|
};
|
|
assert_eq!(
|
|
list(&fields("issue_assign", &d, &before), "assignees_added"),
|
|
["atlas"]
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn with_no_snapshot_every_assignee_is_added_and_none_removed() {
|
|
let d = Delivery::parse(&issue_assign("assigned", &["atlas"])).expect("parses");
|
|
let f = fields("issue_assign", &d, &Snapshot::default());
|
|
assert_eq!(list(&f, "assignees_added"), ["atlas"]);
|
|
assert!(list(&f, "assignees_removed").is_empty());
|
|
}
|
|
|
|
#[test]
|
|
fn a_delivery_that_is_not_an_assign_adds_and_removes_nobody() {
|
|
let d = Delivery::parse(&issue_assign("closed", &["atlas"])).expect("parses");
|
|
let f = fields("issues", &d, &Snapshot::default());
|
|
assert!(list(&f, "assignees_added").is_empty());
|
|
assert!(list(&f, "assignees_removed").is_empty());
|
|
}
|
|
}
|