1use std::collections::{
2 BTreeSet,
3 HashMap,
4};
5use std::sync::Arc;
6
7use async_trait::async_trait;
8use pipedash_plugin_api::*;
9use tracing::{
10 debug,
11 info,
12 warn,
13};
14
15use crate::{
16 client,
17 config,
18 log_dialect,
19 mapper,
20 metadata,
21 types,
22};
23
24const DEFAULT_PAGE_SIZE: usize = 100;
25const LOG_CHUNK_BYTES: usize = 64 * 1024;
26
27pub struct KomodoPlugin {
28 metadata: PluginMetadata,
29 client: Option<client::KomodoClient>,
30 provider_id: Option<i64>,
31 config: HashMap<String, String>,
32 server_url: String,
33}
34
35impl Default for KomodoPlugin {
36 fn default() -> Self {
37 Self::new()
38 }
39}
40
41impl KomodoPlugin {
42 pub fn new() -> Self {
43 Self {
44 metadata: metadata::create_metadata(),
45 client: None,
46 provider_id: None,
47 config: HashMap::new(),
48 server_url: String::new(),
49 }
50 }
51
52 fn client(&self) -> PluginResult<&client::KomodoClient> {
53 self.client
54 .as_ref()
55 .ok_or_else(|| PluginError::Internal("Plugin not initialized".to_string()))
56 }
57
58 fn provider_id(&self) -> PluginResult<i64> {
59 self.provider_id
60 .ok_or_else(|| PluginError::Internal("Provider ID not set".to_string()))
61 }
62
63 async fn update_started_at(
64 &self, stack_id: &str, run_number: i64,
65 ) -> PluginResult<types::UpdateListItem> {
66 let mut matching = self
67 .client()?
68 .updates_started_at(stack_id, run_number)
69 .await?;
70 if matching.len() > 1 {
71 return Err(PluginError::ApiError(format!(
72 "{} updates of stack {stack_id} share run number {run_number}; refusing to pick one",
73 matching.len()
74 )));
75 }
76 matching.pop().ok_or_else(|| {
77 PluginError::PipelineNotFound(format!(
78 "Update {run_number} not found for stack {stack_id}"
79 ))
80 })
81 }
82
83 async fn run_update_logs(
84 &self, run: &RunRef<'_>, stack_id: &str,
85 ) -> PluginResult<(String, types::UpdateLogs)> {
86 let update_id = match run
87 .metadata
88 .get("update_id")
89 .and_then(serde_json::Value::as_str)
90 .filter(|id| !id.is_empty())
91 {
92 Some(id) => id.to_string(),
93 None => self.update_started_at(stack_id, run.run_number).await?.id,
94 };
95 let logs = self.client()?.update_logs(&update_id).await?;
96 Ok((update_id, logs))
97 }
98
99 fn stage_index(node_id: &str) -> PluginResult<usize> {
100 Some(node_id)
101 .filter(|id| id.bytes().all(|b| b.is_ascii_digit()))
102 .and_then(|id| id.parse().ok())
103 .ok_or_else(|| PluginError::InvalidConfig(format!("Invalid log node ID: {node_id}")))
104 }
105
106 async fn write_stage(log: &types::UpdateLog, sink: &mut LogSink) -> PluginResult<()> {
107 let mut chunk = Vec::new();
108 for (marker, line) in log_dialect::stage_lines(log) {
109 log_dialect::push_line(&mut chunk, marker, line);
110 if chunk.len() >= LOG_CHUNK_BYTES {
111 sink.write(std::mem::take(&mut chunk).into()).await?;
112 }
113 }
114 sink.write(chunk.into()).await
115 }
116}
117
118#[async_trait]
119impl Plugin for KomodoPlugin {
120 fn metadata(&self) -> &PluginMetadata {
121 &self.metadata
122 }
123
124 fn provider_type(&self) -> &str {
125 "komodo"
126 }
127
128 fn client_options(&self, config: &HashMap<String, String>) -> ClientOptions {
129 ClientOptions {
130 origin: config::get_server_url(config)
131 .ok()
132 .and_then(|url| reqwest::Url::parse(&url).ok()),
133 accept_invalid_certs: config::is_insecure(config),
134 }
135 }
136
137 fn initialize(
138 &mut self, provider_id: i64, config: HashMap<String, String>,
139 http_client: Option<Arc<reqwest::Client>>,
140 ) -> PluginResult<()> {
141 info!(provider_id, "Initializing Komodo plugin");
142
143 let server_url = config::get_server_url(&config)?;
144 let api_key = config::get_credential(&config, "api_key")?;
145 let api_secret = config::get_credential(&config, "token")?;
146 if config::is_insecure(&config) {
147 warn!("Insecure TLS mode enabled - certificate verification disabled");
148 }
149 let http_client = http_client.ok_or_else(|| {
150 PluginError::Internal("Komodo plugin requires an HTTP client".to_string())
151 })?;
152
153 self.client = Some(client::KomodoClient::new(
154 http_client,
155 server_url.clone(),
156 api_key,
157 api_secret,
158 ));
159 self.provider_id = Some(provider_id);
160 self.server_url = server_url;
161 self.config = config;
162 Ok(())
163 }
164
165 async fn validate_credentials(&self) -> PluginResult<bool> {
166 let version = self.client()?.version().await?;
167 debug!(version, "Komodo credentials validated");
168 Ok(true)
169 }
170
171 async fn fetch_available_pipelines(
172 &self, params: Option<PaginationParams>,
173 ) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
174 self.fetch_available_pipelines_filtered(None, None, params)
175 .await
176 }
177
178 async fn fetch_organizations(&self) -> PluginResult<Vec<Organization>> {
179 let stacks = self.client()?.list_stacks().await?;
180 let hosts: BTreeSet<&str> = stacks
181 .iter()
182 .map(|stack| stack.info.host())
183 .filter(|host| !host.is_empty())
184 .collect();
185
186 Ok(hosts
187 .into_iter()
188 .map(|host| Organization {
189 id: host.to_string(),
190 name: host.to_string(),
191 description: None,
192 })
193 .collect())
194 }
195
196 async fn fetch_available_pipelines_filtered(
197 &self, org: Option<String>, search: Option<String>, params: Option<PaginationParams>,
198 ) -> PluginResult<PaginatedResponse<AvailablePipeline>> {
199 let search = search.map(|term| term.to_lowercase());
200 let matching: Vec<AvailablePipeline> = self
201 .client()?
202 .list_stacks()
203 .await?
204 .iter()
205 .map(mapper::map_available_pipeline)
206 .filter(|stack| org.is_none() || stack.organization == org)
207 .filter(|stack| {
208 search.as_ref().is_none_or(|term| {
209 stack.name.to_lowercase().contains(term)
210 || stack
211 .description
212 .as_ref()
213 .is_some_and(|text| text.to_lowercase().contains(term))
214 })
215 })
216 .collect();
217
218 let total_count = matching.len();
219 let params = params.unwrap_or_default();
220 let page = params.page.max(1);
221 let page_size = if params.page_size == 0 {
222 DEFAULT_PAGE_SIZE
223 } else {
224 params.page_size
225 };
226 let offset = (page - 1).saturating_mul(page_size);
227 let items: Vec<AvailablePipeline> =
228 matching.into_iter().skip(offset).take(page_size).collect();
229
230 let has_more = offset + items.len() < total_count;
231 let mut response = PaginatedResponse::new(items, page, page_size, total_count);
232 response.has_more = has_more;
233 Ok(response)
234 }
235
236 async fn fetch_pipelines(&self) -> PluginResult<Vec<Pipeline>> {
237 let provider_id = self.provider_id()?;
238 let selected = config::parse_selected_items(&self.config);
239 let client = self.client()?;
240 let stacks = client.list_stacks().await?;
241
242 let listed: Vec<&types::StackListItem> = stacks
243 .iter()
244 .filter(|stack| {
245 selected.as_ref().is_none_or(|items| {
246 items
247 .iter()
248 .any(|item| *item == stack.id || *item == stack.name)
249 })
250 })
251 .collect();
252 let stack_ids: Vec<&str> = listed.iter().map(|stack| stack.id.as_str()).collect();
253 let last_starts = client.latest_update_starts(&stack_ids).await?;
254
255 let pipelines: Vec<Pipeline> = listed
256 .iter()
257 .map(|stack| {
258 mapper::map_pipeline(stack, provider_id, last_starts.get(&stack.id).copied())
259 })
260 .collect();
261
262 debug!(
263 stacks = stacks.len(),
264 pipelines = pipelines.len(),
265 "Mapped Komodo stacks to pipelines"
266 );
267 Ok(pipelines)
268 }
269
270 async fn fetch_run_history(
271 &self, pipeline_id: &str, limit: usize,
272 ) -> PluginResult<Vec<PipelineRun>> {
273 let (_, stack_id) = config::parse_pipeline_id(pipeline_id)?;
274 let updates = self.client()?.stack_updates(stack_id, limit).await?;
275
276 Ok(updates
277 .iter()
278 .map(|update| mapper::map_run(pipeline_id, stack_id, &self.server_url, update, None))
279 .collect())
280 }
281
282 async fn fetch_run_details(
283 &self, pipeline_id: &str, run_number: i64,
284 ) -> PluginResult<PipelineRun> {
285 let (_, stack_id) = config::parse_pipeline_id(pipeline_id)?;
286 let client = self.client()?;
287
288 let item = self.update_started_at(stack_id, run_number).await?;
289 let detail = if item.id.is_empty() {
290 None
291 } else {
292 Some(client.update(&item.id).await?)
293 };
294
295 Ok(mapper::map_run(
296 pipeline_id,
297 stack_id,
298 &self.server_url,
299 &item,
300 detail.as_ref(),
301 ))
302 }
303
304 async fn fetch_log_nodes(&self, run: RunRef<'_>) -> PluginResult<Vec<LogNode>> {
305 let (_, stack_id) = config::parse_pipeline_id(run.pipeline_id)?;
306 let (update_id, update) = self.run_update_logs(&run, stack_id).await?;
307
308 Ok(mapper::map_log_nodes(
309 &update,
310 &mapper::run_url(&self.server_url, stack_id, &update_id),
311 ))
312 }
313
314 async fn read_log(
315 &self, run: RunRef<'_>, node_id: &str, sink: &mut LogSink,
316 ) -> PluginResult<()> {
317 let index = Self::stage_index(node_id)?;
318 let (_, stack_id) = config::parse_pipeline_id(run.pipeline_id)?;
319 let (_, update) = self.run_update_logs(&run, stack_id).await?;
320 let log = update.logs.get(index).ok_or_else(|| {
321 PluginError::InvalidConfig(format!("Update has no log stage {index}"))
322 })?;
323
324 Self::write_stage(log, sink).await
325 }
326
327 fn log_dialect(&self) -> &'static dyn LogDialect {
328 &log_dialect::KomodoLogDialect
329 }
330
331 async fn trigger_pipeline(&self, params: TriggerParams) -> PluginResult<String> {
332 let (_, stack_id) = config::parse_pipeline_id(¶ms.workflow_id)?;
333 let update = self.client()?.deploy_stack(stack_id).await?;
334 let update_id = update
335 .id
336 .map(types::MongoId::into_string)
337 .unwrap_or_default();
338 info!(stack_id, update_id, "Triggered Komodo stack deploy");
339
340 Ok(serde_json::json!({
341 "message": format!("Triggered deploy of stack {stack_id}"),
342 "run_number": update.start_ts,
343 "logs_url": mapper::run_url(&self.server_url, stack_id, &update_id),
344 })
345 .to_string())
346 }
347}
348
349#[cfg(test)]
350mod tests {
351 use std::collections::HashSet;
352
353 use serde_json::{
354 json,
355 Value,
356 };
357 use tokio::io::{
358 AsyncReadExt,
359 AsyncWriteExt,
360 };
361 use tokio::sync::Mutex;
362
363 use super::*;
364
365 const API_KEY: &str = "K-stub";
366 const API_SECRET: &str = "S-stub";
367 const PIPELINE: &str = "komodo__1__stk1";
368 const NEWEST_START: i64 = 1_761_660_000_000;
369 const UPDATE_COUNT: usize = 250;
370 const UPDATES_PER_PAGE: usize = 100;
371 const LONG_OUTPUT_LINES: usize = 20_000;
372 const BUSY_START: i64 = NEWEST_START - 5_000_000;
373 const BUSY_UPDATES: usize = 650;
374 const RUNNING_START: i64 = NEWEST_START - 10_000_000;
375 const RUNNING_UPDATES: usize = 150;
376 const DOWN_START: i64 = NEWEST_START - 20_000_000;
377 const STATES: [&str; 11] = [
378 "deploying",
379 "running",
380 "paused",
381 "stopped",
382 "created",
383 "restarting",
384 "dead",
385 "removing",
386 "unhealthy",
387 "down",
388 "unknown",
389 ];
390
391 struct Recorded {
392 path: String,
393 headers: HashMap<String, String>,
394 body: Value,
395 }
396
397 type Requests = Arc<Mutex<Vec<Recorded>>>;
398
399 fn cfg(pairs: &[(&str, &str)]) -> HashMap<String, String> {
400 pairs
401 .iter()
402 .map(|(key, value)| (key.to_string(), value.to_string()))
403 .collect()
404 }
405
406 fn stacks() -> Value {
407 let mut stacks: Vec<Value> = STATES
408 .iter()
409 .map(|state| {
410 json!({
411 "id": format!("id-{state}"), "type": "Stack", "name": format!("stack-{state}"),
412 "template": false, "tags": [],
413 "info": {
414 "server_id": "srv1", "server_name": "prod-1", "repo": "acme/web",
415 "branch": "main", "repo_link": "https://github.com/acme/web",
416 "state": state, "status": "up",
417 "services": [{"service": "app", "image": "web", "update_available": *state == "running"}],
418 "project_missing": false, "missing_files": []
419 }
420 })
421 })
422 .collect();
423 stacks.push(json!({
424 "id": "id-template", "type": "Stack", "name": "stack-template", "template": true,
425 "tags": [], "info": {"state": "running"}
426 }));
427 Value::Array(stacks)
428 }
429
430 fn update_start(index: usize) -> i64 {
431 NEWEST_START - index as i64 * 1_000
432 }
433
434 fn listed_update(index: usize) -> Value {
435 json!({
436 "id": format!("u{index:03}"),
437 "operation": if index % 2 == 0 { "DeployStack" } else { "StartStack" },
438 "start_ts": update_start(index), "success": index % 5 != 0,
439 "username": "ada", "operator": "usr1", "target": {"type": "Stack", "id": "stk1"},
440 "status": if index == 0 { "InProgress" } else { "Complete" }
441 })
442 }
443
444 fn full_update(index: usize) -> Value {
445 json!({
446 "_id": {"$oid": format!("u{index:03}")},
447 "operation": if index % 2 == 0 { "DeployStack" } else { "StartStack" },
448 "start_ts": update_start(index), "success": index % 5 != 0, "operator": "usr1",
449 "target": {"type": "Stack", "id": "stk1"}, "logs": update_logs(index),
450 "end_ts": if index == 0 { Value::Null } else { json!(update_start(index) + 42_500) },
451 "status": if index == 0 { "InProgress" } else { "Complete" },
452 "commit_hash": "abc1234"
453 })
454 }
455
456 fn stage(
457 name: &str, command: &str, stdout: &str, stderr: &str, success: bool, range: (i64, i64),
458 ) -> Value {
459 json!({
460 "stage": name, "command": command, "stdout": stdout, "stderr": stderr,
461 "success": success, "start_ts": range.0, "end_ts": range.1
462 })
463 }
464
465 fn update_logs(index: usize) -> Value {
466 let start = update_start(index);
467 match index {
468 0 => json!([
469 stage(
470 "Clone Repo",
471 "git clone acme/web",
472 "cloned",
473 "",
474 true,
475 (start, start + 1_000)
476 ),
477 stage(
478 "Compose Up",
479 "docker compose up -d",
480 "pulling",
481 "",
482 true,
483 (start + 1_000, start + 2_000)
484 ),
485 ]),
486 1 => json!([stage(
487 "Compose Up",
488 "docker compose up -d",
489 "started",
490 "Network web_default Created",
491 true,
492 (start, start + 1_500)
493 )]),
494 5 => json!([
495 stage(
496 "Clone Repo",
497 "git clone acme/web",
498 "cloned",
499 "",
500 true,
501 (start, start + 1_000)
502 ),
503 stage(
504 "Compose Up",
505 "docker compose up -d",
506 "Pulling 10%\rPulling 100%\nContainer web-app-1 Starting\n",
507 "Error response from daemon: port is already allocated\nfailed to start web",
508 false,
509 (start + 1_000, start + 9_500)
510 ),
511 ]),
512 7 => json!([stage(
513 "Compose Pull",
514 "",
515 &long_output(),
516 "",
517 true,
518 (start, start + 3_000)
519 )]),
520 _ => json!([]),
521 }
522 }
523
524 fn long_output() -> String {
525 (0..LONG_OUTPUT_LINES)
526 .map(|line| format!("line {line:05}\n"))
527 .collect()
528 }
529
530 fn stack_update(stack_id: &str, index: usize, start_ts: i64) -> Value {
531 json!({
532 "id": format!("{stack_id}-u{index}"), "operation": "DeployStack", "start_ts": start_ts,
533 "success": true, "username": "ada", "operator": "usr1",
534 "target": {"type": "Stack", "id": stack_id}, "status": "Complete"
535 })
536 }
537
538 fn updates_of(stack_id: &str) -> Vec<Value> {
539 match stack_id {
540 "stk1" => (0..UPDATE_COUNT).map(listed_update).collect(),
541 "dup" => vec![
542 json!({"id": "d1", "start_ts": 7, "status": "Complete", "success": true}),
543 json!({"id": "d2", "start_ts": 7, "status": "Complete", "success": true}),
544 ],
545 "id-dead" => (0..BUSY_UPDATES)
546 .map(|i| stack_update(stack_id, i, BUSY_START - i as i64 * 1_000))
547 .collect(),
548 "id-running" => (0..RUNNING_UPDATES)
549 .map(|i| stack_update(stack_id, i, RUNNING_START - i as i64 * 1_000))
550 .collect(),
551 "id-down" => vec![stack_update(stack_id, 0, DOWN_START)],
552 _ => Vec::new(),
553 }
554 }
555
556 fn queried_stack_ids(query: &Value) -> Vec<String> {
557 let target = &query["target.id"];
558 match target.as_str() {
559 Some(id) => vec![id.to_string()],
560 None => target["$in"]
561 .as_array()
562 .into_iter()
563 .flatten()
564 .filter_map(|id| id.as_str().map(str::to_string))
565 .collect(),
566 }
567 }
568
569 fn list_updates(query: &Value, page: usize) -> (u16, Value) {
570 if query["target.type"] != "Stack" {
571 return (
572 400,
573 json!({"error": "expected a Stack target", "trace": []}),
574 );
575 }
576 let mut all: Vec<Value> = queried_stack_ids(query)
577 .iter()
578 .flat_map(|id| updates_of(id))
579 .collect();
580 all.sort_by_key(|update| std::cmp::Reverse(update["start_ts"].as_i64()));
581 let updates: Vec<Value> = all
582 .into_iter()
583 .filter(|update| {
584 query
585 .get("start_ts")
586 .is_none_or(|start_ts| *start_ts == update["start_ts"])
587 })
588 .skip(page * UPDATES_PER_PAGE)
589 .take(UPDATES_PER_PAGE)
590 .collect();
591 let next_page = (updates.len() == UPDATES_PER_PAGE).then_some(page + 1);
592 (200, json!({"updates": updates, "next_page": next_page}))
593 }
594
595 fn reply(request: &Recorded) -> (u16, Value) {
596 let header = |name: &str| request.headers.get(name).map(String::as_str);
597 if header("x-api-key") == Some("K-denied") && header("x-api-secret") == Some(API_SECRET) {
598 return (403, json!({"error": "permission denied", "trace": []}));
599 }
600 if header("x-api-key") != Some(API_KEY) || header("x-api-secret") != Some(API_SECRET) {
601 return (
602 401,
603 json!({"error": "invalid api key or secret", "trace": []}),
604 );
605 }
606 let params = &request.body["params"];
607 match (
608 request.path.as_str(),
609 request.body["type"].as_str().unwrap_or_default(),
610 ) {
611 ("/read", "GetVersion") => (200, json!({"version": "2.3.4"})),
612 ("/read", "ListStacks") => (200, stacks()),
613 ("/read", "ListUpdates") => list_updates(
614 ¶ms["query"],
615 params["page"].as_u64().unwrap_or_default() as usize,
616 ),
617 ("/read", "GetUpdate") => {
618 let id = params["id"].as_str().unwrap_or_default();
619 let index = id.strip_prefix('u').and_then(|index| index.parse().ok());
620 match (id, index) {
621 ("u-gone", _) => (
622 500,
623 json!({"error": "no update exists with given id", "trace": ["failed to query to db"]}),
624 ),
625 ("u-down", _) => (503, json!({"error": "database is down", "trace": []})),
626 ("u-busy", _) => (500, json!({"error": BUSY_CORE, "trace": []})),
627 (_, Some(index)) if index < UPDATE_COUNT => (200, full_update(index)),
628 _ => (404, json!({"error": "update not found", "trace": []})),
629 }
630 }
631 ("/execute", "DeployStack") if params["stack"] == "boom" => {
632 (503, json!({"error": "core is restarting", "trace": []}))
633 }
634 ("/execute", "DeployStack") => (
635 200,
636 json!({
637 "_id": {"$oid": "u-new"}, "operation": "DeployStack",
638 "start_ts": NEWEST_START + 5_000, "success": false, "operator": "usr1",
639 "target": {"type": "Stack", "id": params["stack"]}, "logs": [],
640 "end_ts": null, "status": "InProgress"
641 }),
642 ),
643 (path, kind) => (
644 404,
645 json!({"error": format!("unexpected {path} {kind}"), "trace": []}),
646 ),
647 }
648 }
649
650 async fn read_request(stream: &mut tokio::net::TcpStream) -> Option<Recorded> {
651 let mut raw = Vec::new();
652 let mut buffer = [0; 4096];
653 let (head_end, content_length) = loop {
654 let count = stream.read(&mut buffer).await.ok()?;
655 if count == 0 {
656 return None;
657 }
658 raw.extend_from_slice(&buffer[..count]);
659 if let Some(position) = raw.windows(4).position(|window| window == b"\r\n\r\n") {
660 let head = String::from_utf8_lossy(&raw[..position]).to_lowercase();
661 let length = head
662 .lines()
663 .find_map(|line| line.strip_prefix("content-length:"))
664 .and_then(|value| value.trim().parse().ok())
665 .unwrap_or(0);
666 break (position + 4, length);
667 }
668 };
669 while raw.len() < head_end + content_length {
670 let count = stream.read(&mut buffer).await.ok()?;
671 if count == 0 {
672 return None;
673 }
674 raw.extend_from_slice(&buffer[..count]);
675 }
676
677 let head = String::from_utf8_lossy(&raw[..head_end]).into_owned();
678 let mut lines = head.lines();
679 let path = lines.next()?.split_whitespace().nth(1)?.to_string();
680 let headers = lines
681 .filter_map(|line| line.split_once(':'))
682 .map(|(name, value)| (name.trim().to_lowercase(), value.trim().to_string()))
683 .collect();
684 let body = serde_json::from_slice(&raw[head_end..head_end + content_length])
685 .unwrap_or(Value::Null);
686 Some(Recorded {
687 path,
688 headers,
689 body,
690 })
691 }
692
693 const BUSY_CORE: &str = "core is busy";
694
695 async fn stub_komodo() -> (String, Requests) {
696 let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();
697 let addr = listener.local_addr().unwrap();
698 let requests = Requests::default();
699 let recorded = requests.clone();
700 tokio::spawn(async move {
701 while let Ok((mut stream, _)) = listener.accept().await {
702 let recorded = recorded.clone();
703 tokio::spawn(async move {
704 let Some(request) = read_request(&mut stream).await else {
705 return;
706 };
707 let (status, body) = reply(&request);
708 recorded.lock().await.push(request);
709 let retry_after = if body["error"] == BUSY_CORE {
710 "Retry-After: 120\r\n"
711 } else {
712 ""
713 };
714 let body = body.to_string();
715 let response = format!(
716 "HTTP/1.1 {status} X\r\nContent-Type: application/json\r\n{retry_after}Content-Length: {}\r\nConnection: close\r\n\r\n{body}",
717 body.len()
718 );
719 let _ = stream.write_all(response.as_bytes()).await;
720 });
721 }
722 });
723 (format!("http://{addr}"), requests)
724 }
725
726 fn plugin_with(
727 server_url: &str, api_key: &str, secret: &str, extra: &[(&str, &str)],
728 ) -> KomodoPlugin {
729 let _ = rustls::crypto::ring::default_provider().install_default();
730 let mut pairs = vec![
731 ("server_url", server_url),
732 ("api_key", api_key),
733 ("token", secret),
734 ];
735 pairs.extend_from_slice(extra);
736 let mut plugin = KomodoPlugin::new();
737 plugin
738 .initialize(1, cfg(&pairs), Some(Arc::new(reqwest::Client::new())))
739 .unwrap();
740 plugin
741 }
742
743 async fn stub_plugin(extra: &[(&str, &str)]) -> (KomodoPlugin, String, Requests) {
744 let (server_url, requests) = stub_komodo().await;
745 let plugin = plugin_with(&format!("{server_url}/"), API_KEY, API_SECRET, extra);
746 (plugin, server_url, requests)
747 }
748
749 async fn bodies(requests: &Requests, request_type: &str) -> Vec<Value> {
750 requests
751 .lock()
752 .await
753 .iter()
754 .filter(|request| request.body["type"] == request_type)
755 .map(|request| request.body.clone())
756 .collect()
757 }
758
759 async fn list_update_pages(requests: &Requests) -> Vec<u64> {
760 bodies(requests, "ListUpdates")
761 .await
762 .iter()
763 .map(|body| body["params"]["page"].as_u64().unwrap())
764 .collect()
765 }
766
767 #[tokio::test]
768 async fn stacks_become_pipelines_with_every_state_mapped() {
769 let (plugin, _, requests) = stub_plugin(&[]).await;
770
771 let pipelines = plugin.fetch_pipelines().await.unwrap();
772
773 let summary: Vec<_> = pipelines
774 .iter()
775 .map(|pipeline| (pipeline.name.as_str(), pipeline.status))
776 .collect();
777 assert_eq!(
778 summary,
779 [
780 ("stack-deploying", PipelineStatus::Running),
781 ("stack-running", PipelineStatus::Success),
782 ("stack-paused", PipelineStatus::Cancelled),
783 ("stack-stopped", PipelineStatus::Cancelled),
784 ("stack-created", PipelineStatus::Pending),
785 ("stack-restarting", PipelineStatus::Failed),
786 ("stack-dead", PipelineStatus::Failed),
787 ("stack-removing", PipelineStatus::Running),
788 ("stack-unhealthy", PipelineStatus::Failed),
789 ("stack-down", PipelineStatus::Cancelled),
790 ("stack-unknown", PipelineStatus::Pending),
791 ]
792 );
793 assert_eq!(pipelines[0].id, "komodo__1__id-deploying");
794 assert_eq!(pipelines[0].repository, "prod-1/stack-deploying");
795 assert_eq!(pipelines[0].branch.as_deref(), Some("main"));
796 assert_eq!(pipelines[0].metadata["server"], json!("prod-1"));
797 assert_eq!(pipelines[0].metadata["update_available"], json!(false));
798 assert_eq!(pipelines[1].metadata["update_available"], json!(true));
799
800 let requests = requests.lock().await;
801 assert_eq!(requests[0].path, "/read");
802 assert_eq!(
803 requests[0].body,
804 json!({"type": "ListStacks", "params": {"query": {}, "limit": 0}})
805 );
806 assert_eq!(requests[1].body["type"], "ListUpdates");
807 assert_eq!(
808 requests[1].body["params"]["query"]["target.id"]["$in"]
809 .as_array()
810 .map(Vec::len),
811 Some(STATES.len())
812 );
813 }
814
815 #[tokio::test]
816 async fn selection_keeps_stacks_matching_an_id_or_a_name() {
817 let (plugin, _, _) = stub_plugin(&[("selected_items", "stack-running, id-down")]).await;
818 let names: Vec<_> = plugin
819 .fetch_pipelines()
820 .await
821 .unwrap()
822 .into_iter()
823 .map(|pipeline| pipeline.name)
824 .collect();
825 assert_eq!(names, ["stack-running", "stack-down"]);
826
827 let (nothing, _, _) = stub_plugin(&[("selected_items", "")]).await;
828 assert!(nothing.fetch_pipelines().await.unwrap().is_empty());
829 }
830
831 async fn last_runs(extra: &[(&str, &str)]) -> (HashMap<String, Option<i64>>, Vec<u64>, Value) {
832 let (plugin, _, requests) = stub_plugin(extra).await;
833 let pipelines = plugin.fetch_pipelines().await.unwrap();
834 let last_runs = pipelines
835 .into_iter()
836 .map(|pipeline| {
837 (
838 pipeline.name,
839 pipeline.last_run.map(|at| at.timestamp_millis()),
840 )
841 })
842 .collect();
843 let first_lookup = bodies(&requests, "ListUpdates").await.remove(0);
844 (
845 last_runs,
846 list_update_pages(&requests).await,
847 first_lookup["params"]["query"].clone(),
848 )
849 }
850
851 #[tokio::test]
852 async fn last_run_is_the_newest_update_start_and_none_without_updates() {
853 let (last_runs, pages, query) =
854 last_runs(&[("selected_items", "id-running,id-down,id-paused")]).await;
855
856 assert_eq!(last_runs["stack-running"], Some(RUNNING_START));
857 assert_eq!(last_runs["stack-down"], Some(DOWN_START));
858 assert_eq!(last_runs["stack-paused"], None);
859 assert_eq!(pages, [0, 1]);
860 assert_eq!(
861 query,
862 json!({
863 "target.type": "Stack",
864 "target.id": {"$in": ["id-running", "id-paused", "id-down"]}
865 })
866 );
867 }
868
869 #[tokio::test]
870 async fn update_paging_stops_once_every_stack_is_seen() {
871 let (last_runs, pages, _) = last_runs(&[("selected_items", "id-running")]).await;
872
873 assert_eq!(last_runs["stack-running"], Some(RUNNING_START));
874 assert_eq!(pages, [0]);
875 }
876
877 #[tokio::test]
878 async fn update_paging_is_capped_when_a_stack_stays_unseen() {
879 let (last_runs, pages, _) = last_runs(&[("selected_items", "id-dead,id-down")]).await;
880
881 assert_eq!(last_runs["stack-dead"], Some(BUSY_START));
882 assert_eq!(last_runs["stack-down"], None);
883 assert_eq!(pages, [0, 1, 2, 3, 4]);
884 }
885
886 #[tokio::test]
887 async fn no_update_lookup_is_sent_without_stacks() {
888 let (plugin, _, requests) = stub_plugin(&[("selected_items", "")]).await;
889
890 assert!(plugin.fetch_pipelines().await.unwrap().is_empty());
891
892 assert!(bodies(&requests, "ListUpdates").await.is_empty());
893 }
894
895 #[tokio::test]
896 async fn picker_lists_stacks_by_id_with_paging_search_and_server_filter() {
897 let (plugin, _, _) = stub_plugin(&[]).await;
898 let page = |page, page_size| Some(PaginationParams { page, page_size });
899
900 let second = plugin.fetch_available_pipelines(page(2, 5)).await.unwrap();
901 let third = plugin.fetch_available_pipelines(page(3, 5)).await.unwrap();
902 let found = plugin
903 .fetch_available_pipelines_filtered(
904 Some("prod-1".to_string()),
905 Some("DEAD".to_string()),
906 None,
907 )
908 .await
909 .unwrap();
910 let elsewhere = plugin
911 .fetch_available_pipelines_filtered(Some("prod-2".to_string()), None, None)
912 .await
913 .unwrap();
914 let organizations = plugin.fetch_organizations().await.unwrap();
915
916 let ids = |response: &PaginatedResponse<AvailablePipeline>| {
917 response
918 .items
919 .iter()
920 .map(|item| item.id.clone())
921 .collect::<Vec<_>>()
922 };
923 assert_eq!(
924 ids(&second),
925 [
926 "id-restarting",
927 "id-dead",
928 "id-removing",
929 "id-unhealthy",
930 "id-down"
931 ]
932 );
933 assert_eq!(
934 (second.total_count, second.total_pages, second.has_more),
935 (11, 3, true)
936 );
937 assert_eq!(ids(&third), ["id-unknown"]);
938 assert!(!third.has_more);
939 assert_eq!(ids(&found), ["id-dead"]);
940 assert!(elsewhere.items.is_empty());
941 assert_eq!(
942 organizations
943 .iter()
944 .map(|o| o.name.as_str())
945 .collect::<Vec<_>>(),
946 ["prod-1"]
947 );
948 }
949
950 #[tokio::test]
951 async fn run_history_follows_next_page_until_the_limit_is_met() {
952 for (limit, expected_runs, expected_pages) in [
953 (0, 0, vec![]),
954 (1, 1, vec![0]),
955 (100, 100, vec![0]),
956 (101, 101, vec![0, 1]),
957 (1000, UPDATE_COUNT, vec![0, 1, 2]),
958 ] {
959 let (plugin, _, requests) = stub_plugin(&[]).await;
960
961 let runs = plugin.fetch_run_history(PIPELINE, limit).await.unwrap();
962
963 assert_eq!(runs.len(), expected_runs, "limit {limit}");
964 assert_eq!(
965 list_update_pages(&requests).await,
966 expected_pages,
967 "limit {limit}"
968 );
969 assert_eq!(
970 runs.iter()
971 .map(|run| run.run_number)
972 .collect::<HashSet<_>>()
973 .len(),
974 runs.len()
975 );
976 }
977 }
978
979 #[tokio::test]
980 async fn run_history_maps_updates_of_the_stack_newest_first() {
981 let (plugin, server_url, requests) = stub_plugin(&[]).await;
982
983 let runs = plugin.fetch_run_history(PIPELINE, 150).await.unwrap();
984
985 assert!(runs
986 .windows(2)
987 .all(|pair| pair[0].run_number > pair[1].run_number));
988 assert_eq!(runs[0].run_number, NEWEST_START);
989 assert_eq!(runs[0].status, PipelineStatus::Running);
990 assert_eq!(runs[1].status, PipelineStatus::Success);
991 assert_eq!(runs[5].status, PipelineStatus::Failed);
992 assert_eq!(runs[1].actor.as_deref(), Some("ada"));
993 assert_eq!(runs[1].metadata["operation"], json!("StartStack"));
994 assert_eq!(runs[1].pipeline_id, PIPELINE);
995 assert_eq!(runs[1].logs_url, format!("{server_url}/updates/u001"));
996 assert_eq!(runs[1].duration_seconds, None);
997 assert_eq!(
998 bodies(&requests, "ListUpdates").await[0]["params"],
999 json!({"query": {"target.type": "Stack", "target.id": "stk1"}, "page": 0})
1000 );
1001 }
1002
1003 #[tokio::test]
1004 async fn run_details_resolve_a_run_number_to_one_update() {
1005 let (plugin, server_url, requests) = stub_plugin(&[]).await;
1006 let runs = plugin.fetch_run_history(PIPELINE, 10).await.unwrap();
1007 let listed = &runs[4];
1008
1009 let details = plugin
1010 .fetch_run_details(PIPELINE, listed.run_number)
1011 .await
1012 .unwrap();
1013
1014 assert_eq!(details.id, listed.id);
1015 assert_eq!(details.run_number, listed.run_number);
1016 assert_eq!(details.status, PipelineStatus::Success);
1017 assert_eq!(details.actor.as_deref(), Some("ada"));
1018 assert_eq!(details.duration_seconds, Some(42));
1019 assert_eq!(
1020 details.concluded_at.map(|at| at.timestamp_millis()),
1021 Some(listed.run_number + 42_500)
1022 );
1023 assert_eq!(details.commit_sha.as_deref(), Some("abc1234"));
1024 assert_eq!(details.logs_url, format!("{server_url}/updates/u004"));
1025 let lookups = bodies(&requests, "ListUpdates").await;
1026 assert_eq!(
1027 lookups.last().unwrap()["params"]["query"],
1028 json!({"target.type": "Stack", "target.id": "stk1", "start_ts": listed.run_number})
1029 );
1030 assert_eq!(
1031 bodies(&requests, "GetUpdate").await,
1032 [json!({"type": "GetUpdate", "params": {"id": "u004"}})]
1033 );
1034
1035 let running = plugin
1036 .fetch_run_details(PIPELINE, NEWEST_START)
1037 .await
1038 .unwrap();
1039 assert_eq!(running.status, PipelineStatus::Running);
1040 assert_eq!(running.concluded_at, None);
1041 assert_eq!(running.duration_seconds, None);
1042 }
1043
1044 #[tokio::test]
1045 async fn run_details_for_an_unknown_run_number_are_not_found() {
1046 let (plugin, _, requests) = stub_plugin(&[]).await;
1047
1048 let error = plugin
1049 .fetch_run_details(PIPELINE, 12_345)
1050 .await
1051 .unwrap_err();
1052
1053 assert!(matches!(error, PluginError::PipelineNotFound(_)), "{error}");
1054 assert!(bodies(&requests, "GetUpdate").await.is_empty());
1055 }
1056
1057 #[tokio::test]
1058 async fn run_details_refuse_when_two_updates_share_a_run_number() {
1059 let (plugin, _, _) = stub_plugin(&[]).await;
1060
1061 let error = plugin
1062 .fetch_run_details("komodo__1__dup", 7)
1063 .await
1064 .unwrap_err();
1065
1066 assert!(
1067 matches!(&error, PluginError::ApiError(m) if m.contains("share run number")),
1068 "{error}"
1069 );
1070 }
1071
1072 #[tokio::test]
1073 async fn trigger_deploys_the_stack_and_returns_the_new_update() {
1074 let (plugin, server_url, requests) = stub_plugin(&[]).await;
1075
1076 let result = plugin
1077 .trigger_pipeline(TriggerParams {
1078 workflow_id: PIPELINE.to_string(),
1079 inputs: None,
1080 })
1081 .await
1082 .unwrap();
1083
1084 let result: Value = serde_json::from_str(&result).unwrap();
1085 assert_eq!(result["run_number"], json!(NEWEST_START + 5_000));
1086 assert_eq!(
1087 result["logs_url"],
1088 json!(format!("{server_url}/updates/u-new"))
1089 );
1090 let requests = requests.lock().await;
1091 assert_eq!(requests.len(), 1);
1092 assert_eq!(requests[0].path, "/execute");
1093 assert_eq!(
1094 requests[0].body,
1095 json!({"type": "DeployStack", "params": {"stack": "stk1"}})
1096 );
1097 assert_eq!(requests[0].headers["x-api-key"], API_KEY);
1098 assert_eq!(requests[0].headers["x-api-secret"], API_SECRET);
1099 assert_eq!(requests[0].headers["content-type"], "application/json");
1100 }
1101
1102 #[tokio::test]
1103 async fn deploy_is_not_resent_after_a_server_error() {
1104 let (plugin, _, requests) = stub_plugin(&[]).await;
1105
1106 let error = plugin
1107 .trigger_pipeline(TriggerParams {
1108 workflow_id: "komodo__1__boom".to_string(),
1109 inputs: None,
1110 })
1111 .await
1112 .unwrap_err();
1113
1114 assert!(
1115 matches!(error, PluginError::ServiceUnavailable { .. }),
1116 "{error}"
1117 );
1118 assert_eq!(bodies(&requests, "DeployStack").await.len(), 1);
1119 }
1120
1121 #[tokio::test]
1122 async fn validation_calls_get_version_and_rejects_bad_credentials() {
1123 let (plugin, server_url, requests) = stub_plugin(&[]).await;
1124 let wrong_secret = plugin_with(&server_url, API_KEY, "S-wrong", &[]);
1125 let denied = plugin_with(&server_url, "K-denied", API_SECRET, &[]);
1126
1127 assert!(plugin.validate_credentials().await.unwrap());
1128 let unauthorized = wrong_secret.validate_credentials().await.unwrap_err();
1129 let forbidden = denied.validate_credentials().await.unwrap_err();
1130
1131 assert_eq!(
1132 bodies(&requests, "GetVersion").await[0],
1133 json!({"type": "GetVersion", "params": {}})
1134 );
1135 assert!(
1136 matches!(&unauthorized, PluginError::AuthenticationFailed(m) if m.contains("invalid api key")),
1137 "{unauthorized}"
1138 );
1139 assert!(
1140 matches!(forbidden, PluginError::PermissionDenied(_)),
1141 "{forbidden}"
1142 );
1143 }
1144
1145 fn update_metadata(update_id: &str) -> HashMap<String, Value> {
1146 HashMap::from([("update_id".to_string(), json!(update_id))])
1147 }
1148
1149 fn log_run(metadata: &HashMap<String, Value>, run_number: i64) -> RunRef<'_> {
1150 RunRef {
1151 pipeline_id: PIPELINE,
1152 run_number,
1153 run_id: "komodo__1__stk1__run",
1154 metadata,
1155 }
1156 }
1157
1158 async fn read_chunks(
1159 plugin: &KomodoPlugin, run: RunRef<'_>, node_id: &str,
1160 ) -> PluginResult<Vec<Vec<u8>>> {
1161 let (tx, mut rx) = tokio::sync::mpsc::channel(1);
1162 let read = async {
1163 let mut sink = LogSink::new(tx);
1164 plugin.read_log(run, node_id, &mut sink).await
1165 };
1166 let collect = async {
1167 let mut chunks = Vec::new();
1168 while let Some(chunk) = rx.recv().await {
1169 chunks.push(chunk.to_vec());
1170 }
1171 chunks
1172 };
1173 let (result, chunks) = tokio::join!(read, collect);
1174 result.map(|()| chunks)
1175 }
1176
1177 async fn read_stage(
1178 plugin: &KomodoPlugin, run: RunRef<'_>, node_id: &str,
1179 ) -> PluginResult<Vec<u8>> {
1180 read_chunks(plugin, run, node_id)
1181 .await
1182 .map(|chunks| chunks.concat())
1183 }
1184
1185 fn parsed(plugin: &KomodoPlugin, text: &[u8]) -> Vec<(LineMarker, String)> {
1186 let dialect = plugin.log_dialect();
1187 text.strip_suffix(b"\n")
1188 .into_iter()
1189 .flat_map(|text| text.split(|b| *b == b'\n'))
1190 .map(|line| {
1191 let visible = line
1192 .rsplit(|b| *b == b'\r')
1193 .find(|segment| !segment.is_empty())
1194 .unwrap_or_default();
1195 let parsed = dialect.parse(visible);
1196 (
1197 parsed.marker,
1198 String::from_utf8_lossy(parsed.body).into_owned(),
1199 )
1200 })
1201 .collect()
1202 }
1203
1204 fn text(marker: LineMarker, body: &str) -> (LineMarker, String) {
1205 (marker, body.to_string())
1206 }
1207
1208 #[tokio::test]
1209 async fn finished_update_stages_become_job_nodes_read_by_update_id() {
1210 let (plugin, server_url, requests) = stub_plugin(&[]).await;
1211 let metadata = update_metadata("u005");
1212
1213 let nodes = plugin
1214 .fetch_log_nodes(log_run(&metadata, update_start(5)))
1215 .await
1216 .unwrap();
1217
1218 let summary: Vec<_> = nodes
1219 .iter()
1220 .map(|node| {
1221 (
1222 node.id.as_str(),
1223 node.name.as_str(),
1224 node.kind,
1225 node.status,
1226 node.log,
1227 )
1228 })
1229 .collect();
1230 assert_eq!(
1231 summary,
1232 [
1233 (
1234 "0",
1235 "Clone Repo",
1236 LogNodeKind::Job,
1237 PipelineStatus::Success,
1238 Some(LogAvailability::Ready)
1239 ),
1240 (
1241 "1",
1242 "Compose Up",
1243 LogNodeKind::Job,
1244 PipelineStatus::Failed,
1245 Some(LogAvailability::Ready)
1246 ),
1247 ]
1248 );
1249 let millis = |at: Option<chrono::DateTime<chrono::Utc>>| at.map(|at| at.timestamp_millis());
1250 assert_eq!(millis(nodes[1].started_at), Some(update_start(5) + 1_000));
1251 assert_eq!(millis(nodes[1].finished_at), Some(update_start(5) + 9_500));
1252 assert_eq!(
1253 nodes[1].web_url.as_deref(),
1254 Some(format!("{server_url}/updates/u005").as_str())
1255 );
1256 let requests = requests.lock().await;
1257 assert_eq!(requests.len(), 1);
1258 assert_eq!(requests[0].path, "/read");
1259 assert_eq!(
1260 requests[0].body,
1261 json!({"type": "GetUpdate", "params": {"id": "u005"}})
1262 );
1263 }
1264
1265 #[tokio::test]
1266 async fn the_last_stage_of_a_running_update_is_partial_and_open() {
1267 let (plugin, _, _) = stub_plugin(&[]).await;
1268 let metadata = update_metadata("u000");
1269
1270 let nodes = plugin
1271 .fetch_log_nodes(log_run(&metadata, NEWEST_START))
1272 .await
1273 .unwrap();
1274
1275 let summary: Vec<_> = nodes
1276 .iter()
1277 .map(|node| (node.status, node.log, node.finished_at.is_some()))
1278 .collect();
1279 assert_eq!(
1280 summary,
1281 [
1282 (PipelineStatus::Success, Some(LogAvailability::Ready), true),
1283 (
1284 PipelineStatus::Running,
1285 Some(LogAvailability::Partial),
1286 false
1287 ),
1288 ]
1289 );
1290 }
1291
1292 #[tokio::test]
1293 async fn runs_without_an_update_id_are_found_by_their_start_time() {
1294 let (plugin, _, requests) = stub_plugin(&[]).await;
1295 let without_id = HashMap::new();
1296 let empty_id = update_metadata("");
1297
1298 for metadata in [&without_id, &empty_id] {
1299 let nodes = plugin
1300 .fetch_log_nodes(log_run(metadata, update_start(5)))
1301 .await
1302 .unwrap();
1303 assert_eq!(nodes.len(), 2);
1304 }
1305 let stage = read_stage(&plugin, log_run(&without_id, update_start(1)), "0")
1306 .await
1307 .unwrap();
1308
1309 assert!(stage.starts_with(b"$ docker compose up -d\n"));
1310 let lookups = bodies(&requests, "ListUpdates").await;
1311 assert_eq!(lookups.len(), 3);
1312 assert_eq!(
1313 lookups[0]["params"]["query"],
1314 json!({"target.type": "Stack", "target.id": "stk1", "start_ts": update_start(5)})
1315 );
1316 let reads: Vec<_> = bodies(&requests, "GetUpdate")
1317 .await
1318 .iter()
1319 .map(|body| body["params"]["id"].clone())
1320 .collect();
1321 assert_eq!(reads, [json!("u005"), json!("u005"), json!("u001")]);
1322 }
1323
1324 #[tokio::test]
1325 async fn an_unmatched_or_ambiguous_start_time_fails_before_the_update_is_read() {
1326 let (plugin, _, requests) = stub_plugin(&[]).await;
1327 let metadata = HashMap::new();
1328 let ambiguous = RunRef {
1329 pipeline_id: "komodo__1__dup",
1330 run_number: 7,
1331 run_id: "komodo__1__dup__d1",
1332 metadata: &metadata,
1333 };
1334
1335 let unknown = plugin
1336 .fetch_log_nodes(log_run(&metadata, 12_345))
1337 .await
1338 .unwrap_err();
1339 let shared = plugin.fetch_log_nodes(ambiguous).await.unwrap_err();
1340
1341 assert!(
1342 matches!(unknown, PluginError::PipelineNotFound(_)),
1343 "{unknown}"
1344 );
1345 assert!(
1346 matches!(&shared, PluginError::ApiError(m) if m.contains("share run number")),
1347 "{shared}"
1348 );
1349 assert!(bodies(&requests, "GetUpdate").await.is_empty());
1350 }
1351
1352 #[tokio::test]
1353 async fn a_failed_stage_reads_as_command_then_stdout_then_stderr_errors() {
1354 let (plugin, _, _) = stub_plugin(&[]).await;
1355 let metadata = update_metadata("u005");
1356
1357 let stage = read_stage(&plugin, log_run(&metadata, update_start(5)), "1")
1358 .await
1359 .unwrap();
1360
1361 assert_eq!(
1362 String::from_utf8(stage.clone()).unwrap(),
1363 "$ docker compose up -d\n\
1364 . Pulling 10%\r. Pulling 100%\n\
1365 . Container web-app-1 Starting\n\
1366 ! Error response from daemon: port is already allocated\n\
1367 ! failed to start web\n"
1368 );
1369 assert_eq!(
1370 parsed(&plugin, &stage),
1371 [
1372 text(LineMarker::Command, "$ docker compose up -d"),
1373 text(LineMarker::Plain, "Pulling 100%"),
1374 text(LineMarker::Plain, "Container web-app-1 Starting"),
1375 text(
1376 LineMarker::Error,
1377 "Error response from daemon: port is already allocated"
1378 ),
1379 text(LineMarker::Error, "failed to start web"),
1380 ]
1381 );
1382 let first = read_stage(&plugin, log_run(&metadata, update_start(5)), "0")
1383 .await
1384 .unwrap();
1385 assert_eq!(
1386 parsed(&plugin, &first),
1387 [
1388 text(LineMarker::Command, "$ git clone acme/web"),
1389 text(LineMarker::Plain, "cloned"),
1390 ]
1391 );
1392 }
1393
1394 #[tokio::test]
1395 async fn stderr_of_a_successful_stage_reads_as_a_warning() {
1396 let (plugin, _, _) = stub_plugin(&[]).await;
1397 let metadata = update_metadata("u001");
1398
1399 let stage = read_stage(&plugin, log_run(&metadata, update_start(1)), "0")
1400 .await
1401 .unwrap();
1402
1403 assert_eq!(
1404 parsed(&plugin, &stage),
1405 [
1406 text(LineMarker::Command, "$ docker compose up -d"),
1407 text(LineMarker::Plain, "started"),
1408 text(LineMarker::Warning, "Network web_default Created"),
1409 ]
1410 );
1411 }
1412
1413 #[tokio::test]
1414 async fn long_stage_text_is_written_in_chunks_without_losing_a_line() {
1415 let (plugin, _, _) = stub_plugin(&[]).await;
1416 let metadata = update_metadata("u007");
1417
1418 let chunks = read_chunks(&plugin, log_run(&metadata, update_start(7)), "0")
1419 .await
1420 .unwrap();
1421
1422 assert!(chunks.len() >= 3, "{} chunks", chunks.len());
1423 assert!(chunks
1424 .iter()
1425 .all(|chunk| chunk.len() < 2 * LOG_CHUNK_BYTES && chunk.ends_with(b"\n")));
1426 let expected: String = (0..LONG_OUTPUT_LINES)
1427 .map(|line| format!(". line {line:05}\n"))
1428 .collect();
1429 assert_eq!(chunks.concat(), expected.into_bytes());
1430 }
1431
1432 #[tokio::test]
1433 async fn node_ids_that_are_not_stage_numbers_are_rejected_without_a_request() {
1434 let (plugin, _, requests) = stub_plugin(&[]).await;
1435 let metadata = update_metadata("u005");
1436
1437 let rejected = [
1438 "",
1439 "-1",
1440 "+1",
1441 "1.0",
1442 "1 ",
1443 " 1",
1444 "abc",
1445 "1/../../execute",
1446 "1?x=1",
1447 "٤",
1448 "99999999999999999999",
1449 ];
1450 for node_id in rejected {
1451 let error = read_stage(&plugin, log_run(&metadata, update_start(5)), node_id)
1452 .await
1453 .unwrap_err();
1454 assert!(
1455 matches!(error, PluginError::InvalidConfig(_)),
1456 "{node_id:?}"
1457 );
1458 }
1459 let foreign = RunRef {
1460 pipeline_id: "argocd__1__web",
1461 run_number: 1,
1462 run_id: "argocd-sync",
1463 metadata: &metadata,
1464 };
1465 assert!(matches!(
1466 plugin.fetch_log_nodes(foreign).await,
1467 Err(PluginError::InvalidConfig(_))
1468 ));
1469 assert!(requests.lock().await.is_empty());
1470 }
1471
1472 #[tokio::test]
1473 async fn a_stage_number_beyond_the_update_is_rejected_after_reading_it_once() {
1474 let (plugin, _, requests) = stub_plugin(&[]).await;
1475 let metadata = update_metadata("u005");
1476
1477 for node_id in ["2", "7", "18446744073709551615"] {
1478 let error = read_stage(&plugin, log_run(&metadata, update_start(5)), node_id)
1479 .await
1480 .unwrap_err();
1481 assert!(matches!(error, PluginError::InvalidConfig(_)), "{node_id}");
1482 }
1483
1484 assert_eq!(bodies(&requests, "GetUpdate").await.len(), 3);
1485 }
1486
1487 #[tokio::test]
1488 async fn an_update_komodo_no_longer_has_is_not_found_and_other_failures_are_not() {
1489 let (plugin, _, requests) = stub_plugin(&[]).await;
1490
1491 for update_id in ["u999", "u-gone"] {
1492 let metadata = update_metadata(update_id);
1493 let nodes = plugin
1494 .fetch_log_nodes(log_run(&metadata, 1))
1495 .await
1496 .unwrap_err();
1497 let stage = read_stage(&plugin, log_run(&metadata, 1), "0")
1498 .await
1499 .unwrap_err();
1500 for error in [nodes, stage] {
1501 assert!(
1502 matches!(error, PluginError::PipelineNotFound(_)),
1503 "{update_id}: {error}"
1504 );
1505 }
1506 }
1507 let gone_reads = bodies(&requests, "GetUpdate")
1508 .await
1509 .into_iter()
1510 .filter(|body| body["params"]["id"] == "u-gone")
1511 .count();
1512 assert_eq!(gone_reads, 2);
1513 let metadata = update_metadata("u-down");
1514 let down = plugin
1515 .fetch_log_nodes(log_run(&metadata, 1))
1516 .await
1517 .unwrap_err();
1518 assert!(
1519 matches!(down, PluginError::ServiceUnavailable { .. }),
1520 "{down}"
1521 );
1522 }
1523
1524 #[tokio::test]
1525 async fn a_busy_update_read_keeps_the_server_retry_wait() {
1526 let (plugin, _, requests) = stub_plugin(&[]).await;
1527 let metadata = update_metadata("u-busy");
1528
1529 let error = plugin
1530 .fetch_log_nodes(log_run(&metadata, 1))
1531 .await
1532 .unwrap_err();
1533
1534 assert!(
1535 matches!(error, PluginError::ServiceUnavailable { .. }),
1536 "{error}"
1537 );
1538 assert_eq!(
1539 error.retry_after(),
1540 Some(std::time::Duration::from_secs(120))
1541 );
1542 assert_eq!(bodies(&requests, "GetUpdate").await.len(), 1);
1543 }
1544
1545 #[tokio::test]
1546 async fn cancel_is_not_supported_for_stacks() {
1547 let (plugin, _, requests) = stub_plugin(&[]).await;
1548
1549 let error = plugin.cancel_run(PIPELINE, NEWEST_START).await.unwrap_err();
1550
1551 assert!(matches!(error, PluginError::NotSupported(_)), "{error}");
1552 assert!(!plugin.metadata().capabilities.cancel);
1553 assert!(requests.lock().await.is_empty());
1554 }
1555
1556 #[tokio::test]
1557 async fn foreign_pipeline_ids_are_rejected_before_any_request() {
1558 let (plugin, _, requests) = stub_plugin(&[]).await;
1559
1560 let history = plugin.fetch_run_history("argocd__1__web", 10).await;
1561 let details = plugin.fetch_run_details("komodo__x__stk1", 1).await;
1562 let trigger = plugin
1563 .trigger_pipeline(TriggerParams {
1564 workflow_id: "komodo__1".to_string(),
1565 inputs: None,
1566 })
1567 .await;
1568
1569 for result in [history.map(drop), details.map(drop), trigger.map(drop)] {
1570 assert!(matches!(result, Err(PluginError::InvalidConfig(_))));
1571 }
1572 assert!(requests.lock().await.is_empty());
1573 }
1574
1575 #[test]
1576 fn initialize_requires_url_credentials_and_a_client() {
1577 let complete = [
1578 ("server_url", "https://komodo.example.com"),
1579 ("api_key", "K"),
1580 ("token", "S"),
1581 ];
1582 let client = || Some(Arc::new(reqwest::Client::new()));
1583 let _ = rustls::crypto::ring::default_provider().install_default();
1584 let mut plugin = KomodoPlugin::new();
1585
1586 assert!(matches!(
1587 plugin.initialize(1, cfg(&complete), None),
1588 Err(PluginError::Internal(_))
1589 ));
1590 for missing in ["server_url", "api_key", "token"] {
1591 let pairs: Vec<_> = complete
1592 .iter()
1593 .copied()
1594 .filter(|(key, _)| *key != missing)
1595 .collect();
1596 assert!(
1597 matches!(
1598 plugin.initialize(1, cfg(&pairs), client()),
1599 Err(PluginError::InvalidConfig(_))
1600 ),
1601 "{missing}"
1602 );
1603 }
1604 assert!(plugin.initialize(1, cfg(&complete), client()).is_ok());
1605 }
1606
1607 #[tokio::test]
1608 async fn an_uninitialized_plugin_reports_an_internal_error() {
1609 let plugin = KomodoPlugin::new();
1610
1611 assert!(matches!(
1612 plugin.fetch_pipelines().await,
1613 Err(PluginError::Internal(_))
1614 ));
1615 assert!(matches!(
1616 plugin.validate_credentials().await,
1617 Err(PluginError::Internal(_))
1618 ));
1619 }
1620
1621 #[test]
1622 fn client_options_carry_the_origin_and_the_insecure_flag() {
1623 let plugin = KomodoPlugin::new();
1624
1625 let options = plugin.client_options(&cfg(&[
1626 ("server_url", " https://Komodo.example.com:9120/ "),
1627 ("insecure", "true"),
1628 ]));
1629 let defaults = plugin.client_options(&cfg(&[("server_url", "not a url")]));
1630
1631 assert_eq!(
1632 options.origin.as_ref().map(|url| url.as_str()),
1633 Some("https://komodo.example.com:9120/")
1634 );
1635 assert!(options.accept_invalid_certs);
1636 assert_eq!(defaults.origin, None);
1637 assert!(!defaults.accept_invalid_certs);
1638 }
1639}