1use std::collections::HashMap;
21#[cfg(feature = "exochain")]
22use std::sync::Arc;
23use std::time::Duration;
24
25use dashmap::DashMap;
26use serde::{Deserialize, Serialize};
27use tracing::debug;
28
29use async_trait::async_trait;
30
31use crate::health::HealthStatus;
32use crate::service::{ServiceType, SystemService};
33
34#[derive(Debug, Clone, Serialize, Deserialize)]
36pub struct ContainerConfig {
37 #[serde(default = "default_docker_socket")]
40 pub docker_socket: String,
41
42 #[serde(default = "default_network_name")]
45 pub network_name: String,
46
47 #[serde(default)]
49 pub default_restart_policy: RestartPolicy,
50
51 #[serde(default = "default_health_check_interval")]
53 pub health_check_interval_secs: u64,
54}
55
56fn default_docker_socket() -> String {
57 "unix:///var/run/docker.sock".into()
58}
59
60fn default_network_name() -> String {
61 "weftos".into()
62}
63
64fn default_health_check_interval() -> u64 {
65 30
66}
67
68impl Default for ContainerConfig {
69 fn default() -> Self {
70 Self {
71 docker_socket: default_docker_socket(),
72 network_name: default_network_name(),
73 default_restart_policy: RestartPolicy::default(),
74 health_check_interval_secs: default_health_check_interval(),
75 }
76 }
77}
78
79impl ContainerConfig {
80 pub fn health_check_interval(&self) -> Duration {
82 Duration::from_secs(self.health_check_interval_secs)
83 }
84}
85
86#[non_exhaustive]
88#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
89pub enum ContainerState {
90 Pulling,
92 Creating,
94 Running,
96 Stopping,
98 Stopped,
100 Failed(String),
102}
103
104impl std::fmt::Display for ContainerState {
105 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
106 match self {
107 ContainerState::Pulling => write!(f, "pulling"),
108 ContainerState::Creating => write!(f, "creating"),
109 ContainerState::Running => write!(f, "running"),
110 ContainerState::Stopping => write!(f, "stopping"),
111 ContainerState::Stopped => write!(f, "stopped"),
112 ContainerState::Failed(reason) => write!(f, "failed: {reason}"),
113 }
114 }
115}
116
117#[derive(Debug, Clone, Serialize, Deserialize)]
119pub struct PortMapping {
120 pub host_port: u16,
122 pub container_port: u16,
124 #[serde(default = "default_protocol")]
126 pub protocol: String,
127}
128
129fn default_protocol() -> String {
130 "tcp".into()
131}
132
133#[derive(Debug, Clone, Serialize, Deserialize)]
135pub struct VolumeMount {
136 pub host_path: String,
138 pub container_path: String,
140 #[serde(default)]
142 pub read_only: bool,
143}
144
145#[non_exhaustive]
147#[derive(Debug, Clone, Default, Serialize, Deserialize, PartialEq, Eq)]
148pub enum RestartPolicy {
149 #[default]
151 Never,
152 OnFailure {
154 max_retries: u32,
156 },
157 Always,
159}
160
161#[derive(Debug, Clone, Serialize, Deserialize)]
163pub struct ManagedContainer {
164 pub name: String,
166
167 pub image: String,
169
170 #[serde(default, skip_serializing_if = "Option::is_none")]
172 pub container_id: Option<String>,
173
174 #[serde(default = "default_container_state")]
176 pub state: ContainerState,
177
178 #[serde(default)]
180 pub ports: Vec<PortMapping>,
181
182 #[serde(default)]
184 pub env: HashMap<String, String>,
185
186 #[serde(default)]
188 pub volumes: Vec<VolumeMount>,
189
190 #[serde(default, skip_serializing_if = "Option::is_none")]
192 pub health_endpoint: Option<String>,
193
194 #[serde(default, skip_serializing_if = "Option::is_none")]
196 pub restart_policy: Option<RestartPolicy>,
197}
198
199fn default_container_state() -> ContainerState {
200 ContainerState::Stopped
201}
202
203#[derive(Debug, Clone, Serialize, Deserialize)]
205pub struct ContainerHealth {
206 pub container_id: String,
208 pub status: ContainerState,
210 pub healthy: bool,
212 pub message: Option<String>,
214}
215
216#[non_exhaustive]
218#[derive(Debug, thiserror::Error)]
219pub enum ContainerError {
220 #[error("Docker not available: {0}")]
222 DockerNotAvailable(String),
223
224 #[error("image pull failed for '{image}': {reason}")]
226 ImagePullFailed {
227 image: String,
229 reason: String,
231 },
232
233 #[error("container creation failed for '{name}': {reason}")]
235 CreateFailed {
236 name: String,
238 reason: String,
240 },
241
242 #[error("container start failed for '{name}': {reason}")]
244 StartFailed {
245 name: String,
247 reason: String,
249 },
250
251 #[error("port conflict: host port {port} already in use")]
253 PortConflict {
254 port: u16,
256 },
257
258 #[error("container not found: '{name}'")]
260 ContainerNotFound {
261 name: String,
263 },
264
265 #[error("health check failed for '{name}': {reason}")]
267 HealthCheckFailed {
268 name: String,
270 reason: String,
272 },
273
274 #[error("invalid config: {0}")]
276 InvalidConfig(String),
277}
278
279pub const COG_IMAGE_PREFIX: &str = "weftos-cog/";
283
284pub struct ContainerManager {
290 config: ContainerConfig,
291 managed: DashMap<String, ManagedContainer>,
292 simulated: bool,
295 #[cfg(feature = "exochain")]
296 chain_manager: Option<Arc<crate::chain::ChainManager>>,
297}
298
299impl ContainerManager {
300 pub fn new(config: ContainerConfig) -> Self {
305 Self {
306 config,
307 managed: DashMap::new(),
308 simulated: false,
309 #[cfg(feature = "exochain")]
310 chain_manager: None,
311 }
312 }
313
314 pub fn simulated(config: ContainerConfig) -> Self {
318 Self {
319 simulated: true,
320 ..Self::new(config)
321 }
322 }
323
324 pub fn is_simulated(&self) -> bool {
326 self.simulated
327 }
328
329 #[cfg(feature = "exochain")]
331 pub fn set_chain_manager(&mut self, cm: Arc<crate::chain::ChainManager>) {
332 self.chain_manager = Some(cm);
333 }
334
335 pub fn config(&self) -> &ContainerConfig {
337 &self.config
338 }
339
340 pub fn configure(&self, spec: ManagedContainer) -> Result<String, ContainerError> {
351 if spec.image.trim().is_empty() {
353 return Err(ContainerError::InvalidConfig(
354 "image name must not be empty".into(),
355 ));
356 }
357 if spec.name.trim().is_empty() {
359 return Err(ContainerError::InvalidConfig(
360 "container name must not be empty".into(),
361 ));
362 }
363 for pm in &spec.ports {
365 if pm.host_port == 0 {
366 return Err(ContainerError::InvalidConfig(
367 "host port must be > 0".into(),
368 ));
369 }
370 if pm.container_port == 0 {
371 return Err(ContainerError::InvalidConfig(
372 "container port must be > 0".into(),
373 ));
374 }
375 }
376 let name = spec.name.clone();
377 debug!(name = %spec.name, image = %spec.image, "configuring container");
378
379 #[cfg(feature = "exochain")]
380 if let Some(ref cm) = self.chain_manager {
381 cm.append(
382 "container",
383 crate::chain::EVENT_KIND_CONTAINER_CONFIGURE,
384 Some(serde_json::json!({
385 "name": &name,
386 "image": &spec.image,
387 "ports": spec.ports.len(),
388 })),
389 );
390 }
391
392 self.managed.insert(spec.name.clone(), spec);
393 Ok(name)
394 }
395
396 pub fn register(&self, spec: ManagedContainer) {
401 debug!(name = %spec.name, image = %spec.image, "registering container");
402 self.managed.insert(spec.name.clone(), spec);
403 }
404
405 pub fn start_container(&self, name: &str) -> Result<(), ContainerError> {
424 let mut entry =
425 self.managed
426 .get_mut(name)
427 .ok_or_else(|| ContainerError::ContainerNotFound {
428 name: name.to_owned(),
429 })?;
430
431 if entry.image.starts_with(COG_IMAGE_PREFIX) {
434 return Err(ContainerError::StartFailed {
435 name: name.to_owned(),
436 reason: "cog workloads start through workload_runtime::WorkloadHost \
437 (governed, chained engine adapters), not the simulated manager"
438 .into(),
439 });
440 }
441 if !self.simulated {
444 return Err(ContainerError::DockerNotAvailable(format!(
445 "{name}: ContainerManager has no container engine; workloads run through \
446 workload_runtime::WorkloadHost engine adapters"
447 )));
448 }
449
450 match &entry.state {
451 ContainerState::Stopped | ContainerState::Creating | ContainerState::Failed(_) => {
452 debug!(name, "starting container (simulated)");
453 entry.state = ContainerState::Running;
455 if entry.container_id.is_none() {
457 use std::collections::hash_map::DefaultHasher;
458 use std::hash::{Hash, Hasher};
459 let mut h = DefaultHasher::new();
460 name.hash(&mut h);
461 entry.container_id = Some(format!("sim-{:08x}", h.finish() as u32));
462 }
463
464 #[cfg(feature = "exochain")]
465 if let Some(ref cm) = self.chain_manager {
466 cm.append(
467 "container",
468 crate::chain::EVENT_KIND_CONTAINER_START,
469 Some(serde_json::json!({
470 "name": name,
471 "image": &entry.image,
472 })),
473 );
474 }
475
476 Ok(())
477 }
478 ContainerState::Running => {
479 Ok(())
481 }
482 other => Err(ContainerError::StartFailed {
483 name: name.to_owned(),
484 reason: format!("cannot start from state: {other}"),
485 }),
486 }
487 }
488
489 pub fn stop_container(&self, name: &str) -> Result<(), ContainerError> {
493 let mut entry =
494 self.managed
495 .get_mut(name)
496 .ok_or_else(|| ContainerError::ContainerNotFound {
497 name: name.to_owned(),
498 })?;
499
500 debug!(name, "stopping container");
501 entry.state = ContainerState::Stopped;
502
503 #[cfg(feature = "exochain")]
504 if let Some(ref cm) = self.chain_manager {
505 cm.append(
506 "container",
507 crate::chain::EVENT_KIND_CONTAINER_STOP,
508 Some(serde_json::json!({
509 "name": name,
510 })),
511 );
512 }
513
514 Ok(())
515 }
516
517 pub fn container_state(&self, name: &str) -> Option<ContainerState> {
519 self.managed.get(name).map(|e| e.state.clone())
520 }
521
522 pub fn list_containers(&self) -> Vec<(String, ContainerState)> {
524 self.managed
525 .iter()
526 .map(|e| (e.key().clone(), e.value().state.clone()))
527 .collect()
528 }
529
530 pub fn health_check(&self, name: &str) -> Result<HealthStatus, ContainerError> {
532 let entry = self
533 .managed
534 .get(name)
535 .ok_or_else(|| ContainerError::ContainerNotFound {
536 name: name.to_owned(),
537 })?;
538
539 match &entry.state {
540 ContainerState::Running => Ok(HealthStatus::Healthy),
541 ContainerState::Stopped => Ok(HealthStatus::Unhealthy("stopped".into())),
542 ContainerState::Failed(reason) => {
543 Ok(HealthStatus::Unhealthy(format!("failed: {reason}")))
544 }
545 other => Ok(HealthStatus::Degraded(format!("state: {other}"))),
546 }
547 }
548
549 pub fn container_health(&self, name: &str) -> Result<ContainerHealth, ContainerError> {
551 let entry = self
552 .managed
553 .get(name)
554 .ok_or_else(|| ContainerError::ContainerNotFound {
555 name: name.to_owned(),
556 })?;
557
558 let (healthy, message) = match &entry.state {
559 ContainerState::Running => (true, None),
560 ContainerState::Stopped => (false, Some("container is stopped".into())),
561 ContainerState::Failed(reason) => (false, Some(format!("failed: {reason}"))),
562 other => (false, Some(format!("transitional state: {other}"))),
563 };
564
565 Ok(ContainerHealth {
566 container_id: entry.name.clone(),
567 status: entry.state.clone(),
568 healthy,
569 message,
570 })
571 }
572
573 pub fn stop_all(&self) {
575 for mut entry in self.managed.iter_mut() {
576 if matches!(entry.state, ContainerState::Running) {
577 debug!(name = %entry.key(), "stopping container");
578 entry.state = ContainerState::Stopped;
579 }
580 }
581 }
582
583 pub fn len(&self) -> usize {
585 self.managed.len()
586 }
587
588 pub fn is_empty(&self) -> bool {
590 self.managed.is_empty()
591 }
592}
593
594pub struct ContainerService {
601 manager: std::sync::Arc<ContainerManager>,
602}
603
604impl ContainerService {
605 pub fn new(manager: std::sync::Arc<ContainerManager>) -> Self {
607 Self { manager }
608 }
609
610 pub fn manager(&self) -> &std::sync::Arc<ContainerManager> {
612 &self.manager
613 }
614}
615
616#[async_trait]
617impl SystemService for ContainerService {
618 fn name(&self) -> &str {
619 "containers"
620 }
621
622 fn service_type(&self) -> ServiceType {
623 ServiceType::Custom("containers".into())
624 }
625
626 async fn start(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
627 debug!(
628 "container service starting ({} managed)",
629 self.manager.len()
630 );
631 Ok(())
632 }
633
634 async fn stop(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
635 debug!("container service stopping — stopping all containers");
636 self.manager.stop_all();
637 Ok(())
638 }
639
640 async fn health_check(&self) -> HealthStatus {
641 let containers = self.manager.list_containers();
642 if containers.is_empty() {
643 return HealthStatus::Healthy;
644 }
645 let mut unhealthy = Vec::new();
646 for (name, state) in &containers {
647 if !matches!(state, ContainerState::Running) {
648 unhealthy.push(format!("{name}: {state}"));
649 }
650 }
651 if unhealthy.is_empty() {
652 HealthStatus::Healthy
653 } else if unhealthy.len() == containers.len() {
654 HealthStatus::Unhealthy(format!("all containers down: {}", unhealthy.join(", ")))
655 } else {
656 HealthStatus::Degraded(format!(
657 "{}/{} unhealthy: {}",
658 unhealthy.len(),
659 containers.len(),
660 unhealthy.join(", ")
661 ))
662 }
663 }
664}
665
666#[cfg(test)]
667mod tests {
668 use super::*;
669
670 #[test]
671 fn default_config() {
672 let config = ContainerConfig::default();
673 assert!(config.docker_socket.contains("docker.sock"));
674 assert_eq!(config.network_name, "weftos");
675 assert_eq!(config.default_restart_policy, RestartPolicy::Never);
676 assert_eq!(config.health_check_interval_secs, 30);
677 }
678
679 #[test]
680 fn config_serde_roundtrip() {
681 let config = ContainerConfig {
682 docker_socket: "tcp://localhost:2375".into(),
683 network_name: "custom-net".into(),
684 default_restart_policy: RestartPolicy::Always,
685 health_check_interval_secs: 10,
686 };
687 let json = serde_json::to_string(&config).unwrap();
688 let restored: ContainerConfig = serde_json::from_str(&json).unwrap();
689 assert_eq!(restored.network_name, "custom-net");
690 assert_eq!(restored.default_restart_policy, RestartPolicy::Always);
691 }
692
693 #[test]
694 fn health_check_interval_duration() {
695 let config = ContainerConfig {
696 health_check_interval_secs: 15,
697 ..Default::default()
698 };
699 assert_eq!(config.health_check_interval(), Duration::from_secs(15));
700 }
701
702 #[test]
703 fn container_state_display() {
704 assert_eq!(ContainerState::Pulling.to_string(), "pulling");
705 assert_eq!(ContainerState::Running.to_string(), "running");
706 assert_eq!(ContainerState::Stopped.to_string(), "stopped");
707 assert_eq!(
708 ContainerState::Failed("oom".into()).to_string(),
709 "failed: oom"
710 );
711 }
712
713 #[test]
714 fn engineless_manager_never_marks_any_image_running() {
715 let manager = ContainerManager::new(ContainerConfig::default());
716 assert!(!manager.is_simulated());
717 manager.register(ManagedContainer {
718 name: "cog-elsewhere".into(),
719 image: "ghcr.io/x/cog-anomaly-detect:1".into(),
720 container_id: None,
721 state: ContainerState::Stopped,
722 ports: Vec::new(),
723 env: HashMap::new(),
724 volumes: Vec::new(),
725 health_endpoint: None,
726 restart_policy: None,
727 });
728 let e = manager.start_container("cog-elsewhere").unwrap_err();
729 assert!(matches!(e, ContainerError::DockerNotAvailable(_)), "{e}");
730 let (_, state) = manager
731 .list_containers()
732 .into_iter()
733 .find(|c| c.0 == "cog-elsewhere")
734 .unwrap();
735 assert_eq!(state, ContainerState::Stopped, "never faked Running");
736 }
737
738 #[test]
739 fn simulated_start_refuses_cog_workload_images() {
740 let manager = ContainerManager::simulated(ContainerConfig::default());
741 manager.register(ManagedContainer {
742 name: "cog-anomaly-detect".into(),
743 image: format!("{COG_IMAGE_PREFIX}anomaly-detect:0123456789abcdef"),
744 container_id: None,
745 state: ContainerState::Stopped,
746 ports: Vec::new(),
747 env: HashMap::new(),
748 volumes: Vec::new(),
749 health_endpoint: None,
750 restart_policy: None,
751 });
752 let e = manager.start_container("cog-anomaly-detect").unwrap_err();
753 assert!(e.to_string().contains("WorkloadHost"), "{e}");
754 let state = manager
755 .list_containers()
756 .into_iter()
757 .find(|c| c.0 == "cog-anomaly-detect")
758 .unwrap()
759 .1;
760 assert_eq!(state, ContainerState::Stopped, "never faked Running");
761 }
762
763 #[test]
764 fn register_and_list() {
765 let manager = ContainerManager::simulated(ContainerConfig::default());
766 manager.register(ManagedContainer {
767 name: "redis".into(),
768 image: "redis:7-alpine".into(),
769 container_id: None,
770 state: ContainerState::Stopped,
771 ports: vec![PortMapping {
772 host_port: 6379,
773 container_port: 6379,
774 protocol: "tcp".into(),
775 }],
776 env: HashMap::new(),
777 volumes: Vec::new(),
778 health_endpoint: None,
779 restart_policy: None,
780 });
781
782 let containers = manager.list_containers();
783 assert_eq!(containers.len(), 1);
784 assert_eq!(containers[0].0, "redis");
785 }
786
787 #[test]
788 fn stop_container() {
789 let manager = ContainerManager::simulated(ContainerConfig::default());
790 manager.register(ManagedContainer {
791 name: "redis".into(),
792 image: "redis:7-alpine".into(),
793 container_id: None,
794 state: ContainerState::Running,
795 ports: Vec::new(),
796 env: HashMap::new(),
797 volumes: Vec::new(),
798 health_endpoint: None,
799 restart_policy: None,
800 });
801
802 manager.stop_container("redis").unwrap();
803 assert_eq!(
804 manager.container_state("redis"),
805 Some(ContainerState::Stopped)
806 );
807 }
808
809 #[test]
810 fn stop_nonexistent_fails() {
811 let manager = ContainerManager::simulated(ContainerConfig::default());
812 let result = manager.stop_container("nonexistent");
813 assert!(matches!(
814 result,
815 Err(ContainerError::ContainerNotFound { .. })
816 ));
817 }
818
819 #[test]
820 fn health_check_running() {
821 let manager = ContainerManager::simulated(ContainerConfig::default());
822 manager.register(ManagedContainer {
823 name: "redis".into(),
824 image: "redis:7-alpine".into(),
825 container_id: None,
826 state: ContainerState::Running,
827 ports: Vec::new(),
828 env: HashMap::new(),
829 volumes: Vec::new(),
830 health_endpoint: None,
831 restart_policy: None,
832 });
833
834 let health = manager.health_check("redis").unwrap();
835 assert!(matches!(health, HealthStatus::Healthy));
836 }
837
838 #[test]
839 fn health_check_stopped() {
840 let manager = ContainerManager::simulated(ContainerConfig::default());
841 manager.register(ManagedContainer {
842 name: "redis".into(),
843 image: "redis:7-alpine".into(),
844 container_id: None,
845 state: ContainerState::Stopped,
846 ports: Vec::new(),
847 env: HashMap::new(),
848 volumes: Vec::new(),
849 health_endpoint: None,
850 restart_policy: None,
851 });
852
853 let health = manager.health_check("redis").unwrap();
854 assert!(matches!(health, HealthStatus::Unhealthy(_)));
855 }
856
857 #[test]
858 fn health_check_nonexistent() {
859 let manager = ContainerManager::simulated(ContainerConfig::default());
860 assert!(manager.health_check("nope").is_err());
861 }
862
863 #[test]
864 fn stop_all() {
865 let manager = ContainerManager::simulated(ContainerConfig::default());
866 for name in &["redis", "postgres", "memcached"] {
867 manager.register(ManagedContainer {
868 name: (*name).into(),
869 image: format!("{name}:latest"),
870 container_id: None,
871 state: ContainerState::Running,
872 ports: Vec::new(),
873 env: HashMap::new(),
874 volumes: Vec::new(),
875 health_endpoint: None,
876 restart_policy: None,
877 });
878 }
879
880 manager.stop_all();
881
882 for (_, state) in manager.list_containers() {
883 assert_eq!(state, ContainerState::Stopped);
884 }
885 }
886
887 #[test]
888 fn start_container_transitions_to_running() {
889 let manager = ContainerManager::simulated(ContainerConfig::default());
890 manager.register(ManagedContainer {
891 name: "redis".into(),
892 image: "redis:7-alpine".into(),
893 container_id: None,
894 state: ContainerState::Stopped,
895 ports: Vec::new(),
896 env: HashMap::new(),
897 volumes: Vec::new(),
898 health_endpoint: None,
899 restart_policy: None,
900 });
901
902 manager.start_container("redis").unwrap();
903 assert_eq!(
904 manager.container_state("redis"),
905 Some(ContainerState::Running)
906 );
907 }
908
909 #[test]
910 fn start_container_assigns_id() {
911 let manager = ContainerManager::simulated(ContainerConfig::default());
912 manager.register(ManagedContainer {
913 name: "pg".into(),
914 image: "postgres:16".into(),
915 container_id: None,
916 state: ContainerState::Stopped,
917 ports: Vec::new(),
918 env: HashMap::new(),
919 volumes: Vec::new(),
920 health_endpoint: None,
921 restart_policy: None,
922 });
923
924 manager.start_container("pg").unwrap();
925 let entry = manager.managed.get("pg").unwrap();
926 assert!(entry.container_id.is_some());
927 assert!(entry.container_id.as_ref().unwrap().starts_with("sim-"));
928 }
929
930 #[test]
931 fn start_already_running_is_idempotent() {
932 let manager = ContainerManager::simulated(ContainerConfig::default());
933 manager.register(ManagedContainer {
934 name: "redis".into(),
935 image: "redis:7-alpine".into(),
936 container_id: None,
937 state: ContainerState::Running,
938 ports: Vec::new(),
939 env: HashMap::new(),
940 volumes: Vec::new(),
941 health_endpoint: None,
942 restart_policy: None,
943 });
944
945 manager.start_container("redis").unwrap();
947 assert_eq!(
948 manager.container_state("redis"),
949 Some(ContainerState::Running)
950 );
951 }
952
953 #[test]
954 fn managed_container_serde_roundtrip() {
955 let container = ManagedContainer {
956 name: "redis".into(),
957 image: "redis:7-alpine".into(),
958 container_id: Some("abc123".into()),
959 state: ContainerState::Running,
960 ports: vec![PortMapping {
961 host_port: 6379,
962 container_port: 6379,
963 protocol: "tcp".into(),
964 }],
965 env: HashMap::from([("REDIS_PASSWORD".into(), "secret".into())]),
966 volumes: vec![VolumeMount {
967 host_path: "/data".into(),
968 container_path: "/var/lib/redis".into(),
969 read_only: false,
970 }],
971 health_endpoint: Some("http://localhost:6379/ping".into()),
972 restart_policy: Some(RestartPolicy::OnFailure { max_retries: 3 }),
973 };
974
975 let json = serde_json::to_string(&container).unwrap();
976 let restored: ManagedContainer = serde_json::from_str(&json).unwrap();
977 assert_eq!(restored.name, "redis");
978 assert_eq!(restored.ports.len(), 1);
979 assert_eq!(restored.volumes.len(), 1);
980 assert!(!restored.volumes[0].read_only);
981 }
982
983 #[test]
984 fn container_error_display() {
985 let err = ContainerError::DockerNotAvailable("not installed".into());
986 assert!(err.to_string().contains("Docker"));
987
988 let err = ContainerError::ContainerNotFound {
989 name: "redis".into(),
990 };
991 assert!(err.to_string().contains("redis"));
992
993 let err = ContainerError::PortConflict { port: 8080 };
994 assert!(err.to_string().contains("8080"));
995 }
996
997 #[test]
998 fn restart_policy_serde() {
999 let policies = vec![
1000 RestartPolicy::Never,
1001 RestartPolicy::OnFailure { max_retries: 5 },
1002 RestartPolicy::Always,
1003 ];
1004 for policy in policies {
1005 let json = serde_json::to_string(&policy).unwrap();
1006 let restored: RestartPolicy = serde_json::from_str(&json).unwrap();
1007 assert_eq!(restored, policy);
1008 }
1009 }
1010
1011 #[test]
1014 fn container_service_implements_system_service() {
1015 let mgr = std::sync::Arc::new(ContainerManager::simulated(ContainerConfig::default()));
1016 let svc = ContainerService::new(mgr);
1017 assert_eq!(svc.name(), "containers");
1018 assert_eq!(svc.service_type(), ServiceType::Custom("containers".into()));
1019 }
1020
1021 #[tokio::test]
1022 async fn container_service_health_empty_is_healthy() {
1023 let mgr = std::sync::Arc::new(ContainerManager::simulated(ContainerConfig::default()));
1024 let svc = ContainerService::new(mgr);
1025 let health = svc.health_check().await;
1026 assert!(matches!(health, HealthStatus::Healthy));
1027 }
1028
1029 #[tokio::test]
1030 async fn container_service_health_propagates() {
1031 let mgr = std::sync::Arc::new(ContainerManager::simulated(ContainerConfig::default()));
1032 mgr.register(ManagedContainer {
1033 name: "redis".into(),
1034 image: "redis:7-alpine".into(),
1035 container_id: None,
1036 state: ContainerState::Running,
1037 ports: Vec::new(),
1038 env: HashMap::new(),
1039 volumes: Vec::new(),
1040 health_endpoint: None,
1041 restart_policy: None,
1042 });
1043 mgr.register(ManagedContainer {
1044 name: "pg".into(),
1045 image: "postgres:16".into(),
1046 container_id: None,
1047 state: ContainerState::Stopped,
1048 ports: Vec::new(),
1049 env: HashMap::new(),
1050 volumes: Vec::new(),
1051 health_endpoint: None,
1052 restart_policy: None,
1053 });
1054 let svc = ContainerService::new(mgr);
1055 let health = svc.health_check().await;
1056 assert!(matches!(health, HealthStatus::Degraded(_)));
1058 }
1059
1060 #[tokio::test]
1061 async fn container_service_stop_halts_all() {
1062 let mgr = std::sync::Arc::new(ContainerManager::simulated(ContainerConfig::default()));
1063 mgr.register(ManagedContainer {
1064 name: "redis".into(),
1065 image: "redis:7-alpine".into(),
1066 container_id: None,
1067 state: ContainerState::Running,
1068 ports: Vec::new(),
1069 env: HashMap::new(),
1070 volumes: Vec::new(),
1071 health_endpoint: None,
1072 restart_policy: None,
1073 });
1074 let svc = ContainerService::new(mgr.clone());
1075 svc.stop().await.unwrap();
1076 assert_eq!(mgr.container_state("redis"), Some(ContainerState::Stopped));
1077 }
1078
1079 #[test]
1082 fn container_config_validates() {
1083 let manager = ContainerManager::simulated(ContainerConfig::default());
1084 let spec = ManagedContainer {
1085 name: "alpine-test".into(),
1086 image: "alpine:latest".into(),
1087 container_id: None,
1088 state: ContainerState::Stopped,
1089 ports: vec![PortMapping {
1090 host_port: 8080,
1091 container_port: 80,
1092 protocol: "tcp".into(),
1093 }],
1094 env: HashMap::new(),
1095 volumes: Vec::new(),
1096 health_endpoint: None,
1097 restart_policy: None,
1098 };
1099 let id = manager.configure(spec).unwrap();
1100 assert_eq!(id, "alpine-test");
1101 assert_eq!(
1103 manager.container_state("alpine-test"),
1104 Some(ContainerState::Stopped)
1105 );
1106 }
1107
1108 #[test]
1109 fn container_invalid_config_empty_image_rejected() {
1110 let manager = ContainerManager::simulated(ContainerConfig::default());
1111 let spec = ManagedContainer {
1112 name: "bad".into(),
1113 image: "".into(),
1114 container_id: None,
1115 state: ContainerState::Stopped,
1116 ports: Vec::new(),
1117 env: HashMap::new(),
1118 volumes: Vec::new(),
1119 health_endpoint: None,
1120 restart_policy: None,
1121 };
1122 let result = manager.configure(spec);
1123 assert!(matches!(result, Err(ContainerError::InvalidConfig(_))));
1124 }
1125
1126 #[test]
1127 fn container_invalid_config_empty_name_rejected() {
1128 let manager = ContainerManager::simulated(ContainerConfig::default());
1129 let spec = ManagedContainer {
1130 name: "".into(),
1131 image: "alpine:latest".into(),
1132 container_id: None,
1133 state: ContainerState::Stopped,
1134 ports: Vec::new(),
1135 env: HashMap::new(),
1136 volumes: Vec::new(),
1137 health_endpoint: None,
1138 restart_policy: None,
1139 };
1140 let result = manager.configure(spec);
1141 assert!(matches!(result, Err(ContainerError::InvalidConfig(_))));
1142 }
1143
1144 #[test]
1145 fn container_invalid_config_zero_port_rejected() {
1146 let manager = ContainerManager::simulated(ContainerConfig::default());
1147 let spec = ManagedContainer {
1148 name: "bad-port".into(),
1149 image: "alpine:latest".into(),
1150 container_id: None,
1151 state: ContainerState::Stopped,
1152 ports: vec![PortMapping {
1153 host_port: 0,
1154 container_port: 80,
1155 protocol: "tcp".into(),
1156 }],
1157 env: HashMap::new(),
1158 volumes: Vec::new(),
1159 health_endpoint: None,
1160 restart_policy: None,
1161 };
1162 let result = manager.configure(spec);
1163 assert!(matches!(result, Err(ContainerError::InvalidConfig(_))));
1164 }
1165
1166 #[test]
1167 fn container_lifecycle_configure_start_stop() {
1168 let manager = ContainerManager::simulated(ContainerConfig::default());
1169
1170 let spec = ManagedContainer {
1172 name: "lifecycle-test".into(),
1173 image: "redis:7-alpine".into(),
1174 container_id: None,
1175 state: ContainerState::Stopped,
1176 ports: Vec::new(),
1177 env: HashMap::new(),
1178 volumes: Vec::new(),
1179 health_endpoint: None,
1180 restart_policy: None,
1181 };
1182 let name = manager.configure(spec).unwrap();
1183
1184 manager.start_container(&name).unwrap();
1186 assert_eq!(
1187 manager.container_state(&name),
1188 Some(ContainerState::Running)
1189 );
1190
1191 let health = manager.health_check(&name).unwrap();
1193 assert_eq!(health, HealthStatus::Healthy);
1194
1195 manager.stop_container(&name).unwrap();
1197 assert_eq!(
1198 manager.container_state(&name),
1199 Some(ContainerState::Stopped)
1200 );
1201
1202 let health = manager.health_check(&name).unwrap();
1204 assert!(matches!(health, HealthStatus::Unhealthy(_)));
1205 }
1206
1207 #[test]
1208 fn container_health_report_detail() {
1209 let manager = ContainerManager::simulated(ContainerConfig::default());
1210 manager.register(ManagedContainer {
1211 name: "detail".into(),
1212 image: "alpine:latest".into(),
1213 container_id: None,
1214 state: ContainerState::Running,
1215 ports: Vec::new(),
1216 env: HashMap::new(),
1217 volumes: Vec::new(),
1218 health_endpoint: None,
1219 restart_policy: None,
1220 });
1221
1222 let report = manager.container_health("detail").unwrap();
1223 assert!(report.healthy);
1224 assert_eq!(report.status, ContainerState::Running);
1225 assert!(report.message.is_none());
1226
1227 manager.stop_container("detail").unwrap();
1229 let report = manager.container_health("detail").unwrap();
1230 assert!(!report.healthy);
1231 assert_eq!(report.status, ContainerState::Stopped);
1232 assert!(report.message.is_some());
1233 }
1234
1235 #[tokio::test]
1236 async fn container_health_propagates_to_kernel_health_system() {
1237 use crate::health::HealthSystem;
1238 use crate::service::ServiceRegistry;
1239
1240 let mgr = std::sync::Arc::new(ContainerManager::simulated(ContainerConfig::default()));
1241
1242 let spec = ManagedContainer {
1244 name: "redis".into(),
1245 image: "redis:7-alpine".into(),
1246 container_id: None,
1247 state: ContainerState::Stopped,
1248 ports: Vec::new(),
1249 env: HashMap::new(),
1250 volumes: Vec::new(),
1251 health_endpoint: None,
1252 restart_policy: None,
1253 };
1254 mgr.configure(spec).unwrap();
1255 mgr.start_container("redis").unwrap();
1256
1257 let svc = std::sync::Arc::new(ContainerService::new(mgr.clone()));
1259 let registry = std::sync::Arc::new(ServiceRegistry::new());
1260 registry.register(svc).unwrap();
1261
1262 let hs = HealthSystem::new(30);
1264 let (overall, results) = hs.aggregate(®istry).await;
1265 assert!(
1266 matches!(overall, crate::health::OverallHealth::Healthy),
1267 "expected Healthy, got {overall:?}"
1268 );
1269 assert_eq!(results.len(), 1);
1270 assert_eq!(results[0].0, "containers");
1271 assert_eq!(results[0].1, HealthStatus::Healthy);
1272
1273 mgr.stop_container("redis").unwrap();
1275 let (overall, _) = hs.aggregate(®istry).await;
1276 assert!(
1277 matches!(overall, crate::health::OverallHealth::Down),
1278 "expected Down after stopping all containers, got {overall:?}"
1279 );
1280 }
1281
1282 #[test]
1285 fn container_state_serde_roundtrip_all_variants() {
1286 let variants = vec![
1287 ContainerState::Pulling,
1288 ContainerState::Creating,
1289 ContainerState::Running,
1290 ContainerState::Stopping,
1291 ContainerState::Stopped,
1292 ContainerState::Failed("oom killed".into()),
1293 ];
1294 for state in variants {
1295 let json = serde_json::to_string(&state).unwrap();
1296 let restored: ContainerState = serde_json::from_str(&json).unwrap();
1297 assert_eq!(restored, state);
1298 }
1299 }
1300
1301 #[test]
1302 fn port_mapping_serde_roundtrip() {
1303 let pm = PortMapping {
1304 host_port: 8080,
1305 container_port: 80,
1306 protocol: "tcp".into(),
1307 };
1308 let json = serde_json::to_string(&pm).unwrap();
1309 let restored: PortMapping = serde_json::from_str(&json).unwrap();
1310 assert_eq!(restored.host_port, 8080);
1311 assert_eq!(restored.container_port, 80);
1312 assert_eq!(restored.protocol, "tcp");
1313 }
1314
1315 #[test]
1316 fn port_mapping_default_protocol() {
1317 let json = r#"{"host_port": 3000, "container_port": 3000}"#;
1318 let pm: PortMapping = serde_json::from_str(json).unwrap();
1319 assert_eq!(pm.protocol, "tcp");
1320 }
1321
1322 #[test]
1323 fn volume_mount_serde_roundtrip() {
1324 let vm = VolumeMount {
1325 host_path: "/data".into(),
1326 container_path: "/var/data".into(),
1327 read_only: true,
1328 };
1329 let json = serde_json::to_string(&vm).unwrap();
1330 let restored: VolumeMount = serde_json::from_str(&json).unwrap();
1331 assert_eq!(restored.host_path, "/data");
1332 assert_eq!(restored.container_path, "/var/data");
1333 assert!(restored.read_only);
1334 }
1335
1336 #[test]
1337 fn volume_mount_default_read_only() {
1338 let json = r#"{"host_path": "/a", "container_path": "/b"}"#;
1339 let vm: VolumeMount = serde_json::from_str(json).unwrap();
1340 assert!(!vm.read_only);
1341 }
1342
1343 #[test]
1344 fn restart_policy_serde_roundtrip_all_variants() {
1345 let variants = vec![
1346 RestartPolicy::Never,
1347 RestartPolicy::OnFailure { max_retries: 5 },
1348 RestartPolicy::Always,
1349 ];
1350 for policy in variants {
1351 let json = serde_json::to_string(&policy).unwrap();
1352 let restored: RestartPolicy = serde_json::from_str(&json).unwrap();
1353 assert_eq!(restored, policy);
1354 }
1355 }
1356
1357 #[test]
1358 fn restart_policy_default_is_never() {
1359 assert_eq!(RestartPolicy::default(), RestartPolicy::Never);
1360 }
1361
1362 #[test]
1363 fn container_health_serde_roundtrip() {
1364 let health = ContainerHealth {
1365 container_id: "redis-1".into(),
1366 status: ContainerState::Running,
1367 healthy: true,
1368 message: None,
1369 };
1370 let json = serde_json::to_string(&health).unwrap();
1371 let restored: ContainerHealth = serde_json::from_str(&json).unwrap();
1372 assert_eq!(restored.container_id, "redis-1");
1373 assert!(restored.healthy);
1374 assert!(restored.message.is_none());
1375 }
1376
1377 #[test]
1378 fn container_health_with_message_roundtrip() {
1379 let health = ContainerHealth {
1380 container_id: "pg-1".into(),
1381 status: ContainerState::Failed("timeout".into()),
1382 healthy: false,
1383 message: Some("health check failed after 30s".into()),
1384 };
1385 let json = serde_json::to_string(&health).unwrap();
1386 let restored: ContainerHealth = serde_json::from_str(&json).unwrap();
1387 assert!(!restored.healthy);
1388 assert_eq!(restored.message.unwrap(), "health check failed after 30s");
1389 }
1390
1391 #[test]
1392 fn container_state_display_all_variants() {
1393 assert_eq!(ContainerState::Pulling.to_string(), "pulling");
1394 assert_eq!(ContainerState::Creating.to_string(), "creating");
1395 assert_eq!(ContainerState::Running.to_string(), "running");
1396 assert_eq!(ContainerState::Stopping.to_string(), "stopping");
1397 assert_eq!(ContainerState::Stopped.to_string(), "stopped");
1398 assert_eq!(
1399 ContainerState::Failed("oom".into()).to_string(),
1400 "failed: oom"
1401 );
1402 }
1403
1404 #[test]
1405 fn container_config_health_check_interval() {
1406 let cfg = ContainerConfig {
1407 health_check_interval_secs: 10,
1408 ..Default::default()
1409 };
1410 assert_eq!(cfg.health_check_interval(), Duration::from_secs(10));
1411 }
1412
1413 #[test]
1414 fn container_config_defaults_populated() {
1415 let cfg = ContainerConfig::default();
1416 assert_eq!(cfg.docker_socket, "unix:///var/run/docker.sock");
1417 assert_eq!(cfg.network_name, "weftos");
1418 assert_eq!(cfg.default_restart_policy, RestartPolicy::Never);
1419 assert_eq!(cfg.health_check_interval_secs, 30);
1420 }
1421
1422 #[test]
1423 fn managed_container_with_env_and_volumes_roundtrip() {
1424 let mut env = HashMap::new();
1425 env.insert("REDIS_URL".into(), "redis://localhost".into());
1426 env.insert("LOG_LEVEL".into(), "debug".into());
1427
1428 let mc = ManagedContainer {
1429 name: "full-spec".into(),
1430 image: "redis:7".into(),
1431 container_id: Some("abc123".into()),
1432 state: ContainerState::Running,
1433 ports: vec![PortMapping {
1434 host_port: 6379,
1435 container_port: 6379,
1436 protocol: "tcp".into(),
1437 }],
1438 env,
1439 volumes: vec![VolumeMount {
1440 host_path: "/data/redis".into(),
1441 container_path: "/data".into(),
1442 read_only: false,
1443 }],
1444 health_endpoint: Some("http://localhost:6379/ping".into()),
1445 restart_policy: Some(RestartPolicy::Always),
1446 };
1447
1448 let json = serde_json::to_string(&mc).unwrap();
1449 let restored: ManagedContainer = serde_json::from_str(&json).unwrap();
1450 assert_eq!(restored.name, "full-spec");
1451 assert_eq!(restored.container_id, Some("abc123".into()));
1452 assert_eq!(restored.ports.len(), 1);
1453 assert_eq!(restored.env.len(), 2);
1454 assert_eq!(restored.volumes.len(), 1);
1455 assert_eq!(restored.restart_policy, Some(RestartPolicy::Always));
1456 }
1457
1458 #[test]
1459 fn configure_multiple_containers_succeeds() {
1460 let manager = ContainerManager::simulated(ContainerConfig::default());
1461 let spec1 = ManagedContainer {
1462 name: "svc-a".into(),
1463 image: "alpine:latest".into(),
1464 container_id: None,
1465 state: ContainerState::Stopped,
1466 ports: vec![PortMapping {
1467 host_port: 8080,
1468 container_port: 80,
1469 protocol: "tcp".into(),
1470 }],
1471 env: HashMap::new(),
1472 volumes: Vec::new(),
1473 health_endpoint: None,
1474 restart_policy: None,
1475 };
1476 manager.configure(spec1).unwrap();
1477
1478 let spec2 = ManagedContainer {
1479 name: "svc-b".into(),
1480 image: "nginx:latest".into(),
1481 container_id: None,
1482 state: ContainerState::Stopped,
1483 ports: vec![PortMapping {
1484 host_port: 9090,
1485 container_port: 80,
1486 protocol: "tcp".into(),
1487 }],
1488 env: HashMap::new(),
1489 volumes: Vec::new(),
1490 health_endpoint: None,
1491 restart_policy: None,
1492 };
1493 manager.configure(spec2).unwrap();
1494
1495 assert_eq!(
1496 manager.container_state("svc-a"),
1497 Some(ContainerState::Stopped)
1498 );
1499 assert_eq!(
1500 manager.container_state("svc-b"),
1501 Some(ContainerState::Stopped)
1502 );
1503 }
1504}