use std::collections::{
HashMap,
HashSet,
};
use std::sync::Arc;
use std::time::Duration;
use async_trait::async_trait;
use futures::future::try_join_all;
use pipedash_plugin_api::*;
use crate::log_dialect::AzureDevOpsLogDialect;
use crate::{
client,
config,
mapper,
metadata,
types,
};
pub struct AzureDevOpsPlugin {
metadata: PluginMetadata,
client: Option<client::AzureDevOpsClient>,
provider_id: Option<i64>,
organization: String,
config: HashMap<String, String>,
status_source: Option<StatusSource>,
}
impl Default for AzureDevOpsPlugin {
fn default() -> Self {
Self::new()
}
}
impl AzureDevOpsPlugin {
pub fn new() -> Self {
Self {
metadata: metadata::create_metadata(),
client: None,
provider_id: None,
organization: String::new(),
config: HashMap::new(),
status_source: None,
}
}
fn client(&self) -> PluginResult<&client::AzureDevOpsClient> {
self.client
.as_ref()
.ok_or_else(|| PluginError::Internal("Plugin not initialized".to_string()))
}
fn provider_id(&self) -> PluginResult<i64> {
self.provider_id
.ok_or_else(|| PluginError::Internal("Provider ID not set".to_string()))
}
async fn fetch_project_pipelines(
&self, provider_id: i64, project: &str,
) -> PluginResult<Vec<Pipeline>> {
let definitions = self.client()?.list_definitions(project).await?;
Ok(definitions
.iter()
.map(|definition| {
mapper::map_pipeline(provider_id, &self.organization, project, definition)
})
.collect())
}
async fn owned_build(
&self, project: &str, definition_id: i64, pipeline_id: &str, build_id: i64,
) -> PluginResult<types::Build> {
let build = self.client()?.get_build(project, build_id).await?;
let belongs_to_pipeline = build
.definition
.as_ref()
.and_then(|definition| definition.id)
.is_none_or(|id| id == definition_id);
if !belongs_to_pipeline {
return Err(PluginError::PipelineNotFound(format!(
"Run #{build_id} does not belong to {pipeline_id}"
)));
}
Ok(build)
}
fn log_node_record_id(node_id: &str) -> PluginResult<&str> {
Some(node_id)
.filter(|id| !id.is_empty() && id.bytes().all(|b| b.is_ascii_hexdigit() || b == b'-'))
.ok_or_else(|| {
PluginError::InvalidConfig(format!("Invalid Azure DevOps log node ID: {node_id}"))
})
}
}
#[async_trait]
impl Plugin for AzureDevOpsPlugin {
fn metadata(&self) -> &PluginMetadata {
&self.metadata
}
fn status_source(&self) -> Option<&StatusSource> {
self.status_source.as_ref()
}
fn initialize(
&mut self, provider_id: i64, config: HashMap<String, String>,
http_client: Option<Arc<reqwest::Client>>,
) -> PluginResult<()> {
let server_url = config::get_server_url(&config)?;
let token = config
.get("token")
.map(|token| token.trim())
.filter(|token| !token.is_empty())
.ok_or_else(|| {
PluginError::InvalidConfig("Missing Azure DevOps personal access token".to_string())
})?
.to_string();
let status_source = custom_status_source(&config)?;
let http_client = match http_client {
Some(http_client) => http_client,
None => Arc::new(
reqwest::Client::builder()
.use_rustls_tls()
.https_only(cfg!(target_os = "ios"))
.pool_max_idle_per_host(10)
.timeout(Duration::from_secs(30))
.connect_timeout(Duration::from_secs(10))
.tcp_keepalive(Duration::from_secs(60))
.build()
.map_err(|e| {
PluginError::Internal(format!("Failed to build HTTP client: {e}"))
})?,
),
};
self.organization = config::organization_of(&server_url);
self.status_source = status_source;
self.client = Some(client::AzureDevOpsClient::new(
http_client,
server_url,
token,
));
self.provider_id = Some(provider_id);
self.config = config;
Ok(())
}
async fn validate_credentials(&self) -> PluginResult<bool> {
self.client()?.ping().await?;
Ok(true)
}
async fn fetch_available_pipelines(
&self, params: Option<PaginationParams>,
) -> PluginResult<PaginatedAvailablePipelines> {
self.fetch_available_pipelines_filtered(None, None, params)
.await
}
async fn fetch_organizations(&self) -> PluginResult<Vec<Organization>> {
self.client()?;
Ok(vec![Organization {
id: self.organization.clone(),
name: self.organization.clone(),
description: None,
}])
}
async fn ids_with_ci(&self, ids: Vec<String>) -> PluginResult<Vec<String>> {
let client = self.client()?;
Ok(retain_with_ci(ids, |project| async move {
client.has_definitions(&project).await
})
.await)
}
async fn fetch_available_pipelines_filtered(
&self, org: Option<String>, search: Option<String>, params: Option<PaginationParams>,
) -> PluginResult<PaginatedAvailablePipelines> {
let client = self.client()?;
let params = params.unwrap_or_default();
let start = params
.calculate_offset()
.map_err(PluginError::InvalidConfig)?;
let mut projects = if org.is_none_or(|org| org == self.organization) {
client.list_projects().await?
} else {
Vec::new()
};
if let Some(term) = search.as_deref().map(str::to_lowercase) {
projects.retain(|project| {
project.name.to_lowercase().contains(&term)
|| project
.description
.as_deref()
.is_some_and(|description| description.to_lowercase().contains(&term))
});
}
projects.sort_by_cached_key(|project| project.name.to_lowercase());
let total_count = projects.len();
let items = projects
.iter()
.skip(start)
.take(params.page_size)
.map(|project| mapper::map_available_pipeline(project, &self.organization))
.collect::<Vec<_>>();
let mut response =
PaginatedResponse::new(items, params.page, params.page_size, total_count);
response.has_more = start + response.items.len() < total_count;
Ok(response)
}
async fn fetch_pipelines(&self) -> PluginResult<Vec<Pipeline>> {
let provider_id = self.provider_id()?;
let projects = config::get_projects(&self.config);
if projects.is_empty() {
return Err(PluginError::InvalidConfig(
"No projects configured".to_string(),
));
}
let results = try_join_all(
projects
.iter()
.map(|project| self.fetch_project_pipelines(provider_id, project)),
)
.await?;
Ok(results.into_iter().flatten().collect())
}
async fn fetch_run_history(
&self, pipeline_id: &str, limit: usize,
) -> PluginResult<Vec<PipelineRun>> {
let id = config::parse_pipeline_id(pipeline_id)?;
let client = self.client()?;
let builds = client
.list_builds(id.project, id.definition_id, limit)
.await?;
Ok(builds
.iter()
.map(|build| mapper::map_run(build, pipeline_id, id.project, client.server_url()))
.collect())
}
async fn fetch_run_details(
&self, pipeline_id: &str, run_number: i64,
) -> PluginResult<PipelineRun> {
let id = config::parse_pipeline_id(pipeline_id)?;
let build = self
.owned_build(id.project, id.definition_id, pipeline_id, run_number)
.await?;
Ok(mapper::map_run(
&build,
pipeline_id,
id.project,
self.client()?.server_url(),
))
}
async fn fetch_log_nodes(&self, run: RunRef<'_>) -> PluginResult<Vec<LogNode>> {
let id = config::parse_pipeline_id(run.pipeline_id)?;
let timeline = self
.client()?
.get_timeline(id.project, run.run_number)
.await?;
Ok(mapper::timeline_to_log_nodes(&timeline))
}
async fn read_log(
&self, run: RunRef<'_>, node_id: &str, from: Option<&LogCursor>, sink: &mut LogSink,
) -> PluginResult<LogRead> {
let id = config::parse_pipeline_id(run.pipeline_id)?;
let record_id = Self::log_node_record_id(node_id)?;
let client = self.client()?;
let timeline = client.get_timeline(id.project, run.run_number).await?;
let (log_id, attempt) = mapper::job_log_ref(&timeline, record_id).ok_or_else(|| {
PluginError::PipelineNotFound(format!(
"Run #{} has no job log {node_id}",
run.run_number
))
})?;
client
.stream_build_log(id.project, run.run_number, log_id, attempt, from, sink)
.await
}
fn log_dialect(&self) -> &'static dyn LogDialect {
&AzureDevOpsLogDialect
}
async fn fetch_workflow_parameters(
&self, workflow_id: &str,
) -> PluginResult<Vec<WorkflowParameter>> {
let id = config::parse_pipeline_id(workflow_id)?;
let definition = self
.client()?
.get_definition(id.project, id.definition_id)
.await?;
let default_branch = definition
.repository
.and_then(|repository| repository.default_branch)
.filter(|branch| !branch.is_empty())
.map(|branch| mapper::short_branch(&branch).to_string());
let mut parameters = vec![WorkflowParameter {
name: mapper::BRANCH_INPUT.to_string(),
label: Some("Branch".to_string()),
description: Some(
"Branch to build. The definition's default branch is used when empty".to_string(),
),
param_type: WorkflowParameterType::String {
default: default_branch,
},
required: false,
}];
parameters.extend(
definition
.variables
.unwrap_or_default()
.into_iter()
.filter(|(_, variable)| variable.is_overridable())
.map(|(name, variable)| WorkflowParameter {
label: Some(name.clone()),
description: Some("Pipeline variable".to_string()),
param_type: WorkflowParameterType::String {
default: variable.value,
},
required: false,
name,
}),
);
Ok(parameters)
}
async fn trigger_pipeline(&self, params: TriggerParams) -> PluginResult<String> {
let id = config::parse_pipeline_id(¶ms.workflow_id)?;
let client = self.client()?;
let overridable: HashSet<String> = client
.get_definition(id.project, id.definition_id)
.await?
.variables
.unwrap_or_default()
.into_iter()
.filter(|(_, variable)| variable.is_overridable())
.map(|(name, _)| name.to_lowercase())
.collect();
let body = mapper::trigger_body(id.definition_id, params.inputs, &overridable);
let build = client.queue_build(id.project, &body).await?;
let logs_url = build
.web_link()
.map(str::to_string)
.unwrap_or_else(|| config::build_web_url(client.server_url(), id.project, build.id));
let branch = body["sourceBranch"]
.as_str()
.map(mapper::short_branch)
.unwrap_or("the default branch");
Ok(serde_json::json!({
"message": format!("Queued build {} on {branch}", build.id),
"run_number": build.id,
"logs_url": logs_url
})
.to_string())
}
async fn cancel_run(&self, pipeline_id: &str, run_number: i64) -> PluginResult<()> {
let id = config::parse_pipeline_id(pipeline_id)?;
let client = self.client()?;
self.owned_build(id.project, id.definition_id, pipeline_id, run_number)
.await?;
client.cancel_build(id.project, run_number).await
}
}
#[cfg(all(test, not(target_os = "ios")))]
mod tests {
use std::net::SocketAddr;
use std::sync::{
Mutex,
MutexGuard,
};
use serde_json::{
json,
Value,
};
use tokio::io::{
AsyncReadExt,
AsyncWriteExt,
};
use tokio::net::{
TcpListener,
TcpStream,
};
use super::*;
use crate::log_tail::{
LogPosition,
Tally,
BOUNDARY_LINE_LIMIT,
};
const BASIC_AUTH: &str = "Basic OnN0dWItdG9rZW4=";
struct Request {
method: String,
url: reqwest::Url,
authorization: Option<String>,
accept: Option<String>,
body: String,
}
impl Request {
fn query(&self, key: &str) -> Option<String> {
self.url
.query_pairs()
.find(|(name, _)| name == key)
.map(|(_, value)| value.into_owned())
}
}
struct Reply {
status: u16,
content_type: &'static str,
continuation: Option<String>,
body: String,
}
impl Reply {
fn json(status: u16, body: Value) -> Self {
Self {
status,
content_type: "application/json; charset=utf-8",
continuation: None,
body: body.to_string(),
}
}
fn html(status: u16, body: &str) -> Self {
Self {
status,
content_type: "text/html; charset=utf-8",
continuation: None,
body: body.to_string(),
}
}
fn text(status: u16, body: &str) -> Self {
Self {
status,
content_type: "text/plain; charset=utf-8",
continuation: None,
body: body.to_string(),
}
}
fn not_found(request: &Request) -> Self {
Self::json(
404,
json!({"message": format!("unexpected {} {}", request.method, request.url)}),
)
}
fn continued(mut self, token: &str) -> Self {
self.continuation = Some(token.to_string());
self
}
fn to_http(&self) -> String {
let continuation = self
.continuation
.as_ref()
.map(|token| format!("x-ms-continuationtoken: {token}\r\n"))
.unwrap_or_default();
format!(
"HTTP/1.1 {} X\r\nContent-Type: {}\r\n{continuation}Content-Length: {}\r\nConnection: close\r\n\r\n{}",
self.status,
self.content_type,
self.body.len(),
self.body
)
}
}
struct Stub {
base_url: String,
requests: Arc<Mutex<Vec<Request>>>,
server: tokio::task::JoinHandle<()>,
}
impl Stub {
fn requests(&self) -> MutexGuard<'_, Vec<Request>> {
self.requests.lock().unwrap()
}
fn requests_to(&self, method: &str, path: &str) -> Vec<(Option<String>, Option<String>)> {
self.requests()
.iter()
.filter(|request| request.method == method && request.url.path() == path)
.map(|request| (request.query("$top"), request.query("continuationToken")))
.collect()
}
}
impl Drop for Stub {
fn drop(&mut self) {
self.server.abort();
}
}
async fn read_request(stream: &mut TcpStream, addr: SocketAddr) -> Request {
let mut raw = Vec::new();
let mut buffer = [0; 4096];
let header_end = loop {
if let Some(position) = raw.windows(4).position(|window| window == b"\r\n\r\n") {
break position + 4;
}
let count = stream.read(&mut buffer).await.unwrap();
assert!(count > 0, "connection closed before the request head ended");
raw.extend_from_slice(&buffer[..count]);
};
let head = String::from_utf8(raw[..header_end].to_vec()).unwrap();
let header = |name: &str| {
head.lines()
.filter_map(|line| line.split_once(':'))
.find(|(key, _)| key.eq_ignore_ascii_case(name))
.map(|(_, value)| value.trim().to_string())
};
let length: usize = header("content-length")
.map(|value| value.parse().unwrap())
.unwrap_or(0);
while raw.len() < header_end + length {
let count = stream.read(&mut buffer).await.unwrap();
assert!(count > 0, "connection closed before the request body ended");
raw.extend_from_slice(&buffer[..count]);
}
let mut request_line = head.lines().next().unwrap().split_whitespace();
let method = request_line.next().unwrap().to_string();
let target = request_line.next().unwrap();
Request {
method,
url: reqwest::Url::parse(&format!("http://{addr}{target}")).unwrap(),
authorization: header("authorization"),
accept: header("accept"),
body: String::from_utf8(raw[header_end..header_end + length].to_vec()).unwrap(),
}
}
async fn serve(reply: fn(&Request) -> Reply) -> Stub {
let _ = rustls::crypto::ring::default_provider().install_default();
let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
let addr = listener.local_addr().unwrap();
let requests = Arc::new(Mutex::new(Vec::new()));
let recorded = Arc::clone(&requests);
let server = tokio::spawn(async move {
loop {
let (mut stream, _) = listener.accept().await.unwrap();
let request = read_request(&mut stream, addr).await;
let answer = reply(&request);
recorded.lock().unwrap().push(request);
stream.write_all(answer.to_http().as_bytes()).await.unwrap();
}
});
Stub {
base_url: format!("http://{addr}"),
requests,
server,
}
}
fn plugin_for(stub: &Stub, organization_suffix: &str, selected: &str) -> AzureDevOpsPlugin {
let mut plugin = AzureDevOpsPlugin::new();
plugin
.initialize(
1,
HashMap::from([
(
"server_url".to_string(),
format!("{}/acme{organization_suffix}", stub.base_url),
),
("token".to_string(), "stub-token".to_string()),
("selected_items".to_string(), selected.to_string()),
]),
None,
)
.unwrap();
plugin
}
fn build_json(id: i64, status: &str, result: Option<&str>, definition_id: i64) -> Value {
let mut build = json!({
"id": id,
"buildNumber": format!("20261028.{}", id % 100),
"status": status,
"sourceBranch": "refs/heads/main",
"sourceVersion": format!("sha{id}"),
"queueTime": "2026-10-28T10:00:00.1234567Z",
"startTime": "2026-10-28T10:00:05.5Z",
"requestedFor": {"displayName": "Ada Lovelace"},
"definition": {"id": definition_id},
"_links": {"web": {"href": format!("https://dev.azure.com/acme/Alpha/_build/results?buildId={id}")}}
});
if let Some(result) = result {
build["result"] = json!(result);
build["finishTime"] = json!("2026-10-28T10:02:35.5Z");
}
build
}
fn azure_reply(request: &Request) -> Reply {
let token = request.query("continuationToken");
match (request.method.as_str(), request.url.path()) {
("GET", "/acme/_apis/projects") => match token.as_deref() {
None => Reply::json(
200,
json!({"count": 2, "value": [
{"id": "p1", "name": "Alpha", "description": "Main product"},
{"id": "p2", "name": "Beta Team"}
]}),
)
.continued("projects-2"),
Some("projects-2") => {
Reply::json(200, json!({"count": 1, "value": [{"id": "p3", "name": "Empty"}]}))
}
Some(_) => Reply::not_found(request),
},
("GET", "/acme/Alpha/_apis/build/definitions") => match token.as_deref() {
None => Reply::json(
200,
json!({"count": 2, "value": [
{"id": 1, "name": "CI", "latestBuild": build_json(903, "completed", Some("succeeded"), 1)},
{"id": 2, "name": "Docs"}
]}),
)
.continued("defs-2"),
Some("defs-2") => Reply::json(
200,
json!({"count": 2, "value": [
{"id": 3, "name": "Deploy", "latestBuild": build_json(905, "inProgress", None, 3)},
{"id": 4, "name": "Nightly", "latestBuild": build_json(906, "completed", Some("partiallySucceeded"), 4)}
]}),
),
Some(_) => Reply::not_found(request),
},
("GET", "/acme/Beta%20Team/_apis/build/definitions") => Reply::json(
200,
json!({"count": 1, "value": [
{"id": 5, "name": "Mobile", "latestBuild": build_json(907, "completed", Some("canceled"), 5)}
]}),
),
("GET", "/acme/Empty/_apis/build/definitions") => {
Reply::json(200, json!({"count": 0, "value": []}))
}
("GET", "/acme/Alpha/_apis/build/definitions/2") => Reply::json(
200,
json!({
"id": 2, "name": "Docs",
"variables": {"Ref": {"value": "x", "allowOverride": true}}
}),
),
("GET", "/acme/Alpha/_apis/build/definitions/1") => Reply::json(
200,
json!({
"id": 1, "name": "CI",
"repository": {"defaultBranch": "refs/heads/main"},
"variables": {
"deployTarget": {"value": "dev", "allowOverride": true},
"fixed": {"value": "x"},
"apiKey": {"isSecret": true, "allowOverride": true},
"branch": {"value": "legacy", "allowOverride": true}
}
}),
),
("GET", "/acme/Alpha/_apis/build/builds") => {
let history = [
build_json(923, "inProgress", None, 1),
build_json(922, "completed", Some("failed"), 1),
build_json(921, "completed", Some("succeeded"), 1),
];
let start = match token.as_deref() {
None => 0,
Some("builds-2") => 2,
Some(_) => return Reply::not_found(request),
};
let size = request
.query("$top")
.and_then(|top| top.parse().ok())
.unwrap_or(history.len())
.min(2);
let page: Vec<_> = history[start..].iter().take(size).collect();
let reply = Reply::json(200, json!({"count": page.len(), "value": page}));
if start + size < history.len() {
reply.continued("builds-2")
} else {
reply
}
}
("GET", "/acme/Alpha/_apis/build/builds/923") => {
Reply::json(200, build_json(923, "inProgress", None, 1))
}
("GET", "/acme/Alpha/_apis/build/builds/922") => {
Reply::json(200, build_json(922, "completed", Some("failed"), 1))
}
("GET", "/acme/Alpha/_apis/build/builds/777") => {
Reply::json(200, build_json(777, "completed", Some("succeeded"), 9))
}
("POST", "/acme/Alpha/_apis/build/builds") => {
Reply::json(200, build_json(930, "notStarted", None, 1))
}
("PATCH", "/acme/Alpha/_apis/build/builds/923") => {
Reply::json(200, build_json(923, "cancelling", None, 1))
}
_ => Reply::not_found(request),
}
}
#[tokio::test]
async fn projects_are_listed_across_continuation_pages_and_paged_for_the_picker() {
let stub = serve(azure_reply).await;
let plugin = plugin_for(&stub, "", "");
let first = plugin
.fetch_available_pipelines(Some(PaginationParams {
page: 1,
page_size: 2,
}))
.await
.unwrap();
let second = plugin
.fetch_available_pipelines(Some(PaginationParams {
page: 2,
page_size: 2,
}))
.await
.unwrap();
let names = |page: &PaginatedAvailablePipelines| {
page.items
.iter()
.map(|item| item.id.clone())
.collect::<Vec<_>>()
};
assert_eq!(names(&first), ["Alpha", "Beta Team"]);
assert!(first.has_more);
assert_eq!(first.total_count, 3);
assert_eq!(first.items[0].description.as_deref(), Some("Main product"));
assert_eq!(first.items[0].organization.as_deref(), Some("acme"));
assert_eq!(names(&second), ["Empty"]);
assert!(!second.has_more);
assert_eq!(
stub.requests_to("GET", "/acme/_apis/projects"),
[
(Some("100".to_string()), None),
(Some("100".to_string()), Some("projects-2".to_string())),
(Some("100".to_string()), None),
(Some("100".to_string()), Some("projects-2".to_string())),
]
);
assert!(stub.requests().iter().all(|request| {
request.query("api-version").as_deref() == Some("7.0")
&& request.authorization.as_deref() == Some(BASIC_AUTH)
}));
}
#[tokio::test]
async fn picker_filters_by_organization_and_search_term() {
let stub = serve(azure_reply).await;
let plugin = plugin_for(&stub, "", "");
let found = plugin
.fetch_available_pipelines_filtered(
Some("acme".to_string()),
Some("TEAM".to_string()),
None,
)
.await
.unwrap();
assert_eq!(found.items.len(), 1);
assert_eq!(found.items[0].name, "Beta Team");
let requests_before = stub.requests().len();
let other = plugin
.fetch_available_pipelines_filtered(Some("other".to_string()), None, None)
.await
.unwrap();
assert!(other.items.is_empty());
assert_eq!(stub.requests().len(), requests_before);
let organizations = plugin.fetch_organizations().await.unwrap();
assert_eq!(organizations.len(), 1);
assert_eq!(organizations[0].id, "acme");
assert_eq!(organizations[0].name, "acme");
}
#[tokio::test]
async fn definitions_of_selected_projects_become_pipelines_across_continuation_pages() {
let stub = serve(azure_reply).await;
let plugin = plugin_for(&stub, "", "Alpha, Beta Team");
let pipelines = plugin.fetch_pipelines().await.unwrap();
let summary: Vec<_> = pipelines
.iter()
.map(|pipeline| {
(
pipeline.id.as_str(),
pipeline.name.as_str(),
pipeline.status,
pipeline.repository.as_str(),
pipeline.branch.as_deref(),
)
})
.collect();
assert_eq!(
summary,
[
(
"azure-devops__1__Alpha/1",
"CI",
PipelineStatus::Success,
"acme/Alpha",
Some("main")
),
(
"azure-devops__1__Alpha/2",
"Docs",
PipelineStatus::Pending,
"acme/Alpha",
None
),
(
"azure-devops__1__Alpha/3",
"Deploy",
PipelineStatus::Running,
"acme/Alpha",
Some("main")
),
(
"azure-devops__1__Alpha/4",
"Nightly",
PipelineStatus::Failed,
"acme/Alpha",
Some("main")
),
(
"azure-devops__1__Beta Team/5",
"Mobile",
PipelineStatus::Cancelled,
"acme/Beta Team",
Some("main")
),
]
);
assert!(pipelines
.iter()
.all(|pipeline| pipeline.provider_type == "azure-devops" && pipeline.provider_id == 1));
assert_eq!(
stub.requests_to("GET", "/acme/Alpha/_apis/build/definitions"),
[
(Some("100".to_string()), None),
(Some("100".to_string()), Some("defs-2".to_string())),
]
);
assert!(stub
.requests()
.iter()
.filter(|request| request.url.path().ends_with("/build/definitions"))
.all(|request| request.query("includeLatestBuilds").as_deref() == Some("true")));
}
#[tokio::test]
async fn fetching_pipelines_without_selected_projects_is_a_configuration_error() {
let stub = serve(azure_reply).await;
let error = plugin_for(&stub, "", "")
.fetch_pipelines()
.await
.unwrap_err();
assert!(matches!(error, PluginError::InvalidConfig(_)), "{error}");
}
#[tokio::test]
async fn a_trailing_slash_on_the_server_url_does_not_double_the_slash() {
let stub = serve(azure_reply).await;
let plugin = plugin_for(&stub, "/", "Alpha");
plugin.fetch_available_pipelines(None).await.unwrap();
plugin.fetch_pipelines().await.unwrap();
plugin.validate_credentials().await.unwrap();
let paths: Vec<_> = stub
.requests()
.iter()
.map(|request| request.url.path().to_string())
.collect();
assert!(!paths.is_empty());
assert!(paths.iter().all(|path| !path.contains("//")), "{paths:?}");
assert!(
paths.iter().all(|path| path.starts_with("/acme/")),
"{paths:?}"
);
}
#[tokio::test]
async fn run_history_is_newest_first_and_follows_continuation_until_the_limit() {
let stub = serve(azure_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
let runs = plugin
.fetch_run_history("azure-devops__1__Alpha/1", 3)
.await
.unwrap();
let summary: Vec<_> = runs
.iter()
.map(|run| (run.run_number, run.status, run.id.as_str()))
.collect();
assert_eq!(
summary,
[
(923, PipelineStatus::Running, "azure-devops-run-923"),
(922, PipelineStatus::Failed, "azure-devops-run-922"),
(921, PipelineStatus::Success, "azure-devops-run-921"),
]
);
assert_eq!(runs[0].pipeline_id, "azure-devops__1__Alpha/1");
assert_eq!(runs[1].branch.as_deref(), Some("main"));
assert_eq!(runs[1].actor.as_deref(), Some("Ada Lovelace"));
assert_eq!(runs[1].duration_seconds, Some(150));
assert_eq!(runs[1].metadata["build_number"], json!("20261028.22"));
let requests = stub.requests();
assert_eq!(requests.len(), 2);
assert_eq!(requests[0].query("definitions").as_deref(), Some("1"));
assert_eq!(
requests[0].query("queryOrder").as_deref(),
Some("queueTimeDescending")
);
assert_eq!(requests[0].query("$top").as_deref(), Some("3"));
assert_eq!(requests[0].query("continuationToken"), None);
assert_eq!(requests[1].query("$top").as_deref(), Some("1"));
assert_eq!(
requests[1].query("continuationToken").as_deref(),
Some("builds-2")
);
}
#[tokio::test]
async fn run_history_stops_at_the_limit_without_following_the_continuation() {
let stub = serve(azure_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
let runs = plugin
.fetch_run_history("azure-devops__1__Alpha/1", 1)
.await
.unwrap();
assert_eq!(runs.len(), 1);
assert_eq!(runs[0].run_number, 923);
assert_eq!(stub.requests().len(), 1);
assert!(plugin
.fetch_run_history("azure-devops__1__Alpha/1", 0)
.await
.unwrap()
.is_empty());
assert_eq!(stub.requests().len(), 1);
}
#[tokio::test]
async fn run_details_are_fetched_by_build_id_and_checked_against_the_pipeline() {
let stub = serve(azure_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
let run = plugin
.fetch_run_details("azure-devops__1__Alpha/1", 922)
.await
.unwrap();
assert_eq!(run.run_number, 922);
assert_eq!(run.status, PipelineStatus::Failed);
assert_eq!(run.commit_sha.as_deref(), Some("sha922"));
assert_eq!(
run.logs_url,
"https://dev.azure.com/acme/Alpha/_build/results?buildId=922"
);
assert_eq!(run.inputs, Some(json!({"source-branch": "main"})));
let foreign = plugin
.fetch_run_details("azure-devops__1__Alpha/1", 777)
.await
.unwrap_err();
assert!(
matches!(foreign, PluginError::PipelineNotFound(_)),
"{foreign}"
);
let missing = plugin
.fetch_run_details("azure-devops__1__Alpha/1", 404)
.await
.unwrap_err();
assert!(
matches!(missing, PluginError::PipelineNotFound(_)),
"{missing}"
);
let malformed = plugin
.fetch_run_details("github__1__Alpha/1", 922)
.await
.unwrap_err();
assert!(
matches!(malformed, PluginError::InvalidConfig(_)),
"{malformed}"
);
}
#[tokio::test]
async fn workflow_parameters_offer_the_default_branch_and_overridable_non_secret_variables() {
let stub = serve(azure_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
let parameters = plugin
.fetch_workflow_parameters("azure-devops__1__Alpha/1")
.await
.unwrap();
let summary: Vec<_> = parameters
.iter()
.map(|parameter| {
let WorkflowParameterType::String { default } = ¶meter.param_type else {
panic!("expected a string parameter: {}", parameter.name);
};
(
parameter.name.as_str(),
default.as_deref(),
parameter.required,
)
})
.collect();
assert_eq!(
summary,
[
("source-branch", Some("main"), false),
("branch", Some("legacy"), false),
("deployTarget", Some("dev"), false),
]
);
}
#[tokio::test]
async fn trigger_queues_a_build_on_the_branch_with_the_overridden_variable() {
let stub = serve(azure_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
let result = plugin
.trigger_pipeline(TriggerParams {
workflow_id: "azure-devops__1__Alpha/1".to_string(),
inputs: Some(json!({"source-branch": "feature/x", "deployTarget": "staging"})),
})
.await
.unwrap();
let result: Value = serde_json::from_str(&result).unwrap();
assert_eq!(
result,
json!({
"message": "Queued build 930 on feature/x",
"run_number": 930,
"logs_url": "https://dev.azure.com/acme/Alpha/_build/results?buildId=930"
})
);
let requests = stub.requests();
assert_eq!(requests.len(), 2);
assert_eq!(requests[0].method, "GET");
assert_eq!(
requests[0].url.path(),
"/acme/Alpha/_apis/build/definitions/1"
);
let queued = &requests[1];
assert_eq!(queued.method, "POST");
assert_eq!(queued.url.path(), "/acme/Alpha/_apis/build/builds");
assert_eq!(queued.query("api-version").as_deref(), Some("7.0"));
assert_eq!(queued.authorization.as_deref(), Some(BASIC_AUTH));
let body: Value = serde_json::from_str(&queued.body).unwrap();
assert_eq!(
body,
json!({
"definition": {"id": 1},
"sourceBranch": "refs/heads/feature/x",
"parameters": r#"{"deployTarget":"staging"}"#
})
);
}
#[tokio::test]
async fn a_variable_named_branch_is_sent_as_a_variable_not_as_the_source_branch() {
let stub = serve(azure_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
plugin
.trigger_pipeline(TriggerParams {
workflow_id: "azure-devops__1__Alpha/1".to_string(),
inputs: Some(json!({"source-branch": "main", "branch": "legacy"})),
})
.await
.unwrap();
let requests = stub.requests();
assert_eq!(
requests[0].url.path(),
"/acme/Alpha/_apis/build/definitions/1"
);
let body: Value = serde_json::from_str(&requests[1].body).unwrap();
assert_eq!(
body,
json!({
"definition": {"id": 1},
"sourceBranch": "refs/heads/main",
"parameters": r#"{"branch":"legacy"}"#
})
);
}
async fn queued_body(
plugin: &AzureDevOpsPlugin, stub: &Stub, pipeline: &str, inputs: Value,
) -> Value {
plugin
.trigger_pipeline(TriggerParams {
workflow_id: pipeline.to_string(),
inputs: Some(inputs),
})
.await
.unwrap();
let requests = stub.requests();
let queued = requests
.iter()
.find(|request| request.method == "POST")
.unwrap();
serde_json::from_str(&queued.body).unwrap()
}
#[tokio::test]
async fn an_overridable_ref_variable_is_sent_even_when_it_equals_the_branch() {
let stub = serve(azure_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
let body = queued_body(
&plugin,
&stub,
"azure-devops__1__Alpha/2",
json!({"source-branch": "main", "ref": "main"}),
)
.await;
assert_eq!(
body,
json!({
"definition": {"id": 2},
"sourceBranch": "refs/heads/main",
"parameters": r#"{"ref":"main"}"#
})
);
}
#[tokio::test]
async fn the_rerun_ref_echo_is_dropped_when_the_definition_has_no_overridable_ref() {
let stub = serve(azure_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
let body = queued_body(
&plugin,
&stub,
"azure-devops__1__Alpha/1",
json!({"source-branch": "main", "ref": "main"}),
)
.await;
assert_eq!(
body,
json!({"definition": {"id": 1}, "sourceBranch": "refs/heads/main"})
);
}
#[tokio::test]
async fn inputs_that_are_not_overridable_variables_are_not_sent() {
let stub = serve(azure_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
let body = queued_body(
&plugin,
&stub,
"azure-devops__1__Alpha/1",
json!({"DEPLOYtarget": "prod", "fixed": "x", "apiKey": "s3cret", "unknown": "y"}),
)
.await;
assert_eq!(
body,
json!({"definition": {"id": 1}, "parameters": r#"{"DEPLOYtarget":"prod"}"#})
);
}
#[tokio::test]
async fn cancel_patches_the_build_status_to_cancelling() {
let stub = serve(azure_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
plugin
.cancel_run("azure-devops__1__Alpha/1", 923)
.await
.unwrap();
let requests = stub.requests();
assert_eq!(requests.len(), 2);
assert_eq!(requests[0].method, "GET");
assert_eq!(requests[0].url.path(), "/acme/Alpha/_apis/build/builds/923");
assert_eq!(requests[1].method, "PATCH");
assert_eq!(requests[1].url.path(), "/acme/Alpha/_apis/build/builds/923");
assert_eq!(requests[1].query("api-version").as_deref(), Some("7.0"));
let body: Value = serde_json::from_str(&requests[1].body).unwrap();
assert_eq!(body, json!({"status": "cancelling"}));
assert!(plugin.metadata().capabilities.cancel);
}
#[tokio::test]
async fn cancel_refuses_a_build_of_another_definition_without_patching() {
let stub = serve(azure_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
let error = plugin
.cancel_run("azure-devops__1__Alpha/1", 777)
.await
.unwrap_err();
assert!(matches!(error, PluginError::PipelineNotFound(_)), "{error}");
assert!(error.to_string().contains("does not belong"), "{error}");
assert!(stub
.requests()
.iter()
.all(|request| request.method == "GET"));
}
#[tokio::test]
async fn ids_with_ci_keeps_projects_that_have_definitions() {
let stub = serve(azure_reply).await;
let plugin = plugin_for(&stub, "", "");
let kept = plugin
.ids_with_ci(vec!["Alpha".to_string(), "Empty".to_string()])
.await
.unwrap();
assert_eq!(kept, ["Alpha"]);
}
#[tokio::test]
async fn validate_accepts_a_working_token_with_a_single_cheap_call() {
let stub = serve(|_| Reply::json(200, json!({"count": 0, "value": []}))).await;
assert!(plugin_for(&stub, "", "")
.validate_credentials()
.await
.unwrap());
let requests = stub.requests();
assert_eq!(requests.len(), 1);
assert_eq!(requests[0].url.path(), "/acme/_apis/projects");
assert_eq!(requests[0].query("$top").as_deref(), Some("1"));
}
#[tokio::test]
async fn validate_maps_401_to_authentication_failed() {
let stub =
serve(|_| Reply::json(401, json!({"message": "TF400813: not authorized"}))).await;
let error = plugin_for(&stub, "", "")
.validate_credentials()
.await
.unwrap_err();
assert!(
matches!(error, PluginError::AuthenticationFailed(_)),
"{error}"
);
assert_eq!(stub.requests().len(), 1);
}
#[tokio::test]
async fn validate_maps_a_203_sign_in_page_to_authentication_failed() {
let stub = serve(|_| Reply::html(203, "<html><title>Sign In</title></html>")).await;
let error = plugin_for(&stub, "", "")
.validate_credentials()
.await
.unwrap_err();
assert!(
matches!(error, PluginError::AuthenticationFailed(_)),
"{error}"
);
assert!(error.to_string().contains("sign-in page"), "{error}");
assert_eq!(stub.requests().len(), 1);
}
#[tokio::test]
async fn validate_maps_a_200_html_sign_in_page_to_authentication_failed() {
let stub = serve(|_| Reply::html(200, "<html><title>Sign In</title></html>")).await;
let error = plugin_for(&stub, "", "")
.validate_credentials()
.await
.unwrap_err();
assert!(
matches!(error, PluginError::AuthenticationFailed(_)),
"{error}"
);
}
#[tokio::test]
async fn validate_maps_403_to_permission_denied() {
let stub = serve(|_| Reply::json(403, json!({"message": "TF401444: no access"}))).await;
let error = plugin_for(&stub, "", "")
.validate_credentials()
.await
.unwrap_err();
assert!(matches!(error, PluginError::PermissionDenied(_)), "{error}");
}
#[test]
fn initialize_requires_a_server_url_and_a_token() {
let missing_url = AzureDevOpsPlugin::new()
.initialize(
1,
HashMap::from([("token".to_string(), "t".to_string())]),
None,
)
.unwrap_err();
let missing_token = AzureDevOpsPlugin::new()
.initialize(
1,
HashMap::from([(
"server_url".to_string(),
"https://dev.azure.com/acme".to_string(),
)]),
None,
)
.unwrap_err();
assert!(matches!(missing_url, PluginError::InvalidConfig(_)));
assert!(matches!(missing_token, PluginError::InvalidConfig(_)));
}
#[tokio::test]
async fn a_list_cut_short_by_the_page_cap_is_an_error() {
let stub = serve(|_| {
Reply::json(
200,
json!({"count": 1, "value": [{"id": "p", "name": "Alpha"}]}),
)
.continued("again")
})
.await;
let error = plugin_for(&stub, "", "")
.fetch_available_pipelines(None)
.await
.unwrap_err();
assert!(matches!(error, PluginError::ApiError(_)), "{error}");
assert!(error.to_string().contains("more items"), "{error}");
assert!(stub.requests().len() > 1);
}
const STAGE: &str = "a0000000-0000-4000-8000-000000000001";
const PHASE: &str = "b0000000-0000-4000-8000-000000000001";
const JOB_LINUX: &str = "c0000000-0000-4000-8000-000000000001";
const JOB_DOCS: &str = "c0000000-0000-4000-8000-000000000002";
const JOB_PRUNED: &str = "c0000000-0000-4000-8000-000000000003";
const JOB_EXPIRED: &str = "c0000000-0000-4000-8000-000000000004";
const TASK_INIT: &str = "d0000000-0000-4000-8000-000000000001";
const TASK_TESTS: &str = "d0000000-0000-4000-8000-000000000002";
const TASK_PUBLISH: &str = "d0000000-0000-4000-8000-000000000003";
const TASK_DOCS: &str = "d0000000-0000-4000-8000-000000000004";
const CHECKPOINT: &str = "e0000000-0000-4000-8000-000000000001";
const LINUX_LOG: &str =
"\u{feff}2026-10-28T10:00:00.0000000Z ##[section]Starting: Linux tests\n\
2026-10-28T10:00:01.1000000Z ##[section]Starting: Initialize job\n\
2026-10-28T10:00:02.0000000Z ##[section]Finishing: Initialize job\n\
2026-10-28T10:00:03.0000000Z ##[section]Starting: Run tests\n\
2026-10-28T10:00:03.1000000Z ##[command]make test\n\
2026-10-28T10:00:09.0000000Z ##[error]Bash exited with code '2'.\n\
2026-10-28T10:00:09.5000000Z ##[section]Finishing: Run tests\n";
type Outcome = (&'static str, Option<&'static str>);
const SUCCEEDED: Outcome = ("completed", Some("succeeded"));
const FAILED: Outcome = ("completed", Some("failed"));
const SKIPPED: Outcome = ("completed", Some("skipped"));
fn record(
id: &str, parent: Option<&str>, kind: &str, name: &str, order: i64, outcome: Outcome,
log: Option<i64>,
) -> Value {
let mut record = json!({
"id": id,
"parentId": parent,
"type": kind,
"name": name,
"order": order,
"state": outcome.0,
"result": outcome.1,
"startTime": "2026-10-28T10:00:00.5000000Z",
"finishTime": "2026-10-28T10:00:09.5000000Z"
});
if let Some(log) = log {
record["log"] = json!({"id": log, "type": "Container"});
}
record
}
fn timeline_922() -> Value {
let mut publish = record(
TASK_PUBLISH,
Some(JOB_LINUX),
"Task",
"Publish results",
3,
SKIPPED,
None,
);
publish["startTime"] = Value::Null;
publish["finishTime"] = Value::Null;
json!({"id": "f0000000-0000-4000-8000-000000000001", "records": [
record(TASK_TESTS, Some(JOB_LINUX), "Task", "Run tests", 2, FAILED, Some(9)),
record(JOB_DOCS, Some(PHASE), "Job", "Docs", 2, SUCCEEDED, Some(12)),
record(STAGE, None, "Stage", "Build", 1, FAILED, None),
record(PHASE, Some(STAGE), "Phase", "Linux", 1, FAILED, None),
record(JOB_LINUX, Some(PHASE), "Job", "Linux tests", 1, FAILED, Some(7)),
record(CHECKPOINT, Some(STAGE), "Checkpoint", "Checkpoint", 0, SUCCEEDED, None),
record(TASK_INIT, Some(JOB_LINUX), "Task", "Initialize job", 1, SUCCEEDED, Some(8)),
publish,
record(TASK_DOCS, Some(JOB_DOCS), "Task", "Build docs", 1, SUCCEEDED, Some(13))
]})
}
fn timeline_921() -> Value {
json!({"id": "f0000000-0000-4000-8000-000000000002", "records": [
record(PHASE, None, "Phase", "Phase 1", 1, FAILED, None),
record(JOB_PRUNED, Some(PHASE), "Job", "Build", 1, SUCCEEDED, Some(4)),
record(JOB_EXPIRED, Some(PHASE), "Job", "Test", 2, FAILED, Some(5))
]})
}
fn log_reply(request: &Request) -> Reply {
match (request.method.as_str(), request.url.path()) {
("GET", "/acme/Alpha/_apis/build/builds/922/timeline") => {
Reply::json(200, timeline_922())
}
("GET", "/acme/Alpha/_apis/build/builds/922/logs/7") => Reply::text(200, LINUX_LOG),
("GET", "/acme/Alpha/_apis/build/builds/921/timeline") => {
Reply::json(200, timeline_921())
}
("GET", "/acme/Alpha/_apis/build/builds/921/logs/4") => {
Reply::json(404, json!({"message": "TF400898: log not found"}))
}
("GET", "/acme/Alpha/_apis/build/builds/921/logs/5") => {
Reply::json(410, json!({"message": "log expired"}))
}
_ => Reply::not_found(request),
}
}
fn build_of_pipeline(run_number: i64, metadata: &HashMap<String, Value>) -> RunRef<'_> {
RunRef {
pipeline_id: "azure-devops__1__Alpha/1",
run_number,
run_id: "azure-devops-run",
metadata,
}
}
async fn resume_job_log(
plugin: &AzureDevOpsPlugin, run_number: i64, node_id: &str, from: Option<&LogCursor>,
) -> PluginResult<(Vec<u8>, LogRead, bool)> {
let metadata = HashMap::new();
let (tx, mut rx) = tokio::sync::mpsc::channel(16);
let mut sink = LogSink::new(tx);
let result = plugin
.read_log(
build_of_pipeline(run_number, &metadata),
node_id,
from,
&mut sink,
)
.await;
let restarted = sink.restarted();
drop(sink);
let mut bytes = Vec::new();
while let Some(chunk) = rx.recv().await {
bytes.extend_from_slice(&chunk);
}
result.map(|read| (bytes, read, restarted))
}
async fn read_job_log(
plugin: &AzureDevOpsPlugin, run_number: i64, node_id: &str,
) -> PluginResult<Vec<u8>> {
resume_job_log(plugin, run_number, node_id, None)
.await
.map(|(bytes, _, _)| bytes)
}
#[tokio::test]
async fn timeline_becomes_stage_job_and_step_nodes_in_order() {
let stub = serve(log_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
let metadata = HashMap::new();
let nodes = plugin
.fetch_log_nodes(build_of_pipeline(922, &metadata))
.await
.unwrap();
let summary: Vec<_> = nodes
.iter()
.map(|node| {
(
node.id.as_str(),
node.parent_id.as_deref(),
node.kind,
node.name.as_str(),
node.status,
node.log,
)
})
.collect();
assert_eq!(
summary,
[
(
STAGE,
None,
LogNodeKind::Stage,
"Build",
PipelineStatus::Failed,
None
),
(
JOB_LINUX,
Some(STAGE),
LogNodeKind::Job,
"Linux tests",
PipelineStatus::Failed,
Some(LogAvailability::Ready)
),
(
TASK_INIT,
Some(JOB_LINUX),
LogNodeKind::Step,
"Initialize job",
PipelineStatus::Success,
None
),
(
TASK_TESTS,
Some(JOB_LINUX),
LogNodeKind::Step,
"Run tests",
PipelineStatus::Failed,
None
),
(
TASK_PUBLISH,
Some(JOB_LINUX),
LogNodeKind::Step,
"Publish results",
PipelineStatus::Skipped,
None
),
(
JOB_DOCS,
Some(STAGE),
LogNodeKind::Job,
"Docs",
PipelineStatus::Success,
Some(LogAvailability::Ready)
),
(
TASK_DOCS,
Some(JOB_DOCS),
LogNodeKind::Step,
"Build docs",
PipelineStatus::Success,
None
),
]
);
assert_eq!(
nodes[1].started_at.unwrap().to_rfc3339(),
"2026-10-28T10:00:00.500+00:00"
);
assert_eq!(nodes[4].started_at, None);
assert_eq!(nodes[4].finished_at, None);
let requests = stub.requests();
assert_eq!(requests.len(), 1);
assert_eq!(
requests[0].url.path(),
"/acme/Alpha/_apis/build/builds/922/timeline"
);
assert_eq!(requests[0].query("api-version").as_deref(), Some("7.0"));
}
#[tokio::test]
async fn build_without_stages_lists_its_jobs_at_the_top() {
let stub = serve(log_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
let metadata = HashMap::new();
let nodes = plugin
.fetch_log_nodes(build_of_pipeline(921, &metadata))
.await
.unwrap();
let summary: Vec<_> = nodes
.iter()
.map(|node| (node.id.as_str(), node.parent_id.as_deref(), node.kind))
.collect();
assert_eq!(
summary,
[
(JOB_PRUNED, None, LogNodeKind::Job),
(JOB_EXPIRED, None, LogNodeKind::Job)
]
);
}
#[tokio::test]
async fn build_without_a_timeline_has_no_log_nodes() {
let stub = serve(log_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
let metadata = HashMap::new();
let error = plugin
.fetch_log_nodes(build_of_pipeline(999, &metadata))
.await
.unwrap_err();
assert!(matches!(error, PluginError::PipelineNotFound(_)), "{error}");
}
#[tokio::test]
async fn job_log_streams_as_text_through_the_id_the_timeline_lists() {
let stub = serve(log_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
let log = read_job_log(&plugin, 922, JOB_LINUX).await.unwrap();
assert_eq!(log, LINUX_LOG.as_bytes());
let requests = stub.requests();
let paths: Vec<_> = requests
.iter()
.map(|request| request.url.path().to_string())
.collect();
assert_eq!(
paths,
[
"/acme/Alpha/_apis/build/builds/922/timeline",
"/acme/Alpha/_apis/build/builds/922/logs/7"
]
);
assert_eq!(requests[1].accept.as_deref(), Some("text/plain"));
assert_eq!(requests[1].query("api-version").as_deref(), Some("7.0"));
assert!(requests
.iter()
.all(|request| request.authorization.as_deref() == Some(BASIC_AUTH)));
}
#[tokio::test]
async fn log_the_server_no_longer_has_is_not_found() {
let stub = serve(log_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
for job in [JOB_PRUNED, JOB_EXPIRED] {
let error = read_job_log(&plugin, 921, job).await.unwrap_err();
assert!(
matches!(&error, PluginError::PipelineNotFound(message) if message.contains("no longer available")),
"{job}: {error:?}"
);
}
}
#[tokio::test]
async fn node_ids_outside_the_guid_alphabet_are_rejected_without_a_request() {
let stub = serve(log_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
let rejected = [
"",
"../../logs/7",
"c0000000-0000-4000-8000-000000000001/../x",
"x?y=1",
"c0000000 0000",
"c0000000-0000-4000-8000-00000000000g",
"7/../../builds/1",
];
for node_id in rejected {
let error = read_job_log(&plugin, 922, node_id).await.unwrap_err();
assert!(
matches!(error, PluginError::InvalidConfig(_)),
"{node_id}: {error:?}"
);
}
assert!(stub.requests().is_empty());
}
#[tokio::test]
async fn node_the_timeline_does_not_list_as_a_job_log_never_reaches_the_log_endpoint() {
let stub = serve(log_reply).await;
let plugin = plugin_for(&stub, "", "Alpha");
let unlisted = [
TASK_TESTS,
STAGE,
PHASE,
CHECKPOINT,
"ffffffff-0000-4000-8000-000000000000",
"7",
];
for node_id in unlisted {
let error = read_job_log(&plugin, 922, node_id).await.unwrap_err();
assert!(
matches!(error, PluginError::PipelineNotFound(_)),
"{node_id}: {error:?}"
);
}
let requests = stub.requests();
assert_eq!(requests.len(), unlisted.len());
assert!(requests
.iter()
.all(|request| request.url.path().ends_with("/timeline")));
}
const LIVE_LINES: [&str; 5] = [
"2026-10-28T10:00:00.0000000Z ##[section]Starting: Linux tests",
"2026-10-28T10:00:01.0000000Z ##[command]make build",
"2026-10-28T10:00:02.0000000Z compiling",
"2026-10-28T10:00:03.0000000Z compiling",
"2026-10-28T10:00:04.0000000Z ##[section]Finishing: Linux tests",
];
const REPEATED_LINES: [&str; 4] = ["q", "z", "z", "z"];
const LIVE_BUILDS: [(&str, i64, i64); 4] =
[("924", 7, 1), ("925", 8, 2), ("926", 9, 1), ("927", 10, 1)];
fn joined(lines: &[impl AsRef<str>]) -> String {
lines
.iter()
.map(|line| format!("{}\n", line.as_ref()))
.collect()
}
fn live_text(lines: std::ops::Range<usize>) -> String {
joined(&LIVE_LINES[lines])
}
fn oversized_lines() -> Vec<String> {
vec![
"a".to_string(),
"b".to_string(),
"x".repeat(BOUNDARY_LINE_LIMIT + 10),
"new".to_string(),
]
}
fn server_lines(build: &str) -> Vec<String> {
match build {
"926" => REPEATED_LINES.iter().map(|line| line.to_string()).collect(),
"927" => oversized_lines(),
_ => LIVE_LINES.iter().map(|line| line.to_string()).collect(),
}
}
fn job_timeline(log: i64, attempt: i64) -> Value {
let mut job = record(
JOB_LINUX,
None,
"Job",
"Linux tests",
1,
SUCCEEDED,
Some(log),
);
job["attempt"] = json!(attempt);
json!({"records": [job]})
}
fn live_log(request: &Request, first_line: u64) -> Reply {
let path = request.url.path();
let Some((build, log, attempt)) = LIVE_BUILDS.iter().find(|(build, ..)| {
path.starts_with(&format!("/acme/Alpha/_apis/build/builds/{build}/"))
}) else {
return Reply::not_found(request);
};
if path.ends_with("/timeline") {
return Reply::json(200, job_timeline(*log, *attempt));
}
if !path.ends_with(&format!("/logs/{log}")) {
return Reply::not_found(request);
}
let skip = request.query("startLine").map_or(0, |line| {
line.parse::<u64>().unwrap().saturating_sub(first_line) as usize
});
let lines = server_lines(build);
Reply::text(200, &joined(&lines[skip.min(lines.len())..]))
}
fn zero_based_log(request: &Request) -> Reply {
live_log(request, 0)
}
fn one_based_log(request: &Request) -> Reply {
live_log(request, 1)
}
type LiveServer = (fn(&Request) -> Reply, u64);
const LIVE_SERVERS: [LiveServer; 2] = [(zero_based_log, 0), (one_based_log, 1)];
fn position_over(
build: i64, log: i64, attempt: i64, text: &str, first_line: Option<u64>,
) -> LogPosition {
let mut tally = Tally::default();
tally.feed(text.as_bytes());
let mut position = tally.position(build, log, attempt);
if first_line.is_some() {
position.first_line = first_line;
position.anchor = None;
}
position
}
fn cursor_over(
build: i64, log: i64, attempt: i64, text: &str, first_line: Option<u64>,
) -> LogCursor {
LogCursor::new(&position_over(build, log, attempt, text, first_line)).unwrap()
}
fn log_requests(stub: &Stub) -> Vec<Option<String>> {
stub.requests()
.iter()
.filter(|request| request.url.path().contains("/logs/"))
.map(|request| request.query("startLine"))
.collect()
}
fn lines_asked(lines: &[Option<u64>]) -> Vec<Option<String>> {
lines
.iter()
.map(|line| line.map(|line| line.to_string()))
.collect()
}
#[tokio::test]
async fn first_read_is_the_whole_log_with_a_cursor_at_its_end() {
let stub = serve(zero_based_log).await;
let plugin = plugin_for(&stub, "", "Alpha");
let (log, read, restarted) = resume_job_log(&plugin, 924, JOB_LINUX, None).await.unwrap();
assert_eq!(log, live_text(0..5).as_bytes());
assert_eq!(
read,
LogRead::replace(Some(cursor_over(924, 7, 1, &live_text(0..5), None)))
);
assert!(!restarted);
assert_eq!(log_requests(&stub), [None]);
}
#[tokio::test]
async fn resumed_read_appends_only_the_new_lines_whichever_line_the_server_counts_from() {
for (serve_log, first) in LIVE_SERVERS {
let stub = serve(serve_log).await;
let plugin = plugin_for(&stub, "", "Alpha");
let cursor = cursor_over(924, 7, 1, &live_text(0..3), None);
let (log, read, restarted) = resume_job_log(&plugin, 924, JOB_LINUX, Some(&cursor))
.await
.unwrap();
assert_eq!(log, live_text(3..5).as_bytes());
assert_eq!(
read,
LogRead::append(cursor_over(924, 7, 1, &live_text(0..5), Some(first)))
);
assert!(!restarted);
assert_eq!(
log_requests(&stub),
lines_asked(&[Some(2), Some(2 + first)])
);
}
}
#[tokio::test]
async fn once_the_numbering_is_known_a_resume_asks_for_the_boundary_line_only() {
for (serve_log, first) in LIVE_SERVERS {
let stub = serve(serve_log).await;
let plugin = plugin_for(&stub, "", "Alpha");
let cursor = cursor_over(924, 7, 1, &live_text(0..3), Some(first));
let (log, read, restarted) = resume_job_log(&plugin, 924, JOB_LINUX, Some(&cursor))
.await
.unwrap();
assert_eq!(log, live_text(3..5).as_bytes());
assert_eq!(
read,
LogRead::append(cursor_over(924, 7, 1, &live_text(0..5), Some(first)))
);
assert!(!restarted);
assert_eq!(log_requests(&stub), lines_asked(&[Some(2 + first)]));
}
}
#[tokio::test]
async fn repeated_lines_at_the_boundary_do_not_swallow_the_new_line() {
let before = joined(&REPEATED_LINES[..3]);
let after = joined(&REPEATED_LINES);
for (serve_log, first) in LIVE_SERVERS {
for known in [None, Some(first)] {
let stub = serve(serve_log).await;
let plugin = plugin_for(&stub, "", "Alpha");
let cursor = cursor_over(926, 9, 1, &before, known);
let (log, read, restarted) = resume_job_log(&plugin, 926, JOB_LINUX, Some(&cursor))
.await
.unwrap();
assert_eq!(log, b"z\n", "{known:?}");
assert_eq!(
read,
LogRead::append(cursor_over(926, 9, 1, &after, Some(first))),
"{known:?}"
);
assert!(!restarted, "{known:?}");
}
}
}
#[tokio::test]
async fn resumed_read_with_nothing_new_writes_nothing_and_keeps_the_cursor() {
for (serve_log, first) in LIVE_SERVERS {
let stub = serve(serve_log).await;
let plugin = plugin_for(&stub, "", "Alpha");
let cursor = cursor_over(924, 7, 1, &live_text(0..5), None);
let (log, read, restarted) = resume_job_log(&plugin, 924, JOB_LINUX, Some(&cursor))
.await
.unwrap();
assert!(log.is_empty());
assert_eq!(
read,
LogRead::append(cursor_over(924, 7, 1, &live_text(0..5), Some(first)))
);
assert!(!restarted);
}
}
#[tokio::test]
async fn resumed_read_continues_a_last_line_that_had_no_newline_on_a_fresh_line() {
for (serve_log, first) in LIVE_SERVERS {
let stub = serve(serve_log).await;
let plugin = plugin_for(&stub, "", "Alpha");
let unterminated = live_text(0..3);
let unterminated = unterminated.trim_end_matches('\n');
let cursor = cursor_over(924, 7, 1, unterminated, None);
let (log, read, restarted) = resume_job_log(&plugin, 924, JOB_LINUX, Some(&cursor))
.await
.unwrap();
assert_eq!(log, format!("\n{}", live_text(3..5)).as_bytes());
assert_eq!(
read,
LogRead::append(cursor_over(924, 7, 1, &live_text(0..5), Some(first)))
);
assert!(!restarted);
}
}
#[tokio::test]
async fn boundary_line_longer_than_the_limit_restarts_with_a_full_streamed_read() {
let full = joined(&oversized_lines());
let seen = joined(&oversized_lines()[..3]);
for (serve_log, first) in LIVE_SERVERS {
let stub = serve(serve_log).await;
let plugin = plugin_for(&stub, "", "Alpha");
let cursor = cursor_over(927, 10, 1, &seen, Some(first));
let (bytes, read, restarted) = resume_job_log(&plugin, 927, JOB_LINUX, Some(&cursor))
.await
.unwrap();
assert_eq!(bytes, full.as_bytes());
assert_eq!(
read,
LogRead::replace(Some(cursor_over(927, 10, 1, &full, None)))
);
assert!(restarted);
assert_eq!(log_requests(&stub), lines_asked(&[Some(2 + first), None]));
}
}
#[tokio::test]
async fn cursor_that_cannot_continue_the_log_restarts_with_the_whole_log() {
let malformed = LogCursor::new(&"junk").unwrap();
for (serve_log, first) in LIVE_SERVERS {
let cases = [
(
924,
7,
cursor_over(924, 7, 1, &"longer\n".repeat(9), Some(first)),
3,
),
(
924,
7,
cursor_over(924, 7, 1, &"rewritten\n".repeat(5), Some(first)),
3,
),
(924, 7, cursor_over(924, 7, 1, "a\nb\nc\n", None), 3),
(924, 7, cursor_over(924, 7, 1, "z\nz\nz\n", None), 2),
(
924,
7,
cursor_over(924, 6, 1, &live_text(0..3), Some(first)),
2,
),
(
925,
8,
cursor_over(925, 8, 1, &live_text(0..3), Some(first)),
2,
),
(924, 7, malformed.clone(), 2),
];
for (build, log, cursor, requests) in &cases {
let stub = serve(serve_log).await;
let plugin = plugin_for(&stub, "", "Alpha");
let (bytes, read, restarted) =
resume_job_log(&plugin, *build, JOB_LINUX, Some(cursor))
.await
.unwrap();
assert_eq!(bytes, live_text(0..5).as_bytes(), "{cursor:?}");
let attempt = if *build == 925 { 2 } else { 1 };
assert_eq!(
read,
LogRead::replace(Some(cursor_over(
*build,
*log,
attempt,
&live_text(0..5),
None
))),
"{cursor:?}"
);
assert!(restarted, "{cursor:?}");
assert_eq!(stub.requests().len(), *requests, "{cursor:?}");
assert_eq!(log_requests(&stub).last(), Some(&None), "{cursor:?}");
}
}
}
}