groups.rs

// This Source Code Form is subject to the terms of the Mozilla Public
// License, v. 2.0. If a copy of the MPL was not distributed with this
// file, You can obtain one at https://mozilla.org/MPL/2.0/.
//

//! Integration tests for multicast group APIs and IP pool operations.
//!
//! Core multicast functionality tests:
//!
//! - IP pool range validation and allocation
//! - Member operations: add, remove, list, lookup by IP
//! - Instance deletion cleanup (removes multicast memberships)
//! - Automatic pool selection and default pool behavior
//! - Pool exhaustion handling
//! - Pool deletion protection (cannot delete pool with active groups)
//! - DPD(-client) integration: verifies groups are programmed on switches
//! - SSM pool and group tests:
//!   - SSM groups require sources (232/8 for IPv4, ff3x::/32 for IPv6)
//!   - ASM groups have optional source filtering
//!   - Multiple SSM groups can share the same pool

use std::net::{IpAddr, Ipv4Addr, Ipv6Addr};

use dpd_client::types as dpd_types;
use dropshot::HttpErrorResponseBody;
use dropshot::ResultsPage;
use http::{Method, StatusCode};

use crate::integration_tests::instances::{
    instance_simulate, instance_wait_for_state,
};
use nexus_test_utils::dpd_client;
use nexus_test_utils::http_testing::{
    AuthnMode, Collection, NexusRequest, RequestBuilder,
};
use nexus_test_utils::resource_helpers::{
    create_default_ip_pools, create_instance, create_project, link_ip_pool,
    object_create, object_create_error, object_delete, object_delete_error,
    object_get, object_get_error, object_put_error,
};
use nexus_test_utils_macros::nexus_test;
use nexus_types::external_api::ip_pool::{
    IpPool, IpPoolCreate, IpPoolRange, IpRange, IpVersion, Ipv4Range, Ipv6Range,
};
use nexus_types::external_api::multicast::{
    InstanceMulticastGroupJoin, MulticastGroup, MulticastGroupMember,
};
use omicron_common::api::external::{
    IdentityMetadataCreateParams, InstanceState,
};
use omicron_uuid_kinds::InstanceUuid;

use super::*;

/// Test that multicast IP pools reject invalid ranges at the pool level
#[nexus_test]
async fn test_multicast_ip_pool_range_validation(
    cptestctx: &ControlPlaneTestContext,
) {
    let client = &cptestctx.external_client;

    // Create IPv4 multicast pool
    let pool_params = IpPoolCreate::new_multicast(
        IdentityMetadataCreateParams {
            name: "test-v4-pool".parse().unwrap(),
            description: "IPv4 multicast pool for validation tests".to_string(),
        },
        IpVersion::V4,
    );
    object_create::<_, IpPool>(client, "/v1/system/ip-pools", &pool_params)
        .await;

    let range_url = "/v1/system/ip-pools/test-v4-pool/ranges/add";

    // IPv4 non-multicast range should be rejected
    let ipv4_unicast_range = IpRange::V4(
        Ipv4Range::new(
            Ipv4Addr::new(10, 0, 0, 1),
            Ipv4Addr::new(10, 0, 0, 255),
        )
        .unwrap(),
    );
    object_create_error(
        client,
        range_url,
        &ipv4_unicast_range,
        StatusCode::BAD_REQUEST,
    )
    .await;

    // IPv4 link-local multicast range should be rejected
    let ipv4_link_local_range = IpRange::V4(
        Ipv4Range::new(
            Ipv4Addr::new(224, 0, 0, 1),
            Ipv4Addr::new(224, 0, 0, 255),
        )
        .unwrap(),
    );
    object_create_error(
        client,
        range_url,
        &ipv4_link_local_range,
        StatusCode::BAD_REQUEST,
    )
    .await;

    // Valid IPv4 multicast range should be accepted (using ASM range)
    let valid_ipv4_range = IpRange::V4(
        Ipv4Range::new(
            Ipv4Addr::new(224, 1, 0, 1),
            Ipv4Addr::new(224, 1, 0, 255),
        )
        .unwrap(),
    );
    object_create::<_, IpPoolRange>(client, range_url, &valid_ipv4_range).await;

    // Create IPv6 multicast pool
    let ipv6_pool_params = IpPoolCreate::new_multicast(
        IdentityMetadataCreateParams {
            name: "test-v6-pool".parse().unwrap(),
            description: "IPv6 multicast pool for validation tests".to_string(),
        },
        IpVersion::V6,
    );
    object_create::<_, IpPool>(
        client,
        "/v1/system/ip-pools",
        &ipv6_pool_params,
    )
    .await;

    let v6_range_url = "/v1/system/ip-pools/test-v6-pool/ranges/add";

    // IPv6 link-local multicast range (ff02::/16) should be rejected
    let ipv6_link_local_range = IpRange::V6(
        Ipv6Range::new(
            Ipv6Addr::new(0xff02, 0, 0, 0, 0, 0, 0, 1),
            Ipv6Addr::new(0xff02, 0, 0, 0, 0, 0, 0, 255),
        )
        .unwrap(),
    );
    object_create_error(
        client,
        v6_range_url,
        &ipv6_link_local_range,
        StatusCode::BAD_REQUEST,
    )
    .await;

    // Valid IPv6 site-local multicast range (ff05::/16) should be accepted
    let valid_ipv6_range = IpRange::V6(
        Ipv6Range::new(
            Ipv6Addr::new(0xff05, 0, 0, 0, 0, 0, 0, 1),
            Ipv6Addr::new(0xff05, 0, 0, 0, 0, 0, 0, 255),
        )
        .unwrap(),
    );
    object_create::<_, IpPoolRange>(client, v6_range_url, &valid_ipv6_range)
        .await;
}

