tests/test_libjuju.py

   1 # Copyright 2019 Canonical Ltd.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain a copy of the License at
   6 #
   7 #     http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #     Unless required by applicable law or agreed to in writing, software
  10 #     distributed under the License is distributed on an "AS IS" BASIS,
  11 #     WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  12 #     See the License for the specific language governing permissions and
  13 #     limitations under the License.
  14
  15 import base64
  16 import juju
  17 import logging
  18 import n2vc.exceptions
  19 from n2vc.vnf import N2VC  # noqa: F401
  20 import os
  21 import pytest
  22 import re
  23 import ssl
  24 import sys
  25
  26 MODEL_NAME = '5e4e7cb0-5678-4b82-97da-9e4a1b51f5d5'
  27
  28 class TestN2VC(object):
  29
  30     @classmethod
  31     def setup_class(self):
  32         """ setup any state specific to the execution of the given class (which
  33         usually contains tests).
  34         """
  35         # Initialize instance variable(s)
  36         self.log = logging.getLogger()
  37         self.log.level = logging.DEBUG
  38
  39     @classmethod
  40     def teardown_class(self):
  41         """ teardown any state that was previously setup with a call to
  42         setup_class.
  43         """
  44         pass
  45
  46     """Utility functions"""
  47     def get_n2vc(self, params={}):
  48         """Return an instance of N2VC.VNF."""
  49
  50
  51         # Extract parameters from the environment in order to run our test
  52         vca_host = params['VCA_HOST']
  53         vca_port = params['VCA_PORT']
  54         vca_user = params['VCA_USER']
  55         vca_charms = params['VCA_CHARMS']
  56         vca_secret = params['VCA_SECRET']
  57         vca_cacert = params['VCA_CACERT']
  58         vca_public_key = params['VCA_PUBLIC_KEY']
  59
  60         client = n2vc.vnf.N2VC(
  61             log=self.log,
  62             server=vca_host,
  63             port=vca_port,
  64             user=vca_user,
  65             secret=vca_secret,
  66             artifacts=vca_charms,
  67             juju_public_key=vca_public_key,
  68             ca_cert=vca_cacert,
  69         )
  70         return client
  71
  72     """Tests"""
  73
  74     def test_vendored_libjuju(self):
  75         """Test the module import for our vendored version of libjuju.
  76
  77         Test and verify that the version of libjuju being imported by N2VC is our
  78         vendored version, not one installed externally.
  79         """
  80         for name in sys.modules:
  81             if name.startswith("juju"):
  82                 module = sys.modules[name]
  83                 if getattr(module, "__file__"):
  84                     print(getattr(module, "__file__"))
  85                     assert re.search('n2vc', module.__file__, re.IGNORECASE)
  86
  87                     # assert module.__file__.find("N2VC")
  88                     # assert False
  89         return
  90
  91     @pytest.mark.asyncio
  92     async def test_connect_invalid_cacert(self):
  93         params = {
  94             'VCA_HOST': os.getenv('VCA_HOST', '127.0.0.1'),
  95             'VCA_PORT': os.getenv('VCA_PORT', 17070),
  96             'VCA_USER': os.getenv('VCA_USER', 'admin'),
  97             'VCA_SECRET': os.getenv('VCA_SECRET', 'admin'),
  98             'VCA_CHARMS': os.getenv('VCA_CHARMS', None),
  99             'VCA_PUBLIC_KEY': os.getenv('VCA_PUBLIC_KEY', None),
 100             'VCA_CACERT': 'invalidcacert',
 101         }
 102         with pytest.raises(n2vc.exceptions.InvalidCACertificate):
 103             client = self.get_n2vc(params)
 104
 105
 106     @pytest.mark.asyncio
 107     async def test_login(self):
 108         """Test connecting to libjuju."""
 109         params = {
 110             'VCA_HOST': os.getenv('VCA_HOST', '127.0.0.1'),
 111             'VCA_PORT': os.getenv('VCA_PORT', 17070),
 112             'VCA_USER': os.getenv('VCA_USER', 'admin'),
 113             'VCA_SECRET': os.getenv('VCA_SECRET', 'admin'),
 114             'VCA_CHARMS': os.getenv('VCA_CHARMS', None),
 115             'VCA_PUBLIC_KEY': os.getenv('VCA_PUBLIC_KEY', None),
 116             'VCA_CACERT': os.getenv('VCA_CACERT', "invalidcacert"),
 117         }
 118
 119         client = self.get_n2vc(params)
 120
 121         await client.login()
 122         assert client.authenticated
 123
 124         await client.logout()
 125         assert client.authenticated is False
 126
 127     @pytest.mark.asyncio
 128     async def test_model(self):
 129         """Test models."""
 130         params = {
 131             'VCA_HOST': os.getenv('VCA_HOST', '127.0.0.1'),
 132             'VCA_PORT': os.getenv('VCA_PORT', 17070),
 133             'VCA_USER': os.getenv('VCA_USER', 'admin'),
 134             'VCA_SECRET': os.getenv('VCA_SECRET', 'admin'),
 135             'VCA_CHARMS': os.getenv('VCA_CHARMS', None),
 136             'VCA_PUBLIC_KEY': os.getenv('VCA_PUBLIC_KEY', None),
 137             'VCA_CACERT': os.getenv('VCA_CACERT', "invalidcacert"),
 138         }
 139
 140         client = self.get_n2vc(params)
 141
 142         await client.login()
 143         assert client.authenticated
 144
 145         self.log.debug("Creating model {}".format(MODEL_NAME))
 146         await client.CreateNetworkService(MODEL_NAME)
 147
 148         # assert that model exists
 149         model = await client.controller.get_model(MODEL_NAME)
 150         assert model
 151
 152         await client.DestroyNetworkService(MODEL_NAME)
 153
 154         # Wait for model to be destroyed
 155         import time
 156         time.sleep(5)
 157
 158         with pytest.raises(juju.errors.JujuAPIError):
 159             model = await client.controller.get_model(MODEL_NAME)
 160
 161         await client.logout()
 162         assert client.authenticated is False