OXYGENHUB
# da5118d25db2 oxygenflow / crates / flow / src / wait.rs
6.9 KiB · Rust · last changed flow wait: follow the checks a push started on OxygenHub until they finish, against the repository's required checks; exit 0 passed, 1 failed, 2 at the time limit 1 month ago source blame history raw
1 
//! `flow wait`: follow the checks a push started on OxygenHub until they
2 
//! finish, so scripts (and people) can gate on the pipeline: exit 0 when
3 
//! every check passed, 1 when one failed, 2 on the time limit.
4
5 
use anyhow::{bail, Context, Result};
6 
use std::collections::BTreeMap;
7 
use std::time::{Duration, Instant};
8
9 
#[derive(Debug, PartialEq, Eq)]
10 
pub enum Verdict {
11 
    /// Checks still to come or running.
12 
    Pending,
13 
    Passed,
14 
    /// The checks that failed (or, if required, were skipped).
15 
    Failed(Vec<String>),
16 
}
17
18 
/// Judge a state's checks, `name → status`, against the repository's
19 
/// required checks. A required check that has not reported yet is pending;
20 
/// a skipped one (a job it depends on failed) is a failure.
21 
pub fn verdict(required: &[String], checks: &BTreeMap<String, String>) -> Verdict {
22 
    if checks.is_empty() || required.iter().any(|r| !checks.contains_key(r)) {
23 
        return Verdict::Pending;
24 
    }
25 
    if checks.values().any(|s| s == "queued" || s == "running") {
26 
        return Verdict::Pending;
27 
    }
28 
    let failed: Vec<String> = checks
29 
        .iter()
30 
        .filter(|(name, status)| match status.as_str() {
31 
            "passed" => false,
32 
            "skipped" => required.contains(name),
33 
            _ => true,
34 
        })
35 
        .map(|(name, _)| name.clone())
36 
        .collect();
37 
    if failed.is_empty() {
38 
        Verdict::Passed
39 
    } else {
40 
        Verdict::Failed(failed)
41 
    }
42 
}
43
44 
/// `https://hub/owner/name` to the hub's base URL and `owner/name`.
45 
pub fn split_repo_url(url: &str) -> Result<(String, String)> {
46 
    let trimmed = url.trim_end_matches('/');
47 
    let (scheme, rest) = trimmed.split_once("://").context("a repository URL starts with https:// or http://")?;
48 
    let parts: Vec<&str> = rest.split('/').collect();
49 
    if parts.len() < 3 || parts[parts.len() - 2].is_empty() || parts[parts.len() - 1].is_empty() {
50 
        bail!("{url} is not a repository URL (https://hub/owner/name)");
51 
    }
52 
    let n = parts.len();
53 
    Ok((format!("{scheme}://{}", parts[..n - 2].join("/")), format!("{}/{}", parts[n - 2], parts[n - 1])))
54 
}
55
56 
/// Poll every `every` until the checks of `state` are decided or `limit`
57 
/// passes; print each check when its status changes. Returns the process
58 
/// exit code.
59 
pub fn wait(url: &str, state: &str, token: Option<&str>, ca: Option<&std::path::Path>, limit: Duration, every: Duration) -> Result<i32> {
60 
    if state.len() != 64 || !state.chars().all(|c| c.is_ascii_hexdigit()) {
61 
        bail!("the state is a 64-character hex state hash (oxy push --json prints it)");
62 
    }
63 
    let (base, repo) = split_repo_url(url)?;
64 
    let tls = oxy_core::protocol::tls_client_config(ca).map_err(|e| anyhow::anyhow!("{e}"))?;
65 
    let agent = ureq::AgentBuilder::new().tls_config(tls).timeout(Duration::from_secs(30)).build();
66 
    let get = |path: &str| -> Result<serde_json::Value> {
67 
        let url = format!("{base}/api/v1/repos/{repo}{path}");
68 
        let mut req = agent.get(&url);
69 
        if let Some(t) = token {
70 
            req = req.set("Authorization", &format!("Bearer {t}"));
71 
        }
72 
        match req.call() {
73 
            Ok(r) => Ok(r.into_json()?),
74 
            Err(ureq::Error::Status(code, r)) => bail!("{url}: HTTP {code}: {}", r.into_string().unwrap_or_default()),
75 
            Err(e) => bail!("{url}: {e}"),
76 
        }
77 
    };
78 
    let required: Vec<String> = get("/checks/required")?["names"]
79 
        .as_array()
80 
        .map(|a| a.iter().filter_map(|v| v.as_str().map(String::from)).collect())
81 
        .unwrap_or_default();
82 
    let started = Instant::now();
83 
    let mut seen: BTreeMap<String, String> = BTreeMap::new();
84 
    loop {
85 
        let checks: BTreeMap<String, (String, String)> = get(&format!("/checks/{state}"))?
86 
            .as_array()
87 
            .map(|a| {
88 
                a.iter()
89 
                    .filter_map(|c| {
90 
                        Some((
91 
                            c["name"].as_str()?.to_string(),
92 
                            (c["status"].as_str()?.to_string(), c["summary"].as_str().unwrap_or("").to_string()),
93 
                        ))
94 
                    })
95 
                    .collect()
96 
            })
97 
            .unwrap_or_default();
98 
        for (name, (status, summary)) in &checks {
99 
            if seen.get(name) != Some(status) {
100 
                println!("{name:<24} {status}{}", if summary.is_empty() { String::new() } else { format!("  {summary}") });
101 
                seen.insert(name.clone(), status.clone());
102 
            }
103 
        }
104 
        let statuses: BTreeMap<String, String> = checks.into_iter().map(|(n, (s, _))| (n, s)).collect();
105 
        match verdict(&required, &statuses) {
106 
            Verdict::Passed => {
107 
                println!("every check passed");
108 
                return Ok(0);
109 
            }
110 
            Verdict::Failed(names) => {
111 
                println!("failed: {}", names.join(", "));
112 
                return Ok(1);
113 
            }
114 
            Verdict::Pending if started.elapsed() >= limit => {
115 
                println!("still pending after {} min", limit.as_secs() / 60);
116 
                return Ok(2);
117 
            }
118 
            Verdict::Pending => std::thread::sleep(every),
119 
        }
120 
    }
121 
}
122
123 
#[cfg(test)]
124 
mod tests {
125 
    use super::*;
126
127 
    fn checks(pairs: &[(&str, &str)]) -> BTreeMap<String, String> {
128 
        pairs.iter().map(|(n, s)| (n.to_string(), s.to_string())).collect()
129 
    }
130
131 
    #[test]
132 
    fn verdicts() {
133 
        let req = vec!["flow/fmt".to_string(), "flow/test".to_string()];
134 
        assert_eq!(verdict(&req, &checks(&[])), Verdict::Pending, "nothing reported yet");
135 
        assert_eq!(verdict(&req, &checks(&[("flow/fmt", "passed")])), Verdict::Pending, "a required check still to come");
136 
        assert_eq!(verdict(&req, &checks(&[("flow/fmt", "passed"), ("flow/test", "running")])), Verdict::Pending);
137 
        assert_eq!(verdict(&req, &checks(&[("flow/fmt", "passed"), ("flow/test", "passed")])), Verdict::Passed);
138 
        assert_eq!(
139 
            verdict(&req, &checks(&[("flow/fmt", "passed"), ("flow/test", "failed"), ("lint", "passed")])),
140 
            Verdict::Failed(vec!["flow/test".into()])
141 
        );
142 
        assert_eq!(
143 
            verdict(&req, &checks(&[("flow/fmt", "failed"), ("flow/test", "skipped")])),
144 
            Verdict::Failed(vec!["flow/fmt".into(), "flow/test".into()]),
145 
            "a skipped required check fails"
146 
        );
147 
        assert_eq!(
148 
            verdict(&[], &checks(&[("extra", "skipped"), ("other", "passed")])),
149 
            Verdict::Passed,
150 
            "an optional skipped check does not"
151 
        );
152 
        assert_eq!(verdict(&[], &checks(&[("x", "error")])), Verdict::Failed(vec!["x".into()]));
153 
    }
154
155 
    #[test]
156 
    fn repository_urls() {
157 
        assert_eq!(
158 
            split_repo_url("https://earth.example:7343/andy/oxygen").unwrap(),
159 
            ("https://earth.example:7343".into(), "andy/oxygen".into())
160 
        );
161 
        assert_eq!(split_repo_url("http://h/sub/path/a/b/").unwrap(), ("http://h/sub/path".into(), "a/b".into()));
162 
        assert!(split_repo_url("https://h/only").is_err());
163 
        assert!(split_repo_url("h/a/b").is_err());
164 
    }
165 
}