#[nexus_test]
async fn test_multicast_group_member_operations(
    cptestctx: &ControlPlaneTestContext,
) {
    let client = &cptestctx.external_client;
    let project_name = "test-project";
    let group_name = "test-group";
    let instance_name = "test-instance";

    // Create project and IP pools in parallel
    ops::join3(
        create_project(&client, project_name),
        create_default_ip_pools(&client), // For instance networking
        create_multicast_ip_pool_with_range(
            &client,
            "mcast-pool",
            (224, 4, 0, 10),
            (224, 4, 0, 255),
        ),
    )
    .await;

    let instance = create_instance(client, project_name, instance_name).await;

    // Use instance-centric API: PUT /v1/instances/{instance}/multicast-groups/{group}
    let join_url = format!(
        "/v1/instances/{instance_name}/multicast-groups/{group_name}?project={project_name}"
    );
    let join_params =
        InstanceMulticastGroupJoin { source_ips: None, ip_version: None };
    let added_member: MulticastGroupMember =
        put_upsert(client, &join_url, &join_params).await;

    assert_eq!(
        added_member.instance_id.to_string(),
        instance.identity.id.to_string()
    );

    // Wait for member to become joined
    // Member starts in "Joining" state and transitions to "Joined" via reconciler
    // Member only transitions to "Joined" after successful DPD update
    wait_for_member_state(
        cptestctx,
        group_name,
        instance.identity.id,
        nexus_db_model::MulticastGroupMemberState::Joined,
    )
    .await;

    // Test listing members (should have 1 now in Joined state)
    let members = list_multicast_group_members(&client, group_name).await;
    assert_eq!(members.len(), 1, "Expected exactly 1 member");
    assert_eq!(members[0].instance_id, added_member.instance_id);
    assert_eq!(members[0].multicast_group_id, added_member.multicast_group_id);

    // Test listing groups (should include our implicitly created group)
    let groups = list_multicast_groups(&client).await;
    assert!(
        groups.iter().any(|g| g.identity.name == group_name),
        "Expected group {group_name} to appear in group listing"
    );

    // DPD Validation: Verify groups exist in dataplane after member addition
    let dpd_client = dpd_client(cptestctx);
    // Get the multicast IP from the group (since member doesn't have the IP field)
    let group_get_url = mcast_group_url(group_name);
    let group: MulticastGroup = object_get(client, &group_get_url).await;
    let external_multicast_ip = group.multicast_ip;

    // List all groups in DPD to find both external and underlay groups
    let dpd_groups = dpd_client
        .multicast_groups_list(None, None)
        .await
        .expect("Should list DPD groups");

    // Find the external IPv4 group (should exist but may not have members)
    let expect_msg =
        format!("External group {external_multicast_ip} should exist in DPD");
    dpd_groups
        .items
        .iter()
        .find(|g| {
            let ip = match g {
                dpd_types::MulticastGroupResponse::External {
                    group_ip,
                    ..
                } => *group_ip,
                dpd_types::MulticastGroupResponse::Underlay {
                    group_ip,
                    ..
                } => IpAddr::V6(group_ip.0),
            };
            ip == external_multicast_ip
                && matches!(
                    g,
                    dpd_types::MulticastGroupResponse::External { .. }
                )
        })
        .expect(&expect_msg);

    // Directly get the underlay IPv6 group by finding the admin-local address
    // First find the underlay group IP from the list to get the exact IPv6 address
    let underlay_ip = dpd_groups
        .items
        .iter()
        .find_map(|g| {
            match g {
                dpd_types::MulticastGroupResponse::Underlay {
                    group_ip,
                    ..
                } => {
                    // Check if it starts with ff04 (admin-local multicast)
                    if group_ip.0.segments()[0] == 0xff04 {
                        Some(group_ip.clone())
                    } else {
                        None
                    }
                }
                dpd_types::MulticastGroupResponse::External { .. } => None,
            }
        })
        .expect("Should find underlay group IP in DPD response");

    // Get the underlay group directly
    let underlay_group = dpd_client
        .multicast_group_get_underlay(&underlay_ip)
        .await
        .expect("Should get underlay group from DPD");

    assert_eq!(
        underlay_group.members.len(),
        1,
        "Underlay group should have exactly 1 member after member addition"
    );

    // Assert all underlay members use rear (backplane) ports with Underlay direction
    for member in &underlay_group.members {
        assert!(
            matches!(member.port_id, dpd_client::types::PortId::Rear(_)),
            "Underlay member should use rear (backplane) port, got: {:?}",
            member.port_id
        );
        assert_eq!(
            member.direction,
            dpd_client::types::Direction::Underlay,
            "Underlay member should have Underlay direction"
        );
    }

    // Test removing instance from multicast group using instance-centric DELETE
    let member_remove_url = format!(
        "/v1/instances/{instance_name}/multicast-groups/{group_name}?project={project_name}"
    );

    NexusRequest::new(
        RequestBuilder::new(client, http::Method::DELETE, &member_remove_url)
            .expect_status(Some(StatusCode::NO_CONTENT)),
    )
    .authn_as(AuthnMode::PrivilegedUser)
    .execute()
    .await
    .expect("Should remove member from multicast group");

    // Implicit deletion model: group is implicitly deleted when last member is removed
    // Wait for both Nexus group and DPD group to be deleted
    wait_for_group_deleted(cptestctx, group_name).await;
    wait_for_group_deleted_from_dpd(cptestctx, external_multicast_ip).await;
}

