From 5d6a77c1088b37d425d96e305a36f359c9acf5ff Mon Sep 17 00:00:00 2001 From: jblu Date: Sun, 25 Jun 2023 23:46:10 -0500 Subject: [PATCH] updated testing --- cclient.py | 9 +++++++++ test_crane.py | 19 +++++++++---------- 2 files changed, 18 insertions(+), 10 deletions(-) diff --git a/cclient.py b/cclient.py index a25ee5d..d4cc070 100644 --- a/cclient.py +++ b/cclient.py @@ -9,6 +9,15 @@ def c_get_filtered_containers(req_obj, j_obj, host, port, access_token, endpoint c_get_containers_response = req_obj.get(url, headers={"X-API-Key": access_token}, verify=False) return c_get_containers_response.json() +def c_get_container_id(req_obj, j_obj, host, port, access_token, endpoint, containers, statuses): + statuses.append("running") + filter_string = j_obj.dumps({"name":containers,"status":statuses}) + url = f'https://{host}:{port}/api/endpoints/{endpoint}/docker/containers/json?filters={filter_string}' + c_get_container_id_response = req_obj.get(url, headers={"X-API-Key": access_token}, verify=False) + id = c_get_container_id_response.json() + id = id[0]["Id"] + return id + def c_start_container(req_obj, host, port, access_token, endpoint, cid): url = f'https://{host}:{port}/api/endpoints/{endpoint}/docker/containers/{cid}/start' c_start_container_response = req_obj.post(url, headers={"X-API-Key": access_token},verify=False) diff --git a/test_crane.py b/test_crane.py index 63d10ec..2e96a48 100644 --- a/test_crane.py +++ b/test_crane.py @@ -2,9 +2,10 @@ import unittest import requests import json from tomllib import load -from cclient import c_get_filtered_containers, c_start_container, c_stop_container +from cclient import c_get_filtered_containers, c_start_container, c_stop_container, c_get_container_id from cprocess import process_cont_list -# unittest.TestLoader.sortTestMethodsUsing = None +unittest.TestLoader.sortTestMethodsUsing = None +requests.packages.urllib3.disable_warnings() class TestCrane(unittest.TestCase): def setUp(self): @@ -14,32 +15,30 @@ class TestCrane(unittest.TestCase): self.port = self.config["portainer"]["port"] self.access_token = "ptr_ufS1nADXmrU3QSN3bvITLMQ7oOH9yo3ECb/QNwtIYJ4=" self.endpoint = self.config["portainer"]["endpoint"] - self.cid = 'ef8fee86e02b2b82acbddf6f0da1ff023f60bfe52c0b4087cac29c1686ccbac4' self.req_obj = requests self.j_obj = json - self.hypercare_containers = ['hello-world'] + self.hypercare_containers = ['hottub'] self.status_filters = ["paused","dead","created","exited","removing","restarting","created"] - self.cont_obj = c_get_filtered_containers(self.req_obj,self.j_obj, self.host, self.port, self.access_token, self.endpoint,self.hypercare_containers,self.status_filters) def test_c_get_filtered_containers(self): + self.cont_obj = c_get_filtered_containers(self.req_obj,self.j_obj, self.host, self.port, self.access_token, self.endpoint,self.hypercare_containers,self.status_filters) self.assertTrue(self.cont_obj, "No cont object returned by c_get_filtered_containers.") def test_c_start_container(self): + self.cid = c_get_container_id(self.req_obj,self.j_obj, self.host, self.port, self.access_token, self.endpoint, self.hypercare_containers,self.status_filters) + c_stop_container(self.req_obj, self.host, self.port, self.access_token, self.endpoint, self.cid) self.c_start_container_response = c_start_container(self.req_obj, self.host, self.port, self.access_token, self.endpoint, self.cid) - # print(self.c_start_container_response) self.assertTrue(self.c_start_container_response, "No c_start_container_resonse returned by c_start_container.") # 204 success 304 already on def test_c_process_cont_list(self): + self.cont_obj = c_get_filtered_containers(self.req_obj,self.j_obj, self.host, self.port, self.access_token, self.endpoint,self.hypercare_containers,self.status_filters) self.assertTrue(process_cont_list(self.cont_obj, c_start_container, self.req_obj, self.host, self.port, self.access_token, self.endpoint)) def test_c_stop_container(self): + self.cid = c_get_container_id(self.req_obj,self.j_obj, self.host, self.port, self.access_token, self.endpoint, self.hypercare_containers,self.status_filters) self.c_stop_container_response = c_stop_container(self.req_obj, self.host, self.port, self.access_token, self.endpoint, self.cid) - # print(self.c_stop_container_response) self.assertTrue(self.c_stop_container_response, "No c_start_container_resonse returned by c_start_container.") - - def test_z_tear_down(self): - c_stop_container(self.req_obj, self.host, self.port, self.access_token, self.endpoint, self.cid) if __name__ == '__main__':