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