#[nexus_test]
async fn test_instance_multicast_endpoints(
    cptestctx: &ControlPlaneTestContext,
) {
    let client = &cptestctx.external_client;
    let project_name = "test-project";
    let group1_name = "mcast-group-1";
    let group2_name = "mcast-group-2";
    let instance_name = "test-instance";

    // Create project and IP pools in parallel
    ops::join3(
        create_project(&client, project_name),
        create_default_ip_pools(&client),
        create_multicast_ip_pool_with_range(
            &client,
            "mcast-pool",
            (224, 5, 0, 10),
            (224, 5, 0, 255),
        ),
    )
    .await;

    // Implicit deletion model: Groups will implicitly create when first instance joins

    // Create an instance (starts automatically with create_instance helper)
    let instance = create_instance(client, project_name, instance_name).await;
    let instance_id = InstanceUuid::from_untyped_uuid(instance.identity.id);

    // Simulate and wait for instance to be fully running with sled_id assigned
    let nexus = &cptestctx.server.server_context().nexus;
    instance_simulate(nexus, &instance_id).await;
    instance_wait_for_state(client, instance_id, InstanceState::Running).await;
    wait_for_instance_sled_assignment(cptestctx, &instance_id).await;

    // Case: List instance multicast groups (should be empty initially)
    let instance_groups_url = format!(
        "/v1/instances/{instance_name}/multicast-groups?project={project_name}"
    );
    let instance_memberships: ResultsPage<MulticastGroupMember> =
        object_get(client, &instance_groups_url).await;
    assert_eq!(
        instance_memberships.items.len(),
        0,
        "Instance should have no multicast memberships initially"
    );

    // Case: Join group1 using instance-centric endpoint (implicitly creates group1)
    let instance_join_group1_url = format!(
        "/v1/instances/{instance_name}/multicast-groups/{group1_name}?project={project_name}"
    );
    let join_params =
        InstanceMulticastGroupJoin { source_ips: None, ip_version: None };
    // Use PUT method and expect 201 Created (implicitly creating group1)
    let member1: MulticastGroupMember =
        put_upsert(client, &instance_join_group1_url, &join_params).await;
    assert_eq!(member1.instance_id, instance.identity.id);

    // Wait for group1 to become active after implicitly create
    wait_for_group_active(client, group1_name).await;

    // Wait for member to become joined
    wait_for_member_state(
        cptestctx,
        group1_name,
        instance.identity.id,
        nexus_db_model::MulticastGroupMemberState::Joined,
    )
    .await;

    // Case: Verify membership shows up in both endpoints
    // Check group-centric view
    let group1_members =
        list_multicast_group_members(&client, group1_name).await;
    assert_eq!(group1_members.len(), 1);
    assert_eq!(group1_members[0].instance_id, instance.identity.id);

    // Check instance-centric view (test the list endpoint thoroughly)
    let instance_memberships: ResultsPage<MulticastGroupMember> =
        object_get(client, &instance_groups_url).await;
    assert_eq!(
        instance_memberships.items.len(),
        1,
        "Instance should have exactly 1 membership"
    );
    assert_eq!(instance_memberships.items[0].instance_id, instance.identity.id);
    assert_eq!(
        instance_memberships.items[0].multicast_group_id,
        member1.multicast_group_id
    );
    assert_eq!(instance_memberships.items[0].state, "Joined");

    // Join group2 using instance-centric endpoint (implicitly creates group2)
    let join_group2_url = format!(
        "/v1/instances/{instance_name}/multicast-groups/{group2_name}?project={project_name}"
    );
    let join_params2 =
        InstanceMulticastGroupJoin { source_ips: None, ip_version: None };
    let member2: MulticastGroupMember =
        put_upsert(client, &join_group2_url, &join_params2).await;
    assert_eq!(member2.instance_id, instance.identity.id);

    // Wait for group2 to become active after implicitly create
    wait_for_group_active(client, group2_name).await;

    // Wait for member to become joined
    wait_for_member_state(
        cptestctx,
        group2_name,
        instance.identity.id,
        nexus_db_model::MulticastGroupMemberState::Joined,
    )
    .await;

    // Verify instance now belongs to both groups (comprehensive list test)
    let instance_memberships: ResultsPage<MulticastGroupMember> =
        object_get(client, &instance_groups_url).await;
    assert_eq!(
        instance_memberships.items.len(),
        2,
        "Instance should belong to both groups"
    );

    // Verify the list endpoint returns the correct membership details
    let membership_group_ids: Vec<_> = instance_memberships
        .items
        .iter()
        .map(|m| m.multicast_group_id)
        .collect();
    assert!(
        membership_group_ids.contains(&member1.multicast_group_id),
        "List should include group1 membership"
    );
    assert!(
        membership_group_ids.contains(&member2.multicast_group_id),
        "List should include group2 membership"
    );

    // Verify all memberships show correct instance_id and state
    for membership in &instance_memberships.items {
        assert_eq!(membership.instance_id, instance.identity.id);
        assert_eq!(membership.state, "Joined");
    }

    // Verify each group shows the instance as a member
    let group1_members =
        list_multicast_group_members(&client, group1_name).await;
    let group2_members =
        list_multicast_group_members(&client, group2_name).await;
    assert_eq!(group1_members.len(), 1);
    assert_eq!(group2_members.len(), 1);
    assert_eq!(group1_members[0].instance_id, instance.identity.id);
    assert_eq!(group2_members[0].instance_id, instance.identity.id);

    // Leave group1 using instance-centric endpoint
    let instance_leave_group1_url = format!(
        "/v1/instances/{instance_name}/multicast-groups/{group1_name}?project={project_name}"
    );
    object_delete(client, &instance_leave_group1_url).await;

    // Implicit deletion model: group1 should be deleted after last member leaves
    wait_for_group_deleted(cptestctx, group1_name).await;

    // Verify membership removed from both views
    // Check instance-centric view - should only show active memberships (group2)
    let instance_memberships: ResultsPage<MulticastGroupMember> =
        object_get(client, &instance_groups_url).await;
    assert_eq!(
        instance_memberships.items.len(),
        1,
        "Instance should only show active membership (group2)"
    );
    assert_eq!(
        instance_memberships.items[0].multicast_group_id,
        member2.multicast_group_id,
        "Remaining membership should be group2"
    );
    assert_eq!(
        instance_memberships.items[0].state, "Joined",
        "Group2 membership should be Joined"
    );

    // Check group2 still has the member (group1 is already deleted)
    let group2_members =
        list_multicast_group_members(&client, group2_name).await;
    assert_eq!(group2_members.len(), 1, "Group2 should still have 1 member");

    // Leave group2 using instance-centric endpoint
    let member_remove_url = format!(
        "/v1/instances/{instance_name}/multicast-groups/{group2_name}?project={project_name}"
    );

    NexusRequest::new(
        RequestBuilder::new(client, http::Method::DELETE, &member_remove_url)
            .expect_status(Some(StatusCode::NO_CONTENT)),
    )
    .authn_as(AuthnMode::PrivilegedUser)
    .execute()
    .await
    .expect("Should remove member from group2");

    // Wait for reconciler to process the removal
    wait_for_multicast_reconciler(&cptestctx.lockstep_client).await;

    // Verify all memberships are gone
    let instance_memberships: ResultsPage<MulticastGroupMember> =
        object_get(client, &instance_groups_url).await;
    assert_eq!(
        instance_memberships.items.len(),
        0,
        "Instance should have no memberships"
    );

    // Implicit deletion model: Groups should be implicitly deleted after last member removed
    ops::join2(
        wait_for_group_deleted(cptestctx, group1_name),
        wait_for_group_deleted(cptestctx, group2_name),
    )
    .await;
}

