| |
| |
| |
| |
| |
| |
| |
| |
| import os |
| import random |
| |
| |
| |
| import sys |
| import unittest |
|
|
| import docker.errors |
|
|
| panelPath = '/www/server/panel/' |
| os.chdir(panelPath) |
| sys.path.insert(0, panelPath + "class/") |
| import public |
| from btdockerModel import dk_public as dp |
| from btdockerModel import containerModel as dc |
|
|
|
|
| class TestParamManage(): |
| ''' |
| @name 获取字典对象 |
| @author wzz <2023/12/1 下午 2:32> |
| @param 参数名<数据类型> 参数描述 |
| @return 数据类型 |
| ''' |
|
|
| def get_dict_obj(self): |
| return public.dict_obj() |
|
|
| ''' |
| @name 获取所有容器列表 |
| @author wzz <2023/12/1 下午 2:32> |
| @param 参数名<数据类型> 参数描述 |
| @return 数据类型 |
| ''' |
|
|
| def get_container_list(self): |
| return dc.main().get_list(self.get_dict_obj()) |
|
|
| ''' |
| @name 获取随机容器id |
| @author wzz <2023/12/1 下午 2:32> |
| @param 参数名<数据类型> 参数描述 |
| @return 数据类型 |
| ''' |
|
|
| def get_container_id(self): |
| return self.get_container_list()['container_list'][0]['id'] |
|
|
| class TestContainerModel(unittest.TestCase): |
| """ |
| 创建测试用例 |
| """ |
|
|
| def test_connect_client(self): |
| """ |
| 测试docker客户端连接，判断返回值是否是docker.client.DockerClient实例 |
| :return: |
| """ |
| self.assertIsInstance(dp.docker_client(), docker.client.DockerClient) |
|
|
| def test_get_list(self): |
| """ |
| 测试调用containerModel中的get_list方法，获取所有容器列表 |
| @return: |
| """ |
| get = TestParamManage().get_dict_obj() |
| self.assertIsInstance(dc.main().get_list(get), dict) |
|
|
| def test_run(self): |
| """ |
| 测试创建容器 |
| @return: |
| """ |
| get = TestParamManage().get_dict_obj() |
| get.image = 'nginx:latest' |
| get.name = 'test_{}'.format(random.randint(10000, 65535)) |
| get.port = {"80/tcp": format(random.randint(10000, 65535))} |
| get.volumes = {"/www/wwwroot/test": {"bind": "/usr/share/nginx/html", "mode": "rw"}} |
| get.environment = '' |
| get.labels = '' |
| get.network = '' |
| get.cpu_quota = 0 |
| self.assertEqual(dc.main().run(get), {"status": True, "msg": "容器创建成功！"}) |
|
|
| def test_run_cmd(self): |
| ''' |
| @name 测试命令创建容器 |
| @author wzz <2023/11/30 下午 3:07> |
| @param get.cmd 包含docker run的命令，不能含有危险字符 |
| @return {"status": bool, "msg": info} |
| ''' |
| get = TestParamManage().get_dict_obj() |
| get.cmd = 'docker run -d --name test_{} nginx:latest'.format(random.randint(10000, 65535)) |
| self.assertEqual(dc.main().run_cmd(get), {"status": True, "msg": "命令已执行完毕！"}) |
|
|
| def test_upgrade_container(self): |
| ''' |
| @name 测试更新容器 |
| @author wzz <2023/12/1 上午 9:53> |
| @param 参数名<数据类型> 参数描述 |
| @return 数据类型 |
| ''' |
| get = TestParamManage().get_dict_obj() |
| get.image = 'latest' |
| container_list = dc.main().get_list(get) |
|
|
| get.id = container_list['container_list'][0]['id'] |
| get.upgrade = '1' |
| self.assertEqual(dc.main().upgrade_container(get), {"status": True, "msg": "更新成功！"}) |
|
|
| def test_upgrade_container_edit(self): |
| ''' |
| @name 测试编辑容器 |
| @author wzz <2023/12/1 上午 9:53> |
| @param 参数名<数据类型> 参数描述 |
| @return 数据类型 |
| ''' |
| get = TestParamManage().get_dict_obj() |
| get.id = TestParamManage().get_container_id() |
| get.image = 'nginx:latest' |
| get.upgrade = '0' |
| get.new_name = 'test_{}'.format(random.randint(10000, 65535)) |
| get.new_network = 'btnet' |
| get.new_ip_address = '172.18.0.{}'.format(random.randint(2, 245)) |
| get.new_command = '' |
| get.new_entrypoint = '' |
| get.new_auto_remove = '0' |
| get.new_privileged = '1' |
| get.new_restart_policy = {"Name": "always"} |
| get.new_mem_reservation = '0' |
| get.new_cpu_quota = 0 |
| get.new_mem_limit = '0' |
| get.new_labels = 'maintainer=NGINX Docker Maintainers <docker-maint@nginx.com>,,,,,,,,,,,,,,,,,\n' |
| get.new_environment = "PATH=/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin,,,,,,,,,,,,,,,,,\nNGINX_VERSION=1.25.3,,,,,,,,,,,,,,,,,\nNJS_VERSION=0.8.2,,,,,,,,,,,,,,,,,\nPKG_RELEASE=1~bookworm,,,,,,,,,,,,,,,,,\n" |
| get.new_port = {"80/tcp": format(random.randint(10000, 65535))} |
| get.new_publish_all_ports = '0' |
| get.new_volumes = {"/www/wwwroot/test": {"bind": "/usr/share/nginx/html", "mode": "rw"}} |
| result = dc.main().upgrade_container(get) |
| self.assertTrue(result['status'], result['msg']) |
|
|
| def test_commit(self): |
| ''' |
| @name 测试生成镜像和导出成文件 |
| @author wzz <2023/12/1 下午 12:03> |
| @param 参数名<数据类型> 参数描述 |
| @return 数据类型 |
| ''' |
| get = TestParamManage().get_dict_obj() |
| get.id = TestParamManage().get_container_id() |
| chars = 'qwertyuiopasdfghjklzxcvbnm0123456789' |
| get.repository = 'akaishuichi/test{}'.format(random.choice(chars)) |
| get.tag = 'latest' |
| get.message = '' |
| get.author = '' |
| get.path = '/www/dockertest' |
| get.name = 'test_{}'.format(random.randint(10000, 65535)) |
|
|
| result = dc.main().commit(get) |
| self.assertTrue(result['status'], result['msg']) |
|
|
| def test_export(self): |
| ''' |
| @name 测试容器导出文件tar |
| @author wzz <2023/12/1 下午 2:05> |
| @param 参数名<数据类型> 参数描述 |
| @return 数据类型 |
| ''' |
| get = TestParamManage().get_dict_obj() |
| get.id = TestParamManage().get_container_id() |
| get.path = '/www/dockertest' |
| get.name = 'test_{}'.format(random.randint(10000, 65535)) |
| result = dc.main().export(get) |
| self.assertTrue(result['status'], result['msg']) |
|
|
| def test_docker_shell(self): |
| ''' |
| @name 获取容器执行命令 |
| @author wzz <2023/12/1 下午 2:10> |
| @param 参数名<数据类型> 参数描述 |
| @return 数据类型 |
| ''' |
| get = TestParamManage().get_dict_obj() |
| get.id = TestParamManage().get_container_id() |
| result = dc.main().docker_shell(get) |
| print(result['msg']) |
| self.assertTrue(result['status'], result['msg']) |
|
|
| def test_del_container(self): |
| ''' |
| @name 测试删除指定容器 |
| @author wzz <2023/12/1 下午 2:12> |
| @param 参数名<数据类型> 参数描述 |
| @return 数据类型 |
| ''' |
| get = TestParamManage().get_dict_obj() |
| get.id = TestParamManage().get_container_id() |
| result = dc.main().del_container(get) |
| self.assertTrue(result['status'], result['msg']) |
|
|
| def test_set_container_status(self): |
| ''' |
| @name 设置容器状态 |
| @author wzz <2023/12/1 下午 2:14> |
| @param 参数名<数据类型> 参数描述 |
| @return 数据类型 |
| ''' |
| get = TestParamManage().get_dict_obj() |
| get.id = TestParamManage().get_container_id() |
| get.status = 'stop' |
| result = dc.main().set_container_status(get) |
| self.assertTrue(result['status'], result['msg']) |
| get.status = 'start' |
| result = dc.main().set_container_status(get) |
| self.assertTrue(result['status'], result['msg']) |
| get.status = 'pause' |
| result = dc.main().set_container_status(get) |
| self.assertTrue(result['status'], result['msg']) |
| get.status = 'unpause' |
| result = dc.main().set_container_status(get) |
| self.assertTrue(result['status'], result['msg']) |
| get.status = 'reload' |
| result = dc.main().set_container_status(get) |
| self.assertTrue(result['status'], result['msg']) |
| get.status = 'kill' |
| result = dc.main().set_container_status(get) |
| self.assertTrue(result['status'], result['msg']) |
| get.status = 'start' |
| result = dc.main().set_container_status(get) |
| self.assertTrue(result['status'], result['msg']) |
|
|
| |
| |
| |
| |
| |
| |
| |
|
|
|
|
|
|
| if __name__ == '__main__': |
| |
| |
| suite = unittest.TestSuite() |
| suite.addTest(TestContainerModel('test_connect_client')) |
| suite.addTest(TestContainerModel('test_get_list')) |
| suite.addTest(TestContainerModel('test_run')) |
| suite.addTest(TestContainerModel('test_run_cmd')) |
| suite.addTest(TestContainerModel('test_upgrade_container')) |
| suite.addTest(TestContainerModel('test_upgrade_container_edit')) |
| suite.addTest(TestContainerModel('test_commit')) |
| suite.addTest(TestContainerModel('test_export')) |
| suite.addTest(TestContainerModel('test_docker_shell')) |
| suite.addTest(TestContainerModel('test_del_container')) |
| suite.addTest(TestContainerModel('test_set_container_status')) |
|
|
| |
| runner = unittest.TextTestRunner() |
| runner.run(suite) |
|
|
