test(kubernetes): add rbac test coverage for pv listing [C9S-343] (#3350)

2026-08-07 09:04:48 +00:00 · 2026-08-05 12:10:41 +12:00
parent 5d4dbb5a67
commit 935b1d42d0
1 changed files with 242 additions and 0 deletions
@@ -105,6 +105,248 @@ func TestGetPersistentVolumes(t *testing.T) {
 		assert.Equal(t, corev1.VolumeBound, got.Status)
 		assert.Equal(t, []string{"RWO"}, got.AccessModes)
 	})
+
+	t.Run("non-admin sees only PVs bound to their allowed namespaces", func(t *testing.T) {
+		t.Parallel()
+		kcl := &KubeClient{
+			cli:                kfake.NewSimpleClientset(),
+			instanceID:         "instance",
+			isKubeAdmin:        false,
+			nonAdminNamespaces: []string{"nonAdmin"},
+		}
+
+		allowedPV := &corev1.PersistentVolume{
+			ObjectMeta: metav1.ObjectMeta{Name: "pv-allowed"},
+			Spec: corev1.PersistentVolumeSpec{
+				ClaimRef: &corev1.ObjectReference{Namespace: "nonAdmin", Name: "pvc-allowed"},
+			},
+		}
+		_, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), allowedPV, metav1.CreateOptions{})
+		require.NoError(t, err)
+
+		pvs, err := kcl.GetPersistentVolumes()
+		require.NoError(t, err)
+		require.Len(t, pvs, 1)
+		assert.Equal(t, "pv-allowed", pvs[0].Name)
+	})
+
+	t.Run("non-admin does not see PVs bound to another tenant's namespace", func(t *testing.T) {
+		t.Parallel()
+		kcl := &KubeClient{
+			cli:                kfake.NewSimpleClientset(),
+			instanceID:         "instance",
+			isKubeAdmin:        false,
+			nonAdminNamespaces: []string{"nonAdmin"},
+		}
+
+		otherTenantPV := &corev1.PersistentVolume{
+			ObjectMeta: metav1.ObjectMeta{Name: "pv-other-tenant"},
+			Spec: corev1.PersistentVolumeSpec{
+				ClaimRef: &corev1.ObjectReference{Namespace: "otherTenant", Name: "pvc-other-tenant"},
+			},
+		}
+		_, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), otherTenantPV, metav1.CreateOptions{})
+		require.NoError(t, err)
+
+		pvs, err := kcl.GetPersistentVolumes()
+		require.NoError(t, err)
+		assert.Empty(t, pvs)
+	})
+
+	t.Run("non-admin does not see PVs bound to a system namespace even if granted", func(t *testing.T) {
+		t.Parallel()
+		kcl := &KubeClient{
+			cli:                kfake.NewSimpleClientset(),
+			instanceID:         "instance",
+			isKubeAdmin:        false,
+			nonAdminNamespaces: []string{"portainer"},
+		}
+
+		systemPV := &corev1.PersistentVolume{
+			ObjectMeta: metav1.ObjectMeta{Name: "pv-system"},
+			Spec: corev1.PersistentVolumeSpec{
+				ClaimRef: &corev1.ObjectReference{Namespace: "portainer", Name: "pvc-system"},
+			},
+		}
+		_, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), systemPV, metav1.CreateOptions{})
+		require.NoError(t, err)
+
+		pvs, err := kcl.GetPersistentVolumes()
+		require.NoError(t, err)
+		assert.Empty(t, pvs)
+	})
+
+	t.Run("non-admin does not see unbound PVs", func(t *testing.T) {
+		t.Parallel()
+		kcl := &KubeClient{
+			cli:                kfake.NewSimpleClientset(),
+			instanceID:         "instance",
+			isKubeAdmin:        false,
+			nonAdminNamespaces: []string{"nonAdmin"},
+		}
+
+		unboundPV := &corev1.PersistentVolume{
+			ObjectMeta: metav1.ObjectMeta{Name: "pv-unbound"},
+			Spec:       corev1.PersistentVolumeSpec{},
+		}
+		_, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), unboundPV, metav1.CreateOptions{})
+		require.NoError(t, err)
+
+		pvs, err := kcl.GetPersistentVolumes()
+		require.NoError(t, err)
+		assert.Empty(t, pvs)
+	})
+
+	t.Run("non-admin with multiple allowed namespaces sees PVs from all of them", func(t *testing.T) {
+		t.Parallel()
+		kcl := &KubeClient{
+			cli:                kfake.NewSimpleClientset(),
+			instanceID:         "instance",
+			isKubeAdmin:        false,
+			nonAdminNamespaces: []string{"nonAdminOne", "nonAdminTwo"},
+		}
+
+		pvOne := &corev1.PersistentVolume{
+			ObjectMeta: metav1.ObjectMeta{Name: "pv-one"},
+			Spec: corev1.PersistentVolumeSpec{
+				ClaimRef: &corev1.ObjectReference{Namespace: "nonAdminOne", Name: "pvc-one"},
+			},
+		}
+		pvTwo := &corev1.PersistentVolume{
+			ObjectMeta: metav1.ObjectMeta{Name: "pv-two"},
+			Spec: corev1.PersistentVolumeSpec{
+				ClaimRef: &corev1.ObjectReference{Namespace: "nonAdminTwo", Name: "pvc-two"},
+			},
+		}
+		_, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), pvOne, metav1.CreateOptions{})
+		require.NoError(t, err)
+		_, err = kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), pvTwo, metav1.CreateOptions{})
+		require.NoError(t, err)
+
+		pvs, err := kcl.GetPersistentVolumes()
+		require.NoError(t, err)
+		require.Len(t, pvs, 2)
+
+		names := []string{pvs[0].Name, pvs[1].Name}
+		assert.ElementsMatch(t, []string{"pv-one", "pv-two"}, names)
+	})
+
+	t.Run("non-admin with no allowed namespaces gets empty list", func(t *testing.T) {
+		t.Parallel()
+		kcl := &KubeClient{
+			cli:                kfake.NewSimpleClientset(),
+			instanceID:         "instance",
+			isKubeAdmin:        false,
+			nonAdminNamespaces: []string{},
+		}
+
+		pv := &corev1.PersistentVolume{
+			ObjectMeta: metav1.ObjectMeta{Name: "pv-somewhere"},
+			Spec: corev1.PersistentVolumeSpec{
+				ClaimRef: &corev1.ObjectReference{Namespace: "somewhere", Name: "pvc-somewhere"},
+			},
+		}
+		_, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), pv, metav1.CreateOptions{})
+		require.NoError(t, err)
+
+		pvs, err := kcl.GetPersistentVolumes()
+		require.NoError(t, err)
+		assert.Empty(t, pvs)
+	})
+}
+
+func TestGetPersistentVolume(t *testing.T) {
+	t.Parallel()
+
+	t.Run("non-admin can get a PV bound to an allowed namespace", func(t *testing.T) {
+		t.Parallel()
+		kcl := &KubeClient{
+			cli:                kfake.NewSimpleClientset(),
+			instanceID:         "instance",
+			isKubeAdmin:        false,
+			nonAdminNamespaces: []string{"nonAdmin"},
+		}
+
+		pv := &corev1.PersistentVolume{
+			ObjectMeta: metav1.ObjectMeta{Name: "pv-allowed"},
+			Spec: corev1.PersistentVolumeSpec{
+				ClaimRef: &corev1.ObjectReference{Namespace: "nonAdmin", Name: "pvc-allowed"},
+			},
+		}
+		_, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), pv, metav1.CreateOptions{})
+		require.NoError(t, err)
+
+		got, err := kcl.GetPersistentVolume("pv-allowed")
+		require.NoError(t, err)
+		assert.Equal(t, "pv-allowed", got.Name)
+	})
+
+	t.Run("non-admin is denied access to a PV bound to another tenant's namespace", func(t *testing.T) {
+		t.Parallel()
+		kcl := &KubeClient{
+			cli:                kfake.NewSimpleClientset(),
+			instanceID:         "instance",
+			isKubeAdmin:        false,
+			nonAdminNamespaces: []string{"nonAdmin"},
+		}
+
+		pv := &corev1.PersistentVolume{
+			ObjectMeta: metav1.ObjectMeta{Name: "pv-other-tenant"},
+			Spec: corev1.PersistentVolumeSpec{
+				ClaimRef: &corev1.ObjectReference{Namespace: "otherTenant", Name: "pvc-other-tenant"},
+			},
+		}
+		_, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), pv, metav1.CreateOptions{})
+		require.NoError(t, err)
+
+		_, err = kcl.GetPersistentVolume("pv-other-tenant")
+		require.Error(t, err)
+		assert.ErrorIs(t, err, ErrUnauthorized)
+	})
+
+	t.Run("non-admin is denied access to a PV bound to a system namespace", func(t *testing.T) {
+		t.Parallel()
+		kcl := &KubeClient{
+			cli:                kfake.NewSimpleClientset(),
+			instanceID:         "instance",
+			isKubeAdmin:        false,
+			nonAdminNamespaces: []string{"portainer"},
+		}
+
+		pv := &corev1.PersistentVolume{
+			ObjectMeta: metav1.ObjectMeta{Name: "pv-system"},
+			Spec: corev1.PersistentVolumeSpec{
+				ClaimRef: &corev1.ObjectReference{Namespace: "portainer", Name: "pvc-system"},
+			},
+		}
+		_, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), pv, metav1.CreateOptions{})
+		require.NoError(t, err)
+
+		_, err = kcl.GetPersistentVolume("pv-system")
+		require.Error(t, err)
+		assert.ErrorIs(t, err, ErrUnauthorized)
+	})
+
+	t.Run("non-admin is denied access to an unbound PV", func(t *testing.T) {
+		t.Parallel()
+		kcl := &KubeClient{
+			cli:                kfake.NewSimpleClientset(),
+			instanceID:         "instance",
+			isKubeAdmin:        false,
+			nonAdminNamespaces: []string{"nonAdmin"},
+		}
+
+		pv := &corev1.PersistentVolume{
+			ObjectMeta: metav1.ObjectMeta{Name: "pv-unbound"},
+			Spec:       corev1.PersistentVolumeSpec{},
+		}
+		_, err := kcl.cli.CoreV1().PersistentVolumes().Create(t.Context(), pv, metav1.CreateOptions{})
+		require.NoError(t, err)
+
+		_, err = kcl.GetPersistentVolume("pv-unbound")
+		require.Error(t, err)
+		assert.ErrorIs(t, err, ErrUnauthorized)
+	})
 }

 func TestDeletePersistentVolumes(t *testing.T) {