#[nexus_test]
async fn test_multicast_group_member_errors(
    cptestctx: &ControlPlaneTestContext,
) {
    let client = &cptestctx.external_client;
    let project_name = "test-project";
    let group_name = "test-group";
    let nonexistent_instance = "nonexistent-instance";

    // Create project and IP pools in parallel
    ops::join3(
        create_project(&client, project_name),
        create_default_ip_pools(&client),
        create_multicast_ip_pool_with_range(
            &client,
            "mcast-pool",
            (224, 6, 0, 10),
            (224, 6, 0, 255),
        ),
    )
    .await;

    // Implicitly create a multicast group by adding an instance as first member
    let instance_name = "test-instance";
    create_instance(client, project_name, instance_name).await;

    // Use instance-centric API to join group (implicitly creates group)
    multicast_group_attach(cptestctx, project_name, instance_name, group_name)
        .await;

    // Wait for group to become active before testing error cases
    wait_for_group_active(&client, group_name).await;

    // Test joining with nonexistent instance - should fail with NOT_FOUND
    let bad_join_url = format!(
        "/v1/instances/{nonexistent_instance}/multicast-groups/{group_name}?project={project_name}"
    );
    let join_params =
        InstanceMulticastGroupJoin { source_ips: None, ip_version: None };
    object_put_error(
        client,
        &bad_join_url,
        &join_params,
        StatusCode::NOT_FOUND,
    )
    .await;

    cleanup_instances(cptestctx, client, project_name, &[instance_name]).await;
    wait_for_group_deleted(cptestctx, group_name).await;
}

#[nexus_test]
async fn test_instance_deletion_removes_multicast_memberships(
    cptestctx: &ControlPlaneTestContext,
) {
    let client = &cptestctx.external_client;
    let project_name = "springfield-squidport"; // Use the same project name as instance helpers
    let group_name = "instance-deletion-group";
    let instance_name = "deletion-test-instance";

    // Create project and IP pools in parallel
    ops::join3(
        create_project(&client, project_name),
        create_default_ip_pools(&client),
        create_multicast_ip_pool_with_range(
            &client,
            "mcast-pool",
            (224, 9, 0, 10),
            (224, 9, 0, 255),
        ),
    )
    .await;

    // Implicitly create multicast group by adding instance as first member
    let instance = create_instance(client, project_name, instance_name).await;

    // Use instance-centric API to join group (implicit group creation)
    multicast_group_attach(cptestctx, project_name, instance_name, group_name)
        .await;

    // Wait for group to become active after implicitly create
    wait_for_group_active(&client, group_name).await;

    // Get the group to find its auto-allocated IP address (needed for DPD check)
    let created_group: MulticastGroup =
        object_get(client, &mcast_group_url(group_name)).await;
    let multicast_ip = created_group.multicast_ip;

    // Wait for member to join
    wait_for_member_state(
        cptestctx,
        group_name,
        instance.identity.id,
        nexus_db_model::MulticastGroupMemberState::Joined,
    )
    .await;

    // Verify member was added
    let members = list_multicast_group_members(&client, group_name).await;
    assert_eq!(members.len(), 1, "Instance should be a member of the group");
    assert_eq!(members[0].instance_id, instance.identity.id);

    // Case: Instance deletion should clean up multicast memberships
    cleanup_instances(cptestctx, client, project_name, &[instance_name]).await;

    // Verify instance is gone
    let instance_url =
        format!("/v1/instances/{instance_name}?project={project_name}");

    object_get_error(client, &instance_url, StatusCode::NOT_FOUND).await;

    // Implicit model: group is implicitly deleted when last member (instance) is removed
    wait_for_group_deleted(cptestctx, group_name).await;

    // Wait for reconciler to clean up DPD state (activates reconciler repeatedly until DPD confirms deletion)
    wait_for_group_deleted_from_dpd(cptestctx, multicast_ip).await;
}

/// Test that the multicast_ip field is correctly populated in MulticastGroupMember API responses.
/// This validates the denormalized multicast_ip field added for API ergonomics.
#[nexus_test]
async fn test_member_response_includes_multicast_ip(
    cptestctx: &ControlPlaneTestContext,
) {
    let client = &cptestctx.external_client;
    let project_name = "multicast-ip-test";
    let group_name = "test-group";
    let instance_name = "test-instance";

    // Create project and IP pools in parallel
    ops::join3(
        create_project(&client, project_name),
        create_default_ip_pools(&client),
        create_multicast_ip_pool_with_range(
            &client,
            "test-pool",
            (224, 30, 0, 1),
            (224, 30, 0, 10),
        ),
    )
    .await;

    // Create instance for implicit group creation
    create_instance(client, project_name, instance_name).await;

    // Implicitly create group via instance-centric API
    let join_url = format!(
        "/v1/instances/{instance_name}/multicast-groups/{group_name}?project={project_name}"
    );
    let join_params =
        InstanceMulticastGroupJoin { source_ips: None, ip_version: None };

    // Add member and verify multicast_ip field is present in response
    let added_member: MulticastGroupMember =
        put_upsert(client, &join_url, &join_params).await;

    // Wait for group to become active
    wait_for_group_active(client, group_name).await;

    // Get the group to verify its multicast_ip
    let group: MulticastGroup =
        object_get(client, &mcast_group_url(group_name)).await;

    // Verify multicast_ip field is present in member response
    assert_eq!(
        added_member.multicast_ip, group.multicast_ip,
        "MulticastGroupMember API response should include multicast_ip field that matches the group's IP"
    );

    // Verify multicast_ip is in expected range from the pool
    let member_ip_str = added_member.multicast_ip.to_string();
    assert!(
        member_ip_str.starts_with("224.30.0."),
        "Member multicast_ip should be allocated from the pool range, got: {member_ip_str}"
    );

    // Case: List members and verify multicast_ip in all responses
    let members_list_url = format!(
        "{}?project={project_name}",
        mcast_group_members_url(group_name)
    );
    let members: ResultsPage<MulticastGroupMember> =
        object_get(client, &members_list_url).await;

    assert_eq!(members.items.len(), 1, "Should have exactly one member");
    assert_eq!(
        members.items[0].multicast_ip, group.multicast_ip,
        "Listed member should also include multicast_ip field"
    );

    // Case: Remove and re-add member (reactivation) - verify field preserved
    let member_remove_url = format!(
        "/v1/instances/{instance_name}/multicast-groups/{group_name}?project={project_name}"
    );
    NexusRequest::new(
        RequestBuilder::new(client, http::Method::DELETE, &member_remove_url)
            .expect_status(Some(StatusCode::NO_CONTENT)),
    )
    .authn_as(AuthnMode::PrivilegedUser)
    .execute()
    .await
    .expect("Should remove member");

    wait_for_group_deleted(cptestctx, group_name).await;

    // Re-create group by adding member again
    let readded_member: MulticastGroupMember =
        put_upsert(client, &join_url, &join_params).await;

    wait_for_group_active(client, group_name).await;

    let new_group: MulticastGroup =
        object_get(client, &mcast_group_url(group_name)).await;

    // Verify multicast_ip field is present in re-added member
    assert_eq!(
        readded_member.multicast_ip, new_group.multicast_ip,
        "Re-added member should also have multicast_ip field"
    );

    NexusRequest::new(
        RequestBuilder::new(client, http::Method::DELETE, &member_remove_url)
            .expect_status(Some(StatusCode::NO_CONTENT)),
    )
    .authn_as(AuthnMode::PrivilegedUser)
    .execute()
    .await
    .expect("Should remove member for cleanup");

    wait_for_group_deleted(cptestctx, group_name).await;
}

