1use std::collections::HashMap;
2use std::time::Duration;
3
4use async_trait::async_trait;
5use futures::future::join_all;
6use pipedash_plugin_api::*;
7
8use crate::log_dialect::GitlabLogDialect;
9use crate::{
10 client,
11 config,
12 mapper,
13 metadata,
14 types,
15};
16
17pub struct GitLabPlugin {
18 metadata: PluginMetadata,
19 client: Option<client::GitLabClient>,
20 provider_id: Option<i64>,
21 config: HashMap<String, String>,
22 status_source: Option<StatusSource>,
23}
24
25impl Default for GitLabPlugin {
26 fn default() -> Self {
27 Self::new()
28 }
29}
30
31impl GitLabPlugin {
32 pub fn new() -> Self {
33 Self {
34 metadata: metadata::create_metadata(),
35 client: None,
36 provider_id: None,
37 config: HashMap::new(),
38 status_source: None,
39 }
40 }
41
42 fn client(&self) -> PluginResult<&client::GitLabClient> {
43 self.client
44 .as_ref()
45 .ok_or_else(|| PluginError::Internal("Plugin not initialized".to_string()))
46 }
47
48 async fn fetch_all_projects(&self) -> PluginResult<Vec<types::Project>> {
49 let client = self.client()?;
50 let mut all_projects = Vec::new();
51 let mut page = 1;
52
53 loop {
54 let params = PaginationParams {
55 page,
56 page_size: 100,
57 };
58 let response = client.list_projects(¶ms).await?;
59 if response.items.is_empty() {
60 break;
61 }
62 all_projects.extend(response.items);
63
64 if !response.has_more || page > 100 {
65 break;
66 }
67 page += 1;
68 }
69
70 if let Some(selected_paths) = config::parse_selected_items(&self.config) {
71 Ok(all_projects
72 .into_iter()
73 .filter(|p| {
74 let normalized = p.name_with_namespace.replace(" ", "");
75 selected_paths.contains(&normalized)
76 })
77 .collect())
78 } else {
79 Ok(all_projects)
80 }
81 }
82
83 async fn fetch_projects_paginated(
84 &self, params: &PaginationParams,
85 ) -> PluginResult<PaginatedResponse<types::Project>> {
86 let client = self.client()?;
87 let response = client.list_projects(params).await?;
88
89 if let Some(selected_paths) = config::parse_selected_items(&self.config) {
90 let filtered_items: Vec<_> = response
91 .items
92 .into_iter()
93 .filter(|p| {
94 let normalized = p.name_with_namespace.replace(" ", "");
95 selected_paths.contains(&normalized)
96 })
97 .collect();
98 let filtered_count = filtered_items.len();
99 Ok(PaginatedResponse::new(
100 filtered_items,
101 params.page,
102 params.page_size,
103 filtered_count,
104 ))
105 } else {
106 Ok(response)
107 }
108 }
109}
110
111fn collect_partial<T>(results: Vec<PluginResult<T>>, item: &str) -> PluginResult<Vec<T>> {
112 let mut kept = Vec::with_capacity(results.len());
113 for result in results {
114 match result {
115 Ok(value) => kept.push(value),
116 Err(e) if e.is_transient() => return Err(e),
117 Err(e) => tracing::warn!(error = %e, item, "Skipping failed GitLab item"),
118 }
119 }
120 Ok(kept)
121}
122
123#[async_trait]
124impl Plugin for GitLabPlugin {
125 fn metadata(&self) -> &PluginMetadata {
126 &self.metadata
127 }
128
129 fn provider_type(&self) -> &str {
130 "gitlab"
131 }
132
133 fn status_source(&self) -> Option<&StatusSource> {
134 self.status_source.as_ref()
135 }
136
137 fn initialize(
138 &mut self, provider_id: i64, config: HashMap<String, String>,
139 http_client: Option<std::sync::Arc<reqwest::Client>>,
140 ) -> PluginResult<()> {
141 let token = config
142 .get("token")
143 .ok_or_else(|| PluginError::InvalidConfig("Missing GitLab access token".to_string()))?
144 .to_string();
145
146 let status_source = config::status_source(&config)?;
147 let base_url = config::get_base_url(&config);
148 let api_url = config::build_api_url(&base_url);
149
150 let client = http_client.unwrap_or_else(|| {
151 std::sync::Arc::new(
152 reqwest::Client::builder()
153 .use_rustls_tls()
154 .pool_max_idle_per_host(10)
155 .timeout(Duration::from_secs(30))
156 .connect_timeout(Duration::from_secs(10))
157 .tcp_keepalive(Duration::from_secs(60))
158 .build()
159 .expect("Failed to build HTTP client"),
160 )
161 });
162
163 self.status_source = status_source;
164 self.client = Some(client::GitLabClient::new(client, api_url, token));
165 self.provider_id = Some(provider_id);
166 self.config = config;
167
168 Ok(())
169 }
170
171 async fn validate_credentials(&self) -> PluginResult<bool> {
172 let client = self.client()?;
173 client.get_user().await?;
174 Ok(true)
175 }
176
177 async fn fetch_available_pipelines(
178 &self, params: Option<PaginationParams>,
179 ) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
180 let params = params.unwrap_or_default();
181 let response = self.fetch_projects_paginated(¶ms).await?;
182
183 let available_pipelines = response
184 .items
185 .iter()
186 .map(mapper::map_available_pipeline)
187 .collect();
188
189 Ok(PaginatedResponse::new(
190 available_pipelines,
191 response.page,
192 response.page_size,
193 response.total_count,
194 ))
195 }
196
197 async fn fetch_organizations(&self) -> PluginResult<Vec<pipedash_plugin_api::Organization>> {
198 let client = self.client()?;
199 client.list_groups().await
200 }
201
202 async fn ids_with_ci(&self, ids: Vec<String>) -> PluginResult<Vec<String>> {
203 let client = self.client()?;
204 let projects: HashMap<String, types::Project> = self
205 .fetch_all_projects()
206 .await?
207 .into_iter()
208 .map(|project| (project.name_with_namespace.replace(' ', ""), project))
209 .collect();
210 Ok(retain_with_ci(ids, |id| {
211 let project = projects.get(&id).cloned();
212 async move {
213 let Some(project) = project else {
214 return Ok(true);
215 };
216 if !client
217 .get_project_pipelines(project.id, 1)
218 .await?
219 .is_empty()
220 {
221 return Ok(true);
222 }
223 client.has_ci_config(&project).await
224 }
225 })
226 .await)
227 }
228
229 async fn fetch_available_pipelines_filtered(
230 &self, org: Option<String>, search: Option<String>, params: Option<PaginationParams>,
231 ) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
232 let params = params.unwrap_or_default();
233 let client = self.client()?;
234 let response = client.list_projects_filtered(org, search, ¶ms).await?;
235
236 let available_pipelines = response
237 .items
238 .iter()
239 .map(mapper::map_available_pipeline)
240 .collect();
241
242 Ok(PaginatedResponse::new(
243 available_pipelines,
244 response.page,
245 response.page_size,
246 response.total_count,
247 ))
248 }
249
250 async fn fetch_pipelines(&self) -> PluginResult<Vec<Pipeline>> {
251 let provider_id = self
252 .provider_id
253 .ok_or_else(|| PluginError::Internal("Provider ID not set".to_string()))?;
254
255 let client = self.client()?;
256 let projects = self.fetch_all_projects().await?;
257
258 let pipeline_futures = projects.iter().map(|project| async move {
259 let pipelines = client.get_project_pipelines(project.id, 1).await?;
260 let latest_pipeline = pipelines.first();
261 if latest_pipeline.is_none() && !client.has_ci_config(project).await.unwrap_or(true) {
262 return Ok(None);
263 }
264 Ok(Some(mapper::map_pipeline(
265 project,
266 latest_pipeline,
267 provider_id,
268 )))
269 });
270
271 Ok(
272 collect_partial(join_all(pipeline_futures).await, "project")?
273 .into_iter()
274 .flatten()
275 .collect(),
276 )
277 }
278
279 async fn fetch_run_history(
280 &self, pipeline_id: &str, limit: usize,
281 ) -> PluginResult<Vec<PipelineRun>> {
282 let (provider_id, project_id) = config::parse_pipeline_id(pipeline_id)?;
283 let client = self.client()?;
284
285 let project = client.get_project(project_id).await?;
286
287 let pipeline_list = client.get_project_pipelines(project_id, limit).await?;
288
289 let parts: Vec<&str> = project.name_with_namespace.split('/').collect();
290 let namespace = if parts.len() >= 2 {
291 Some(parts[..parts.len() - 1].join("/"))
292 } else {
293 None
294 };
295
296 let detailed_pipeline_futures = pipeline_list
297 .iter()
298 .map(|p| async move { client.get_pipeline(project_id, p.id).await });
299
300 let detailed_pipelines =
301 collect_partial(join_all(detailed_pipeline_futures).await, "pipeline run")?;
302
303 Ok(detailed_pipelines
304 .into_iter()
305 .map(|p| mapper::map_pipeline_run(&p, project_id, provider_id, namespace.as_deref()))
306 .collect())
307 }
308
309 async fn fetch_run_details(
310 &self, pipeline_id: &str, run_number: i64,
311 ) -> PluginResult<PipelineRun> {
312 let (provider_id, project_id) = config::parse_pipeline_id(pipeline_id)?;
313 let client = self.client()?;
314
315 let project = client.get_project(project_id).await?;
316 let pipeline = client.get_pipeline(project_id, run_number).await?;
317
318 let parts: Vec<&str> = project.name_with_namespace.split('/').collect();
319 let namespace = if parts.len() >= 2 {
320 Some(parts[..parts.len() - 1].join("/"))
321 } else {
322 None
323 };
324
325 Ok(mapper::map_pipeline_run(
326 &pipeline,
327 project_id,
328 provider_id,
329 namespace.as_deref(),
330 ))
331 }
332
333 async fn fetch_log_nodes(&self, run: RunRef<'_>) -> PluginResult<Vec<LogNode>> {
334 let (_, project_id) = config::parse_pipeline_id(run.pipeline_id)?;
335 let jobs = self
336 .client()?
337 .get_pipeline_jobs(project_id, run.run_number)
338 .await?;
339
340 Ok(mapper::jobs_to_log_nodes(&jobs))
341 }
342
343 async fn read_log(
344 &self, run: RunRef<'_>, node_id: &str, sink: &mut LogSink,
345 ) -> PluginResult<()> {
346 if node_id.is_empty() || !node_id.bytes().all(|b| b.is_ascii_digit()) {
347 return Err(PluginError::InvalidConfig(format!(
348 "Invalid GitLab job ID: {node_id}"
349 )));
350 }
351 let (_, project_id) = config::parse_pipeline_id(run.pipeline_id)?;
352
353 self.client()?
354 .stream_job_log(project_id, node_id, sink)
355 .await
356 }
357
358 fn log_dialect(&self) -> &'static dyn LogDialect {
359 &GitlabLogDialect
360 }
361
362 async fn fetch_workflow_parameters(
363 &self, _workflow_id: &str,
364 ) -> PluginResult<Vec<WorkflowParameter>> {
365 Ok(vec![WorkflowParameter {
366 name: "ref".to_string(),
367 label: Some("Ref".to_string()),
368 description: Some("Branch, tag, or commit SHA to run pipeline on".to_string()),
369 param_type: WorkflowParameterType::String {
370 default: Some("main".to_string()),
371 },
372 required: true,
373 }])
374 }
375
376 async fn trigger_pipeline(&self, params: TriggerParams) -> PluginResult<String> {
377 let (_provider_id, project_id) = config::parse_pipeline_id(¶ms.workflow_id)?;
378
379 let client = self.client()?;
380
381 let ref_name = params
382 .inputs
383 .as_ref()
384 .and_then(|inputs| inputs.get("ref"))
385 .and_then(|v| v.as_str())
386 .unwrap_or("main")
387 .to_string();
388
389 let variables = params.inputs.as_ref().and_then(|inputs| {
390 inputs.get("variables").and_then(|vars| {
391 vars.as_object().map(|obj| {
392 obj.iter()
393 .map(|(key, value)| types::PipelineVariable {
394 key: key.clone(),
395 value: value.as_str().unwrap_or("").to_string(),
396 variable_type: Some("env_var".to_string()),
397 })
398 .collect()
399 })
400 })
401 });
402
403 let pipeline = client
404 .trigger_pipeline(project_id, ref_name, variables)
405 .await?;
406 Ok(pipeline.web_url)
407 }
408
409 async fn cancel_run(&self, pipeline_id: &str, run_number: i64) -> PluginResult<()> {
410 let (_, project_id) = config::parse_pipeline_id(pipeline_id)?;
411 let client = self.client()?;
412 client.cancel_pipeline(project_id, run_number).await?;
413 Ok(())
414 }
415
416 async fn fetch_agents(&self) -> PluginResult<Vec<BuildAgent>> {
417 Err(PluginError::NotSupported(
418 "GitLab runners monitoring not implemented".to_string(),
419 ))
420 }
421
422 async fn fetch_artifacts(&self, _run_id: &str) -> PluginResult<Vec<BuildArtifact>> {
423 Err(PluginError::NotSupported(
424 "Artifacts download not implemented".to_string(),
425 ))
426 }
427
428 async fn fetch_queues(&self) -> PluginResult<Vec<BuildQueue>> {
429 Err(PluginError::NotSupported(
430 "Build queues not supported by GitLab".to_string(),
431 ))
432 }
433
434 fn get_migrations(&self) -> Vec<String> {
435 vec![]
436 }
437}
438
439#[cfg(test)]
440mod tests {
441 use tokio::io::{
442 AsyncReadExt,
443 AsyncWriteExt,
444 };
445
446 use super::*;
447
448 fn stub_pipeline(id: i64) -> serde_json::Value {
449 serde_json::json!({
450 "id": id, "project_id": 200, "status": "success", "ref": "main", "sha": "abc",
451 "web_url": format!("https://gitlab.example/team/app/-/pipelines/{id}"),
452 "created_at": "2026-10-01T00:00:00Z", "updated_at": "2026-10-01T00:00:00Z",
453 "started_at": null, "finished_at": null, "duration": null, "user": null
454 })
455 }
456
457 async fn stub_gitlab() -> (GitLabPlugin, tokio::task::JoinHandle<()>) {
458 let _ = rustls::crypto::ring::default_provider().install_default();
459 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
460 let addr = listener.local_addr().unwrap();
461 let server = tokio::spawn(async move {
462 loop {
463 let (mut stream, _) = listener.accept().await.unwrap();
464 let mut request = Vec::new();
465 let mut buffer = [0; 4096];
466 while !request.windows(4).any(|window| window == b"\r\n\r\n") {
467 let count = stream.read(&mut buffer).await.unwrap();
468 if count == 0 {
469 break;
470 }
471 request.extend_from_slice(&buffer[..count]);
472 }
473 let request = String::from_utf8(request).unwrap();
474 let target = request.split_whitespace().nth(1).unwrap();
475 let url = reqwest::Url::parse(&format!("http://{addr}{target}")).unwrap();
476 let query: HashMap<_, _> = url.query_pairs().into_owned().collect();
477 let mut total = 0;
478 let mut next_page = String::new();
479 let mut status = "200 OK";
480 let body = match url.path() {
481 "/api/v4/user" => serde_json::json!({
482 "id": 7, "username": "alice", "name": "Alice"
483 }),
484 "/api/v4/projects/102/pipelines" => serde_json::json!([{
485 "id": 9, "status": "success", "ref": "main", "sha": "abc",
486 "web_url": format!("http://{addr}/team/backend/api/-/pipelines/9"),
487 "created_at": "2026-10-26T12:00:00Z", "updated_at": "2026-10-26T12:05:00Z"
488 }]),
489 "/api/v4/projects/101/pipelines"
490 | "/api/v4/projects/103/pipelines"
491 | "/api/v4/projects/104/pipelines" => {
492 serde_json::json!([])
493 }
494 "/api/v4/projects/104/repository/files/.gitlab-ci.yml" => {
495 serde_json::json!({"file_name": ".gitlab-ci.yml"})
496 }
497 "/api/v4/projects/101/repository/files/.gitlab-ci.yml"
498 | "/api/v4/projects/103/repository/files/.gitlab-ci.yml" => {
499 status = "404 Not Found";
500 serde_json::json!({"message": "404 File Not Found"})
501 }
502 "/api/v4/projects/200" => serde_json::json!({
503 "id": 200, "name": "app", "name_with_namespace": "team / app",
504 "description": null, "web_url": format!("http://{addr}/team/app")
505 }),
506 "/api/v4/projects/200/pipelines" => {
507 total = 150_usize;
508 let page = query.get("page").map_or(1, |p| p.parse::<usize>().unwrap());
509 let page_size = query
510 .get("per_page")
511 .map_or(20, |p| p.parse::<usize>().unwrap())
512 .min(100);
513 if page < total.div_ceil(page_size) {
514 next_page = (page + 1).to_string();
515 }
516 serde_json::Value::Array(
517 (1..=150)
518 .rev()
519 .skip((page - 1) * page_size)
520 .take(page_size)
521 .map(stub_pipeline)
522 .collect(),
523 )
524 }
525 path if path.starts_with("/api/v4/projects/200/pipelines/") => {
526 stub_pipeline(path.rsplit('/').next().unwrap().parse().unwrap())
527 }
528 "/api/v4/groups/42/projects"
529 | "/api/v4/projects"
530 | "/api/v4/users/7/projects" => {
531 let recursive = url.path() == "/api/v4/projects"
532 || query
533 .get("include_subgroups")
534 .is_some_and(|value| value == "true");
535 let projects: Vec<_> = [
536 (101, "team / root"),
537 (102, "team / backend / api"),
538 (103, "team / frontend / api"),
539 (104, "team / backend / nested / worker"),
540 ]
541 .into_iter()
542 .filter(|(_, path)| recursive || *path == "team / root")
543 .filter(|(_, path)| {
544 query
545 .get("search")
546 .is_none_or(|search| path.contains(search))
547 })
548 .map(|(id, path)| {
549 serde_json::json!({
550 "id": id,
551 "name": path.rsplit(" / ").next().unwrap(),
552 "name_with_namespace": path,
553 "web_url": format!("http://{addr}/{}", path.replace(" / ", "/"))
554 })
555 })
556 .collect();
557 total = projects.len();
558 let page = query.get("page").unwrap().parse::<usize>().unwrap();
559 let page_size = query.get("per_page").unwrap().parse::<usize>().unwrap();
560 serde_json::Value::Array(
561 projects
562 .into_iter()
563 .skip((page - 1) * page_size)
564 .take(page_size)
565 .collect(),
566 )
567 }
568 "/api/v4/groups" => {
569 let groups: Vec<_> = (1..=105)
570 .map(|id| {
571 let (name, full_path) = match id {
572 1 => ("backend".to_string(), "team-a/backend".to_string()),
573 2 => ("backend".to_string(), "team-b/backend".to_string()),
574 _ => (format!("group-{id}"), format!("group-{id}")),
575 };
576 serde_json::json!({"id": id, "name": name, "full_path": full_path})
577 })
578 .collect();
579 total = groups.len();
580 let page = query.get("page").map_or(1, |p| p.parse::<usize>().unwrap());
581 let page_size = query
582 .get("per_page")
583 .map_or(20, |p| p.parse::<usize>().unwrap())
584 .min(100);
585 if page < total.div_ceil(page_size) {
586 next_page = (page + 1).to_string();
587 }
588 serde_json::Value::Array(
589 groups
590 .into_iter()
591 .skip((page - 1) * page_size)
592 .take(page_size)
593 .collect(),
594 )
595 }
596 path => panic!("Unexpected stub path: {path}"),
597 }
598 .to_string();
599 let response = format!(
600 "HTTP/1.1 {status}\r\nContent-Type: application/json\r\nContent-Length: {}\r\nX-Total: {total}\r\nX-Next-Page: {next_page}\r\nConnection: close\r\n\r\n{body}",
601 body.len()
602 );
603 stream.write_all(response.as_bytes()).await.unwrap();
604 }
605 });
606 let mut plugin = GitLabPlugin::new();
607 plugin
608 .initialize(
609 1,
610 HashMap::from([
611 ("base_url".to_string(), format!("http://{addr}")),
612 ("token".to_string(), "stub-token".to_string()),
613 ]),
614 None,
615 )
616 .unwrap();
617 (plugin, server)
618 }
619
620 #[tokio::test]
621 async fn parent_group_discovers_nested_projects_with_distinct_paths() {
622 let (plugin, server) = stub_gitlab().await;
623 let response = plugin
624 .fetch_available_pipelines_filtered(Some("42".to_string()), None, None)
625 .await
626 .unwrap();
627
628 let paths: Vec<_> = response.items.iter().map(|item| item.id.as_str()).collect();
629 assert_eq!(
630 paths,
631 [
632 "team/root",
633 "team/backend/api",
634 "team/frontend/api",
635 "team/backend/nested/worker"
636 ]
637 );
638 let displayed_paths: Vec<_> = response
639 .items
640 .iter()
641 .map(|item| {
642 format!(
643 "{}/{}",
644 item.organization.as_deref().unwrap(),
645 item.repository.as_deref().unwrap()
646 )
647 .replace(' ', "")
648 })
649 .collect();
650 assert_eq!(displayed_paths, paths);
651 server.abort();
652 }
653
654 #[tokio::test]
655 async fn subgroup_search_keeps_pagination_and_total_count() {
656 let (plugin, server) = stub_gitlab().await;
657 for (page, expected, has_more) in [
658 (1, vec!["team/backend/api"], true),
659 (2, vec!["team/frontend/api"], true),
660 (3, vec![], false),
661 ] {
662 let response = plugin
663 .fetch_available_pipelines_filtered(
664 Some("42".to_string()),
665 Some("api".to_string()),
666 Some(PaginationParams { page, page_size: 1 }),
667 )
668 .await
669 .unwrap();
670
671 assert_eq!(
672 response
673 .items
674 .iter()
675 .map(|item| item.id.as_str())
676 .collect::<Vec<_>>(),
677 expected
678 );
679 assert_eq!(response.total_count, 2);
680 assert_eq!(response.page, page);
681 assert_eq!(response.has_more, has_more);
682 }
683 server.abort();
684 }
685
686 #[tokio::test]
687 async fn saved_subgroup_selection_still_matches_during_refresh() {
688 let (mut plugin, server) = stub_gitlab().await;
689 plugin
690 .config
691 .insert("selected_items".to_string(), "team/backend/api".to_string());
692 let pipelines = plugin.fetch_pipelines().await.unwrap();
693
694 assert_eq!(
695 pipelines
696 .iter()
697 .map(|pipeline| pipeline.id.as_str())
698 .collect::<Vec<_>>(),
699 ["gitlab__1__102"]
700 );
701 assert_eq!(pipelines[0].repository, "team/backend/api");
702 server.abort();
703 }
704
705 #[tokio::test]
706 async fn only_projects_without_runs_and_ci_file_are_dropped_from_a_selection() {
707 let (mut plugin, server) = stub_gitlab().await;
708 let selection = "team/backend/api,team/frontend/api,team/backend/nested/worker,gone/repo";
709 plugin
710 .config
711 .insert("selected_items".to_string(), selection.to_string());
712
713 let kept = plugin
714 .ids_with_ci(selection.split(',').map(String::from).collect())
715 .await
716 .unwrap();
717 server.abort();
718
719 assert_eq!(
720 kept,
721 [
722 "team/backend/api",
723 "team/backend/nested/worker",
724 "gone/repo"
725 ]
726 );
727 }
728
729 #[tokio::test]
730 async fn tracking_every_project_lists_those_with_runs_or_a_ci_file() {
731 let (plugin, server) = stub_gitlab().await;
732
733 let pipelines = plugin.fetch_pipelines().await.unwrap();
734 server.abort();
735
736 assert_eq!(
737 pipelines
738 .iter()
739 .map(|pipeline| (pipeline.id.as_str(), pipeline.status))
740 .collect::<Vec<_>>(),
741 [
742 ("gitlab__1__102", PipelineStatus::Success),
743 ("gitlab__1__104", PipelineStatus::Pending)
744 ]
745 );
746 }
747
748 #[tokio::test]
749 async fn group_picker_lists_every_group_by_full_path() {
750 let (plugin, server) = stub_gitlab().await;
751 let orgs = plugin.fetch_organizations().await.unwrap();
752 server.abort();
753
754 assert_eq!(orgs.len(), 106);
755 assert_eq!(orgs[0].name, "alice");
756 assert_eq!(orgs[1].name, "team-a/backend");
757 assert_eq!(orgs[2].name, "team-b/backend");
758 assert!(orgs.iter().any(|org| org.id == "105"));
759 }
760
761 #[tokio::test]
762 async fn run_history_reads_past_the_first_100_pipelines() {
763 let (plugin, server) = stub_gitlab().await;
764 let all = plugin
765 .fetch_run_history("gitlab__1__200", 200)
766 .await
767 .unwrap();
768 let some = plugin
769 .fetch_run_history("gitlab__1__200", 120)
770 .await
771 .unwrap();
772 let latest = plugin.fetch_run_history("gitlab__1__200", 1).await.unwrap();
773 server.abort();
774
775 let numbers = |runs: &[PipelineRun]| runs.iter().map(|r| r.run_number).collect::<Vec<_>>();
776 assert_eq!(numbers(&all), (1..=150).rev().collect::<Vec<_>>());
777 assert_eq!(numbers(&some), (31..=150).rev().collect::<Vec<_>>());
778 assert_eq!(numbers(&latest), vec![150]);
779 }
780
781 type Requests = std::sync::Arc<tokio::sync::Mutex<Vec<String>>>;
782
783 const PIPELINE_JOBS: &str = r#"[
784 {"id":41,"name":"compile","stage":"build","status":"success","started_at":"2026-10-27T10:05:00Z","finished_at":"2026-10-27T10:06:00Z","web_url":"https://gitlab.example/team/app/-/jobs/41"},
785 {"id":40,"name":"unit","stage":"test","status":"running","started_at":"2026-10-27T10:06:10Z","finished_at":null,"web_url":"https://gitlab.example/team/app/-/jobs/40"},
786 {"id":14,"name":"docs","stage":"deploy","status":"skipped","started_at":null,"finished_at":null,"web_url":"https://gitlab.example/team/app/-/jobs/14"},
787 {"id":13,"name":"ship","stage":"deploy","status":"manual","started_at":null,"finished_at":null,"web_url":"https://gitlab.example/team/app/-/jobs/13"},
788 {"id":12,"name":"lint","stage":"test","status":"success","started_at":"2026-10-27T10:01:00Z","finished_at":"2026-10-27T10:02:00Z","web_url":"https://gitlab.example/team/app/-/jobs/12"},
789 {"id":11,"name":"unit","stage":"test","status":"failed","started_at":"2026-10-27T10:01:00Z","finished_at":"2026-10-27T10:03:00Z","web_url":"https://gitlab.example/team/app/-/jobs/11"},
790 {"id":10,"name":"compile","stage":"build","status":"failed","started_at":"2026-10-27T10:00:00Z","finished_at":"2026-10-27T10:00:30Z","web_url":"https://gitlab.example/team/app/-/jobs/10"}
791 ]"#;
792
793 const JOB_TRACE: &[u8] = b"Running with gitlab-runner 17.0.0\n\
794 section_start:1700000000:step_script\r\x1b[0K\x1b[36;1mExecuting step_script\x1b[0;m\n\
795 $ make test\n\
796 section_end:1700000007:step_script\r\x1b[0K\n";
797
798 async fn stub_jobs_and_trace() -> (GitLabPlugin, Requests, tokio::task::JoinHandle<()>) {
799 let _ = rustls::crypto::ring::default_provider().install_default();
800 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
801 let addr = listener.local_addr().unwrap();
802 let requests = Requests::default();
803 let seen = requests.clone();
804 let server = tokio::spawn(async move {
805 loop {
806 let (mut stream, _) = listener.accept().await.unwrap();
807 let mut request = Vec::new();
808 let mut buffer = [0; 4096];
809 while !request.windows(4).any(|window| window == b"\r\n\r\n") {
810 let count = stream.read(&mut buffer).await.unwrap();
811 if count == 0 {
812 break;
813 }
814 request.extend_from_slice(&buffer[..count]);
815 }
816 let request = String::from_utf8(request).unwrap();
817 seen.lock().await.push(request.clone());
818 let target = request.split_whitespace().nth(1).unwrap();
819 let url = reqwest::Url::parse(&format!("http://{addr}{target}")).unwrap();
820 let query: HashMap<_, _> = url.query_pairs().into_owned().collect();
821 let mut next_page = String::new();
822 let (status, content_type, body) = match url.path() {
823 "/api/v4/projects/200/pipelines/77/jobs" => {
824 let jobs: Vec<serde_json::Value> =
825 serde_json::from_str(PIPELINE_JOBS).unwrap();
826 let page = query["page"].parse::<usize>().unwrap();
827 let page_size = query["per_page"].parse::<usize>().unwrap().min(2);
828 if page * page_size < jobs.len() {
829 next_page = (page + 1).to_string();
830 }
831 let items: Vec<_> = jobs
832 .into_iter()
833 .skip((page - 1) * page_size)
834 .take(page_size)
835 .collect();
836 (
837 "200 OK",
838 "application/json",
839 serde_json::Value::Array(items).to_string().into_bytes(),
840 )
841 }
842 "/api/v4/projects/200/jobs/41/trace" => {
843 ("200 OK", "text/plain", JOB_TRACE.to_vec())
844 }
845 _ => (
846 "404 Not Found",
847 "application/json",
848 br#"{"message":"404 Not found"}"#.to_vec(),
849 ),
850 };
851 let head = format!(
852 "HTTP/1.1 {status}\r\nContent-Type: {content_type}\r\nContent-Length: {}\r\nX-Next-Page: {next_page}\r\nConnection: close\r\n\r\n",
853 body.len()
854 );
855 stream.write_all(head.as_bytes()).await.unwrap();
856 stream.write_all(&body).await.unwrap();
857 }
858 });
859 let mut plugin = GitLabPlugin::new();
860 plugin
861 .initialize(
862 1,
863 HashMap::from([
864 ("base_url".to_string(), format!("http://{addr}")),
865 ("token".to_string(), "stub-token".to_string()),
866 ]),
867 None,
868 )
869 .unwrap();
870 (plugin, requests, server)
871 }
872
873 fn run_77(metadata: &HashMap<String, serde_json::Value>) -> RunRef<'_> {
874 RunRef {
875 pipeline_id: "gitlab__1__200",
876 run_number: 77,
877 run_id: "gitlab__1__200_77",
878 metadata,
879 }
880 }
881
882 async fn read_job(plugin: &GitLabPlugin, node_id: &str) -> PluginResult<Vec<u8>> {
883 let metadata = HashMap::new();
884 let (tx, mut rx) = tokio::sync::mpsc::channel(16);
885 let mut sink = LogSink::new(tx);
886 let result = plugin.read_log(run_77(&metadata), node_id, &mut sink).await;
887 drop(sink);
888 let mut bytes = Vec::new();
889 while let Some(chunk) = rx.recv().await {
890 bytes.extend_from_slice(&chunk);
891 }
892 result.map(|()| bytes)
893 }
894
895 async fn targets(requests: &Requests) -> Vec<String> {
896 requests
897 .lock()
898 .await
899 .iter()
900 .map(|request| request.split_whitespace().nth(1).unwrap().to_string())
901 .collect()
902 }
903
904 #[tokio::test]
905 async fn pipeline_jobs_become_stages_holding_their_latest_jobs() {
906 let (plugin, requests, server) = stub_jobs_and_trace().await;
907 let metadata = HashMap::new();
908
909 let nodes = plugin.fetch_log_nodes(run_77(&metadata)).await.unwrap();
910 server.abort();
911
912 let summary: Vec<_> = nodes
913 .iter()
914 .map(|node| {
915 (
916 node.id.as_str(),
917 node.parent_id.as_deref(),
918 node.kind,
919 node.name.as_str(),
920 node.status,
921 node.log,
922 )
923 })
924 .collect();
925 assert_eq!(
926 summary,
927 [
928 (
929 "stage:build",
930 None,
931 LogNodeKind::Stage,
932 "build",
933 PipelineStatus::Success,
934 None
935 ),
936 (
937 "41",
938 Some("stage:build"),
939 LogNodeKind::Job,
940 "compile",
941 PipelineStatus::Success,
942 Some(LogAvailability::Ready)
943 ),
944 (
945 "stage:test",
946 None,
947 LogNodeKind::Stage,
948 "test",
949 PipelineStatus::Running,
950 None
951 ),
952 (
953 "40",
954 Some("stage:test"),
955 LogNodeKind::Job,
956 "unit",
957 PipelineStatus::Running,
958 Some(LogAvailability::Partial)
959 ),
960 (
961 "12",
962 Some("stage:test"),
963 LogNodeKind::Job,
964 "lint",
965 PipelineStatus::Success,
966 Some(LogAvailability::Ready)
967 ),
968 (
969 "stage:deploy",
970 None,
971 LogNodeKind::Stage,
972 "deploy",
973 PipelineStatus::Pending,
974 None
975 ),
976 (
977 "13",
978 Some("stage:deploy"),
979 LogNodeKind::Job,
980 "ship",
981 PipelineStatus::Pending,
982 Some(LogAvailability::Pending)
983 ),
984 (
985 "14",
986 Some("stage:deploy"),
987 LogNodeKind::Job,
988 "docs",
989 PipelineStatus::Skipped,
990 None
991 ),
992 ]
993 );
994 assert_eq!(
995 nodes[1].web_url.as_deref(),
996 Some("https://gitlab.example/team/app/-/jobs/41")
997 );
998 assert_eq!(
999 targets(&requests).await,
1000 (1..=4)
1001 .map(|page| format!(
1002 "/api/v4/projects/200/pipelines/77/jobs?per_page=100&include_retried=true&page={page}"
1003 ))
1004 .collect::<Vec<_>>()
1005 );
1006 }
1007
1008 #[tokio::test]
1009 async fn job_trace_streams_the_raw_text() {
1010 let (plugin, requests, server) = stub_jobs_and_trace().await;
1011
1012 let trace = read_job(&plugin, "41").await.unwrap();
1013 server.abort();
1014
1015 assert_eq!(trace, JOB_TRACE);
1016 assert_eq!(
1017 targets(&requests).await,
1018 ["/api/v4/projects/200/jobs/41/trace"]
1019 );
1020 assert!(requests.lock().await[0]
1021 .to_lowercase()
1022 .contains("private-token: stub-token"));
1023 }
1024
1025 #[tokio::test]
1026 async fn missing_job_trace_is_not_found() {
1027 let (plugin, _, server) = stub_jobs_and_trace().await;
1028
1029 let error = read_job(&plugin, "999").await.unwrap_err();
1030 server.abort();
1031
1032 assert!(
1033 matches!(error, PluginError::PipelineNotFound(_)),
1034 "{error:?}"
1035 );
1036 }
1037
1038 #[tokio::test]
1039 async fn job_ids_that_are_not_numbers_are_rejected_without_a_request() {
1040 let (plugin, requests, server) = stub_jobs_and_trace().await;
1041
1042 for node_id in [
1043 "",
1044 "../../jobs/41",
1045 "41/trace",
1046 "41?x=1",
1047 "-41",
1048 "4 1",
1049 "stage:test",
1050 "４１",
1051 ] {
1052 let error = read_job(&plugin, node_id).await.unwrap_err();
1053 assert!(
1054 matches!(error, PluginError::InvalidConfig(_)),
1055 "{node_id}: {error:?}"
1056 );
1057 }
1058 server.abort();
1059
1060 assert!(requests.lock().await.is_empty());
1061 }
1062}