harmony/fleet/harmony-fleet-e2e/tests/operator.rs

use harmony_fleet_deploy::operator::SERVICE_ACCOUNT;
use harmony_fleet_e2e::{StackOptions, shared_stack};
use harmony_fleet_operator::crd::{
    Deployment, DeploymentSpec, Device, DeviceSpec, PodmanService, PodmanV0Score, ReconcileScore,
    Rollout, RolloutStrategy,
};
use harmony_reconciler_contracts::{BUCKET_DESIRED_STATE, DeploymentName, desired_state_key};
use k8s_openapi::api::authorization::v1::{
    ResourceAttributes, SubjectAccessReview, SubjectAccessReviewSpec,
};
use k8s_openapi::api::core::v1::Namespace;
use k8s_openapi::apimachinery::pkg::apis::meta::v1::LabelSelector;
use kube::Client;
use kube::api::{Api, DeleteParams, ObjectMeta, Patch, PatchParams, PostParams};
use std::sync::Arc;
use std::time::{Duration, Instant};

const E2E_ENV: &str = "HARMONY_FLEET_E2E";

fn e2e_enabled() -> bool {
    matches!(std::env::var(E2E_ENV).as_deref(), Ok("1" | "true"))
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn operator_adds_finalizer_to_fleet_deployment() -> anyhow::Result<()> {
    if !e2e_enabled() {
        skip_e2e();
        return Ok(());
    }

    let stack = operator_stack().await?;
    let deployments = fleet_deployments(&stack.namespace).await?;

    create_fleet_deployment(&deployments, "finalizer-test").await?;
    wait_for_finalizer(&deployments, "finalizer-test").await?;

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn operator_writes_desired_state_for_matching_device() -> anyhow::Result<()> {
    if !e2e_enabled() {
        skip_e2e();
        return Ok(());
    }

    let stack = operator_stack().await?;

    let client = Client::try_default().await?;
    let devices: Api<Device> = Api::namespaced(client.clone(), &stack.namespace);
    let deployments: Api<Deployment> = Api::namespaced(client, &stack.namespace);

    create_device(&devices, "desired-state-device").await?;
    create_fleet_deployment(&deployments, "desired-state-test").await?;
    wait_for_desired_state_entry(&stack, "desired-state-device", "desired-state-test", true)
        .await?;

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn operator_deletes_desired_state_when_deployment_is_deleted() -> anyhow::Result<()> {
    if !e2e_enabled() {
        skip_e2e();
        return Ok(());
    }

    let stack = operator_stack().await?;

    let client = Client::try_default().await?;
    let devices: Api<Device> = Api::namespaced(client.clone(), &stack.namespace);
    let deployments: Api<Deployment> = Api::namespaced(client, &stack.namespace);

    create_device(&devices, "cleanup-device").await?;
    create_fleet_deployment(&deployments, "cleanup-test").await?;
    wait_for_finalizer(&deployments, "cleanup-test").await?;
    wait_for_desired_state_entry(&stack, "cleanup-device", "cleanup-test", true).await?;

    deployments
        .delete("cleanup-test", &DeleteParams::default())
        .await?;

    wait_for_desired_state_entry(&stack, "cleanup-device", "cleanup-test", false).await?;

    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn operator_deletes_desired_state_when_selector_stops_matching() -> anyhow::Result<()> {
    if !e2e_enabled() {
        skip_e2e();
        return Ok(());
    }

    let stack = operator_stack().await?;
    let client = Client::try_default().await?;
    let devices: Api<Device> = Api::namespaced(client.clone(), &stack.namespace);
    let deployments: Api<Deployment> = Api::namespaced(client, &stack.namespace);

    create_device(&devices, "retarget-device").await?;
    create_fleet_deployment(&deployments, "retarget-test").await?;
    wait_for_desired_state_entry(&stack, "retarget-device", "retarget-test", true).await?;

    deployments
        .patch(
            "retarget-test",
            &PatchParams::default(),
            &Patch::Merge(serde_json::json!({
                "spec": {
                    "targetSelector": {
                        "matchLabels": { "device-id": "missing" }
                    }
                }
            })),
        )
        .await?;

    wait_for_desired_state_entry(&stack, "retarget-device", "retarget-test", false).await?;
    Ok(())
}

#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
async fn operator_ignores_other_tenant_namespaces() -> anyhow::Result<()> {
    if !e2e_enabled() {
        skip_e2e();
        return Ok(());
    }

    let stack = operator_stack().await?;
    let client = Client::try_default().await?;
    let namespace = format!(
        "e2e-isolation-{}",
        &uuid::Uuid::new_v4().simple().to_string()[..8]
    );
    let namespaces: Api<Namespace> = Api::all(client.clone());
    namespaces
        .create(
            &PostParams::default(),
            &Namespace {
                metadata: ObjectMeta {
                    name: Some(namespace.clone()),
                    labels: Some(std::collections::BTreeMap::from([(
                        "harmony.io/managed-by".to_string(),
                        "fleet-e2e".to_string(),
                    )])),
                    ..Default::default()
                },
                ..Default::default()
            },
        )
        .await?;

    let devices: Api<Device> = Api::namespaced(client.clone(), &namespace);
    let deployments: Api<Deployment> = Api::namespaced(client.clone(), &namespace);
    create_device(&devices, "other-tenant-device").await?;
    create_fleet_deployment(&deployments, "other-tenant-deployment").await?;

    let sentinel = format!(
        "isolation-sentinel-{}",
        &uuid::Uuid::new_v4().simple().to_string()[..8]
    );
    let tenant_devices: Api<Device> = Api::namespaced(client.clone(), &stack.namespace);
    let tenant_deployments: Api<Deployment> = Api::namespaced(client.clone(), &stack.namespace);
    create_device(&tenant_devices, &sentinel).await?;
    create_fleet_deployment(&tenant_deployments, &sentinel).await?;
    wait_for_desired_state_entry(&stack, &sentinel, &sentinel, true).await?;

    let deployment = deployments.get("other-tenant-deployment").await?;
    assert!(
        deployment
            .metadata
            .finalizers
            .unwrap_or_default()
            .is_empty()
    );
    wait_for_desired_state_entry(
        &stack,
        "other-tenant-device",
        "other-tenant-deployment",
        false,
    )
    .await?;

    let reviews: Api<SubjectAccessReview> = Api::all(client);
    let review = reviews
        .create(
            &PostParams::default(),
            &SubjectAccessReview {
                metadata: ObjectMeta::default(),
                spec: SubjectAccessReviewSpec {
                    resource_attributes: Some(ResourceAttributes {
                        group: Some("fleet.nationtech.io".to_string()),
                        namespace: Some(namespace.clone()),
                        resource: Some("devices".to_string()),
                        verb: Some("list".to_string()),
                        ..Default::default()
                    }),
                    user: Some(format!(
                        "system:serviceaccount:{}:{SERVICE_ACCOUNT}",
                        stack.namespace
                    )),
                    ..Default::default()
                },
                status: None,
            },
        )
        .await?;
    assert!(!review.status.is_some_and(|status| status.allowed));
    tenant_deployments
        .delete(&sentinel, &DeleteParams::default())
        .await?;
    tenant_devices
        .delete(&sentinel, &DeleteParams::default())
        .await?;
    namespaces
        .delete(&namespace, &DeleteParams::default())
        .await?;

    Ok(())
}

async fn operator_stack() -> anyhow::Result<Arc<harmony_fleet_e2e::Stack>> {
    let _ = tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .try_init();

    let stack = shared_stack(StackOptions {
        deploy_agent: false,
        deploy_operator: true,
        ..StackOptions::default()
    })
    .await?;

    stack.print_debug_info();
    assert!(stack.device_ids.is_empty());

    Ok(stack)
}

fn skip_e2e() {
    eprintln!(
        "skipping {E2E_ENV}-gated e2e test (set {E2E_ENV}=1 to run; \
         requires k3d + podman on PATH)"
    );
}

async fn fleet_deployments(namespace: &str) -> anyhow::Result<Api<Deployment>> {
    let client = Client::try_default().await?;
    Ok(Api::namespaced(client, namespace))
}

async fn create_device(devices: &Api<Device>, name: &str) -> anyhow::Result<()> {
    let device = Device::new(
        name,
        DeviceSpec {
            inventory: None,
            agent_upgrade: None,
        },
    );
    devices.create(&PostParams::default(), &device).await?;
    Ok(())
}

async fn create_fleet_deployment(deployments: &Api<Deployment>, name: &str) -> anyhow::Result<()> {
    let deployment = Deployment {
        metadata: ObjectMeta {
            name: Some(name.to_string()),
            ..Default::default()
        },
        spec: DeploymentSpec {
            allowed_groups: None,
            target_selector: LabelSelector::default(),
            score: ReconcileScore::PodmanV0(PodmanV0Score {
                services: vec![PodmanService {
                    name: "hello".to_string(),
                    image: "docker.io/library/hello-world:latest".to_string(),
                    ports: vec![],
                    env: vec![],
                    secret_env: vec![],
                    volumes: vec![],
                    restart_policy: Default::default(),
                }],
            }),
            rollout: Rollout {
                strategy: RolloutStrategy::Immediate,
            },
        },
        status: None,
    };

    deployments
        .create(&PostParams::default(), &deployment)
        .await?;
    Ok(())
}

async fn wait_for_finalizer(deployments: &Api<Deployment>, name: &str) -> anyhow::Result<()> {
    let deadline = Instant::now() + Duration::from_secs(30);
    loop {
        let created = deployments.get(name).await?;
        let finalizers = created.metadata.finalizers.unwrap_or_default();
        if finalizers
            .iter()
            .any(|finalizer| finalizer == "fleet.nationtech.io/finalizer")
        {
            break;
        }
        if Instant::now() >= deadline {
            anyhow::bail!(
                "timed out waiting for fleet finalizer; current finalizers: {:?}",
                finalizers
            );
        }
        tokio::time::sleep(Duration::from_secs(1)).await;
    }
    Ok(())
}

async fn wait_for_desired_state_entry(
    stack: &harmony_fleet_e2e::Stack,
    device_id: &str,
    deployment: &str,
    expect_present: bool,
) -> anyhow::Result<()> {
    let nats_client = connect_admin_nats(stack).await?;
    let js = async_nats::jetstream::new(nats_client);
    let desired_state = js.get_key_value(BUCKET_DESIRED_STATE).await?;
    let deployment_name = DeploymentName::try_new(deployment)?;
    let desired_state_key = desired_state_key(device_id, &deployment_name);

    let deadline = Instant::now() + Duration::from_secs(30);
    loop {
        let exists = desired_state.get(&desired_state_key).await?.is_some();
        if exists == expect_present {
            break;
        }
        if Instant::now() >= deadline {
            let state = if expect_present { "present" } else { "deleted" };
            anyhow::bail!(
                "timed out waiting for desired-state key {desired_state_key} to be {state}"
            );
        }
        tokio::time::sleep(Duration::from_secs(1)).await;
    }
    Ok(())
}

async fn connect_admin_nats(
    stack: &harmony_fleet_e2e::Stack,
) -> anyhow::Result<async_nats::Client> {
    async_nats::ConnectOptions::new()
        .user_and_password(stack.admin_user.clone(), stack.admin_pass.clone())
        .connect(&stack.nats_url)
        .await
        .map_err(Into::into)
}