/// Test that we cannot delete a multicast IP pool when multicast groups are
/// linked to it (allocated IPs from it).
///
/// With implicit groups:
/// - Groups implicitly create when instances join
/// - Groups hold IPs from pools while they exist
/// - Pool should be protected while groups exist
/// - After groups are implicitly deleted (last member leaves), pool can be deleted
#[nexus_test]
async fn test_cannot_delete_multicast_pool_with_groups(
    cptestctx: &ControlPlaneTestContext,
) {
    let client = &cptestctx.external_client;
    let project_name = "test-project";
    let pool_name = "mcast-pool-delete-test";
    let group_name = "mcast-group-blocks-delete";
    let instance_name = "pool-test-instance";

    // Create project and IP pools in parallel
    ops::join3(
        create_project(&client, project_name),
        create_default_ip_pools(&client),
        create_multicast_ip_pool_with_range(
            client,
            pool_name,
            (224, 10, 0, 1),
            (224, 10, 0, 10),
        ),
    )
    .await;

    let pool_url = format!("/v1/system/ip-pools/{pool_name}");
    let range_url = format!("/v1/system/ip-pools/{pool_name}/ranges/remove");

    let range = IpRange::V4(
        Ipv4Range::new(
            std::net::Ipv4Addr::new(224, 10, 0, 1),
            std::net::Ipv4Addr::new(224, 10, 0, 10),
        )
        .unwrap(),
    );

    // Verify we can't delete the pool while it has ranges
    let error: HttpErrorResponseBody =
        object_delete_error(client, &pool_url, StatusCode::BAD_REQUEST).await;
    assert_eq!(
        error.message,
        "IP Pool cannot be deleted while it contains IP ranges"
    );

    // Create instance and implicitly create group via instance-centric API
    create_instance(client, project_name, instance_name).await;

    multicast_group_attach(cptestctx, project_name, instance_name, group_name)
        .await;

    // Wait for group to become active
    wait_for_group_active(client, group_name).await;

    // Verify we can't delete the range while groups are allocated from it
    let error: HttpErrorResponseBody = NexusRequest::new(
        RequestBuilder::new(client, Method::POST, &range_url)
            .body(Some(&range))
            .expect_status(Some(StatusCode::BAD_REQUEST)),
    )
    .authn_as(AuthnMode::PrivilegedUser)
    .execute()
    .await
    .unwrap()
    .parsed_body()
    .unwrap();
    assert_eq!(
        error.message,
        "IP pool ranges cannot be deleted while multicast groups are allocated from them"
    );

    // Verify we still can't delete the pool (indirectly protected by ranges)
    let error: HttpErrorResponseBody =
        object_delete_error(client, &pool_url, StatusCode::BAD_REQUEST).await;
    assert_eq!(
        error.message,
        "IP Pool cannot be deleted while it contains IP ranges"
    );

    // Verify we can't unlink the pool from the silo while groups are
    // allocated from it.
    let unlink_url = format!(
        "/v1/system/ip-pools/{pool_name}/silos/{}",
        DEFAULT_SILO.name().as_str()
    );
    object_delete_error(client, &unlink_url, StatusCode::BAD_REQUEST).await;

    cleanup_instances(cptestctx, client, project_name, &[instance_name]).await;
    wait_for_group_deleted(cptestctx, group_name).await;

    // Now we should be able to delete the range
    NexusRequest::new(
        RequestBuilder::new(client, Method::POST, &range_url)
            .body(Some(&range))
            .expect_status(Some(StatusCode::NO_CONTENT)),
    )
    .authn_as(AuthnMode::PrivilegedUser)
    .execute()
    .await
    .expect(
        "Should be able to delete range after groups are implicitly deleted",
    );

    // And we can unlink the pool from the silo
    object_delete(client, &unlink_url).await;

    // And now we should be able to delete the pool
    NexusRequest::object_delete(client, &pool_url)
        .authn_as(AuthnMode::PrivilegedUser)
        .execute()
        .await
        .expect("Should be able to delete pool after ranges are deleted");
}

