jblu/crane
1
0
Fork 0
Code Issues 4 Pull Requests Packages Projects Releases Wiki Activity

updated testing
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
jblu 2023-06-25 23:46:10 -05:00
parent f60c3be987
commit 5d6a77c108
2 changed files with 18 additions and 10 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
cclient.py
View File

@ -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)

19
test_crane.py
View File

@ -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__':