tests/test_libjuju.py

# Copyright 2019 Canonical Ltd.
#
# Licensed 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 base64
import juju
import logging
import n2vc.exceptions
from n2vc.vnf import N2VC  # noqa: F401
import os
import pytest
import re
import ssl
import sys

MODEL_NAME = '5e4e7cb0-5678-4b82-97da-9e4a1b51f5d5'

class TestN2VC(object):

    @classmethod
    def setup_class(self):
        """ setup any state specific to the execution of the given class (which
        usually contains tests).
        """
        # Initialize instance variable(s)
        self.log = logging.getLogger()
        self.log.level = logging.DEBUG

    @classmethod
    def teardown_class(self):
        """ teardown any state that was previously setup with a call to
        setup_class.
        """
        pass

    """Utility functions"""
    def get_n2vc(self, params={}):
        """Return an instance of N2VC.VNF."""


        # Extract parameters from the environment in order to run our test
        vca_host = params['VCA_HOST']
        vca_port = params['VCA_PORT']
        vca_user = params['VCA_USER']
        vca_charms = params['VCA_CHARMS']
        vca_secret = params['VCA_SECRET']
        vca_cacert = params['VCA_CACERT']
        vca_public_key = params['VCA_PUBLIC_KEY']

        client = n2vc.vnf.N2VC(
            log=self.log,
            server=vca_host,
            port=vca_port,
            user=vca_user,
            secret=vca_secret,
            artifacts=vca_charms,
            juju_public_key=vca_public_key,
            ca_cert=vca_cacert,
        )
        return client

    """Tests"""

    def test_vendored_libjuju(self):
        """Test the module import for our vendored version of libjuju.

        Test and verify that the version of libjuju being imported by N2VC is our
        vendored version, not one installed externally.
        """
        for name in sys.modules:
            if name.startswith("juju"):
                module = sys.modules[name]
                if getattr(module, "__file__"):
                    print(getattr(module, "__file__"))
                    assert re.search('n2vc', module.__file__, re.IGNORECASE)

                    # assert module.__file__.find("N2VC")
                    # assert False
        return

    @pytest.mark.asyncio
    async def test_connect_invalid_cacert(self):
        params = {
            'VCA_HOST': os.getenv('VCA_HOST', '127.0.0.1'),
            'VCA_PORT': os.getenv('VCA_PORT', 17070),
            'VCA_USER': os.getenv('VCA_USER', 'admin'),
            'VCA_SECRET': os.getenv('VCA_SECRET', 'admin'),
            'VCA_CHARMS': os.getenv('VCA_CHARMS', None),
            'VCA_PUBLIC_KEY': os.getenv('VCA_PUBLIC_KEY', None),
            'VCA_CACERT': 'invalidcacert',
        }
        with pytest.raises(n2vc.exceptions.InvalidCACertificate):
            client = self.get_n2vc(params)


    @pytest.mark.asyncio
    async def test_login(self):
        """Test connecting to libjuju."""
        params = {
            'VCA_HOST': os.getenv('VCA_HOST', '127.0.0.1'),
            'VCA_PORT': os.getenv('VCA_PORT', 17070),
            'VCA_USER': os.getenv('VCA_USER', 'admin'),
            'VCA_SECRET': os.getenv('VCA_SECRET', 'admin'),
            'VCA_CHARMS': os.getenv('VCA_CHARMS', None),
            'VCA_PUBLIC_KEY': os.getenv('VCA_PUBLIC_KEY', None),
            'VCA_CACERT': os.getenv('VCA_CACERT', "invalidcacert"),
        }

        client = self.get_n2vc(params)

        await client.login()
        assert client.authenticated

        await client.logout()
        assert client.authenticated is False

    @pytest.mark.asyncio
    async def test_model(self):
        """Test models."""
        params = {
            'VCA_HOST': os.getenv('VCA_HOST', '127.0.0.1'),
            'VCA_PORT': os.getenv('VCA_PORT', 17070),
            'VCA_USER': os.getenv('VCA_USER', 'admin'),
            'VCA_SECRET': os.getenv('VCA_SECRET', 'admin'),
            'VCA_CHARMS': os.getenv('VCA_CHARMS', None),
            'VCA_PUBLIC_KEY': os.getenv('VCA_PUBLIC_KEY', None),
            'VCA_CACERT': os.getenv('VCA_CACERT', "invalidcacert"),
        }

        client = self.get_n2vc(params)

        await client.login()
        assert client.authenticated

        self.log.debug("Creating model {}".format(MODEL_NAME))
        await client.CreateNetworkService(MODEL_NAME)

        # assert that model exists
        model = await client.controller.get_model(MODEL_NAME)
        assert model

        await client.DestroyNetworkService(MODEL_NAME)

        # Wait for model to be destroyed
        import time
        time.sleep(5)

        with pytest.raises(juju.errors.JujuAPIError):
            model = await client.controller.get_model(MODEL_NAME)

        await client.logout()
        assert client.authenticated is False