/// Consolidated test for SSM (Source-Specific Multicast) source IP behavior.
///
/// This test covers:
/// - Source IP union on group view (multiple members with different sources)
/// - IPv4/IPv6 source address family validation
/// - Multiple SSM groups from same pool with different sources
///
/// SSM groups (232.0.0.0/8 for IPv4, ff3x::/32 for IPv6) require source IPs
/// to be specified on join. The group's `source_ips` field shows the union
/// of all member sources.
#[nexus_test]
async fn test_ssm_source_ip_behavior(cptestctx: &ControlPlaneTestContext) {
    let client = &cptestctx.external_client;
    let project_name = "ssm-source-ip-test";

    // Create project and IP pools in parallel
    // SSM pool uses 232.x.x.x range with enough IPs for multiple groups
    let (_, _, ssm_pool) = ops::join3(
        create_project(&client, project_name),
        create_default_ip_pools(&client),
        create_multicast_ip_pool_with_range(
            &client,
            "ssm-test-pool",
            (232, 1, 0, 1),
            (232, 1, 0, 100),
        ),
    )
    .await;

    // Also create IPv6 pool for address family validation tests
    create_multicast_ip_pool_v6(&client, "ssm-test-pool-v6").await;

    // Create instances for all test cases
    let instance_names = [
        "ssm-inst-1",
        "ssm-inst-2",
        "ssm-inst-3",
        "ssm-inst-4",
        "ssm-inst-5",
        "ssm-inst-6",
    ];
    for name in &instance_names {
        create_instance(client, project_name, name).await;
    }

    // Case: Source IP union on group view
    // Multiple members join with different sources, group view shows union
    let ssm_union_ip = "232.1.0.10";
    let source1 = IpAddr::V4(Ipv4Addr::new(10, 0, 0, 1));
    let source2 = IpAddr::V4(Ipv4Addr::new(10, 0, 0, 2));
    let source3 = IpAddr::V4(Ipv4Addr::new(10, 0, 0, 3));

    // First instance joins with source1 - creates SSM group
    let join_url1 = format!(
        "/v1/instances/{}/multicast-groups/{ssm_union_ip}?project={project_name}",
        instance_names[0]
    );
    let join_body1 = InstanceMulticastGroupJoin {
        source_ips: Some(vec![source1]),
        ip_version: None,
    };
    put_upsert::<_, MulticastGroupMember>(client, &join_url1, &join_body1)
        .await;

    // Verify group source_ips shows source1
    let group: MulticastGroup =
        object_get(client, &format!("/v1/multicast-groups/{ssm_union_ip}"))
            .await;
    assert_eq!(group.source_ips, vec![source1], "Group should show source1");

    // Second instance joins with source1 and source2
    let join_url2 = format!(
        "/v1/instances/{}/multicast-groups/{ssm_union_ip}?project={project_name}",
        instance_names[1]
    );
    let join_body2 = InstanceMulticastGroupJoin {
        source_ips: Some(vec![source1, source2]),
        ip_version: None,
    };
    put_upsert::<_, MulticastGroupMember>(client, &join_url2, &join_body2)
        .await;

    // Verify group source_ips is union of member sources (sorted for comparison)
    let group: MulticastGroup =
        object_get(client, &format!("/v1/multicast-groups/{ssm_union_ip}"))
            .await;
    let mut actual_sources = group.source_ips.clone();
    actual_sources.sort();
    let mut expected_sources = vec![source1, source2];
    expected_sources.sort();
    assert_eq!(
        actual_sources, expected_sources,
        "Group source_ips should be union of all member sources"
    );

    // Third instance joins with source3 - union should now include all three
    let join_url3 = format!(
        "/v1/instances/{}/multicast-groups/{ssm_union_ip}?project={project_name}",
        instance_names[2]
    );
    let join_body3 = InstanceMulticastGroupJoin {
        source_ips: Some(vec![source3]),
        ip_version: None,
    };
    put_upsert::<_, MulticastGroupMember>(client, &join_url3, &join_body3)
        .await;

    // Verify group source_ips is union of all three members' sources
    let group: MulticastGroup =
        object_get(client, &format!("/v1/multicast-groups/{ssm_union_ip}"))
            .await;
    let mut actual_sources = group.source_ips.clone();
    actual_sources.sort();
    let mut expected_union_sources = vec![source1, source2, source3];
    expected_union_sources.sort();
    assert_eq!(
        actual_sources, expected_union_sources,
        "Group source_ips should be union of all member sources"
    );

    // Verify list endpoint also returns groups with source_ips as union
    let groups: Collection<MulticastGroup> =
        NexusRequest::iter_collection_authn(
            client,
            "/v1/multicast-groups",
            "",
            None,
        )
        .await
        .expect("Should list multicast groups");

    let listed_group = groups
        .all_items
        .iter()
        .find(|g| g.multicast_ip.to_string() == ssm_union_ip)
        .expect("SSM group should appear in list");
    let mut listed_sources = listed_group.source_ips.clone();
    listed_sources.sort();
    assert_eq!(
        listed_sources, expected_union_sources,
        "List endpoint should also return source_ips as union"
    );

    // Wait for reconciler to sync all member sources to DPD
    wait_for_multicast_reconciler(&cptestctx.lockstep_client).await;

    // Verify DPD external group contains the union of sources
    let multicast_ip: IpAddr = ssm_union_ip.parse().unwrap();
    let dpd_response = dpd_client(cptestctx)
        .multicast_group_get(&multicast_ip)
        .await
        .expect("DPD should have external group")
        .into_inner();

    let dpd_sources = match dpd_response {
        dpd_types::MulticastGroupResponse::External { sources, .. } => sources,
        dpd_types::MulticastGroupResponse::Underlay { .. } => {
            panic!("Expected External group from DPD, got Underlay");
        }
    };

    // Convert DPD sources (IpSrc) to IpAddr for comparison
    let mut dpd_source_ips: Vec<IpAddr> = dpd_sources
        .iter()
        .flatten()
        .filter_map(|src| match src {
            dpd_types::IpSrc::Exact(ip) => Some(*ip),
            dpd_types::IpSrc::Any => None, // Any-source (ASM) wildcard
        })
        .collect();
    dpd_source_ips.sort();

    assert_eq!(
        dpd_source_ips, expected_union_sources,
        "DPD external group sources should be union of all member sources"
    );

    // Case: IPv6 source with IPv4 group should fail
    let ipv4_ssm_ip = "232.1.0.20";
    let ipv6_source: IpAddr = "2001:db8::1".parse().unwrap();
    let join_url_v4_with_v6 = format!(
        "/v1/instances/{}/multicast-groups/{ipv4_ssm_ip}?project={project_name}",
        instance_names[3]
    );
    let join_body_v4_with_v6 = InstanceMulticastGroupJoin {
        source_ips: Some(vec![ipv6_source]),
        ip_version: None,
    };

    let error = NexusRequest::new(
        RequestBuilder::new(client, Method::PUT, &join_url_v4_with_v6)
            .body(Some(&join_body_v4_with_v6))
            .expect_status(Some(StatusCode::BAD_REQUEST)),
    )
    .authn_as(AuthnMode::PrivilegedUser)
    .execute()
    .await
    .expect("Request should complete")
    .parsed_body::<dropshot::HttpErrorResponseBody>()
    .expect("Should parse error body");

    assert_eq!(
        error.error_code,
        Some("InvalidRequest".to_string()),
        "Expected InvalidRequest for IPv6 source with IPv4 group, got: {:?}",
        error.error_code
    );

    // Case: IPv4 source with IPv6 group should fail
    let ipv4_source: IpAddr = "10.0.0.1".parse().unwrap();
    let join_url_v6_with_v4 = format!(
        "/v1/instances/{}/multicast-groups/ipv6-mismatch-group?project={project_name}",
        instance_names[3]
    );
    let join_body_v6_with_v4 = InstanceMulticastGroupJoin {
        source_ips: Some(vec![ipv4_source]),
        ip_version: Some(IpVersion::V6),
    };

    let error_v6 = NexusRequest::new(
        RequestBuilder::new(client, Method::PUT, &join_url_v6_with_v4)
            .body(Some(&join_body_v6_with_v4))
            .expect_status(Some(StatusCode::BAD_REQUEST)),
    )
    .authn_as(AuthnMode::PrivilegedUser)
    .execute()
    .await
    .expect("Request should complete")
    .parsed_body::<dropshot::HttpErrorResponseBody>()
    .expect("Should parse error body");

    assert_eq!(
        error_v6.error_code,
        Some("InvalidRequest".to_string()),
        "Expected InvalidRequest for IPv4 source with IPv6 group, got: {:?}",
        error_v6.error_code
    );

    // Case: Multiple SSM groups from same pool with different sources
    // Each instance joins a different SSM group with unique sources
    let group_configs = [
        ("ssm-multi-group-1", "10.1.1.1"),
        ("ssm-multi-group-2", "10.2.2.2"),
        ("ssm-multi-group-3", "10.3.3.3"),
    ];

    // Create 3 different SSM groups from the same pool
    for (i, (group_name, source_ip)) in group_configs.iter().enumerate() {
        let instance_name = instance_names[i + 3]; // Use instances 3, 4, 5
        let join_url = format!(
            "/v1/instances/{instance_name}/multicast-groups/{group_name}?project={project_name}"
        );
        let join_params = InstanceMulticastGroupJoin {
            source_ips: Some(vec![source_ip.parse().unwrap()]),
            ip_version: None,
        };
        put_upsert::<_, MulticastGroupMember>(client, &join_url, &join_params)
            .await;

        // Wait for group to become active
        wait_for_group_active(client, group_name).await;
    }

    // Verify all groups exist with correct properties
    let mut allocated_ips = Vec::new();
    for (group_name, expected_source) in &group_configs {
        let group: MulticastGroup =
            object_get(client, &mcast_group_url(group_name)).await;

        // Verify pool reference
        assert_eq!(
            group.ip_pool_id, ssm_pool.identity.id,
            "Group {group_name} should reference the shared SSM pool"
        );

        // Verify SSM range (232.x.x.x)
        if let IpAddr::V4(ip) = group.multicast_ip {
            assert_eq!(
                ip.octets()[0],
                232,
                "Group {group_name} should have IP in SSM range (232.x.x.x)"
            );
            assert_eq!(
                ip.octets()[1],
                1,
                "Group {group_name} should have IP from pool range (232.1.x.x)"
            );
        } else {
            panic!("Expected IPv4 multicast address for group {group_name}");
        }

        // Verify source IPs
        assert_eq!(
            group.source_ips.len(),
            1,
            "Group {group_name} should have exactly 1 source IP"
        );
        assert_eq!(
            group.source_ips[0].to_string(),
            *expected_source,
            "Group {group_name} should have correct source IP"
        );

        // Collect allocated IP for uniqueness check
        allocated_ips.push(group.multicast_ip);
    }

    // Verify all allocated IPs are unique
    let unique_ips: std::collections::HashSet<_> =
        allocated_ips.iter().collect();
    assert_eq!(
        unique_ips.len(),
        allocated_ips.len(),
        "All SSM groups should have unique multicast IPs from the pool"
    );

    // Verify we can list all members for each group
    for (group_name, _) in &group_configs {
        let members = list_multicast_group_members(&client, group_name).await;
        assert_eq!(
            members.len(),
            1,
            "Group {group_name} should have exactly 1 member"
        );
    }

    // Cleanup all instances and wait for groups to be deleted
    cleanup_instances(cptestctx, client, project_name, &instance_names).await;

    // Wait for all groups to be deleted
    wait_for_group_deleted(cptestctx, ssm_union_ip).await;
    for (group_name, _) in &group_configs {
        wait_for_group_deleted(cptestctx, group_name).await;
    }
}

