tests/test_lxd.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 """
  16 This test exercises LXD, to make sure that we can:
  17 1. Create a container profile
  18 2. Launch a container with a profile
  19 3. Stop a container
  20 4. Destroy a container
  21 5. Delete a container profile
  22
  23 """
  24 import logging
  25 # import os
  26 import pytest
  27 from . import base
  28 import subprocess
  29 import shlex
  30 import tempfile
  31
  32
  33 @pytest.mark.asyncio
  34 async def test_lxd():
  35
  36     container = base.create_lxd_container(name="test-lxd")
  37     assert container is not None
  38
  39     # Get the hostname of the container
  40     hostname = container.name
  41
  42     # Delete the container
  43     base.destroy_lxd_container(container)
  44
  45     # Verify the container is deleted
  46     client = base.get_lxd_client()
  47     assert client.containers.exists(hostname) is False
  48
  49
  50 @pytest.mark.asyncio
  51 async def test_lxd_ssh():
  52
  53     with tempfile.TemporaryDirectory() as tmp:
  54         try:
  55             # Create a temporary keypair
  56             cmd = shlex.split(
  57                 "ssh-keygen -t rsa -b 4096 -N '' -f {}/id_lxd_rsa".format(
  58                     tmp,
  59                 )
  60             )
  61             subprocess.check_call(cmd)
  62         except subprocess.CalledProcessError as e:
  63             logging.debug(e)
  64             assert False
  65
  66         # Slurp the public key
  67         public_key = None
  68         with open("{}/id_lxd_rsa.pub".format(tmp), "r") as f:
  69             public_key = f.read()
  70
  71         assert public_key is not None
  72
  73         # Create the container with the keypair injected via profile
  74         container = base.create_lxd_container(
  75             public_key=public_key,
  76             name="test-lxd"
  77         )
  78         assert container is not None
  79
  80         # Get the hostname of the container
  81         hostname = container.name
  82
  83         addresses = container.state().network['eth0']['addresses']
  84         # The interface may have more than one address, but we only need
  85         # the first one for testing purposes.
  86         ipaddr = addresses[0]['address']
  87
  88         # Verify we can SSH into container
  89         try:
  90             cmd = shlex.split(
  91                 "ssh -i {}/id_lxd_rsa {} root@{} hostname".format(
  92                     tmp,
  93                     "-oStrictHostKeyChecking=no",
  94                     ipaddr,
  95                 )
  96             )
  97             subprocess.check_call(cmd)
  98         except subprocess.CalledProcessError as e:
  99             logging.debug(e)
 100             assert False
 101
 102         # Delete the container
 103         base.destroy_lxd_container(container)
 104
 105         # Verify the container is deleted
 106         client = base.get_lxd_client()
 107         assert client.containers.exists(hostname) is False
 108
 109         # Verify the container profile is deleted
 110         assert client.profiles.exists(hostname) is False