tests/test_k8s_juju_conn.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 argparse
  16 import asyncio
  17 import logging
  18 import n2vc.k8s_juju_conn
  19 from base import get_juju_public_key
  20 import os
  21 from osm_common.fslocal import FsLocal
  22 import subprocess
  23 import yaml
  24
  25
  26 def get_args():
  27     parser = argparse.ArgumentParser()
  28     parser.add_argument("--cluster_uuid", help='The UUID of an existing cluster to use', default=None)
  29     parser.add_argument("--reset", action="store_true")
  30     return parser.parse_args()
  31
  32 async def main():
  33
  34     args = get_args()
  35
  36     reuse_cluster_uuid = args.cluster_uuid
  37
  38     log = logging.getLogger()
  39     log.level = logging.DEBUG
  40
  41     # Extract parameters from the environment in order to run our tests
  42     vca_host = os.getenv('VCA_HOST', '127.0.0.1')
  43     vca_port = os.getenv('VCA_PORT', 17070)
  44     vca_user = os.getenv('VCA_USER', 'admin')
  45     vca_charms = os.getenv('VCA_CHARMS', None)
  46     vca_secret = os.getenv('VCA_SECRET', None)
  47     vca_ca_cert = os.getenv('VCA_CACERT', None)
  48
  49     # Get the Juju Public key
  50     juju_public_key = get_juju_public_key()
  51     if juju_public_key:
  52         with open(juju_public_key, 'r') as f:
  53             juju_public_key = f.read()
  54     else:
  55         raise Exception("No Juju Public Key found")
  56
  57     storage = {
  58         'driver': 'local',
  59         'path': '/srv/app/storage'
  60     }
  61     fs = FsLocal()
  62     fs.fs_connect(storage)
  63
  64     client = n2vc.k8s_juju_conn.K8sJujuConnector(
  65         kubectl_command='/snap/bin/kubectl',
  66         juju_command='/snap/bin/juju',
  67         fs=fs,
  68         db=None,
  69     )
  70
  71     # kubectl config view --raw
  72     # microk8s.config
  73
  74     # if microk8s then
  75     # kubecfg = subprocess.getoutput('microk8s.config')
  76     # else
  77     kubecfg = subprocess.getoutput('kubectl config view --raw')
  78     # print(kubecfg)
  79
  80     # k8screds = yaml.load(kubecfg, Loader=yaml.FullLoader)
  81     namespace = 'testing'
  82     kdu_model = "./tests/bundles/k8s-zookeeper.yaml"
  83
  84     """init_env"""
  85     cluster_uuid, _ = await client.init_env(kubecfg, namespace, reuse_cluster_uuid=reuse_cluster_uuid)
  86     print(cluster_uuid)
  87
  88     if not reuse_cluster_uuid:
  89         # This is a new cluster, so install to it
  90
  91         """install"""
  92         # async def install(self, cluster_uuid, kdu_model, atomic=True, timeout=None, params=None):
  93         # TODO: Re-add storage declaration to bundle. The API doesn't support the storage option yet. David is investigating.
  94
  95         # Deploy the bundle
  96         kdu_instance = await client.install(cluster_uuid, kdu_model, atomic=True, timeout=600)
  97
  98         if kdu_instance:
  99             # Inspect
 100             print("Getting status")
 101             status = await client.status_kdu(cluster_uuid, kdu_instance)
 102             print(status)
 103
 104     # Inspect the bundle
 105     config = await client.inspect_kdu(kdu_model)
 106     print(config)
 107
 108     readme = await client.help_kdu(kdu_model)
 109     # print(readme)
 110
 111
 112     """upgrade
 113     Upgrade to a newer version of the bundle
 114     """
 115     kdu_model_upgrade = "./tests/bundles/k8s-zookeeper-upgrade.yaml"
 116     upgraded = await client.upgrade(cluster_uuid, namespace, kdu_model=kdu_model_upgrade)
 117
 118     kdu_model_upgrade = "./tests/bundles/k8s-zookeeper-downgrade.yaml"
 119     upgraded = await client.upgrade(cluster_uuid, namespace, kdu_model=kdu_model_upgrade)
 120
 121     """uninstall"""
 122
 123     """reset"""
 124     if args.reset:
 125         await client.reset(cluster_uuid)
 126
 127     await client.logout()
 128
 129     print("Done")
 130
 131 if __name__ == "__main__":
 132     asyncio.run(main())