Repository: kubernetes/kubernetes
File: test/e2e/network/netpol/network_policy.go (Go)

Code before:

Each test specifies the exact pod topology it needs (for example: x/a, x/b, y/a, z/a).
*/

var _ = common.SIGDescribe("Netpol", func() {
	f := framework.NewDefaultFramework("netpol")
	f.SkipNamespaceCreation = true // we create our own 3 test namespaces, we don't need the default one
	f.NamespacePodSecurityLevel = admissionapi.LevelBaseline

	ginkgo.Context("NetworkPolicy between server and client", func() {
		var k8s *kubeManager

		f.It("should support a 'default-deny-ingress' policy", feature.NetworkPolicy, func(ctx context.Context) {

			// Only poll TCP
			protocols := []v1.Protocol{protocolTCP}

			// Only testing port 80
			ports := []int32{80}

			k8s = initializeResources(ctx, f, protocols, ports, "x/a", "x/b", "y/a", "y/b")

			// Only going to make a policy in namespace X
			nsX, _, _ := getK8sNamespaces(k8s)
			policy := GenNetworkPolicyWithNameAndPodSelector("deny-ingress", metav1.LabelSelector{}, SetSpecIngressRules())

			// Create the policy
			CreatePolicy(ctx, k8s, policy, nsX)

			// Make a truth table of connectivity for all pods in ns x y z
			reachability := NewReachability(k8s.AllPodStrings(), true)
			// Set the nsX as false, since it has a policy that blocks traffic
			reachability.ExpectPeer(&Peer{}, &Peer{Namespace: nsX}, false)

			// Confirm that the real world connectivity matches our matrix
			ValidateOrFail(k8s, &TestCase{ToPort: 80, Protocol: v1.ProtocolTCP, Reachability: reachability})
		})

		f.It("should support a 'default-deny-all' policy", feature.NetworkPolicy, func(ctx context.Context) {
			protocols := []v1.Protocol{protocolTCP}
			ports := []int32{80}
			k8s = initializeResources(ctx, f, protocols, ports, "x/a", "x/b", "y/a")
			nsX, _, _ := getK8sNamespaces(k8s)

			policy := GenNetworkPolicyWithNameAndPodSelector("deny-all", metav1.LabelSelector{}, SetSpecIngressRules(), SetSpecEgressRules())
			CreatePolicy(ctx, k8s, policy, nsX)

			reachability := NewReachability(k8s.AllPodStrings(), true)
			reachability.ExpectPeer(&Peer{}, &Peer{Namespace: nsX}, false)
			reachability.ExpectPeer(&Peer{Namespace: nsX}, &Peer{}, false)


Review comment (suggestion):
Each test case should have an explanation of why you chose that particular set of pods (and not more or less).

So in this case, for instance:

```
// Namespace X has a default-deny, so we need to test that connectivity isn't
// allowed between 2 pods in X, or from a pod in Y to a pod in X, but it is
// allowed between 2 pods in Y, and from a pod in X to a pod in Y.
```

Code after:
showing the connectivity of pods in other namespaces is key information to show whether a network policy is working as intended or not.

Each test specifies the exact pod topology it needs (for example: x/a, x/b, y/a, z/a).
*/

var _ = common.SIGDescribe("Netpol", func() {
	f := framework.NewDefaultFramework("netpol")
	f.SkipNamespaceCreation = true // we create our own 3 test namespaces, we don't need the default one
	f.NamespacePodSecurityLevel = admissionapi.LevelBaseline

	ginkgo.Context("NetworkPolicy between server and client", func() {
		var k8s *kubeManager

		f.It("should support a 'default-deny-ingress' policy", feature.NetworkPolicy, func(ctx context.Context) {

			// Only poll TCP
			protocols := []v1.Protocol{protocolTCP}

			// Only testing port 80
			ports := []int32{80}

			// Namespace X has a default-deny-ingress policy, so we need 2 pods in X to
			// verify X-to-X ingress is blocked, and 2 pods in Y to verify Y-to-X is
			// also blocked while X-to-Y and Y-to-Y traffic remain allowed.
			// (In later tests, we just assume that a policy in X will not affect Y-to-Y traffic.)
			k8s = initializeResources(ctx, f, protocols, ports, "x/a", "x/b", "y/a", "y/b")

			// Only going to make a policy in namespace X
			nsX, _, _ := getK8sNamespaces(k8s)
			policy := GenNetworkPolicyWithNameAndPodSelector("deny-ingress", metav1.LabelSelector{}, SetSpecIngressRules())

			// Create the policy
			CreatePolicy(ctx, k8s, policy, nsX)

			// Make a truth table of connectivity for all pods in ns x y z
			reachability := NewReachability(k8s.AllPodStrings(), true)
			// Set the nsX as false, since it has a policy that blocks traffic
			reachability.ExpectPeer(&Peer{}, &Peer{Namespace: nsX}, false)

			// Confirm that the real world connectivity matches our matrix
			ValidateOrFail(k8s, &TestCase{ToPort: 80, Protocol: v1.ProtocolTCP, Reachability: reachability})
		})

		f.It("should support a 'default-deny-all' policy", feature.NetworkPolicy, func(ctx context.Context) {
			protocols := []v1.Protocol{protocolTCP}
			ports := []int32{80}
			// Namespace X has a default-deny for both ingress and egress, so we need 2
			// pods in X to verify X-to-X traffic is blocked, and a pod in Y to verify
			// X<->Y traffic is blocked in both directions.
			k8s = initializeResources(ctx, f, protocols, ports, "x/a", "x/b", "y/a")
			nsX, _, _ := getK8sNamespaces(k8s)