1use std::collections::{
2 HashMap,
3 HashSet,
4};
5use std::sync::Arc;
6use std::time::Duration;
7
8use async_trait::async_trait;
9use futures::future::try_join_all;
10use pipedash_plugin_api::*;
11
12use crate::log_dialect::AzureDevOpsLogDialect;
13use crate::{
14 client,
15 config,
16 mapper,
17 metadata,
18 types,
19};
20
21pub struct AzureDevOpsPlugin {
22 metadata: PluginMetadata,
23 client: Option<client::AzureDevOpsClient>,
24 provider_id: Option<i64>,
25 organization: String,
26 config: HashMap<String, String>,
27 status_source: Option<StatusSource>,
28}
29
30impl Default for AzureDevOpsPlugin {
31 fn default() -> Self {
32 Self::new()
33 }
34}
35
36impl AzureDevOpsPlugin {
37 pub fn new() -> Self {
38 Self {
39 metadata: metadata::create_metadata(),
40 client: None,
41 provider_id: None,
42 organization: String::new(),
43 config: HashMap::new(),
44 status_source: None,
45 }
46 }
47
48 fn client(&self) -> PluginResult<&client::AzureDevOpsClient> {
49 self.client
50 .as_ref()
51 .ok_or_else(|| PluginError::Internal("Plugin not initialized".to_string()))
52 }
53
54 fn provider_id(&self) -> PluginResult<i64> {
55 self.provider_id
56 .ok_or_else(|| PluginError::Internal("Provider ID not set".to_string()))
57 }
58
59 async fn fetch_project_pipelines(
60 &self, provider_id: i64, project: &str,
61 ) -> PluginResult<Vec<Pipeline>> {
62 let definitions = self.client()?.list_definitions(project).await?;
63 Ok(definitions
64 .iter()
65 .map(|definition| {
66 mapper::map_pipeline(provider_id, &self.organization, project, definition)
67 })
68 .collect())
69 }
70
71 async fn owned_build(
72 &self, project: &str, definition_id: i64, pipeline_id: &str, build_id: i64,
73 ) -> PluginResult<types::Build> {
74 let build = self.client()?.get_build(project, build_id).await?;
75 let belongs_to_pipeline = build
76 .definition
77 .as_ref()
78 .and_then(|definition| definition.id)
79 .is_none_or(|id| id == definition_id);
80 if !belongs_to_pipeline {
81 return Err(PluginError::PipelineNotFound(format!(
82 "Run #{build_id} does not belong to {pipeline_id}"
83 )));
84 }
85 Ok(build)
86 }
87
88 fn log_node_record_id(node_id: &str) -> PluginResult<&str> {
89 Some(node_id)
90 .filter(|id| !id.is_empty() && id.bytes().all(|b| b.is_ascii_hexdigit() || b == b'-'))
91 .ok_or_else(|| {
92 PluginError::InvalidConfig(format!("Invalid Azure DevOps log node ID: {node_id}"))
93 })
94 }
95}
96
97#[async_trait]
98impl Plugin for AzureDevOpsPlugin {
99 fn metadata(&self) -> &PluginMetadata {
100 &self.metadata
101 }
102
103 fn status_source(&self) -> Option<&StatusSource> {
104 self.status_source.as_ref()
105 }
106
107 fn initialize(
108 &mut self, provider_id: i64, config: HashMap<String, String>,
109 http_client: Option<Arc<reqwest::Client>>,
110 ) -> PluginResult<()> {
111 let server_url = config::get_server_url(&config)?;
112 let token = config
113 .get("token")
114 .map(|token| token.trim())
115 .filter(|token| !token.is_empty())
116 .ok_or_else(|| {
117 PluginError::InvalidConfig("Missing Azure DevOps personal access token".to_string())
118 })?
119 .to_string();
120 let status_source = custom_status_source(&config)?;
121
122 let http_client = match http_client {
123 Some(http_client) => http_client,
124 None => Arc::new(
125 reqwest::Client::builder()
126 .use_rustls_tls()
127 .https_only(cfg!(target_os = "ios"))
128 .pool_max_idle_per_host(10)
129 .timeout(Duration::from_secs(30))
130 .connect_timeout(Duration::from_secs(10))
131 .tcp_keepalive(Duration::from_secs(60))
132 .build()
133 .map_err(|e| {
134 PluginError::Internal(format!("Failed to build HTTP client: {e}"))
135 })?,
136 ),
137 };
138
139 self.organization = config::organization_of(&server_url);
140 self.status_source = status_source;
141 self.client = Some(client::AzureDevOpsClient::new(
142 http_client,
143 server_url,
144 token,
145 ));
146 self.provider_id = Some(provider_id);
147 self.config = config;
148 Ok(())
149 }
150
151 async fn validate_credentials(&self) -> PluginResult<bool> {
152 self.client()?.ping().await?;
153 Ok(true)
154 }
155
156 async fn fetch_available_pipelines(
157 &self, params: Option<PaginationParams>,
158 ) -> PluginResult<PaginatedAvailablePipelines> {
159 self.fetch_available_pipelines_filtered(None, None, params)
160 .await
161 }
162
163 async fn fetch_organizations(&self) -> PluginResult<Vec<Organization>> {
164 self.client()?;
165 Ok(vec![Organization {
166 id: self.organization.clone(),
167 name: self.organization.clone(),
168 description: None,
169 }])
170 }
171
172 async fn ids_with_ci(&self, ids: Vec<String>) -> PluginResult<Vec<String>> {
173 let client = self.client()?;
174 Ok(retain_with_ci(ids, |project| async move {
175 client.has_definitions(&project).await
176 })
177 .await)
178 }
179
180 async fn fetch_available_pipelines_filtered(
181 &self, org: Option<String>, search: Option<String>, params: Option<PaginationParams>,
182 ) -> PluginResult<PaginatedAvailablePipelines> {
183 let client = self.client()?;
184 let params = params.unwrap_or_default();
185 let start = params
186 .calculate_offset()
187 .map_err(PluginError::InvalidConfig)?;
188
189 let mut projects = if org.is_none_or(|org| org == self.organization) {
190 client.list_projects().await?
191 } else {
192 Vec::new()
193 };
194 if let Some(term) = search.as_deref().map(str::to_lowercase) {
195 projects.retain(|project| {
196 project.name.to_lowercase().contains(&term)
197 || project
198 .description
199 .as_deref()
200 .is_some_and(|description| description.to_lowercase().contains(&term))
201 });
202 }
203 projects.sort_by_cached_key(|project| project.name.to_lowercase());
204
205 let total_count = projects.len();
206 let items = projects
207 .iter()
208 .skip(start)
209 .take(params.page_size)
210 .map(|project| mapper::map_available_pipeline(project, &self.organization))
211 .collect::<Vec<_>>();
212 let mut response =
213 PaginatedResponse::new(items, params.page, params.page_size, total_count);
214 response.has_more = start + response.items.len() < total_count;
215 Ok(response)
216 }
217
218 async fn fetch_pipelines(&self) -> PluginResult<Vec<Pipeline>> {
219 let provider_id = self.provider_id()?;
220 let projects = config::get_projects(&self.config);
221 if projects.is_empty() {
222 return Err(PluginError::InvalidConfig(
223 "No projects configured".to_string(),
224 ));
225 }
226
227 let results = try_join_all(
228 projects
229 .iter()
230 .map(|project| self.fetch_project_pipelines(provider_id, project)),
231 )
232 .await?;
233
234 Ok(results.into_iter().flatten().collect())
235 }
236
237 async fn fetch_run_history(
238 &self, pipeline_id: &str, limit: usize,
239 ) -> PluginResult<Vec<PipelineRun>> {
240 let id = config::parse_pipeline_id(pipeline_id)?;
241 let client = self.client()?;
242 let builds = client
243 .list_builds(id.project, id.definition_id, limit)
244 .await?;
245 Ok(builds
246 .iter()
247 .map(|build| mapper::map_run(build, pipeline_id, id.project, client.server_url()))
248 .collect())
249 }
250
251 async fn fetch_run_details(
252 &self, pipeline_id: &str, run_number: i64,
253 ) -> PluginResult<PipelineRun> {
254 let id = config::parse_pipeline_id(pipeline_id)?;
255 let build = self
256 .owned_build(id.project, id.definition_id, pipeline_id, run_number)
257 .await?;
258 Ok(mapper::map_run(
259 &build,
260 pipeline_id,
261 id.project,
262 self.client()?.server_url(),
263 ))
264 }
265
266 async fn fetch_log_nodes(&self, run: RunRef<'_>) -> PluginResult<Vec<LogNode>> {
267 let id = config::parse_pipeline_id(run.pipeline_id)?;
268 let timeline = self
269 .client()?
270 .get_timeline(id.project, run.run_number)
271 .await?;
272 Ok(mapper::timeline_to_log_nodes(&timeline))
273 }
274
275 async fn read_log(
276 &self, run: RunRef<'_>, node_id: &str, sink: &mut LogSink,
277 ) -> PluginResult<()> {
278 let id = config::parse_pipeline_id(run.pipeline_id)?;
279 let record_id = Self::log_node_record_id(node_id)?;
280 let client = self.client()?;
281 let timeline = client.get_timeline(id.project, run.run_number).await?;
282 let log_id = mapper::job_log_id(&timeline, record_id).ok_or_else(|| {
283 PluginError::PipelineNotFound(format!(
284 "Run #{} has no job log {node_id}",
285 run.run_number
286 ))
287 })?;
288 client
289 .stream_build_log(id.project, run.run_number, log_id, sink)
290 .await
291 }
292
293 fn log_dialect(&self) -> &'static dyn LogDialect {
294 &AzureDevOpsLogDialect
295 }
296
297 async fn fetch_workflow_parameters(
298 &self, workflow_id: &str,
299 ) -> PluginResult<Vec<WorkflowParameter>> {
300 let id = config::parse_pipeline_id(workflow_id)?;
301 let definition = self
302 .client()?
303 .get_definition(id.project, id.definition_id)
304 .await?;
305
306 let default_branch = definition
307 .repository
308 .and_then(|repository| repository.default_branch)
309 .filter(|branch| !branch.is_empty())
310 .map(|branch| mapper::short_branch(&branch).to_string());
311 let mut parameters = vec![WorkflowParameter {
312 name: mapper::BRANCH_INPUT.to_string(),
313 label: Some("Branch".to_string()),
314 description: Some(
315 "Branch to build. The definition's default branch is used when empty".to_string(),
316 ),
317 param_type: WorkflowParameterType::String {
318 default: default_branch,
319 },
320 required: false,
321 }];
322 parameters.extend(
323 definition
324 .variables
325 .unwrap_or_default()
326 .into_iter()
327 .filter(|(_, variable)| variable.is_overridable())
328 .map(|(name, variable)| WorkflowParameter {
329 label: Some(name.clone()),
330 description: Some("Pipeline variable".to_string()),
331 param_type: WorkflowParameterType::String {
332 default: variable.value,
333 },
334 required: false,
335 name,
336 }),
337 );
338 Ok(parameters)
339 }
340
341 async fn trigger_pipeline(&self, params: TriggerParams) -> PluginResult<String> {
342 let id = config::parse_pipeline_id(¶ms.workflow_id)?;
343 let client = self.client()?;
344 let overridable: HashSet<String> = client
345 .get_definition(id.project, id.definition_id)
346 .await?
347 .variables
348 .unwrap_or_default()
349 .into_iter()
350 .filter(|(_, variable)| variable.is_overridable())
351 .map(|(name, _)| name.to_lowercase())
352 .collect();
353 let body = mapper::trigger_body(id.definition_id, params.inputs, &overridable);
354 let build = client.queue_build(id.project, &body).await?;
355 let logs_url = build
356 .web_link()
357 .map(str::to_string)
358 .unwrap_or_else(|| config::build_web_url(client.server_url(), id.project, build.id));
359 let branch = body["sourceBranch"]
360 .as_str()
361 .map(mapper::short_branch)
362 .unwrap_or("the default branch");
363 Ok(serde_json::json!({
364 "message": format!("Queued build {} on {branch}", build.id),
365 "run_number": build.id,
366 "logs_url": logs_url
367 })
368 .to_string())
369 }
370
371 async fn cancel_run(&self, pipeline_id: &str, run_number: i64) -> PluginResult<()> {
372 let id = config::parse_pipeline_id(pipeline_id)?;
373 let client = self.client()?;
374 self.owned_build(id.project, id.definition_id, pipeline_id, run_number)
375 .await?;
376 client.cancel_build(id.project, run_number).await
377 }
378}
379
380#[cfg(all(test, not(target_os = "ios")))]
381mod tests {
382 use std::net::SocketAddr;
383 use std::sync::{
384 Mutex,
385 MutexGuard,
386 };
387
388 use serde_json::{
389 json,
390 Value,
391 };
392 use tokio::io::{
393 AsyncReadExt,
394 AsyncWriteExt,
395 };
396 use tokio::net::{
397 TcpListener,
398 TcpStream,
399 };
400
401 use super::*;
402
403 const BASIC_AUTH: &str = "Basic OnN0dWItdG9rZW4=";
404
405 struct Request {
406 method: String,
407 url: reqwest::Url,
408 authorization: Option<String>,
409 accept: Option<String>,
410 body: String,
411 }
412
413 impl Request {
414 fn query(&self, key: &str) -> Option<String> {
415 self.url
416 .query_pairs()
417 .find(|(name, _)| name == key)
418 .map(|(_, value)| value.into_owned())
419 }
420 }
421
422 struct Reply {
423 status: u16,
424 content_type: &'static str,
425 continuation: Option<String>,
426 body: String,
427 }
428
429 impl Reply {
430 fn json(status: u16, body: Value) -> Self {
431 Self {
432 status,
433 content_type: "application/json; charset=utf-8",
434 continuation: None,
435 body: body.to_string(),
436 }
437 }
438
439 fn html(status: u16, body: &str) -> Self {
440 Self {
441 status,
442 content_type: "text/html; charset=utf-8",
443 continuation: None,
444 body: body.to_string(),
445 }
446 }
447
448 fn text(status: u16, body: &str) -> Self {
449 Self {
450 status,
451 content_type: "text/plain; charset=utf-8",
452 continuation: None,
453 body: body.to_string(),
454 }
455 }
456
457 fn not_found(request: &Request) -> Self {
458 Self::json(
459 404,
460 json!({"message": format!("unexpected {} {}", request.method, request.url)}),
461 )
462 }
463
464 fn continued(mut self, token: &str) -> Self {
465 self.continuation = Some(token.to_string());
466 self
467 }
468
469 fn to_http(&self) -> String {
470 let continuation = self
471 .continuation
472 .as_ref()
473 .map(|token| format!("x-ms-continuationtoken: {token}\r\n"))
474 .unwrap_or_default();
475 format!(
476 "HTTP/1.1 {} X\r\nContent-Type: {}\r\n{continuation}Content-Length: {}\r\nConnection: close\r\n\r\n{}",
477 self.status,
478 self.content_type,
479 self.body.len(),
480 self.body
481 )
482 }
483 }
484
485 struct Stub {
486 base_url: String,
487 requests: Arc<Mutex<Vec<Request>>>,
488 server: tokio::task::JoinHandle<()>,
489 }
490
491 impl Stub {
492 fn requests(&self) -> MutexGuard<'_, Vec<Request>> {
493 self.requests.lock().unwrap()
494 }
495
496 fn requests_to(&self, method: &str, path: &str) -> Vec<(Option<String>, Option<String>)> {
497 self.requests()
498 .iter()
499 .filter(|request| request.method == method && request.url.path() == path)
500 .map(|request| (request.query("$top"), request.query("continuationToken")))
501 .collect()
502 }
503 }
504
505 impl Drop for Stub {
506 fn drop(&mut self) {
507 self.server.abort();
508 }
509 }
510
511 async fn read_request(stream: &mut TcpStream, addr: SocketAddr) -> Request {
512 let mut raw = Vec::new();
513 let mut buffer = [0; 4096];
514 let header_end = loop {
515 if let Some(position) = raw.windows(4).position(|window| window == b"\r\n\r\n") {
516 break position + 4;
517 }
518 let count = stream.read(&mut buffer).await.unwrap();
519 assert!(count > 0, "connection closed before the request head ended");
520 raw.extend_from_slice(&buffer[..count]);
521 };
522 let head = String::from_utf8(raw[..header_end].to_vec()).unwrap();
523 let header = |name: &str| {
524 head.lines()
525 .filter_map(|line| line.split_once(':'))
526 .find(|(key, _)| key.eq_ignore_ascii_case(name))
527 .map(|(_, value)| value.trim().to_string())
528 };
529 let length: usize = header("content-length")
530 .map(|value| value.parse().unwrap())
531 .unwrap_or(0);
532 while raw.len() < header_end + length {
533 let count = stream.read(&mut buffer).await.unwrap();
534 assert!(count > 0, "connection closed before the request body ended");
535 raw.extend_from_slice(&buffer[..count]);
536 }
537
538 let mut request_line = head.lines().next().unwrap().split_whitespace();
539 let method = request_line.next().unwrap().to_string();
540 let target = request_line.next().unwrap();
541 Request {
542 method,
543 url: reqwest::Url::parse(&format!("http://{addr}{target}")).unwrap(),
544 authorization: header("authorization"),
545 accept: header("accept"),
546 body: String::from_utf8(raw[header_end..header_end + length].to_vec()).unwrap(),
547 }
548 }
549
550 async fn serve(reply: fn(&Request) -> Reply) -> Stub {
551 let _ = rustls::crypto::ring::default_provider().install_default();
552 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
553 let addr = listener.local_addr().unwrap();
554 let requests = Arc::new(Mutex::new(Vec::new()));
555 let recorded = Arc::clone(&requests);
556 let server = tokio::spawn(async move {
557 loop {
558 let (mut stream, _) = listener.accept().await.unwrap();
559 let request = read_request(&mut stream, addr).await;
560 let answer = reply(&request);
561 recorded.lock().unwrap().push(request);
562 stream.write_all(answer.to_http().as_bytes()).await.unwrap();
563 }
564 });
565 Stub {
566 base_url: format!("http://{addr}"),
567 requests,
568 server,
569 }
570 }
571
572 fn plugin_for(stub: &Stub, organization_suffix: &str, selected: &str) -> AzureDevOpsPlugin {
573 let mut plugin = AzureDevOpsPlugin::new();
574 plugin
575 .initialize(
576 1,
577 HashMap::from([
578 (
579 "server_url".to_string(),
580 format!("{}/acme{organization_suffix}", stub.base_url),
581 ),
582 ("token".to_string(), "stub-token".to_string()),
583 ("selected_items".to_string(), selected.to_string()),
584 ]),
585 None,
586 )
587 .unwrap();
588 plugin
589 }
590
591 fn build_json(id: i64, status: &str, result: Option<&str>, definition_id: i64) -> Value {
592 let mut build = json!({
593 "id": id,
594 "buildNumber": format!("20261028.{}", id % 100),
595 "status": status,
596 "sourceBranch": "refs/heads/main",
597 "sourceVersion": format!("sha{id}"),
598 "queueTime": "2026-10-28T10:00:00.1234567Z",
599 "startTime": "2026-10-28T10:00:05.5Z",
600 "requestedFor": {"displayName": "Ada Lovelace"},
601 "definition": {"id": definition_id},
602 "_links": {"web": {"href": format!("https://dev.azure.com/acme/Alpha/_build/results?buildId={id}")}}
603 });
604 if let Some(result) = result {
605 build["result"] = json!(result);
606 build["finishTime"] = json!("2026-10-28T10:02:35.5Z");
607 }
608 build
609 }
610
611 fn azure_reply(request: &Request) -> Reply {
612 let token = request.query("continuationToken");
613 match (request.method.as_str(), request.url.path()) {
614 ("GET", "/acme/_apis/projects") => match token.as_deref() {
615 None => Reply::json(
616 200,
617 json!({"count": 2, "value": [
618 {"id": "p1", "name": "Alpha", "description": "Main product"},
619 {"id": "p2", "name": "Beta Team"}
620 ]}),
621 )
622 .continued("projects-2"),
623 Some("projects-2") => {
624 Reply::json(200, json!({"count": 1, "value": [{"id": "p3", "name": "Empty"}]}))
625 }
626 Some(_) => Reply::not_found(request),
627 },
628 ("GET", "/acme/Alpha/_apis/build/definitions") => match token.as_deref() {
629 None => Reply::json(
630 200,
631 json!({"count": 2, "value": [
632 {"id": 1, "name": "CI", "latestBuild": build_json(903, "completed", Some("succeeded"), 1)},
633 {"id": 2, "name": "Docs"}
634 ]}),
635 )
636 .continued("defs-2"),
637 Some("defs-2") => Reply::json(
638 200,
639 json!({"count": 2, "value": [
640 {"id": 3, "name": "Deploy", "latestBuild": build_json(905, "inProgress", None, 3)},
641 {"id": 4, "name": "Nightly", "latestBuild": build_json(906, "completed", Some("partiallySucceeded"), 4)}
642 ]}),
643 ),
644 Some(_) => Reply::not_found(request),
645 },
646 ("GET", "/acme/Beta%20Team/_apis/build/definitions") => Reply::json(
647 200,
648 json!({"count": 1, "value": [
649 {"id": 5, "name": "Mobile", "latestBuild": build_json(907, "completed", Some("canceled"), 5)}
650 ]}),
651 ),
652 ("GET", "/acme/Empty/_apis/build/definitions") => {
653 Reply::json(200, json!({"count": 0, "value": []}))
654 }
655 ("GET", "/acme/Alpha/_apis/build/definitions/2") => Reply::json(
656 200,
657 json!({
658 "id": 2, "name": "Docs",
659 "variables": {"Ref": {"value": "x", "allowOverride": true}}
660 }),
661 ),
662 ("GET", "/acme/Alpha/_apis/build/definitions/1") => Reply::json(
663 200,
664 json!({
665 "id": 1, "name": "CI",
666 "repository": {"defaultBranch": "refs/heads/main"},
667 "variables": {
668 "deployTarget": {"value": "dev", "allowOverride": true},
669 "fixed": {"value": "x"},
670 "apiKey": {"isSecret": true, "allowOverride": true},
671 "branch": {"value": "legacy", "allowOverride": true}
672 }
673 }),
674 ),
675 ("GET", "/acme/Alpha/_apis/build/builds") => {
676 let history = [
677 build_json(923, "inProgress", None, 1),
678 build_json(922, "completed", Some("failed"), 1),
679 build_json(921, "completed", Some("succeeded"), 1),
680 ];
681 let start = match token.as_deref() {
682 None => 0,
683 Some("builds-2") => 2,
684 Some(_) => return Reply::not_found(request),
685 };
686 let size = request
687 .query("$top")
688 .and_then(|top| top.parse().ok())
689 .unwrap_or(history.len())
690 .min(2);
691 let page: Vec<_> = history[start..].iter().take(size).collect();
692 let reply = Reply::json(200, json!({"count": page.len(), "value": page}));
693 if start + size < history.len() {
694 reply.continued("builds-2")
695 } else {
696 reply
697 }
698 }
699 ("GET", "/acme/Alpha/_apis/build/builds/923") => {
700 Reply::json(200, build_json(923, "inProgress", None, 1))
701 }
702 ("GET", "/acme/Alpha/_apis/build/builds/922") => {
703 Reply::json(200, build_json(922, "completed", Some("failed"), 1))
704 }
705 ("GET", "/acme/Alpha/_apis/build/builds/777") => {
706 Reply::json(200, build_json(777, "completed", Some("succeeded"), 9))
707 }
708 ("POST", "/acme/Alpha/_apis/build/builds") => {
709 Reply::json(200, build_json(930, "notStarted", None, 1))
710 }
711 ("PATCH", "/acme/Alpha/_apis/build/builds/923") => {
712 Reply::json(200, build_json(923, "cancelling", None, 1))
713 }
714 _ => Reply::not_found(request),
715 }
716 }
717
718 #[tokio::test]
719 async fn projects_are_listed_across_continuation_pages_and_paged_for_the_picker() {
720 let stub = serve(azure_reply).await;
721 let plugin = plugin_for(&stub, "", "");
722
723 let first = plugin
724 .fetch_available_pipelines(Some(PaginationParams {
725 page: 1,
726 page_size: 2,
727 }))
728 .await
729 .unwrap();
730 let second = plugin
731 .fetch_available_pipelines(Some(PaginationParams {
732 page: 2,
733 page_size: 2,
734 }))
735 .await
736 .unwrap();
737
738 let names = |page: &PaginatedAvailablePipelines| {
739 page.items
740 .iter()
741 .map(|item| item.id.clone())
742 .collect::<Vec<_>>()
743 };
744 assert_eq!(names(&first), ["Alpha", "Beta Team"]);
745 assert!(first.has_more);
746 assert_eq!(first.total_count, 3);
747 assert_eq!(first.items[0].description.as_deref(), Some("Main product"));
748 assert_eq!(first.items[0].organization.as_deref(), Some("acme"));
749 assert_eq!(names(&second), ["Empty"]);
750 assert!(!second.has_more);
751
752 assert_eq!(
753 stub.requests_to("GET", "/acme/_apis/projects"),
754 [
755 (Some("100".to_string()), None),
756 (Some("100".to_string()), Some("projects-2".to_string())),
757 (Some("100".to_string()), None),
758 (Some("100".to_string()), Some("projects-2".to_string())),
759 ]
760 );
761 assert!(stub.requests().iter().all(|request| {
762 request.query("api-version").as_deref() == Some("7.0")
763 && request.authorization.as_deref() == Some(BASIC_AUTH)
764 }));
765 }
766
767 #[tokio::test]
768 async fn picker_filters_by_organization_and_search_term() {
769 let stub = serve(azure_reply).await;
770 let plugin = plugin_for(&stub, "", "");
771
772 let found = plugin
773 .fetch_available_pipelines_filtered(
774 Some("acme".to_string()),
775 Some("TEAM".to_string()),
776 None,
777 )
778 .await
779 .unwrap();
780 assert_eq!(found.items.len(), 1);
781 assert_eq!(found.items[0].name, "Beta Team");
782
783 let requests_before = stub.requests().len();
784 let other = plugin
785 .fetch_available_pipelines_filtered(Some("other".to_string()), None, None)
786 .await
787 .unwrap();
788 assert!(other.items.is_empty());
789 assert_eq!(stub.requests().len(), requests_before);
790
791 let organizations = plugin.fetch_organizations().await.unwrap();
792 assert_eq!(organizations.len(), 1);
793 assert_eq!(organizations[0].id, "acme");
794 assert_eq!(organizations[0].name, "acme");
795 }
796
797 #[tokio::test]
798 async fn definitions_of_selected_projects_become_pipelines_across_continuation_pages() {
799 let stub = serve(azure_reply).await;
800 let plugin = plugin_for(&stub, "", "Alpha, Beta Team");
801
802 let pipelines = plugin.fetch_pipelines().await.unwrap();
803
804 let summary: Vec<_> = pipelines
805 .iter()
806 .map(|pipeline| {
807 (
808 pipeline.id.as_str(),
809 pipeline.name.as_str(),
810 pipeline.status,
811 pipeline.repository.as_str(),
812 pipeline.branch.as_deref(),
813 )
814 })
815 .collect();
816 assert_eq!(
817 summary,
818 [
819 (
820 "azure-devops__1__Alpha/1",
821 "CI",
822 PipelineStatus::Success,
823 "acme/Alpha",
824 Some("main")
825 ),
826 (
827 "azure-devops__1__Alpha/2",
828 "Docs",
829 PipelineStatus::Pending,
830 "acme/Alpha",
831 None
832 ),
833 (
834 "azure-devops__1__Alpha/3",
835 "Deploy",
836 PipelineStatus::Running,
837 "acme/Alpha",
838 Some("main")
839 ),
840 (
841 "azure-devops__1__Alpha/4",
842 "Nightly",
843 PipelineStatus::Failed,
844 "acme/Alpha",
845 Some("main")
846 ),
847 (
848 "azure-devops__1__Beta Team/5",
849 "Mobile",
850 PipelineStatus::Cancelled,
851 "acme/Beta Team",
852 Some("main")
853 ),
854 ]
855 );
856 assert!(pipelines
857 .iter()
858 .all(|pipeline| pipeline.provider_type == "azure-devops" && pipeline.provider_id == 1));
859 assert_eq!(
860 stub.requests_to("GET", "/acme/Alpha/_apis/build/definitions"),
861 [
862 (Some("100".to_string()), None),
863 (Some("100".to_string()), Some("defs-2".to_string())),
864 ]
865 );
866 assert!(stub
867 .requests()
868 .iter()
869 .filter(|request| request.url.path().ends_with("/build/definitions"))
870 .all(|request| request.query("includeLatestBuilds").as_deref() == Some("true")));
871 }
872
873 #[tokio::test]
874 async fn fetching_pipelines_without_selected_projects_is_a_configuration_error() {
875 let stub = serve(azure_reply).await;
876 let error = plugin_for(&stub, "", "")
877 .fetch_pipelines()
878 .await
879 .unwrap_err();
880 assert!(matches!(error, PluginError::InvalidConfig(_)), "{error}");
881 }
882
883 #[tokio::test]
884 async fn a_trailing_slash_on_the_server_url_does_not_double_the_slash() {
885 let stub = serve(azure_reply).await;
886 let plugin = plugin_for(&stub, "/", "Alpha");
887
888 plugin.fetch_available_pipelines(None).await.unwrap();
889 plugin.fetch_pipelines().await.unwrap();
890 plugin.validate_credentials().await.unwrap();
891
892 let paths: Vec<_> = stub
893 .requests()
894 .iter()
895 .map(|request| request.url.path().to_string())
896 .collect();
897 assert!(!paths.is_empty());
898 assert!(paths.iter().all(|path| !path.contains("//")), "{paths:?}");
899 assert!(
900 paths.iter().all(|path| path.starts_with("/acme/")),
901 "{paths:?}"
902 );
903 }
904
905 #[tokio::test]
906 async fn run_history_is_newest_first_and_follows_continuation_until_the_limit() {
907 let stub = serve(azure_reply).await;
908 let plugin = plugin_for(&stub, "", "Alpha");
909
910 let runs = plugin
911 .fetch_run_history("azure-devops__1__Alpha/1", 3)
912 .await
913 .unwrap();
914
915 let summary: Vec<_> = runs
916 .iter()
917 .map(|run| (run.run_number, run.status, run.id.as_str()))
918 .collect();
919 assert_eq!(
920 summary,
921 [
922 (923, PipelineStatus::Running, "azure-devops-run-923"),
923 (922, PipelineStatus::Failed, "azure-devops-run-922"),
924 (921, PipelineStatus::Success, "azure-devops-run-921"),
925 ]
926 );
927 assert_eq!(runs[0].pipeline_id, "azure-devops__1__Alpha/1");
928 assert_eq!(runs[1].branch.as_deref(), Some("main"));
929 assert_eq!(runs[1].actor.as_deref(), Some("Ada Lovelace"));
930 assert_eq!(runs[1].duration_seconds, Some(150));
931 assert_eq!(runs[1].metadata["build_number"], json!("20261028.22"));
932
933 let requests = stub.requests();
934 assert_eq!(requests.len(), 2);
935 assert_eq!(requests[0].query("definitions").as_deref(), Some("1"));
936 assert_eq!(
937 requests[0].query("queryOrder").as_deref(),
938 Some("queueTimeDescending")
939 );
940 assert_eq!(requests[0].query("$top").as_deref(), Some("3"));
941 assert_eq!(requests[0].query("continuationToken"), None);
942 assert_eq!(requests[1].query("$top").as_deref(), Some("1"));
943 assert_eq!(
944 requests[1].query("continuationToken").as_deref(),
945 Some("builds-2")
946 );
947 }
948
949 #[tokio::test]
950 async fn run_history_stops_at_the_limit_without_following_the_continuation() {
951 let stub = serve(azure_reply).await;
952 let plugin = plugin_for(&stub, "", "Alpha");
953
954 let runs = plugin
955 .fetch_run_history("azure-devops__1__Alpha/1", 1)
956 .await
957 .unwrap();
958
959 assert_eq!(runs.len(), 1);
960 assert_eq!(runs[0].run_number, 923);
961 assert_eq!(stub.requests().len(), 1);
962 assert!(plugin
963 .fetch_run_history("azure-devops__1__Alpha/1", 0)
964 .await
965 .unwrap()
966 .is_empty());
967 assert_eq!(stub.requests().len(), 1);
968 }
969
970 #[tokio::test]
971 async fn run_details_are_fetched_by_build_id_and_checked_against_the_pipeline() {
972 let stub = serve(azure_reply).await;
973 let plugin = plugin_for(&stub, "", "Alpha");
974
975 let run = plugin
976 .fetch_run_details("azure-devops__1__Alpha/1", 922)
977 .await
978 .unwrap();
979 assert_eq!(run.run_number, 922);
980 assert_eq!(run.status, PipelineStatus::Failed);
981 assert_eq!(run.commit_sha.as_deref(), Some("sha922"));
982 assert_eq!(
983 run.logs_url,
984 "https://dev.azure.com/acme/Alpha/_build/results?buildId=922"
985 );
986 assert_eq!(run.inputs, Some(json!({"source-branch": "main"})));
987
988 let foreign = plugin
989 .fetch_run_details("azure-devops__1__Alpha/1", 777)
990 .await
991 .unwrap_err();
992 assert!(
993 matches!(foreign, PluginError::PipelineNotFound(_)),
994 "{foreign}"
995 );
996 let missing = plugin
997 .fetch_run_details("azure-devops__1__Alpha/1", 404)
998 .await
999 .unwrap_err();
1000 assert!(
1001 matches!(missing, PluginError::PipelineNotFound(_)),
1002 "{missing}"
1003 );
1004 let malformed = plugin
1005 .fetch_run_details("github__1__Alpha/1", 922)
1006 .await
1007 .unwrap_err();
1008 assert!(
1009 matches!(malformed, PluginError::InvalidConfig(_)),
1010 "{malformed}"
1011 );
1012 }
1013
1014 #[tokio::test]
1015 async fn workflow_parameters_offer_the_default_branch_and_overridable_non_secret_variables() {
1016 let stub = serve(azure_reply).await;
1017 let plugin = plugin_for(&stub, "", "Alpha");
1018
1019 let parameters = plugin
1020 .fetch_workflow_parameters("azure-devops__1__Alpha/1")
1021 .await
1022 .unwrap();
1023
1024 let summary: Vec<_> = parameters
1025 .iter()
1026 .map(|parameter| {
1027 let WorkflowParameterType::String { default } = ¶meter.param_type else {
1028 panic!("expected a string parameter: {}", parameter.name);
1029 };
1030 (
1031 parameter.name.as_str(),
1032 default.as_deref(),
1033 parameter.required,
1034 )
1035 })
1036 .collect();
1037 assert_eq!(
1038 summary,
1039 [
1040 ("source-branch", Some("main"), false),
1041 ("branch", Some("legacy"), false),
1042 ("deployTarget", Some("dev"), false),
1043 ]
1044 );
1045 }
1046
1047 #[tokio::test]
1048 async fn trigger_queues_a_build_on_the_branch_with_the_overridden_variable() {
1049 let stub = serve(azure_reply).await;
1050 let plugin = plugin_for(&stub, "", "Alpha");
1051
1052 let result = plugin
1053 .trigger_pipeline(TriggerParams {
1054 workflow_id: "azure-devops__1__Alpha/1".to_string(),
1055 inputs: Some(json!({"source-branch": "feature/x", "deployTarget": "staging"})),
1056 })
1057 .await
1058 .unwrap();
1059
1060 let result: Value = serde_json::from_str(&result).unwrap();
1061 assert_eq!(
1062 result,
1063 json!({
1064 "message": "Queued build 930 on feature/x",
1065 "run_number": 930,
1066 "logs_url": "https://dev.azure.com/acme/Alpha/_build/results?buildId=930"
1067 })
1068 );
1069 let requests = stub.requests();
1070 assert_eq!(requests.len(), 2);
1071 assert_eq!(requests[0].method, "GET");
1072 assert_eq!(
1073 requests[0].url.path(),
1074 "/acme/Alpha/_apis/build/definitions/1"
1075 );
1076 let queued = &requests[1];
1077 assert_eq!(queued.method, "POST");
1078 assert_eq!(queued.url.path(), "/acme/Alpha/_apis/build/builds");
1079 assert_eq!(queued.query("api-version").as_deref(), Some("7.0"));
1080 assert_eq!(queued.authorization.as_deref(), Some(BASIC_AUTH));
1081 let body: Value = serde_json::from_str(&queued.body).unwrap();
1082 assert_eq!(
1083 body,
1084 json!({
1085 "definition": {"id": 1},
1086 "sourceBranch": "refs/heads/feature/x",
1087 "parameters": r#"{"deployTarget":"staging"}"#
1088 })
1089 );
1090 }
1091
1092 #[tokio::test]
1093 async fn a_variable_named_branch_is_sent_as_a_variable_not_as_the_source_branch() {
1094 let stub = serve(azure_reply).await;
1095 let plugin = plugin_for(&stub, "", "Alpha");
1096
1097 plugin
1098 .trigger_pipeline(TriggerParams {
1099 workflow_id: "azure-devops__1__Alpha/1".to_string(),
1100 inputs: Some(json!({"source-branch": "main", "branch": "legacy"})),
1101 })
1102 .await
1103 .unwrap();
1104
1105 let requests = stub.requests();
1106 assert_eq!(
1107 requests[0].url.path(),
1108 "/acme/Alpha/_apis/build/definitions/1"
1109 );
1110 let body: Value = serde_json::from_str(&requests[1].body).unwrap();
1111 assert_eq!(
1112 body,
1113 json!({
1114 "definition": {"id": 1},
1115 "sourceBranch": "refs/heads/main",
1116 "parameters": r#"{"branch":"legacy"}"#
1117 })
1118 );
1119 }
1120
1121 async fn queued_body(
1122 plugin: &AzureDevOpsPlugin, stub: &Stub, pipeline: &str, inputs: Value,
1123 ) -> Value {
1124 plugin
1125 .trigger_pipeline(TriggerParams {
1126 workflow_id: pipeline.to_string(),
1127 inputs: Some(inputs),
1128 })
1129 .await
1130 .unwrap();
1131 let requests = stub.requests();
1132 let queued = requests
1133 .iter()
1134 .find(|request| request.method == "POST")
1135 .unwrap();
1136 serde_json::from_str(&queued.body).unwrap()
1137 }
1138
1139 #[tokio::test]
1140 async fn an_overridable_ref_variable_is_sent_even_when_it_equals_the_branch() {
1141 let stub = serve(azure_reply).await;
1142 let plugin = plugin_for(&stub, "", "Alpha");
1143
1144 let body = queued_body(
1145 &plugin,
1146 &stub,
1147 "azure-devops__1__Alpha/2",
1148 json!({"source-branch": "main", "ref": "main"}),
1149 )
1150 .await;
1151
1152 assert_eq!(
1153 body,
1154 json!({
1155 "definition": {"id": 2},
1156 "sourceBranch": "refs/heads/main",
1157 "parameters": r#"{"ref":"main"}"#
1158 })
1159 );
1160 }
1161
1162 #[tokio::test]
1163 async fn the_rerun_ref_echo_is_dropped_when_the_definition_has_no_overridable_ref() {
1164 let stub = serve(azure_reply).await;
1165 let plugin = plugin_for(&stub, "", "Alpha");
1166
1167 let body = queued_body(
1168 &plugin,
1169 &stub,
1170 "azure-devops__1__Alpha/1",
1171 json!({"source-branch": "main", "ref": "main"}),
1172 )
1173 .await;
1174
1175 assert_eq!(
1176 body,
1177 json!({"definition": {"id": 1}, "sourceBranch": "refs/heads/main"})
1178 );
1179 }
1180
1181 #[tokio::test]
1182 async fn inputs_that_are_not_overridable_variables_are_not_sent() {
1183 let stub = serve(azure_reply).await;
1184 let plugin = plugin_for(&stub, "", "Alpha");
1185
1186 let body = queued_body(
1187 &plugin,
1188 &stub,
1189 "azure-devops__1__Alpha/1",
1190 json!({"DEPLOYtarget": "prod", "fixed": "x", "apiKey": "s3cret", "unknown": "y"}),
1191 )
1192 .await;
1193
1194 assert_eq!(
1195 body,
1196 json!({"definition": {"id": 1}, "parameters": r#"{"DEPLOYtarget":"prod"}"#})
1197 );
1198 }
1199
1200 #[tokio::test]
1201 async fn cancel_patches_the_build_status_to_cancelling() {
1202 let stub = serve(azure_reply).await;
1203 let plugin = plugin_for(&stub, "", "Alpha");
1204
1205 plugin
1206 .cancel_run("azure-devops__1__Alpha/1", 923)
1207 .await
1208 .unwrap();
1209
1210 let requests = stub.requests();
1211 assert_eq!(requests.len(), 2);
1212 assert_eq!(requests[0].method, "GET");
1213 assert_eq!(requests[0].url.path(), "/acme/Alpha/_apis/build/builds/923");
1214 assert_eq!(requests[1].method, "PATCH");
1215 assert_eq!(requests[1].url.path(), "/acme/Alpha/_apis/build/builds/923");
1216 assert_eq!(requests[1].query("api-version").as_deref(), Some("7.0"));
1217 let body: Value = serde_json::from_str(&requests[1].body).unwrap();
1218 assert_eq!(body, json!({"status": "cancelling"}));
1219 assert!(plugin.metadata().capabilities.cancel);
1220 }
1221
1222 #[tokio::test]
1223 async fn cancel_refuses_a_build_of_another_definition_without_patching() {
1224 let stub = serve(azure_reply).await;
1225 let plugin = plugin_for(&stub, "", "Alpha");
1226
1227 let error = plugin
1228 .cancel_run("azure-devops__1__Alpha/1", 777)
1229 .await
1230 .unwrap_err();
1231
1232 assert!(matches!(error, PluginError::PipelineNotFound(_)), "{error}");
1233 assert!(error.to_string().contains("does not belong"), "{error}");
1234 assert!(stub
1235 .requests()
1236 .iter()
1237 .all(|request| request.method == "GET"));
1238 }
1239
1240 #[tokio::test]
1241 async fn ids_with_ci_keeps_projects_that_have_definitions() {
1242 let stub = serve(azure_reply).await;
1243 let plugin = plugin_for(&stub, "", "");
1244
1245 let kept = plugin
1246 .ids_with_ci(vec!["Alpha".to_string(), "Empty".to_string()])
1247 .await
1248 .unwrap();
1249
1250 assert_eq!(kept, ["Alpha"]);
1251 }
1252
1253 #[tokio::test]
1254 async fn validate_accepts_a_working_token_with_a_single_cheap_call() {
1255 let stub = serve(|_| Reply::json(200, json!({"count": 0, "value": []}))).await;
1256
1257 assert!(plugin_for(&stub, "", "")
1258 .validate_credentials()
1259 .await
1260 .unwrap());
1261
1262 let requests = stub.requests();
1263 assert_eq!(requests.len(), 1);
1264 assert_eq!(requests[0].url.path(), "/acme/_apis/projects");
1265 assert_eq!(requests[0].query("$top").as_deref(), Some("1"));
1266 }
1267
1268 #[tokio::test]
1269 async fn validate_maps_401_to_authentication_failed() {
1270 let stub =
1271 serve(|_| Reply::json(401, json!({"message": "TF400813: not authorized"}))).await;
1272
1273 let error = plugin_for(&stub, "", "")
1274 .validate_credentials()
1275 .await
1276 .unwrap_err();
1277
1278 assert!(
1279 matches!(error, PluginError::AuthenticationFailed(_)),
1280 "{error}"
1281 );
1282 assert_eq!(stub.requests().len(), 1);
1283 }
1284
1285 #[tokio::test]
1286 async fn validate_maps_a_203_sign_in_page_to_authentication_failed() {
1287 let stub = serve(|_| Reply::html(203, "<html><title>Sign In</title></html>")).await;
1288
1289 let error = plugin_for(&stub, "", "")
1290 .validate_credentials()
1291 .await
1292 .unwrap_err();
1293
1294 assert!(
1295 matches!(error, PluginError::AuthenticationFailed(_)),
1296 "{error}"
1297 );
1298 assert!(error.to_string().contains("sign-in page"), "{error}");
1299 assert_eq!(stub.requests().len(), 1);
1300 }
1301
1302 #[tokio::test]
1303 async fn validate_maps_a_200_html_sign_in_page_to_authentication_failed() {
1304 let stub = serve(|_| Reply::html(200, "<html><title>Sign In</title></html>")).await;
1305
1306 let error = plugin_for(&stub, "", "")
1307 .validate_credentials()
1308 .await
1309 .unwrap_err();
1310
1311 assert!(
1312 matches!(error, PluginError::AuthenticationFailed(_)),
1313 "{error}"
1314 );
1315 }
1316
1317 #[tokio::test]
1318 async fn validate_maps_403_to_permission_denied() {
1319 let stub = serve(|_| Reply::json(403, json!({"message": "TF401444: no access"}))).await;
1320
1321 let error = plugin_for(&stub, "", "")
1322 .validate_credentials()
1323 .await
1324 .unwrap_err();
1325
1326 assert!(matches!(error, PluginError::PermissionDenied(_)), "{error}");
1327 }
1328
1329 #[test]
1330 fn initialize_requires_a_server_url_and_a_token() {
1331 let missing_url = AzureDevOpsPlugin::new()
1332 .initialize(
1333 1,
1334 HashMap::from([("token".to_string(), "t".to_string())]),
1335 None,
1336 )
1337 .unwrap_err();
1338 let missing_token = AzureDevOpsPlugin::new()
1339 .initialize(
1340 1,
1341 HashMap::from([(
1342 "server_url".to_string(),
1343 "https://dev.azure.com/acme".to_string(),
1344 )]),
1345 None,
1346 )
1347 .unwrap_err();
1348
1349 assert!(matches!(missing_url, PluginError::InvalidConfig(_)));
1350 assert!(matches!(missing_token, PluginError::InvalidConfig(_)));
1351 }
1352
1353 #[tokio::test]
1354 async fn a_list_cut_short_by_the_page_cap_is_an_error() {
1355 let stub = serve(|_| {
1356 Reply::json(
1357 200,
1358 json!({"count": 1, "value": [{"id": "p", "name": "Alpha"}]}),
1359 )
1360 .continued("again")
1361 })
1362 .await;
1363
1364 let error = plugin_for(&stub, "", "")
1365 .fetch_available_pipelines(None)
1366 .await
1367 .unwrap_err();
1368
1369 assert!(matches!(error, PluginError::ApiError(_)), "{error}");
1370 assert!(error.to_string().contains("more items"), "{error}");
1371 assert!(stub.requests().len() > 1);
1372 }
1373
1374 const STAGE: &str = "a0000000-0000-4000-8000-000000000001";
1375 const PHASE: &str = "b0000000-0000-4000-8000-000000000001";
1376 const JOB_LINUX: &str = "c0000000-0000-4000-8000-000000000001";
1377 const JOB_DOCS: &str = "c0000000-0000-4000-8000-000000000002";
1378 const JOB_PRUNED: &str = "c0000000-0000-4000-8000-000000000003";
1379 const JOB_EXPIRED: &str = "c0000000-0000-4000-8000-000000000004";
1380 const TASK_INIT: &str = "d0000000-0000-4000-8000-000000000001";
1381 const TASK_TESTS: &str = "d0000000-0000-4000-8000-000000000002";
1382 const TASK_PUBLISH: &str = "d0000000-0000-4000-8000-000000000003";
1383 const TASK_DOCS: &str = "d0000000-0000-4000-8000-000000000004";
1384 const CHECKPOINT: &str = "e0000000-0000-4000-8000-000000000001";
1385 const LINUX_LOG: &str =
1386 "\u{feff}2026-10-28T10:00:00.0000000Z ##[section]Starting: Linux tests\n\
1387 2026-10-28T10:00:01.1000000Z ##[section]Starting: Initialize job\n\
1388 2026-10-28T10:00:02.0000000Z ##[section]Finishing: Initialize job\n\
1389 2026-10-28T10:00:03.0000000Z ##[section]Starting: Run tests\n\
1390 2026-10-28T10:00:03.1000000Z ##[command]make test\n\
1391 2026-10-28T10:00:09.0000000Z ##[error]Bash exited with code '2'.\n\
1392 2026-10-28T10:00:09.5000000Z ##[section]Finishing: Run tests\n";
1393
1394 type Outcome = (&'static str, Option<&'static str>);
1395
1396 const SUCCEEDED: Outcome = ("completed", Some("succeeded"));
1397 const FAILED: Outcome = ("completed", Some("failed"));
1398 const SKIPPED: Outcome = ("completed", Some("skipped"));
1399
1400 fn record(
1401 id: &str, parent: Option<&str>, kind: &str, name: &str, order: i64, outcome: Outcome,
1402 log: Option<i64>,
1403 ) -> Value {
1404 let mut record = json!({
1405 "id": id,
1406 "parentId": parent,
1407 "type": kind,
1408 "name": name,
1409 "order": order,
1410 "state": outcome.0,
1411 "result": outcome.1,
1412 "startTime": "2026-10-28T10:00:00.5000000Z",
1413 "finishTime": "2026-10-28T10:00:09.5000000Z"
1414 });
1415 if let Some(log) = log {
1416 record["log"] = json!({"id": log, "type": "Container"});
1417 }
1418 record
1419 }
1420
1421 fn timeline_922() -> Value {
1422 let mut publish = record(
1423 TASK_PUBLISH,
1424 Some(JOB_LINUX),
1425 "Task",
1426 "Publish results",
1427 3,
1428 SKIPPED,
1429 None,
1430 );
1431 publish["startTime"] = Value::Null;
1432 publish["finishTime"] = Value::Null;
1433 json!({"id": "f0000000-0000-4000-8000-000000000001", "records": [
1434 record(TASK_TESTS, Some(JOB_LINUX), "Task", "Run tests", 2, FAILED, Some(9)),
1435 record(JOB_DOCS, Some(PHASE), "Job", "Docs", 2, SUCCEEDED, Some(12)),
1436 record(STAGE, None, "Stage", "Build", 1, FAILED, None),
1437 record(PHASE, Some(STAGE), "Phase", "Linux", 1, FAILED, None),
1438 record(JOB_LINUX, Some(PHASE), "Job", "Linux tests", 1, FAILED, Some(7)),
1439 record(CHECKPOINT, Some(STAGE), "Checkpoint", "Checkpoint", 0, SUCCEEDED, None),
1440 record(TASK_INIT, Some(JOB_LINUX), "Task", "Initialize job", 1, SUCCEEDED, Some(8)),
1441 publish,
1442 record(TASK_DOCS, Some(JOB_DOCS), "Task", "Build docs", 1, SUCCEEDED, Some(13))
1443 ]})
1444 }
1445
1446 fn timeline_921() -> Value {
1447 json!({"id": "f0000000-0000-4000-8000-000000000002", "records": [
1448 record(PHASE, None, "Phase", "Phase 1", 1, FAILED, None),
1449 record(JOB_PRUNED, Some(PHASE), "Job", "Build", 1, SUCCEEDED, Some(4)),
1450 record(JOB_EXPIRED, Some(PHASE), "Job", "Test", 2, FAILED, Some(5))
1451 ]})
1452 }
1453
1454 fn log_reply(request: &Request) -> Reply {
1455 match (request.method.as_str(), request.url.path()) {
1456 ("GET", "/acme/Alpha/_apis/build/builds/922/timeline") => {
1457 Reply::json(200, timeline_922())
1458 }
1459 ("GET", "/acme/Alpha/_apis/build/builds/922/logs/7") => Reply::text(200, LINUX_LOG),
1460 ("GET", "/acme/Alpha/_apis/build/builds/921/timeline") => {
1461 Reply::json(200, timeline_921())
1462 }
1463 ("GET", "/acme/Alpha/_apis/build/builds/921/logs/4") => {
1464 Reply::json(404, json!({"message": "TF400898: log not found"}))
1465 }
1466 ("GET", "/acme/Alpha/_apis/build/builds/921/logs/5") => {
1467 Reply::json(410, json!({"message": "log expired"}))
1468 }
1469 _ => Reply::not_found(request),
1470 }
1471 }
1472
1473 fn build_of_pipeline(run_number: i64, metadata: &HashMap<String, Value>) -> RunRef<'_> {
1474 RunRef {
1475 pipeline_id: "azure-devops__1__Alpha/1",
1476 run_number,
1477 run_id: "azure-devops-run",
1478 metadata,
1479 }
1480 }
1481
1482 async fn read_job_log(
1483 plugin: &AzureDevOpsPlugin, run_number: i64, node_id: &str,
1484 ) -> PluginResult<Vec<u8>> {
1485 let metadata = HashMap::new();
1486 let (tx, mut rx) = tokio::sync::mpsc::channel(16);
1487 let mut sink = LogSink::new(tx);
1488 let result = plugin
1489 .read_log(build_of_pipeline(run_number, &metadata), node_id, &mut sink)
1490 .await;
1491 drop(sink);
1492 let mut bytes = Vec::new();
1493 while let Some(chunk) = rx.recv().await {
1494 bytes.extend_from_slice(&chunk);
1495 }
1496 result.map(|()| bytes)
1497 }
1498
1499 #[tokio::test]
1500 async fn timeline_becomes_stage_job_and_step_nodes_in_order() {
1501 let stub = serve(log_reply).await;
1502 let plugin = plugin_for(&stub, "", "Alpha");
1503 let metadata = HashMap::new();
1504
1505 let nodes = plugin
1506 .fetch_log_nodes(build_of_pipeline(922, &metadata))
1507 .await
1508 .unwrap();
1509
1510 let summary: Vec<_> = nodes
1511 .iter()
1512 .map(|node| {
1513 (
1514 node.id.as_str(),
1515 node.parent_id.as_deref(),
1516 node.kind,
1517 node.name.as_str(),
1518 node.status,
1519 node.log,
1520 )
1521 })
1522 .collect();
1523 assert_eq!(
1524 summary,
1525 [
1526 (
1527 STAGE,
1528 None,
1529 LogNodeKind::Stage,
1530 "Build",
1531 PipelineStatus::Failed,
1532 None
1533 ),
1534 (
1535 JOB_LINUX,
1536 Some(STAGE),
1537 LogNodeKind::Job,
1538 "Linux tests",
1539 PipelineStatus::Failed,
1540 Some(LogAvailability::Ready)
1541 ),
1542 (
1543 TASK_INIT,
1544 Some(JOB_LINUX),
1545 LogNodeKind::Step,
1546 "Initialize job",
1547 PipelineStatus::Success,
1548 None
1549 ),
1550 (
1551 TASK_TESTS,
1552 Some(JOB_LINUX),
1553 LogNodeKind::Step,
1554 "Run tests",
1555 PipelineStatus::Failed,
1556 None
1557 ),
1558 (
1559 TASK_PUBLISH,
1560 Some(JOB_LINUX),
1561 LogNodeKind::Step,
1562 "Publish results",
1563 PipelineStatus::Skipped,
1564 None
1565 ),
1566 (
1567 JOB_DOCS,
1568 Some(STAGE),
1569 LogNodeKind::Job,
1570 "Docs",
1571 PipelineStatus::Success,
1572 Some(LogAvailability::Ready)
1573 ),
1574 (
1575 TASK_DOCS,
1576 Some(JOB_DOCS),
1577 LogNodeKind::Step,
1578 "Build docs",
1579 PipelineStatus::Success,
1580 None
1581 ),
1582 ]
1583 );
1584 assert_eq!(
1585 nodes[1].started_at.unwrap().to_rfc3339(),
1586 "2026-10-28T10:00:00.500+00:00"
1587 );
1588 assert_eq!(nodes[4].started_at, None);
1589 assert_eq!(nodes[4].finished_at, None);
1590 let requests = stub.requests();
1591 assert_eq!(requests.len(), 1);
1592 assert_eq!(
1593 requests[0].url.path(),
1594 "/acme/Alpha/_apis/build/builds/922/timeline"
1595 );
1596 assert_eq!(requests[0].query("api-version").as_deref(), Some("7.0"));
1597 }
1598
1599 #[tokio::test]
1600 async fn build_without_stages_lists_its_jobs_at_the_top() {
1601 let stub = serve(log_reply).await;
1602 let plugin = plugin_for(&stub, "", "Alpha");
1603 let metadata = HashMap::new();
1604
1605 let nodes = plugin
1606 .fetch_log_nodes(build_of_pipeline(921, &metadata))
1607 .await
1608 .unwrap();
1609
1610 let summary: Vec<_> = nodes
1611 .iter()
1612 .map(|node| (node.id.as_str(), node.parent_id.as_deref(), node.kind))
1613 .collect();
1614 assert_eq!(
1615 summary,
1616 [
1617 (JOB_PRUNED, None, LogNodeKind::Job),
1618 (JOB_EXPIRED, None, LogNodeKind::Job)
1619 ]
1620 );
1621 }
1622
1623 #[tokio::test]
1624 async fn build_without_a_timeline_has_no_log_nodes() {
1625 let stub = serve(log_reply).await;
1626 let plugin = plugin_for(&stub, "", "Alpha");
1627 let metadata = HashMap::new();
1628
1629 let error = plugin
1630 .fetch_log_nodes(build_of_pipeline(999, &metadata))
1631 .await
1632 .unwrap_err();
1633
1634 assert!(matches!(error, PluginError::PipelineNotFound(_)), "{error}");
1635 }
1636
1637 #[tokio::test]
1638 async fn job_log_streams_as_text_through_the_id_the_timeline_lists() {
1639 let stub = serve(log_reply).await;
1640 let plugin = plugin_for(&stub, "", "Alpha");
1641
1642 let log = read_job_log(&plugin, 922, JOB_LINUX).await.unwrap();
1643
1644 assert_eq!(log, LINUX_LOG.as_bytes());
1645 let requests = stub.requests();
1646 let paths: Vec<_> = requests
1647 .iter()
1648 .map(|request| request.url.path().to_string())
1649 .collect();
1650 assert_eq!(
1651 paths,
1652 [
1653 "/acme/Alpha/_apis/build/builds/922/timeline",
1654 "/acme/Alpha/_apis/build/builds/922/logs/7"
1655 ]
1656 );
1657 assert_eq!(requests[1].accept.as_deref(), Some("text/plain"));
1658 assert_eq!(requests[1].query("api-version").as_deref(), Some("7.0"));
1659 assert!(requests
1660 .iter()
1661 .all(|request| request.authorization.as_deref() == Some(BASIC_AUTH)));
1662 }
1663
1664 #[tokio::test]
1665 async fn log_the_server_no_longer_has_is_not_found() {
1666 let stub = serve(log_reply).await;
1667 let plugin = plugin_for(&stub, "", "Alpha");
1668
1669 for job in [JOB_PRUNED, JOB_EXPIRED] {
1670 let error = read_job_log(&plugin, 921, job).await.unwrap_err();
1671 assert!(
1672 matches!(&error, PluginError::PipelineNotFound(message) if message.contains("no longer available")),
1673 "{job}: {error:?}"
1674 );
1675 }
1676 }
1677
1678 #[tokio::test]
1679 async fn node_ids_outside_the_guid_alphabet_are_rejected_without_a_request() {
1680 let stub = serve(log_reply).await;
1681 let plugin = plugin_for(&stub, "", "Alpha");
1682
1683 let rejected = [
1684 "",
1685 "../../logs/7",
1686 "c0000000-0000-4000-8000-000000000001/../x",
1687 "x?y=1",
1688 "c0000000 0000",
1689 "c0000000-0000-4000-8000-00000000000g",
1690 "7/../../builds/1",
1691 ];
1692 for node_id in rejected {
1693 let error = read_job_log(&plugin, 922, node_id).await.unwrap_err();
1694 assert!(
1695 matches!(error, PluginError::InvalidConfig(_)),
1696 "{node_id}: {error:?}"
1697 );
1698 }
1699 assert!(stub.requests().is_empty());
1700 }
1701
1702 #[tokio::test]
1703 async fn node_the_timeline_does_not_list_as_a_job_log_never_reaches_the_log_endpoint() {
1704 let stub = serve(log_reply).await;
1705 let plugin = plugin_for(&stub, "", "Alpha");
1706
1707 let unlisted = [
1708 TASK_TESTS,
1709 STAGE,
1710 PHASE,
1711 CHECKPOINT,
1712 "ffffffff-0000-4000-8000-000000000000",
1713 "7",
1714 ];
1715 for node_id in unlisted {
1716 let error = read_job_log(&plugin, 922, node_id).await.unwrap_err();
1717 assert!(
1718 matches!(error, PluginError::PipelineNotFound(_)),
1719 "{node_id}: {error:?}"
1720 );
1721 }
1722 let requests = stub.requests();
1723 assert_eq!(requests.len(), unlisted.len());
1724 assert!(requests
1725 .iter()
1726 .all(|request| request.url.path().ends_with("/timeline")));
1727 }
1728}