Skip to main content

pipedash_plugin_azure_devops/
plugin.rs

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(&params.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 } = &parameter.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}