public_git
/
postgres-operator
mirror of https://github.com/zalando/postgres-operator.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

add a test for multiple namespaces

This commit is contained in:
Sergey Dudoladov 2019-05-24 15:53:36 +02:00
parent 4bb20da4f5
commit 7075d11975
1 changed files with 24 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

26
e2e/tests/test_e2e.py
View File

@ -32,6 +32,12 @@ class EndToEndTestCase(unittest.TestCase):
# set a single k8s wrapper for all tests # set a single k8s wrapper for all tests
k8s = cls.k8s = K8s() k8s = cls.k8s = K8s()
# operator deploys pod service account there on start up
# needed for test_multi_namespace_support()
cls.namespace = "test"
v1_namespace = client.V1Namespace(metadata=client.V1ObjectMeta(name=cls.namespace))
k8s.api.core_v1.create_namespace(v1_namespace)
# submit the most recent operator image built on the Docker host # submit the most recent operator image built on the Docker host
with open("manifests/postgres-operator.yaml", 'r+') as f: with open("manifests/postgres-operator.yaml", 'r+') as f:
operator_deployment = yaml.load(f, Loader=yaml.Loader) operator_deployment = yaml.load(f, Loader=yaml.Loader)
@ -52,6 +58,22 @@ class EndToEndTestCase(unittest.TestCase):
k8s.create_with_kubectl("manifests/minimal-postgres-manifest.yaml") k8s.create_with_kubectl("manifests/minimal-postgres-manifest.yaml")
k8s.wait_for_pod_start('spilo-role=master') k8s.wait_for_pod_start('spilo-role=master')
@timeout_decorator.timeout(TEST_TIMEOUT_SEC)
def test_multi_namespace_support(self):
'''
Create a customized Postgres cluster in a non-default namespace.
'''
k8s = self.k8s
with open("manifests/complete-postgres-manifest.yaml", 'r+') as f:
pg_manifest = yaml.load(f, Loader=yaml.Loader)
pg_manifest["metadata"]["namespace"] = self.namespace
yaml.dump(pg_manifest, f, Dumper=yaml.Dumper)
k8s.create_with_kubectl("manifests/complete-postgres-manifest.yaml")
k8s.wait_for_pod_start("spilo-role=master", self.namespace)
self.assert_master_is_unique(self.namespace, version="acid-test-cluster")
@timeout_decorator.timeout(TEST_TIMEOUT_SEC) @timeout_decorator.timeout(TEST_TIMEOUT_SEC)
def test_scaling(self): def test_scaling(self):
""" """
@ -190,14 +212,14 @@ class EndToEndTestCase(unittest.TestCase):
self.assertTrue(0 == len(jobs), self.assertTrue(0 == len(jobs),
"Expected 0 logical backup jobs, found {}".format(len(jobs))) "Expected 0 logical backup jobs, found {}".format(len(jobs)))
def assert_master_is_unique(self, namespace='default'): def assert_master_is_unique(self, namespace='default', version="acid-minimal-cluster"):
""" """
Check that there is a single pod in the k8s cluster with the label "spilo-role=master" Check that there is a single pod in the k8s cluster with the label "spilo-role=master"
To be called manually after operations that affect pods To be called manually after operations that affect pods
""" """
k8s = self.k8s k8s = self.k8s
labels = 'spilo-role=master,version=acid-minimal-cluster' labels = 'spilo-role=master,version=' + version
num_of_master_pods = k8s.count_pods_with_label(labels, namespace) num_of_master_pods = k8s.count_pods_with_label(labels, namespace)
self.assertEqual(num_of_master_pods, 1, "Expected 1 master pod, found {}".format(num_of_master_pods)) self.assertEqual(num_of_master_pods, 1, "Expected 1 master pod, found {}".format(num_of_master_pods))