1#@!#!123s

: / opt / imh-python / lib / python3.9 / site-packages / libcloud / test / container /
Filename : test_rancher.py
back
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import sys

from libcloud.test import unittest

from libcloud.container.base import ContainerImage
from libcloud.container.drivers.rancher import RancherContainerDriver

from libcloud.utils.py3 import httplib
from libcloud.test.secrets import CONTAINER_PARAMS_RANCHER
from libcloud.test.file_fixtures import ContainerFileFixtures
from libcloud.test import MockHttp


# --------------------------------------------------------------------------- #
# Mock Classes

class RancherMockHttp(MockHttp):
    fixtures = ContainerFileFixtures('rancher')

    def _v1_environments(self, method, url, body, headers):
        if method == 'GET':
            return (httplib.OK, self.fixtures.load('ex_list_stacks.json'), {},
                    httplib.responses[httplib.OK])
        else:
            return (httplib.OK, self.fixtures.load('ex_deploy_stack.json'), {},
                    httplib.responses[httplib.OK])

    def _v1_environments_1e9(self, method, url, body, headers):
        return (httplib.OK, self.fixtures.load('ex_deploy_stack.json'), {},
                httplib.responses[httplib.OK])

    def _v1_environments_1e10(self, method, url, body, headers):
        return (httplib.OK, self.fixtures.load('ex_destroy_stack.json'), {},
                httplib.responses[httplib.OK])

    def _v1_environments_1e1(self, method, url, body, headers):
        return (httplib.OK, self.fixtures.load('ex_activate_stack.json'), {},
                httplib.responses[httplib.OK])

    def _v1_services(self, method, url, body, headers):
        if '?healthState=healthy' in url:
            return (httplib.OK, self.fixtures.load('ex_search_services.json'),
                    {}, httplib.responses[httplib.OK])
        elif method == 'GET':
            return (httplib.OK, self.fixtures.load('ex_list_services.json'),
                    {}, httplib.responses[httplib.OK])
        else:
            return (httplib.OK, self.fixtures.load('ex_deploy_service.json'),
                    {}, httplib.responses[httplib.OK])

    def _v1_services_1s13(self, method, url, body, headers):
        if method == 'GET':
            return (httplib.OK, self.fixtures.load('ex_deploy_service.json'),
                    {}, httplib.responses[httplib.OK])
        elif method == 'DELETE':
            return (httplib.OK, self.fixtures.load('ex_destroy_service.json'),
                    {}, httplib.responses[httplib.OK])

    def _v1_services_1s6(self, method, url, body, headers):
        return (httplib.OK, self.fixtures.load('ex_activate_service.json'), {},
                httplib.responses[httplib.OK])

    def _v1_containers(self, method, url, body, headers):
        if '?state=running' in url:
            return (httplib.OK,
                    self.fixtures.load('ex_search_containers.json'), {},
                    httplib.responses[httplib.OK])
        elif method == 'POST':
            return (httplib.OK, self.fixtures.load('deploy_container.json'),
                    {}, httplib.responses[httplib.OK])
        return (httplib.OK, self.fixtures.load('list_containers.json'), {},
                httplib.responses[httplib.OK])

    def _v1_containers_1i31(self, method, url, body, headers):
        if method == 'GET':
            return (httplib.OK, self.fixtures.load('deploy_container.json'),
                    {}, httplib.responses[httplib.OK])
        elif method == 'DELETE' or '?action=stop' in url:
            return (httplib.OK, self.fixtures.load('stop_container.json'), {},
                    httplib.responses[httplib.OK])
        elif '?action=start' in url:
            return (httplib.OK, self.fixtures.load('start_container.json'), {},
                    httplib.responses[httplib.OK])
        else:
            return (httplib.OK, self.fixtures.load('deploy_container.json'),
                    {}, httplib.responses[httplib.OK])


RancherContainerDriver.connectionCls.conn_class = RancherMockHttp
RancherMockHttp.type = None
RancherMockHttp.use_param = 'a'


