Add new k8s genie-cni calico+flannel test
Change-Id: I9441cf3224358d272abe7d647d7209463f7cae0c
Related-PROD: PROD-21550
diff --git a/tcp_tests/tests/system/test_k8s_actions.py b/tcp_tests/tests/system/test_k8s_actions.py
index 1d28a5f..c0f1b94 100644
--- a/tcp_tests/tests/system/test_k8s_actions.py
+++ b/tcp_tests/tests/system/test_k8s_actions.py
@@ -13,6 +13,8 @@
# under the License.
import pytest
+import netaddr
+import os
from tcp_tests import logger
from tcp_tests import settings
@@ -140,16 +142,9 @@
pytest.skip("Test requires metallb addon enabled")
show_step(2)
- pods = k8s_deployed.api.pods.list(namespace="metallb-system")
-
- def is_pod_exists_with_prefix(prefix):
- for pod in pods:
- if pod.name.startswith(prefix) and pod.phase == 'Running':
- return True
- return False
-
- assert is_pod_exists_with_prefix("controller")
- assert is_pod_exists_with_prefix("speaker")
+ ns = "metallb-system"
+ assert k8s_deployed.is_pod_exists_with_prefix("controller", ns)
+ assert k8s_deployed.is_pod_exists_with_prefix("speaker", ns)
show_step(3)
samples = []
@@ -175,3 +170,111 @@
show_step(7)
for sample in samples:
assert sample.is_service_available(external=True)
+
+ @pytest.mark.grap_versions
+ @pytest.mark.fail_snapshot
+ def test_k8s_genie_flannel(self, show_step, underlay, salt_deployed,
+ k8s_deployed, k8s_copy_sample_testdata):
+ """Test genie-cni+flannel cni setup
+
+ Scenario:
+ 1. Setup Kubernetes cluster with genie cni and flannel
+ 2. Check that flannel pods created in kube-system namespace
+ 3. Create sample deployment with flannel cni annotation
+ 4. Check that the deployment have 1 ip addresses from cni provider
+ 5. Create sample deployment with calico cni annotation
+ 6. Check that the deployment have 1 ip addresses from cni provider
+ 7. Create sample deployment with multi-cni annotation
+ 8. Check that the deployment have 2 ip addresses from different
+ cni providers
+ 9. Create sample deployment without cni annotation
+ 10. Check that the deployment have 1 ip address
+ 11. Check pods availability
+ 12. Run conformance
+ 13. Check pods availability
+ """
+ show_step(1)
+
+ # Find out calico and flannel networks
+ tgt_k8s_control = "I@kubernetes:control:enabled:True"
+
+ flannel_pillar = salt_deployed.get_pillar(
+ tgt=tgt_k8s_control,
+ pillar="kubernetes:master:network:flannel:private_ip_range")[0]
+ flannel_network = netaddr.IPNetwork(flannel_pillar.values()[0])
+ LOG.info("Flannel network: {}".format(flannel_network))
+
+ calico_network_pillar = salt_deployed.get_pillar(
+ tgt=tgt_k8s_control, pillar="_param:calico_private_network")[0]
+ calico_netmask_pillar = salt_deployed.get_pillar(
+ tgt=tgt_k8s_control, pillar="_param:calico_private_netmask")[0]
+ calico_network = netaddr.IPNetwork(
+ "{0}/{1}".format(calico_network_pillar.values()[0],
+ calico_netmask_pillar.values()[0]))
+ LOG.info("Calico network: {}".format(calico_network))
+
+ show_step(2)
+ assert k8s_deployed.is_pod_exists_with_prefix("kube-flannel-",
+ "kube-system")
+
+ data_dir = os.path.join(os.path.dirname(__file__), 'testdata/k8s')
+ show_step(3)
+ flannel_pod = k8s_deployed.create_pod_from_file(
+ os.path.join(data_dir, 'pod-sample-flannel.yaml'))
+
+ show_step(4)
+ flannel_ips = k8s_deployed.get_pod_ips_from_container(flannel_pod.name)
+ assert len(flannel_ips) == 1
+ assert netaddr.IPAddress(flannel_ips[0]) in flannel_network
+
+ show_step(5)
+ calico_pod = k8s_deployed.create_pod_from_file(
+ os.path.join(data_dir, 'pod-sample-calico.yaml'))
+
+ show_step(6)
+ calico_ips = k8s_deployed.get_pod_ips_from_container(calico_pod.name)
+ assert len(calico_ips) == 1
+ assert netaddr.IPAddress(calico_ips[0]) in calico_network
+
+ show_step(7)
+ multicni_pod = k8s_deployed.create_pod_from_file(
+ os.path.join(data_dir, 'pod-sample-multicni.yaml'))
+
+ show_step(8)
+ multicni_ips = \
+ k8s_deployed.get_pod_ips_from_container(multicni_pod.name)
+ assert len(multicni_ips) == 2
+ for net in [calico_network, flannel_network]:
+ assert netaddr.IPAddress(multicni_ips[0]) in net or \
+ netaddr.IPAddress(multicni_ips[1]) in net
+
+ show_step(9)
+ nocni_pod = k8s_deployed.create_pod_from_file(
+ os.path.join(data_dir, 'pod-sample.yaml'))
+
+ show_step(10)
+ nocni_ips = k8s_deployed.get_pod_ips_from_container(nocni_pod.name)
+ assert len(nocni_ips) == 1
+ assert (netaddr.IPAddress(nocni_ips[0]) in calico_network or
+ netaddr.IPAddress(nocni_ips[0]) in flannel_network)
+
+ show_step(11)
+
+ def check_pod_availability(ip):
+ assert "Hello Kubernetes!" in k8s_deployed.curl(
+ "http://{}:8080".format(ip))
+
+ def check_pods_availability():
+ check_pod_availability(flannel_ips[0])
+ check_pod_availability(calico_ips[0])
+ check_pod_availability(multicni_ips[0])
+ check_pod_availability(multicni_ips[1])
+ check_pod_availability(nocni_ips[0])
+
+ check_pods_availability()
+
+ show_step(12)
+ k8s_deployed.run_conformance()
+
+ show_step(13)
+ check_pods_availability()