tests: Add Pod MAC collision detection E2E tests

Add E2E tests for Pod-Pod and cross-type Pod-VMI MAC collision
detection, mirroring the existing VMI collision test structure.

Pod-Pod tests cover: inter-Pod collisions (with MAC format
variations), out-of-range MACs, 3+ Pods, partial collisions on
multi-interface Pods, multiple colliding MACs, and collision
clearing on Pod deletion.

Cross-type tests cover: detecting collisions between a Pod and a
VMI sharing the same MAC, and verifying collision clears when
either the Pod or the VMI is deleted.

Assisted-by: Claude Opus 4.6 <noreply@anthropic.com>
Signed-off-by: Ram Lavi <ralavi@redhat.com>

Author: RamLavi

tests/pod.go

@@ -0,0 +1,90 @@
+package tests
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"time"
+
+	. "github.com/onsi/gomega"
+
+	corev1 "k8s.io/api/core/v1"
+	metav1 "k8s.io/apimachinery/pkg/apis/meta/v1"
+
+	networkv1 "github.com/k8snetworkplumbingwg/network-attachment-definition-client/pkg/apis/k8s.cni.cncf.io/v1"
+	multus "gopkg.in/k8snetworkplumbingwg/multus-cni.v3/pkg/types"
+)
+
+const (
+	podCollisionCleanupTimeout  = 2 * time.Minute
+	podCollisionCleanupInterval = 5 * time.Second
+)
+
+type PodOption func(*corev1.Pod)
+
+func WithMultusNetwork(nadName, mac string) PodOption {
+	return func(pod *corev1.Pod) {
+		var networks []*multus.NetworkSelectionElement
+		if existing, ok := pod.Annotations[networkv1.NetworkAttachmentAnnot]; ok {
+			_ = json.Unmarshal([]byte(existing), &networks)
+		}
+		networks = append(networks, &multus.NetworkSelectionElement{
+			Name:       nadName,
+			Namespace:  pod.Namespace,
+			MacRequest: mac,
+		})
+		data, _ := json.Marshal(networks)
+		pod.Annotations[networkv1.NetworkAttachmentAnnot] = string(data)
+	}
+}
+
+func NewCollisionPod(namespace, name string, opts ...PodOption) *corev1.Pod {
+	pod := &corev1.Pod{
+		ObjectMeta: metav1.ObjectMeta{
+			Name:        randName(name),
+			Namespace:   namespace,
+			Annotations: map[string]string{},
+		},
+		Spec: corev1.PodSpec{
+			TerminationGracePeriodSeconds: &gracePeriodSeconds,
+			Containers: []corev1.Container{
+				{
+					Name:    "test",
+					Image:   "quay.io/centos/centos:stream9",
+					Command: []string{"/bin/bash", "-c", "sleep INF"},
+				},
+			},
+		},
+	}
+	for _, opt := range opts {
+		opt(pod)
+	}
+	return pod
+}
+
+func getPodEvents(namespace, podName string) (*corev1.EventList, error) {
+	return testClient.K8sClient.CoreV1().Events(namespace).List(context.TODO(), metav1.ListOptions{
+		FieldSelector: fmt.Sprintf("involvedObject.name=%s,involvedObject.kind=Pod", podName),
+	})
+}
+
+func cleanupTestPodsInNamespaces(namespaces []string) {
+	for _, namespace := range namespaces {
+		podList, err := testClient.K8sClient.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{})
+		Expect(err).ToNot(HaveOccurred())
+
+		for _, pod := range podList.Items {
+			err = testClient.K8sClient.CoreV1().Pods(namespace).Delete(context.TODO(), pod.Name, metav1.DeleteOptions{})
+			if err != nil {
+				continue
+			}
+		}
+
+		Eventually(func() int {
+			list, listErr := testClient.K8sClient.CoreV1().Pods(namespace).List(context.TODO(), metav1.ListOptions{})
+			Expect(listErr).ToNot(HaveOccurred())
+			return len(list.Items)
+		}).WithTimeout(podCollisionCleanupTimeout).WithPolling(podCollisionCleanupInterval).Should(Equal(0),
+			"All test pods in namespace %s should be deleted", namespace)
+	}
+}