# --------------------------------------------------------------------------- #
# Test Cases


class RancherContainerDriverInitTestCase(unittest.TestCase):
    """
    Tests for testing the different permutations of the driver initialization
    string.
    """

    def test_full_url_string(self):
        """
        Test a 'full' URL string, which contains a scheme, port, and base path.
        """
        path = "http://myhostname:1234/base"
        driver = RancherContainerDriver(*CONTAINER_PARAMS_RANCHER, host=path)

        self.assertEqual(driver.secure, False)
        self.assertEqual(driver.connection.host, "myhostname")
        self.assertEqual(driver.connection.port, 1234)
        self.assertEqual(driver.baseuri, "/base")

    def test_url_string_no_port(self):
        """
        Test a partial URL string, which contains a scheme, and base path.
        """
        path = "http://myhostname/base"
        driver = RancherContainerDriver(*CONTAINER_PARAMS_RANCHER, host=path,
                                        port=1234)

        self.assertEqual(driver.secure, False)
        self.assertEqual(driver.connection.host, "myhostname")
        self.assertEqual(driver.connection.port, 1234)
        self.assertEqual(driver.baseuri, "/base")

    def test_url_string_no_scheme(self):
        """
        Test a partial URL string, which contains a port, and base path.
        """
        path = "myhostname:1234/base"
        driver = RancherContainerDriver(*CONTAINER_PARAMS_RANCHER, host=path)

        self.assertEqual(driver.secure, True)
        self.assertEqual(driver.connection.host, "myhostname")
        self.assertEqual(driver.connection.port, 1234)
        self.assertEqual(driver.baseuri, "/base")

    def test_url_string_no_base_path(self):
        """
        Test a partial URL string, which contains a scheme, and a port.
        """
        path = "http://myhostname:1234"
        driver = RancherContainerDriver(*CONTAINER_PARAMS_RANCHER, host=path)

        self.assertEqual(driver.secure, False)
        self.assertEqual(driver.connection.host, "myhostname")
        self.assertEqual(driver.connection.port, 1234)
        self.assertEqual(driver.baseuri, "/v%s" % driver.version)