/// Test default pool behavior when no pool is specified on member join.
///
/// When a member joins a group without specifying a pool:
/// - If a default multicast pool exists, use it
/// - If no default pool, fail with appropriate error
#[nexus_test]
async fn test_default_pool_on_implicit_creation(
    cptestctx: &ControlPlaneTestContext,
) {
    let client = &cptestctx.external_client;
    let group_name = "default-pool-test-group";
    let group_name2 = "default-pool-test-group-2";
    let project_name = "default-pool-test-project";
    let instance_name = "default-pool-test-instance";

    // Setup: project and default IP pool in parallel (but no multicast pool yet)
    ops::join2(
        create_project(&client, project_name),
        create_default_ip_pools(&client),
    )
    .await;
    create_instance(client, project_name, instance_name).await;

    // Case: Joining when no multicast pool exists - should fail with 400 (no pool available)
    let join_url = format!(
        "/v1/instances/{instance_name}/multicast-groups/{group_name}?project={project_name}"
    );
    let join_params =
        InstanceMulticastGroupJoin { source_ips: None, ip_version: None };
    object_put_error(client, &join_url, &join_params, StatusCode::BAD_REQUEST)
        .await;

    // Create a multicast pool using resource_helpers (doesn't auto-link)
    let (mcast_pool, _) =
        nexus_test_utils::resource_helpers::create_multicast_ip_pool(
            &client,
            "default-mcast-pool",
            None,
        )
        .await;

    // Link as default so it can be auto-discovered when joining by name
    link_ip_pool(&client, "default-mcast-pool", &DEFAULT_SILO.id(), true).await;

    // Case: Joining when multicast pool exists - should succeed (pool auto-discovered)
    multicast_group_attach(cptestctx, project_name, instance_name, group_name2)
        .await;

    // Verify group was allocated from default pool
    let group: MulticastGroup =
        object_get(client, &mcast_group_url(group_name2)).await;
    assert_eq!(group.ip_pool_id, mcast_pool.identity.id);

    // Cleanup
    cleanup_instances(cptestctx, client, project_name, &[instance_name]).await;
    wait_for_group_deleted(cptestctx, group_name2).await;
}

