diff --git a/Cargo.lock b/Cargo.lock index 6edeb727..db8ce7f9 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -4850,6 +4850,7 @@ dependencies = [ "swarm-queue-client", "swarm-secret-client", "tokio", + "tower", "tracing", "url", "utoipa", diff --git a/swarm-controller/Cargo.toml b/swarm-controller/Cargo.toml index dde94f96..c99cb3da 100644 --- a/swarm-controller/Cargo.toml +++ b/swarm-controller/Cargo.toml @@ -94,10 +94,13 @@ swarm-authelia-bridge-sock.workspace = true # config the crate spells. # `agent-term`: likewise every agent's terminal stream, plus the consumer it # replays that stream through. +# `schema`: the review classifier try-out API serves the shapes the queue +# carries. swarm-queue-client = { workspace = true, features = [ "kv", "subagent-term", "agent-term", + "schema", ] } # `matrix_account.rs` writes the credential this daemon's route accepts. Same # crate the hive reads it back with, which is the point: the path, the field @@ -132,6 +135,8 @@ x509-cert.workspace = true # tests wait out its five-minute bound on paused time. [dev-dependencies] tokio = { workspace = true, features = ["test-util"] } +# `ServiceExt::oneshot`, to drive a route through the real router in tests. +tower = { version = "0.5", features = ["util"] } [lints] workspace = true diff --git a/swarm-controller/src/forge.rs b/swarm-controller/src/forge.rs index 89a9efd1..d408130f 100644 --- a/swarm-controller/src/forge.rs +++ b/swarm-controller/src/forge.rs @@ -24,8 +24,9 @@ use forgejo_api::structs::{ ChangeFileOperationOperation, ChangeFilesOptions, CreateBranchProtectionOption, CreateHookOption, CreateHookOptionConfig, CreateHookOptionType, CreateRepoOption, CreateUserOption, EditUserOption, IssueListIssuesQuery, IssueListIssuesQueryState, - IssueListIssuesQueryType, RepoGetContentsQuery, RepoListPullRequestsQuery, - RepoListPullRequestsQueryState, RepoSearchQuery, StateType, + IssueListIssuesQueryType, RepoDownloadPullDiffOrPatchQuery, RepoGetContentsQuery, + RepoGetPullRequestCommitsQuery, RepoListPullRequestsQuery, RepoListPullRequestsQueryState, + RepoSearchQuery, StateType, }; use forgejo_api::{ApiErrorKind, Auth, Forgejo, ForgejoError}; use futures_util::{StreamExt as _, TryStreamExt as _}; @@ -36,6 +37,7 @@ use std::sync::{Arc, OnceLock}; use utoipa::ToSchema; use crate::webhook::DeliveryKind; +use swarm_queue_client::review_classifier::Commit; pub mod agent_token; pub mod docs_grant; @@ -55,6 +57,14 @@ pub struct ConfigPrStatus { pub html_url: Option, } +/// A PR as [`Client::pull_for_review`] reads it for the review classifier. +#[derive(Clone, Debug, PartialEq, Eq)] +pub struct PullForReview { + pub head_sha: String, + pub commits: Vec, + pub diff: String, +} + /// One row of [`Client::issue_report`] — an open issue plus the two facts /// derived from its dependency graph that swarm-ui's issue-report page /// sorts/filters on. See that function's doc comment for how both are @@ -1031,6 +1041,67 @@ impl Client { Ok(rows.into_iter().flatten().collect()) } + /// What the review classifier reads of `owner/repo`'s PR `number`: its + /// head commit, its commits oldest first with their full messages, and + /// its unified diff. + /// + /// # Errors + /// A forge failure, with the [`ForgejoError`] underneath; a 404 on the + /// first call means the repo or the PR does not exist + /// ([`is_repo_not_found`]). + pub async fn pull_for_review( + &self, + owner: &str, + repo: &str, + number: i64, + ) -> Result { + let pull = self + .api + .repo_get_pull_request(owner, repo, number) + .await + .with_context(|| format!("read PR {number} of {owner}/{repo}"))?; + let head_sha = pull + .head + .and_then(|h| h.sha) + .with_context(|| format!("PR {number} of {owner}/{repo} names no head commit"))?; + let commits = self + .api + .repo_get_pull_request_commits( + owner, + repo, + number, + RepoGetPullRequestCommitsQuery { + verification: Some(false), + files: Some(false), + }, + ) + .all() + .await + .with_context(|| format!("list the commits of PR {number} of {owner}/{repo}"))? + .into_iter() + .map(|c| Commit { + sha: c.sha.unwrap_or_default(), + message: c.commit.and_then(|m| m.message).unwrap_or_default(), + }) + .collect(); + let diff = self + .api + .repo_download_pull_diff_or_patch( + owner, + repo, + number, + "diff", + RepoDownloadPullDiffOrPatchQuery::default(), + ) + .await + .with_context(|| format!("read the diff of PR {number} of {owner}/{repo}"))?; + Ok(PullForReview { + head_sha, + commits, + diff, + }) + } + /// Register the swarm-wide hooks against this controller, so a real /// forge event reaches [`crate::webhook`] instead of the endpoint only /// being reachable by hand. diff --git a/swarm-controller/src/forge_account.rs b/swarm-controller/src/forge_account.rs index f9059351..afdc1d7d 100644 --- a/swarm-controller/src/forge_account.rs +++ b/swarm-controller/src/forge_account.rs @@ -164,6 +164,7 @@ mod tests { forge: std::sync::Arc::default(), create_gate: std::sync::Arc::default(), accounts: Accounts::Fake(Arc::clone(store)), + try_outs: super::super::review_classifier::TryOuts::default(), } } diff --git a/swarm-controller/src/github_account.rs b/swarm-controller/src/github_account.rs index 3a1634a1..60e6d924 100644 --- a/swarm-controller/src/github_account.rs +++ b/swarm-controller/src/github_account.rs @@ -134,6 +134,7 @@ mod tests { forge: std::sync::Arc::default(), create_gate: std::sync::Arc::default(), accounts: Accounts::Fake(Arc::clone(store)), + try_outs: super::super::review_classifier::TryOuts::default(), } } diff --git a/swarm-controller/src/linked_accounts.rs b/swarm-controller/src/linked_accounts.rs index 4b31adf3..b37bc251 100644 --- a/swarm-controller/src/linked_accounts.rs +++ b/swarm-controller/src/linked_accounts.rs @@ -969,6 +969,7 @@ pub(crate) mod tests { forge: std::sync::Arc::default(), create_gate: std::sync::Arc::default(), accounts: super::Accounts::Fake(Arc::clone(store)), + try_outs: super::super::review_classifier::TryOuts::default(), } } diff --git a/swarm-controller/src/main.rs b/swarm-controller/src/main.rs index a2038ed0..15018542 100644 --- a/swarm-controller/src/main.rs +++ b/swarm-controller/src/main.rs @@ -59,6 +59,7 @@ mod matrix_account; mod otel_http_client; mod queue_identity; mod read_policy; +mod review_classifier; mod schedules; mod status; mod store; @@ -735,6 +736,7 @@ fn socket_path() -> PathBuf { (name = "repos", description = "forge repo browsing (swarm-ui's issue-report page)"), (name = "forge", description = "forge site admins (`swarmctl forge make-admin`)"), (name = "schedules", description = "scheduled prompts (`swarmctl schedule`)"), + (name = "review-classifier", description = "trying review classifications against a PR (swarm-ui's try-out page)"), ) )] struct ApiDoc; @@ -857,6 +859,9 @@ struct AppState { create_gate: Arc>, /// The store the linked-account routes read and write. accounts: linked_accounts::Accounts, + /// The review classifier try-outs started here, until their results + /// expire. + try_outs: review_classifier::TryOuts, } /// Env var the controller's NixOS module sets from @@ -3208,6 +3213,7 @@ async fn main() -> Result<()> { forge: forge_client, create_gate: Arc::default(), accounts: linked_accounts::Accounts::Live, + try_outs: review_classifier::TryOuts::default(), }; let app = build_app(state); @@ -3273,6 +3279,8 @@ fn build_app(state: AppState) -> axum::Router { .routes(routes!(issue_report::get_repos)) .routes(routes!(issue_report::get_issue_report_all)) .routes(routes!(issue_report::get_issue_report)) + .routes(routes!(review_classifier::post_try_out)) + .routes(routes!(review_classifier::get_try_out)) .routes(routes!(webhook::post_webhook_forge)) .split_for_parts(); // Just the JSON, not the UI — Swagger UI itself is nginx-hosted from @@ -3361,7 +3369,7 @@ mod tests { >, >; - fn state_with_roster() -> (super::AppState, SharedSched) { + pub(crate) fn state_with_roster() -> (super::AppState, SharedSched) { let sched = std::sync::Arc::new(std::sync::Mutex::new(hive_jobq::scheduler::Scheduler::new( hive_jobq::Graph::new(), @@ -3392,6 +3400,7 @@ mod tests { forge: std::sync::Arc::default(), create_gate: std::sync::Arc::default(), accounts: super::linked_accounts::Accounts::Fake(std::sync::Arc::default()), + try_outs: super::review_classifier::TryOuts::default(), }; (state, sched) } diff --git a/swarm-controller/src/matrix_account.rs b/swarm-controller/src/matrix_account.rs index 82b60847..b29b549b 100644 --- a/swarm-controller/src/matrix_account.rs +++ b/swarm-controller/src/matrix_account.rs @@ -578,6 +578,7 @@ mod tests { forge: std::sync::Arc::default(), create_gate: std::sync::Arc::default(), accounts: Accounts::Fake(Arc::clone(store)), + try_outs: super::super::review_classifier::TryOuts::default(), } } diff --git a/swarm-controller/src/review_classifier.rs b/swarm-controller/src/review_classifier.rs new file mode 100644 index 00000000..de3af7d2 --- /dev/null +++ b/swarm-controller/src/review_classifier.rs @@ -0,0 +1,731 @@ +//! `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. +//! +//! 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. Nothing reaches the forge. +//! +//! The operator is the `Remote-User` header the gateway sets, as for +//! `agent_input`; the classifier names them in its journal line. + +use std::collections::{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; +use axum::extract::{Path, 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::ToSchema; + +use crate::forge::{self, PullForReview}; +use crate::{AppState, error_problem}; + +/// 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"; + +/// 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, +} + +#[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`. + #[serde(skip_serializing_if = "Option::is_none")] + overall: Option, + /// Set when `failed`. + #[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 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: a repo that is not \ + owner/name, no PR number, no classification or more than ten, a classification \ + without a name or description, two with one 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 = "no forge access or no swarm queue on this host, or the \ + forge failed to answer (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() { + return Box::pin(start( + &state.try_outs, + &*fake, + (*fake).clone(), + operator, + run, + )) + .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), + ) + })?; + Box::pin(start(&state.try_outs, forge.as_ref(), queue, operator, run)).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 or settled", body = TryOutStatus), + (status = 404, description = "no run has this id, or its result expired \ + (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", + ) + }) +} + +/// 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) = req + .repo + .split_once('/') + .filter(|(o, n)| segment(o) && segment(n)) + .ok_or_else(|| format!("repo {:?} is not owner/name", 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))?; + let count = req.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 &req.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(Self { + owner: owner.to_owned(), + name: name.to_owned(), + pr, + classifications: req.classifications, + }) + } +} + +/// 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; the reply settles the run when +/// it arrives. +async fn start( + try_outs: &TryOuts, + pulls: &impl PullSource, + sink: impl RunSink, + operator: String, + run: Run, +) -> Result<(StatusCode, Json), ProblemDetails> { + 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, + }; + 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); + let (runs, settle_id) = (try_outs.clone(), id.clone()); + tokio::spawn(async move { + let reply = sink.run(payload).await; + runs.settle(&settle_id, reply); + }); + tracing::info!(%id, repo = %request.repo, pr = request.pr, "review try-out started"); + Ok((StatusCode::ACCEPTED, Json(TryOutAccepted { id }))) +} + +/// 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) + } + } +} + +/// 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::*; + + /// A forge holding one PR, `o/r` 7, and a classifier answering every run + /// with `reply`. + #[derive(Clone)] + pub(crate) struct Fake { + reply: Reply, + } + + 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(), + }) + } else { + Err(ForgejoError::ApiError(ApiErrorKind::NotFound { errors: None }.into()).into()) + } + } + } + + impl RunSink for Fake { + async fn run(&self, _request: Vec) -> Reply { + 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 { + let (mut state, _) = crate::tests::state_with_roster(); + state.try_outs.fake = Some(Arc::new(Fake { reply })); + 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_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_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))); + } +}