class RancherContainerDriverTestCase(unittest.TestCase):

    def setUp(self):
        self.driver = RancherContainerDriver(*CONTAINER_PARAMS_RANCHER)

    # Stacks
    def test_ex_list_stacks(self):
        stacks = self.driver.ex_list_stacks()
        self.assertEqual(len(stacks), 6)
        self.assertEqual(stacks[0]['id'], "1e1")

    def test_ex_deploy_stack(self):
        stack = self.driver.ex_deploy_stack(name="newstack",
                                            environment={
                                                "root_password": "password"
                                            })
        self.assertEqual(stack['id'], "1e9")
        self.assertEqual(stack['environment']['root_password'], "password")

    def test_ex_get_stack(self):
        # also uses ex_deploy_stack.json
        stack = self.driver.ex_get_stack("1e9")
        self.assertEqual(stack['id'], "1e9")
        self.assertEqual(stack['environment']['root_password'], "password")

    def test_ex_search_stacks(self):
        stacks = self.driver.ex_search_stacks({"healthState": "healthy"})
        self.assertEqual(len(stacks), 6)
        self.assertEqual(stacks[0]['healthState'], "healthy")

    def test_ex_destroy_stack(self):
        response = self.driver.ex_destroy_stack("1e10")
        self.assertEqual(response, True)

    def test_ex_activate_stack(self):
        response = self.driver.ex_activate_stack("1e1")
        self.assertEqual(response, True)

    def test_ex_deactivate_stack(self):
        # also uses ex_activate_stack.json
        response = self.driver.ex_activate_stack("1e1")
        self.assertEqual(response, True)

    def test_ex_list_services(self):
        services = self.driver.ex_list_services()
        self.assertEqual(len(services), 4)
        self.assertEqual(services[0]['id'], "1s1")

    def test_ex_deploy_service(self):
        image = ContainerImage(
            id="hastebin",
            name="hastebin",
            path="rlister/hastebin",
            version="latest",
            driver=None
        )
        service = self.driver.ex_deploy_service(name="newservice",
                                                environment_id="1e1",
                                                image=image,
                                                environment={
                                                    "root_password": "password"
                                                })
        self.assertEqual(service['id'], "1s13")
        self.assertEqual(service['environmentId'], "1e6")
        self.assertEqual(service['launchConfig']['environment']
                         ['root_password'], "password")
        self.assertEqual(service['launchConfig']['imageUuid'],
                         "docker:rlister/hastebin:latest")

    def test_ex_get_service(self):
        # also uses ex_deploy_service.json
        service = self.driver.ex_get_service("1s13")
        self.assertEqual(service['id'], "1s13")
        self.assertEqual(service['environmentId'], "1e6")
        self.assertEqual(service['launchConfig']['environment']
                         ['root_password'], "password")

    def test_ex_search_services(self):
        services = self.driver.ex_search_services({"healthState": "healthy"})
        self.assertEqual(len(services), 2)
        self.assertEqual(services[0]['healthState'], "healthy")

    def test_ex_destroy_service(self):
        # Not sure how to do these with returns in mockhttp
        response = self.driver.ex_destroy_service("1s13")
        self.assertEqual(response, True)

    def test_ex_activate_service(self):
        response = self.driver.ex_activate_service("1s6")
        self.assertEqual(response, True)

    def test_ex_deactivate_service(self):
        # also uses ex_activate_service.json
        response = self.driver.ex_activate_service("1s6")
        self.assertEqual(response, True)

    def test_list_containers(self):
        containers = self.driver.list_containers()
        self.assertEqual(len(containers), 2)
        self.assertEqual(containers[0].id, "1i1")

    def test_deploy_container(self):
        container = self.driver.deploy_container(
            name='newcontainer',
            image=ContainerImage(
                id="hastebin",
                name="hastebin",
                path="rlister/hastebin",
                version="latest",
                driver=None
            ),
            environment={"STORAGE_TYPE": "file"},
            networkMode="managed"
        )
        self.assertEqual(container.id, '1i31')
        self.assertEqual(container.name, 'newcontainer')
        self.assertEqual(container.extra['environment'],
                         {'STORAGE_TYPE': 'file'})

    def test_get_container(self):
        # also uses ex_deploy_container.json
        container = self.driver.get_container("1i31")
        self.assertEqual(container.id, '1i31')
        self.assertEqual(container.name, 'newcontainer')
        self.assertEqual(container.extra['environment'],
                         {'STORAGE_TYPE': 'file'})

    def test_start_container(self):
        container = self.driver.get_container("1i31")
        started = container.start()
        self.assertEqual(started.id, "1i31")
        self.assertEqual(started.name, "newcontainer")
        self.assertEqual(started.state, "pending")
        self.assertEqual(started.extra['state'], "starting")

    def test_stop_container(self):
        container = self.driver.get_container("1i31")
        stopped = container.stop()
        self.assertEqual(stopped.id, "1i31")
        self.assertEqual(stopped.name, "newcontainer")
        self.assertEqual(stopped.state, "pending")
        self.assertEqual(stopped.extra['state'], "stopping")

    def test_ex_search_containers(self):
        containers = self.driver.ex_search_containers({"state": "running"})
        self.assertEqual(len(containers), 1)

    def test_destroy_container(self):
        container = self.driver.get_container("1i31")
        destroyed = container.destroy()
        self.assertEqual(destroyed.id, "1i31")
        self.assertEqual(destroyed.name, "newcontainer")
        self.assertEqual(destroyed.state, "pending")
        self.assertEqual(destroyed.extra['state'], "stopping")


if __name__ == '__main__':
    sys.exit(unittest.main())