/// Test pool range allocation for multicast groups.
///
/// Verifies that multicast IPs are correctly allocated from the specified
/// pool's ranges, and that exhausted ranges are handled properly.
#[nexus_test]
async fn test_pool_range_allocation(cptestctx: &ControlPlaneTestContext) {
    let client = &cptestctx.external_client;
    let project_name = "pool-range-test-project";

    // Create project and IP pools in parallel
    // Multicast pool has small range (3 IPs: 224.10.0.1-224.10.0.3)
    ops::join3(
        create_project(&client, project_name),
        create_default_ip_pools(&client),
        create_multicast_ip_pool_with_range(
            &client,
            "small-range-pool",
            (224, 10, 0, 1),
            (224, 10, 0, 3),
        ),
    )
    .await;

    // Create 4 instances for testing
    let instance_names =
        ["range-inst-1", "range-inst-2", "range-inst-3", "range-inst-4"];
    for name in &instance_names {
        create_instance(client, project_name, name).await;
    }

    // Case: Create 3 groups - should succeed (uses all IPs in range)
    for i in 0..3 {
        let group_name = format!("range-group-{i}");
        let instance_name = instance_names[i];
        multicast_group_attach(
            cptestctx,
            project_name,
            instance_name,
            &group_name,
        )
        .await;
    }

    // Case: Try to create 4th group - should fail (range exhausted)
    let group_name4 = "range-group-3";
    let instance_name4 = instance_names[3];
    let join_url4 = format!(
        "/v1/instances/{instance_name4}/multicast-groups/{group_name4}?project={project_name}"
    );
    let join_params4 =
        InstanceMulticastGroupJoin { source_ips: None, ip_version: None };
    let error = object_put_error(
        client,
        &join_url4,
        &join_params4,
        StatusCode::INSUFFICIENT_STORAGE, // or appropriate error code
    )
    .await;
    assert_eq!(
        error.error_code,
        Some("InsufficientCapacity".to_string()),
        "Expected InsufficientCapacity for pool exhaustion, got: {:?}",
        error.error_code
    );

    // Case: Delete one group (by removing all members)
    cleanup_instances(cptestctx, client, project_name, &[instance_names[0]])
        .await;
    wait_for_group_deleted(cptestctx, "range-group-0").await;

    // Case: Create new group - should succeed (IP reclaimed)
    let group_name_new = "range-group-new";
    multicast_group_attach(
        cptestctx,
        project_name,
        instance_name4,
        group_name_new,
    )
    .await;

    // Cleanup
    cleanup_instances(
        cptestctx,
        client,
        project_name,
        &[instance_names[1], instance_names[2], instance_names[3]],
    )
    .await;
}

/// Test multicast pool selection with custom ASM range.
///
/// Verifies that joining a multicast group by name correctly allocates
/// from a linked multicast pool, and that joining by explicit IP also works.
///
/// Note: The V4+V6 conflict scenario (where both default pools exist and the
/// system cannot determine which to use) is tested at the datastore level in
/// `nexus/db-queries/src/db/datastore/ip_pool.rs::test_ip_pools_fetch_asm_multicast_version_conflict`
/// because IPv6 ranges cannot be added via the HTTP API (they're rejected).
#[nexus_test]
async fn test_multicast_group_ip_version_conflict(
    cptestctx: &ControlPlaneTestContext,
) {
    let client = &cptestctx.external_client;

    // Setup: create project and default unicast IP pool
    let project_name = "ip-version-conflict-project";
    let instance_name = "ip-version-conflict-instance";
    ops::join2(
        create_project(client, project_name),
        create_default_ip_pools(client),
    )
    .await;

    // Create IPv4 multicast pool using local helper (creates + adds range + links)
    let v4_pool = create_multicast_ip_pool_with_range(
        client,
        "v4-mcast-conflict-pool",
        (224, 11, 0, 1),
        (224, 11, 0, 100),
    )
    .await;

    // Create instance for joining
    create_instance(client, project_name, instance_name).await;

    // Join by name should succeed (V4 pool auto-discovered)
    let join_url = format!(
        "/v1/instances/{instance_name}/multicast-groups/conflict-test-group?project={project_name}"
    );
    put_upsert::<_, MulticastGroupMember>(
        client,
        &join_url,
        &InstanceMulticastGroupJoin { source_ips: None, ip_version: None },
    )
    .await;

    // Join by explicit IP should also succeed (pool auto-discovered from IP)
    let explicit_ip = "224.11.0.50";
    let explicit_ip_join_url = format!(
        "/v1/instances/{instance_name}/multicast-groups/{explicit_ip}?project={project_name}"
    );
    put_upsert::<_, MulticastGroupMember>(
        client,
        &explicit_ip_join_url,
        &InstanceMulticastGroupJoin { source_ips: None, ip_version: None },
    )
    .await;

    // Wait for group to become active
    wait_for_multicast_reconciler(&cptestctx.lockstep_client).await;

    // Verify the group was created with IPv4 address from V4 pool
    let groups: ResultsPage<MulticastGroup> =
        object_get(client, "/v1/multicast-groups").await;
    let group = groups
        .items
        .iter()
        .find(|g| g.multicast_ip == explicit_ip.parse::<IpAddr>().unwrap())
        .expect("Should find group with explicit IP");

    assert!(group.multicast_ip.is_ipv4(), "Expected IPv4 address");
    assert_eq!(group.ip_pool_id, v4_pool.identity.id, "Should use V4 pool");

    // Cleanup
    let ip_based_group_name =
        format!("mcast-{}", explicit_ip.replace('.', "-"));
    cleanup_instances(cptestctx, client, project_name, &[instance_name]).await;

    // Wait for both groups to be deleted (test creates two groups)
    wait_for_group_deleted(cptestctx, "conflict-test-group").await;
    wait_for_group_deleted(cptestctx, &ip_based_group_name).await;
}