//! `POST /api/review-classifier/try-out` and //! `GET /api/review-classifier/try-out/{id}`: run classifications against a //! PR from the swarm UI, then read the result back. //! `GET /api/review-classifier/rules`: a repo's own classifications, from //! the rules file on its default branch. [`open_pr`] puts classifications on //! a new branch of the repo, for a PR. //! //! The POST reads the PR with this daemon's forge token, sends one request to //! the review classifier on [`SUBJECT`] and answers 202 with an id at once: //! one model call over a big diff can outlast the 60 seconds the gateway //! holds a request open. The reply is kept in memory, where the GET reads it //! until [`RESULT_TTL`] after it arrived, and recorded in [`runs`]. Nothing //! reaches the forge. //! //! The rules are read with the same token on every request, with no cache. //! //! The operator is the `Remote-User` header the gateway sets, as for //! `agent_input`; the classifier names them in its journal line. //! //! [`review`] runs the same rules on a review request and posts the result. use std::collections::{BTreeMap, HashMap, HashSet}; use std::fmt::Write as _; use std::future::Future; use std::sync::{Arc, Mutex}; use std::time::{Duration, Instant}; use axum::Json; use axum::extract::rejection::{JsonRejection, QueryRejection}; use axum::extract::{Path, Query, State}; use axum::http::{HeaderMap, StatusCode}; use problem_details::ProblemDetails; use serde::{Deserialize, Serialize}; use swarm_queue_client::review_classifier::{ Classification, MAX_CLASSIFICATIONS, Overall, RUN_DEADLINE, Reply, Row, SUBJECT, TryOut, }; use utoipa::{IntoParams, ToSchema}; use self::settings::{Settings, SettingsSource}; use crate::forge::{self, DefaultBranchFiles, PullForReview}; use crate::{AppState, error_problem}; pub(crate) mod open_pr; pub(crate) mod review; pub(crate) mod runs; pub(crate) mod settings; /// How long a finished run's result stays readable. pub(crate) const RESULT_TTL: Duration = Duration::from_hours(1); /// The header the gateway names the session's user in. const OPERATOR_HEADER: &str = "remote-user"; /// Where a repo keeps its rules, as YAML or as JSON. A repo holds one of the /// two. const RULES_PATHS: [&str; 2] = [ ".hyperhive/review-classifier.yml", ".hyperhive/review-classifier.json", ]; /// A run to start. #[derive(Debug, Deserialize, ToSchema)] pub(crate) struct TryOutRequest { /// `owner/name`. repo: String, pr: u64, /// One to ten, each with a distinct name. classifications: Vec, } /// The started run's handle. #[derive(Debug, Serialize, ToSchema)] pub(crate) struct TryOutAccepted { id: String, } /// Whose rules to read. #[derive(Debug, Deserialize, IntoParams)] pub(crate) struct RulesQuery { /// `owner/name`. repo: String, } /// A repo's classifications, as its default branch holds them. #[derive(Debug, Serialize, ToSchema)] pub(crate) struct Rules { /// The file they were read from: `.hyperhive/review-classifier.yml` or /// `.hyperhive/review-classifier.json`. path: String, /// The repo's default branch. branch: String, /// The commit on `branch` the file was read at. commit: String, /// One to ten, each with a distinct name, in the file's order. classifications: Vec, } /// A rules file: the one list, and nothing beside it. #[derive(Deserialize)] #[serde(deny_unknown_fields)] struct RulesFile { classifications: Vec, } /// One entry of a rules file. `unknown` collects every key a /// [`Classification`] doesn't have, which `deny_unknown_fields` can't do /// beside a `flatten`. #[derive(Deserialize)] struct Rule { #[serde(flatten)] classification: Classification, #[serde(flatten)] unknown: BTreeMap, } /// `running` lasts at most the run deadline, 31 minutes: ten /// classifications at three minutes each, one after another, and a minute /// more. #[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, ToSchema)] #[serde(rename_all = "lowercase")] pub(crate) enum RunState { Running, Done, Failed, } /// A run as `GET /api/review-classifier/try-out/{id}` reports it. #[derive(Debug, Clone, PartialEq, Eq, Serialize, ToSchema)] pub(crate) struct TryOutStatus { state: RunState, /// The commit the PR was read at. head_sha: String, /// Set when `done`. #[serde(skip_serializing_if = "Option::is_none")] rows: Option>, /// Set when `done`. Rows with an `error` don't count, so `approve` can /// stand beside them. #[serde(skip_serializing_if = "Option::is_none")] overall: Option, /// Set when `failed`: no API key, no classification answered, no /// classifier listening, or no answer within the run deadline. #[serde(skip_serializing_if = "Option::is_none")] error: Option, } impl TryOutStatus { fn settled(head_sha: String, reply: Reply) -> Self { match reply { Reply::Done { rows, overall } => Self { state: RunState::Done, head_sha, rows: Some(rows), overall: Some(overall), error: None, }, Reply::Failed { error } => Self { state: RunState::Failed, head_sha, rows: None, overall: None, error: Some(error), }, } } } /// The runs this daemon holds, by id. #[derive(Clone, Default)] pub(crate) struct TryOuts { runs: Arc>>, /// Stands in for the forge and the queue when set. #[cfg(test)] fake: Option>, } struct Entry { status: TryOutStatus, started: Instant, settled: Option, } impl Entry { /// Expired once [`RESULT_TTL`] has passed since the reply, or since the /// latest a reply can come. fn expired(&self, now: Instant) -> bool { let since = self.settled.unwrap_or(self.started + RUN_DEADLINE); now.saturating_duration_since(since) > RESULT_TTL } } impl TryOuts { fn lock(&self) -> std::sync::MutexGuard<'_, HashMap> { self.runs .lock() .unwrap_or_else(std::sync::PoisonError::into_inner) } fn insert(&self, id: String, head_sha: String) { let now = Instant::now(); let mut runs = self.lock(); runs.retain(|_, e| !e.expired(now)); runs.insert( id, Entry { status: TryOutStatus { state: RunState::Running, head_sha, rows: None, overall: None, error: None, }, started: now, settled: None, }, ); } fn settle(&self, id: &str, reply: Reply) { if let Some(entry) = self.lock().get_mut(id) { entry.status = TryOutStatus::settled(entry.status.head_sha.clone(), reply); entry.settled = Some(Instant::now()); } } fn get(&self, id: &str) -> Option { let now = Instant::now(); self.lock() .get(id) .filter(|e| !e.expired(now)) .map(|e| e.status.clone()) } } /// Where the PR is read from: the forge in production, a fake in the tests. pub(crate) trait PullSource { fn pull( &self, owner: &str, repo: &str, number: i64, ) -> impl Future> + Send; } impl PullSource for forge::Client { async fn pull(&self, owner: &str, repo: &str, number: i64) -> anyhow::Result { self.pull_for_review(owner, repo, number).await } } /// Where a repo's files are read from: the forge in production, a fake in /// the tests. pub(crate) trait FileSource { fn default_branch_files( &self, owner: &str, repo: &str, paths: &[&str], ) -> impl Future>> + Send; } impl FileSource for forge::Client { async fn default_branch_files( &self, owner: &str, repo: &str, paths: &[&str], ) -> anyhow::Result> { forge::Client::default_branch_files(self, owner, repo, paths).await } } /// Where a run is sent: the queue in production, a fake in the tests. pub(crate) trait RunSink: Clone + Send + Sync + 'static { fn run(&self, request: Vec) -> impl Future + Send; } impl RunSink for async_nats::Client { async fn run(&self, request: Vec) -> Reply { use async_nats::RequestErrorKind; let sent = async_nats::Request::new() .payload(request.into()) .timeout(Some(RUN_DEADLINE)); let error = match self.send_request(SUBJECT, sent).await { Ok(msg) => { return serde_json::from_slice(&msg.payload).unwrap_or_else(|e| Reply::Failed { error: format!("the classifier's reply did not decode: {e}"), }); } Err(e) => match e.kind() { RequestErrorKind::NoResponders => { "no review classifier is listening: deploy.review-classifier.enable is set on \ no host, or its service is not connected to the swarm queue" .to_owned() } RequestErrorKind::TimedOut => format!( "the review classifier sent no answer within {} minutes", RUN_DEADLINE.as_secs() / 60 ), _ => format!("sending the run to the review classifier failed: {e}"), }, }; Reply::Failed { error } } } /// Start a run against a PR. #[utoipa::path( post, path = "/api/review-classifier/try-out", request_body = TryOutRequest, responses( (status = 202, description = "the run started; read it at \ `/api/review-classifier/try-out/{id}`", body = TryOutAccepted), (status = 400, description = "the body is not a valid run: not JSON or not this \ shape (an unknown polarity, a negative or non-integer number, a threshold over \ 255); a repo that is not owner/name with segments of `[A-Za-z0-9._-]` other than \ `.` and `..`; a PR number of 0; no classification or more than ten; a \ classification whose trimmed name or description is empty; two with one trimmed \ name; or a threshold over 100 (problem+json)", body = String), (status = 401, description = "the request carries no `Remote-User`, so there is no \ operator to name (problem+json)", body = String), (status = 404, description = "the forge knows no such repo or PR (problem+json)", body = String), (status = 503, description = "this host has no forge token yet, no swarm queue \ configured, or a queue that is not connected; or the forge failed or did not \ answer within two minutes (problem+json)", body = String), (status = 500, description = "the review classifier's settings could not be read \ (problem+json)", body = String), ), tag = "review-classifier" )] pub(crate) async fn post_try_out( State(state): State, headers: HeaderMap, body: Result, JsonRejection>, ) -> Result<(StatusCode, Json), ProblemDetails> { let operator = operator(&headers).ok_or_else(|| { error_problem( StatusCode::UNAUTHORIZED, "no Remote-User header: the gateway names the operator, and this request did not \ come through it", ) })?; let Json(req) = body.map_err(|e| error_problem(StatusCode::BAD_REQUEST, &e.body_text()))?; let run = Run::checked(req).map_err(|d| error_problem(StatusCode::BAD_REQUEST, &d))?; #[cfg(test)] if let Some(fake) = state.try_outs.fake.clone() { let settings = fake.settings().await.map_err(|e| settings_problem(&e))?; return Box::pin(start( &state.try_outs, &*fake, (*fake).clone(), state.runs.clone(), operator, run, settings, )) .await; } let Some(forge) = state.forge.get() else { return Err(error_problem( StatusCode::SERVICE_UNAVAILABLE, "no forge access on this host yet", )); }; let Some(status) = state.status.as_ref() else { return Err(error_problem( StatusCode::SERVICE_UNAVAILABLE, "no swarm queue is configured on this host", )); }; let queue = status.queue_client(); swarm_queue_client::ensure_connected(&queue).map_err(|e| { error_problem( StatusCode::SERVICE_UNAVAILABLE, &swarm_queue_client::chain(&e), ) })?; let runs = state.runs.clone(); let settings = queue.settings().await.map_err(|e| settings_problem(&e))?; Box::pin(start( &state.try_outs, forge.as_ref(), queue, runs, operator, run, settings, )) .await } /// Read a run's result. #[utoipa::path( get, path = "/api/review-classifier/try-out/{id}", params(("id" = String, Path, description = "the id the POST answered with")), responses( (status = 200, description = "the run: `running` for at most 31 minutes, then \ `done` or `failed`", body = TryOutStatus), (status = 404, description = "no run has this id: it never existed, its result \ expired an hour after it settled, or the controller restarted since the POST, \ which loses every run (problem+json)", body = String), ), tag = "review-classifier" )] pub(crate) async fn get_try_out( State(state): State, Path(id): Path, ) -> Result, ProblemDetails> { state.try_outs.get(&id).map(Json).ok_or_else(|| { error_problem( StatusCode::NOT_FOUND, "no try-out has this id: it never existed, or its result expired", ) }) } /// Read a repo's classifications from its default branch. #[utoipa::path( get, path = "/api/review-classifier/rules", params(RulesQuery), responses( (status = 200, description = "the classifications the default branch's rules file \ holds, and the commit it was read at", body = Rules), (status = 400, description = "no `repo`, or a repo that is not owner/name with \ segments of `[A-Za-z0-9._-]` other than `.` and `..` (problem+json)", body = String), (status = 404, description = "the forge knows no such repo; its default branch has no \ commit, as in an empty repo or one whose default branch doesn't exist; or its \ default branch holds neither `.hyperhive/review-classifier.yml` nor \ `.hyperhive/review-classifier.json` (problem+json)", body = String), (status = 422, description = "the default branch holds both files; or the file does \ not parse as its extension says; or it holds a key other than `classifications`, \ or a classification with a key other than `name`, `description`, `polarity` and \ `threshold`; or it fails a check the try-out POST makes: no classification or \ more than ten, an empty trimmed name or description, two with one trimmed name, \ or a threshold over 100 (problem+json)", body = String), (status = 503, description = "this host has no forge token yet; or the forge failed \ or did not answer within two minutes (problem+json)", body = String), ), tag = "review-classifier" )] pub(crate) async fn get_rules( State(state): State, query: Result, QueryRejection>, ) -> Result, ProblemDetails> { let Query(query) = query.map_err(|e| error_problem(StatusCode::BAD_REQUEST, &e.body_text()))?; let (owner, name) = split_repo(&query.repo).map_err(|d| error_problem(StatusCode::BAD_REQUEST, &d))?; #[cfg(test)] if let Some(fake) = state.try_outs.fake.clone() { return rules(&*fake, owner, name).await; } let Some(forge) = state.forge.get() else { return Err(error_problem( StatusCode::SERVICE_UNAVAILABLE, "no forge access on this host yet", )); }; rules(forge.as_ref(), owner, name).await } /// Read the rules file of `owner/name` and parse it. async fn rules( files: &impl FileSource, owner: &str, name: &str, ) -> Result, ProblemDetails> { let repo = format!("{owner}/{name}"); let read = forge::within_forge_deadline( "review rules", files.default_branch_files(owner, name, &RULES_PATHS), ) .await .ok_or_else(|| { error_problem( StatusCode::SERVICE_UNAVAILABLE, "the forge did not answer within the deadline", ) })? .map_err(|e| rules_problem(&e, &repo))? .ok_or_else(|| { error_problem( StatusCode::NOT_FOUND, &format!("{repo} has no commit on its default branch, so no rules file"), ) })?; let [yml, json] = RULES_PATHS; let (path, bytes) = match read.files.as_slice() { [] => { return Err(error_problem( StatusCode::NOT_FOUND, &format!( "{repo}'s default branch {} holds neither {yml} nor {json}", read.branch ), )); } [file] => (&file.path, &file.bytes), _ => { return Err(error_problem( StatusCode::UNPROCESSABLE_ENTITY, &format!( "{repo}'s default branch {} holds both {yml} and {json}: keep one", read.branch ), )); } }; let classifications = parse_rules(path, bytes) .map_err(|d| error_problem(StatusCode::UNPROCESSABLE_ENTITY, &d))?; tracing::info!(%repo, branch = %read.branch, commit = %read.commit, %path, "review rules read"); Ok(Json(Rules { path: path.clone(), branch: read.branch, commit: read.commit, classifications, })) } /// The classifications in the rules file at `path`, parsed as JSON for a /// `.json` path and as YAML otherwise, or what is wrong with them. fn parse_rules(path: &str, bytes: &[u8]) -> Result, String> { let file: RulesFile = if std::path::Path::new(path) .extension() .is_some_and(|e| e == "json") { serde_json::from_slice(bytes).map_err(|e| format!("{path}: {e}"))? } else { serde_norway::from_slice(bytes).map_err(|e| format!("{path}: {e}"))? }; let mut classifications = Vec::with_capacity(file.classifications.len()); for rule in file.classifications { if let Some(key) = rule.unknown.keys().next() { return Err(format!( "{path}: classification {:?} has an unknown field {key:?}", rule.classification.name.trim() )); } classifications.push(rule.classification); } check_classifications(&classifications).map_err(|e| format!("{path}: {e}"))?; Ok(classifications) } /// A request that passed every check, with its repo split. #[derive(Debug)] struct Run { owner: String, name: String, pr: i64, classifications: Vec, } impl Run { /// The request, or what is wrong with it. fn checked(req: TryOutRequest) -> Result { let (owner, name) = split_repo(&req.repo)?; let pr = i64::try_from(req.pr) .ok() .filter(|n| *n > 0) .ok_or_else(|| format!("{} is not a PR number", req.pr))?; check_classifications(&req.classifications)?; Ok(Self { owner: owner.to_owned(), name: name.to_owned(), pr, classifications: req.classifications, }) } } /// `repo` as its owner and name, or what is wrong with it. pub(crate) fn split_repo(repo: &str) -> Result<(&str, &str), String> { repo.split_once('/') .filter(|(o, n)| segment(o) && segment(n)) .ok_or_else(|| format!("repo {repo:?} is not owner/name")) } /// What is wrong with `classifications` as one run's, if anything. fn check_classifications(classifications: &[Classification]) -> Result<(), String> { let count = classifications.len(); if count == 0 || count > MAX_CLASSIFICATIONS { return Err(format!( "{count} classifications: a run carries 1 to {MAX_CLASSIFICATIONS}" )); } let mut names = HashSet::new(); for c in classifications { let name = c.name.trim(); if name.is_empty() { return Err("a classification has no name".to_owned()); } if c.description.trim().is_empty() { return Err(format!("classification {name} has no description")); } if c.threshold > 100 { return Err(format!( "classification {name}'s threshold {} is over 100", c.threshold )); } if !names.insert(name) { return Err(format!("two classifications are named {name}")); } } Ok(()) } /// One `owner` or `name` of a repo, as the forge allows them. fn segment(s: &str) -> bool { !s.is_empty() && s != "." && s != ".." && s.bytes() .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'-' | b'_' | b'.')) } fn operator(headers: &HeaderMap) -> Option { headers .get(OPERATOR_HEADER) .and_then(|v| v.to_str().ok()) .map(str::trim) .filter(|s| !s.is_empty()) .map(str::to_owned) } /// Read the PR, record the run and send it on `settings`' model and effort; /// the reply settles the run when it arrives, and the settled run goes into /// `runs`. async fn start( try_outs: &TryOuts, pulls: &impl PullSource, sink: impl RunSink, runs: runs::Runs, operator: String, run: Run, settings: Settings, ) -> Result<(StatusCode, Json), ProblemDetails> { let started_unix = crate::agent_renewal::unix_now(); let runtime = settings .runtime() .map_err(|e| error_problem(StatusCode::INTERNAL_SERVER_ERROR, &e))?; let repo = format!("{}/{}", run.owner, run.name); let pull = forge::within_forge_deadline("review try-out", pulls.pull(&run.owner, &run.name, run.pr)) .await .ok_or_else(|| { error_problem( StatusCode::SERVICE_UNAVAILABLE, "the forge did not answer within the deadline", ) })? .map_err(|e| pull_problem(&e, &repo, run.pr))?; let request = TryOut { operator, repo, pr: run.pr.unsigned_abs(), head_sha: pull.head_sha.clone(), commits: pull.commits, diff: pull.diff, classifications: run.classifications, model: settings.model, effort: settings.effort, runtime, }; let payload = serde_json::to_vec(&request).map_err(|e| { error_problem( StatusCode::INTERNAL_SERVER_ERROR, &format!("encoding the run failed: {e}"), ) })?; let id = mint_id().map_err(|e| error_problem(StatusCode::INTERNAL_SERVER_ERROR, &e))?; try_outs.insert(id.clone(), pull.head_sha.clone()); let try_outs = try_outs.clone(); let (settle_id, repo, pr, requester) = ( id.clone(), request.repo.clone(), request.pr, request.operator.clone(), ); tokio::spawn(async move { let reply = sink.run(payload).await; let (rows, overall) = runs::answer(&reply); let (outcome, detail) = match &reply { Reply::Done { .. } => (runs::RunOutcome::Done, None), Reply::Failed { error } => (runs::RunOutcome::Failed, Some(error.clone())), }; try_outs.settle(&settle_id, reply); let record = runs::RunRecord { id: settle_id, trigger: runs::Trigger::TryOut, repo, pr, requester, head_sha: Some(pull.head_sha), started_unix, finished_unix: crate::agent_renewal::unix_now(), outcome, detail, rows, overall, review_url: None, }; runs.record(record).await; }); tracing::info!(%id, repo = %request.repo, pr = request.pr, "review try-out started"); Ok((StatusCode::ACCEPTED, Json(TryOutAccepted { id }))) } /// The settings could not be read: 503 when the queue is not connected, 500 /// otherwise. fn settings_problem(e: &anyhow::Error) -> ProblemDetails { let detail = format!("{e:#}"); tracing::warn!(error = %detail, "reading the review classifier's settings for a try-out failed"); error_problem(crate::wanted_error_status(e), &detail) } /// 404 when the forge knows no such repo or PR, 503 for anything else. fn pull_problem(e: &anyhow::Error, repo: &str, pr: i64) -> ProblemDetails { match e.downcast_ref::() { Some(fe) if forge::is_repo_not_found(fe) => { error_problem(StatusCode::NOT_FOUND, &format!("no PR {pr} in {repo}")) } _ => { let detail = format!("{e:#}"); tracing::warn!(error = %detail, %repo, pr, "reading a PR for a review try-out failed"); error_problem(StatusCode::SERVICE_UNAVAILABLE, &detail) } } } /// 404 when the forge knows no such repo, 503 for anything else. fn rules_problem(e: &anyhow::Error, repo: &str) -> ProblemDetails { match e.downcast_ref::() { Some(fe) if forge::is_repo_not_found(fe) => { error_problem(StatusCode::NOT_FOUND, &format!("no repo {repo}")) } _ => { let detail = format!("{e:#}"); tracing::warn!(error = %detail, %repo, "reading a repo's review rules failed"); error_problem(StatusCode::SERVICE_UNAVAILABLE, &detail) } } } /// A fresh run id: 32 hex digits from the kernel's CSPRNG. fn mint_id() -> Result { let mut bytes = [0u8; 16]; getrandom::fill(&mut bytes).map_err(|e| format!("drawing a run id: {e}"))?; Ok(bytes.iter().fold(String::with_capacity(32), |mut id, b| { let _ = write!(id, "{b:02x}"); id })) } #[cfg(test)] pub(crate) mod tests { use std::sync::Arc; use axum::body::Body; use axum::http::{Request, StatusCode}; use forgejo_api::{ApiErrorKind, ForgejoError}; use swarm_queue_client::review_classifier::{Overall, Polarity, Reply, Row, Verdict}; use tower::ServiceExt as _; use super::*; use crate::forge::BranchFile; use crate::forge::classifier_account::{BranchCommit, CommitAuthor}; /// A forge holding one PR, `o/r` 7, with `files` on `o/r`'s default /// branch, `settings` stored, and a classifier answering every run with /// `reply` and keeping its request. `o/empty` has no commit, `o/down` /// fails with something other than a 404, and `o/hang` never answers. /// The forge has the accounts `alice` and `bob`, and keeps each branch /// pushed with its repo, or refuses it with `push_refusal`. #[derive(Clone)] pub(crate) struct Fake { reply: Reply, files: Vec<(&'static str, String)>, pub(crate) settings: Settings, pub(crate) sent: Arc>>, pub(crate) pushed: Arc>>, pub(crate) push_refusal: Option, pub(crate) narrowed: Option, } impl Fake { pub(crate) fn new(reply: Reply, files: Vec<(&'static str, String)>) -> Self { Self { reply, files, settings: Settings::default(), sent: Arc::default(), pushed: Arc::default(), push_refusal: None, narrowed: None, } } } impl open_pr::BranchSink for Fake { async fn commit_author(&self, login: &str) -> anyhow::Result> { let name = match login { "alice" => "Alice", "bob" => "bob", _ => return Ok(None), }; Ok(Some(CommitAuthor { name: name.to_owned(), email: format!("{login}@example.test"), })) } async fn push(&self, owner: &str, repo: &str, commit: BranchCommit) -> anyhow::Result<()> { if let Some(status) = self.push_refusal { return Err(ForgejoError::UnexpectedStatusCode(status).into()); } self.pushed .lock() .expect("unpoisoned") .push((format!("{owner}/{repo}"), commit)); Ok(()) } } impl SettingsSource for Fake { async fn settings(&self) -> anyhow::Result { Ok(self.settings.clone()) } } impl PullSource for Fake { async fn pull( &self, owner: &str, repo: &str, number: i64, ) -> anyhow::Result { if (owner, repo, number) == ("o", "r", 7) { Ok(PullForReview { head_sha: "abc123".to_owned(), commits: Vec::new(), diff: String::new(), narrowed: self.narrowed.clone(), }) } else { Err(ForgejoError::ApiError(ApiErrorKind::NotFound { errors: None }.into()).into()) } } } impl FileSource for Fake { async fn default_branch_files( &self, owner: &str, repo: &str, paths: &[&str], ) -> anyhow::Result> { match (owner, repo) { ("o", "r") => Ok(Some(DefaultBranchFiles { branch: "main".to_owned(), commit: "def456".to_owned(), web_url: Some(url::Url::parse("https://forge.test/o/r").expect("a URL")), files: paths .iter() .filter_map(|p| self.files.iter().find(|(f, _)| f == p)) .map(|(f, body)| BranchFile { path: (*f).to_owned(), sha: format!("sha-{f}"), bytes: body.clone().into_bytes(), }) .collect(), })), ("o", "empty") => Ok(None), ("o", "down") => Err(anyhow::anyhow!("the forge answered 500")), ("o", "hang") => std::future::pending().await, _ => Err( ForgejoError::ApiError(ApiErrorKind::NotFound { errors: None }.into()).into(), ), } } } impl RunSink for Fake { async fn run(&self, request: Vec) -> Reply { if let Ok(sent) = serde_json::from_slice(&request) { self.sent.lock().expect("unpoisoned").push(sent); } self.reply.clone() } } fn done() -> Reply { Reply::Done { rows: vec![Row { name: "datetime".to_owned(), polarity: Polarity::Positive, verdict: Verdict::Match, certainty: 90, threshold: 80, counted: true, passed: true, reason: "Zones everywhere.".to_owned(), error: None, }], overall: Overall::Approve, } } fn app(reply: Reply) -> axum::Router { app_with_files(reply, Vec::new()) } fn app_with_files(reply: Reply, files: Vec<(&'static str, String)>) -> axum::Router { let (mut state, _) = crate::tests::state_with_roster(); state.try_outs.fake = Some(Arc::new(Fake::new(reply, files))); crate::build_app(state) } fn body(repo: &str, pr: u64) -> String { serde_json::json!({ "repo": repo, "pr": pr, "classifications": [{ "name": "datetime", "description": "All datetimes carry a zone.", "polarity": "positive", "threshold": 80, }], }) .to_string() } async fn post(app: &axum::Router, body: String) -> (StatusCode, serde_json::Value) { let req = Request::post("/api/review-classifier/try-out") .header("content-type", "application/json") .header("remote-user", "alice") .body(Body::from(body)) .expect("a request"); send(app, req).await } async fn get(app: &axum::Router, id: &str) -> (StatusCode, serde_json::Value) { let req = Request::get(format!("/api/review-classifier/try-out/{id}")) .body(Body::empty()) .expect("a request"); send(app, req).await } async fn send(app: &axum::Router, req: Request) -> (StatusCode, serde_json::Value) { let res = app.clone().oneshot(req).await.expect("infallible"); let status = res.status(); let bytes = axum::body::to_bytes(res.into_body(), usize::MAX) .await .expect("a body"); (status, serde_json::from_slice(&bytes).unwrap_or_default()) } /// The run the GET reports once the fake's reply has settled it. async fn settled(app: &axum::Router, id: &str) -> serde_json::Value { for _ in 0..100 { let (status, run) = get(app, id).await; assert_eq!(status, StatusCode::OK, "{run}"); if run["state"] != "running" { return run; } tokio::task::yield_now().await; } panic!("run {id} never settled"); } #[tokio::test] async fn a_valid_run_is_202_with_an_id_the_get_reads_back() { let app = app(done()); let (status, accepted) = post(&app, body("o/r", 7)).await; assert_eq!(status, StatusCode::ACCEPTED, "{accepted}"); let id = accepted["id"].as_str().expect("an id"); let run = settled(&app, id).await; assert_eq!(run["state"], "done"); assert_eq!(run["head_sha"], "abc123"); assert_eq!(run["overall"], "approve"); assert_eq!(run["rows"][0]["verdict"], "match"); assert!(run.get("error").is_none(), "{run}"); } #[tokio::test] async fn a_try_out_runs_on_the_stored_runtime_model_and_effort() { use swarm_queue_client::agent_inference::Effort; use swarm_queue_client::review_classifier::Runtime; let acp = swarm_queue_client::review_classifier::Acp { base_url: "https://llm.example/v1".to_owned(), model: "qwen".to_owned(), prices: swarm_queue_client::review_classifier::Prices { input: 1.0, output: 2.0, cache_read: None, cache_write: None, }, }; let (mut state, _) = crate::tests::state_with_roster(); let fake = Arc::new(Fake { settings: Settings { model: "opus".to_owned(), effort: Some(Effort::High), max_concurrent_reviews: 2, runtime: settings::RuntimeKind::Acp, acp: Some(acp.clone()), }, ..Fake::new(done(), Vec::new()) }); state.try_outs.fake = Some(Arc::clone(&fake)); let app = crate::build_app(state); let (status, accepted) = post(&app, body("o/r", 7)).await; assert_eq!(status, StatusCode::ACCEPTED, "{accepted}"); settled(&app, accepted["id"].as_str().expect("an id")).await; let sent = fake.sent.lock().expect("unpoisoned"); assert_eq!(sent.len(), 1); assert_eq!( (sent[0].model.as_str(), sent[0].effort), ("opus", Some(Effort::High)) ); assert_eq!(sent[0].runtime, Runtime::Acp(acp)); } #[tokio::test] async fn a_failed_run_reports_its_error_and_no_verdict() { let app = app(Reply::Failed { error: "no API key".to_owned(), }); let (_, accepted) = post(&app, body("o/r", 7)).await; let run = settled(&app, accepted["id"].as_str().expect("an id")).await; assert_eq!(run["state"], "failed"); assert_eq!(run["error"], "no API key"); assert!( run.get("overall").is_none() && run.get("rows").is_none(), "{run}" ); } #[tokio::test] async fn a_settled_try_out_is_recorded_as_done_or_failed() { let failed = Reply::Failed { error: "no API key".to_owned(), }; for (reply, outcome, detail) in [ (done(), runs::RunOutcome::Done, None), (failed, runs::RunOutcome::Failed, Some("no API key")), ] { let doc = Arc::new(runs::tests::FakeDoc::default()); let (mut state, _) = crate::tests::state_with_roster(); state.try_outs.fake = Some(Arc::new(Fake::new(reply.clone(), Vec::new()))); state.runs = runs::Runs::Fake(Arc::clone(&doc)); let app = crate::build_app(state); let (_, accepted) = post(&app, body("o/r", 7)).await; let id = accepted["id"].as_str().expect("an id"); settled(&app, id).await; let mut recorded = Vec::new(); for _ in 0..100 { recorded = doc.runs(); if !recorded.is_empty() { break; } tokio::task::yield_now().await; } assert_eq!(recorded.len(), 1, "{outcome:?}"); let run = &recorded[0]; assert_eq!(run.id, id); assert_eq!(run.trigger, runs::Trigger::TryOut); assert_eq!((run.repo.as_str(), run.pr), ("o/r", 7)); assert_eq!(run.requester, "alice"); assert_eq!(run.head_sha.as_deref(), Some("abc123")); assert_eq!(run.outcome, outcome); assert_eq!(run.detail.as_deref(), detail); assert_eq!((run.rows.clone(), run.overall), runs::answer(&reply)); assert!(run.started_unix <= run.finished_unix); } } #[tokio::test] async fn a_bad_body_is_400() { let app = app(done()); let mut eleven = serde_json::from_str::(&body("o/r", 7)).expect("json"); let one = eleven["classifications"][0].clone(); eleven["classifications"] = (0..11) .map(|i| { let mut c = one.clone(); c["name"] = format!("c{i}").into(); c }) .collect(); let with = |field: &str, value: serde_json::Value| { let mut b = serde_json::from_str::(&body("o/r", 7)).expect("json"); b["classifications"][0][field] = value; b.to_string() }; let twice = { let mut b = serde_json::from_str::(&body("o/r", 7)).expect("json"); let c = b["classifications"][0].clone(); b["classifications"] = serde_json::json!([c.clone(), c]); b.to_string() }; for bad in [ "not json".to_owned(), r#"{"repo":"o/r","pr":7}"#.to_owned(), body("o", 7), body("o/r/x", 7), body("../r", 7), body("o/r", 0), r#"{"repo":"o/r","pr":7,"classifications":[]}"#.to_owned(), eleven.to_string(), with("polarity", "sideways".into()), with("threshold", 101.into()), with("name", " ".into()), with("description", "".into()), twice, ] { let (status, problem) = post(&app, bad.clone()).await; assert_eq!(status, StatusCode::BAD_REQUEST, "{bad}: {problem}"); assert!(problem["detail"].is_string(), "{bad}: {problem}"); } } #[tokio::test] async fn an_unknown_repo_or_pr_is_404() { let app = app(done()); for (repo, pr) in [("o/r", 8), ("o/elsewhere", 7)] { let (status, problem) = post(&app, body(repo, pr)).await; assert_eq!(status, StatusCode::NOT_FOUND, "{repo} {pr}: {problem}"); } } #[tokio::test] async fn an_unknown_id_is_404() { let (status, problem) = get(&app(done()), "0123456789abcdef").await; assert_eq!(status, StatusCode::NOT_FOUND, "{problem}"); } #[tokio::test] async fn a_run_without_an_operator_is_401() { let req = Request::post("/api/review-classifier/try-out") .header("content-type", "application/json") .body(Body::from(body("o/r", 7))) .expect("a request"); let (status, _) = send(&app(done()), req).await; assert_eq!(status, StatusCode::UNAUTHORIZED); } /// The queue half against a real server: no classifier listening is an /// error naming the option, and a listener's reply comes back decoded. #[tokio::test] async fn a_run_reaches_a_listening_classifier_and_fails_without_one() { use futures_util::StreamExt as _; let nats = crate::test_nats::TestNats::start(); let client = nats.client().await; let unanswered = client.run(b"{}".to_vec()).await; assert!( matches!(&unanswered, Reply::Failed { error } if error.contains("deploy.review-classifier.enable")), "{unanswered:?}" ); let classifier = nats.client().await; let mut requests = classifier .queue_subscribe( SUBJECT, swarm_queue_client::review_classifier::QUEUE_GROUP.to_owned(), ) .await .expect("subscribe"); // `flush` only writes the buffer. A message the classifier sends itself // and receives back proves the server processed the subscription // before it, since the server reads one connection in order. let mut probe = classifier .subscribe("review-classifier-test.probe") .await .expect("subscribe"); classifier .publish("review-classifier-test.probe", "".into()) .await .expect("publish"); probe.next().await.expect("the probe comes back"); tokio::spawn(async move { let msg = requests.next().await.expect("a request"); let reply = serde_json::to_vec(&done()).expect("encodes"); classifier .publish(msg.reply.expect("a reply subject"), reply.into()) .await .expect("publish"); }); assert_eq!(client.run(b"{}".to_vec()).await, done()); } #[test] fn a_result_expires_after_its_ttl() { let now = Instant::now(); let entry = |settled: Option| Entry { status: TryOutStatus::settled("abc".to_owned(), done()), started: now, settled, }; assert!(!entry(Some(now)).expired(now + RESULT_TTL)); assert!(entry(Some(now)).expired(now + RESULT_TTL + Duration::from_secs(1))); // A run whose reply never came stays until the latest it could have. assert!(!entry(None).expired(now + RUN_DEADLINE + RESULT_TTL)); assert!(entry(None).expired(now + RUN_DEADLINE + RESULT_TTL + Duration::from_secs(1))); } /// A rules file holding `classifications`, and `extra` beside them. fn rules_value( classifications: &serde_json::Value, extra: &[(&str, &str)], ) -> serde_json::Value { let mut file = serde_json::json!({ "classifications": classifications }); for (key, value) in extra { file[*key] = (*value).into(); } file } fn datetime() -> serde_json::Value { serde_json::json!({ "name": "datetime", "description": "All datetimes carry time zone information.", "polarity": "positive", "threshold": 80, }) } /// A JSON rules file holding the one `datetime` classification. pub(crate) fn rules_json() -> (&'static str, String) { let [_, json] = RULES_PATHS; ( json, rules_value(&serde_json::json!([datetime()]), &[]).to_string(), ) } /// `file` as both formats, each with its path. fn both_formats(file: &serde_json::Value) -> [(&'static str, String); 2] { let [yml, json] = RULES_PATHS; [ (yml, serde_norway::to_string(file).expect("encodes as YAML")), (json, file.to_string()), ] } #[test] fn a_rules_file_parses_the_same_as_yaml_and_as_json() { let yaml = "\ classifications: - name: datetime description: All datetimes carry time zone information. polarity: positive threshold: 80 - name: no-ads description: >- A commit message contains a co-authored-by line or other advertising for an AI model. polarity: negative threshold: 90 "; let json = r#"{"classifications": [ {"name": "datetime", "description": "All datetimes carry time zone information.", "polarity": "positive", "threshold": 80}, {"name": "no-ads", "description": "A commit message contains a co-authored-by line or other advertising for an AI model.", "polarity": "negative", "threshold": 90} ]}"#; let want = vec![ Classification { name: "datetime".to_owned(), description: "All datetimes carry time zone information.".to_owned(), polarity: Polarity::Positive, threshold: 80, }, Classification { name: "no-ads".to_owned(), description: "A commit message contains a co-authored-by line or other \ advertising for an AI model." .to_owned(), polarity: Polarity::Negative, threshold: 90, }, ]; let [yml_path, json_path] = RULES_PATHS; assert_eq!(parse_rules(yml_path, yaml.as_bytes()), Ok(want.clone())); assert_eq!(parse_rules(json_path, json.as_bytes()), Ok(want)); } #[test] fn this_repos_own_rules_file_loads() { let [yml_path, _] = RULES_PATHS; let rules = parse_rules( yml_path, include_str!("../../.hyperhive/review-classifier.yml").as_bytes(), ) .expect("the repo's rules file parses"); let names: Vec<_> = rules.iter().map(|c| c.name.as_str()).collect(); assert_eq!( names, [ "no-ads", "no-secrets", "no-lint-suppression", "no-debug-leftovers", "no-dead-code", "tests-with-behaviour", "docs-updated", "one-purpose", ] ); } #[test] fn a_bad_rules_file_is_an_error_in_either_format() { let mut unknown_rule_field = datetime(); unknown_rule_field["threshhold"] = 80.into(); let mut over_100 = datetime(); over_100["threshold"] = 101.into(); let mut no_name = datetime(); no_name["name"] = " ".into(); let mut bad_polarity = datetime(); bad_polarity["polarity"] = "sideways".into(); let mut no_polarity = datetime(); no_polarity .as_object_mut() .expect("an object") .remove("polarity"); let eleven: Vec<_> = (0..11) .map(|i| { let mut c = datetime(); c["name"] = format!("c{i}").into(); c }) .collect(); for (case, file, says) in [ ("no list", serde_json::json!({}), "classifications"), ( "an empty list", rules_value(&serde_json::json!([]), &[]), "0 classifications", ), ( "an unknown top-level field", rules_value(&serde_json::json!([datetime()]), &[("version", "1")]), "version", ), ( "an unknown classification field", rules_value(&serde_json::json!([unknown_rule_field]), &[]), "threshhold", ), ( "eleven classifications", rules_value(&eleven.into(), &[]), "11 classifications", ), ( "a threshold over 100", rules_value(&serde_json::json!([over_100]), &[]), "over 100", ), ( "an empty name", rules_value(&serde_json::json!([no_name]), &[]), "no name", ), ( "no polarity", rules_value(&serde_json::json!([no_polarity]), &[]), "polarity", ), ( "a bad polarity", rules_value(&serde_json::json!([bad_polarity]), &[]), "sideways", ), ( "one name twice", rules_value(&serde_json::json!([datetime(), datetime()]), &[]), "two classifications", ), ] { for (path, body) in both_formats(&file) { let err = parse_rules(path, body.as_bytes()) .expect_err(&format!("{case} in {path} must not parse")); assert!( err.starts_with(path) && err.contains(says), "{case} in {path}: {err}" ); } } let [yml, json] = RULES_PATHS; for (path, body) in [(yml, "classifications: [\n"), (json, "{")] { let err = parse_rules(path, body.as_bytes()).expect_err("must not parse"); assert!(err.starts_with(path), "{err}"); } } async fn get_rules(app: &axum::Router, query: &str) -> (StatusCode, serde_json::Value) { let req = Request::get(format!("/api/review-classifier/rules{query}")) .body(Body::empty()) .expect("a request"); send(app, req).await } #[tokio::test] async fn a_repos_rules_are_200_with_the_commit_they_were_read_at() { let file = rules_value(&serde_json::json!([datetime()]), &[]); for (path, body) in both_formats(&file) { let app = app_with_files(done(), vec![(path, body)]); let (status, rules) = get_rules(&app, "?repo=o/r").await; assert_eq!(status, StatusCode::OK, "{path}: {rules}"); assert_eq!(rules["path"], path); assert_eq!(rules["branch"], "main"); assert_eq!(rules["commit"], "def456"); assert_eq!(rules["classifications"], serde_json::json!([datetime()])); } } #[tokio::test] async fn a_repo_without_rules_or_with_bad_ones_is_an_error() { let good = rules_value(&serde_json::json!([datetime()]), &[]); let bad = rules_value(&serde_json::json!([]), &[]); let mut sideways = datetime(); sideways["polarity"] = "sideways".into(); let sideways = rules_value(&serde_json::json!([sideways]), &[]); let [yml, json] = RULES_PATHS; let one = |path: &'static str, file: &serde_json::Value| vec![(path, file.to_string())]; for (query, files, want, says) in [ ("", Vec::new(), StatusCode::BAD_REQUEST, "repo"), ("?repo=o", Vec::new(), StatusCode::BAD_REQUEST, "owner/name"), ( "?repo=o/elsewhere", Vec::new(), StatusCode::NOT_FOUND, "no repo o/elsewhere", ), ( "?repo=o/empty", Vec::new(), StatusCode::NOT_FOUND, "no commit on its default branch", ), ("?repo=o/r", Vec::new(), StatusCode::NOT_FOUND, "neither"), ( "?repo=o/down", Vec::new(), StatusCode::SERVICE_UNAVAILABLE, "500", ), ( "?repo=o/r", [one(yml, &good), one(json, &good)].concat(), StatusCode::UNPROCESSABLE_ENTITY, "keep one", ), ( "?repo=o/r", one(json, &bad), StatusCode::UNPROCESSABLE_ENTITY, "0 classifications", ), ( "?repo=o/r", one(yml, &sideways), StatusCode::UNPROCESSABLE_ENTITY, "sideways", ), ] { let app = app_with_files(done(), files); let (status, problem) = get_rules(&app, query).await; assert_eq!(status, want, "{query}: {problem}"); let detail = problem["detail"].as_str().expect("a problem+json detail"); assert!(detail.contains(says), "{query}: {detail}"); } } #[tokio::test] async fn rules_without_forge_access_on_this_host_are_503() { let (state, _) = crate::tests::state_with_roster(); let (status, problem) = get_rules(&crate::build_app(state), "?repo=o/r").await; assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "{problem}"); assert_eq!(problem["detail"], "no forge access on this host yet"); } /// The paused clock jumps to the deadline once nothing else can run, so /// this takes no wall time. #[tokio::test(start_paused = true)] async fn rules_from_a_forge_that_never_answers_are_503_at_the_deadline() { let start = tokio::time::Instant::now(); let (status, problem) = get_rules(&app(done()), "?repo=o/hang").await; assert_eq!(status, StatusCode::SERVICE_UNAVAILABLE, "{problem}"); assert_eq!( problem["detail"], "the forge did not answer within the deadline" ); assert_eq!(start.elapsed(), forge::FORGE_DEADLINE); } }