from juju.client.client import ConfigValue, ApplicationFacade
from juju.model import Model, ModelObserver
from juju.utils import block_until, run_with_interrupt
+from juju.errors import JujuError
+
+import os
+import pylxd
+import time
+import uuid
import pytest
@base.bootstrapped
@pytest.mark.asyncio
async def test_deploy_local_bundle(event_loop):
- from pathlib import Path
tests_dir = Path(__file__).absolute().parent.parent
bundle_path = tests_dir / 'bundle'
mini_bundle_file_path = bundle_path / 'mini-bundle.yaml'
assert app in model.applications
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_deploy_invalid_bundle(event_loop):
+ tests_dir = Path(__file__).absolute().parent.parent
+ bundle_path = tests_dir / 'bundle' / 'invalid.yaml'
+ async with base.CleanModel() as model:
+ with pytest.raises(JujuError):
+ await model.deploy(str(bundle_path))
+
+
@base.bootstrapped
@pytest.mark.asyncio
async def test_deploy_local_charm(event_loop):
assert len(model.machines) == 0
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_add_manual_machine_ssh(event_loop):
+
+ # Verify controller is localhost
+ async with base.CleanController() as controller:
+ cloud = await controller.get_cloud()
+ if cloud != "localhost":
+ pytest.skip('Skipping because test requires lxd.')
+
+ async with base.CleanModel() as model:
+ private_key_path = os.path.expanduser(
+ "~/.local/share/juju/ssh/juju_id_rsa"
+ )
+ public_key_path = os.path.expanduser(
+ "~/.local/share/juju/ssh/juju_id_rsa.pub"
+ )
+
+ # Use the self-signed cert generated by lxc on first run
+ crt = os.path.expanduser('~/snap/lxd/current/.config/lxc/client.crt')
+ assert os.path.exists(crt)
+
+ key = os.path.expanduser('~/snap/lxd/current/.config/lxc/client.key')
+ assert os.path.exists(key)
+
+ client = pylxd.Client(
+ endpoint="https://127.0.0.1:8443",
+ cert=(crt, key),
+ verify=False,
+ )
+
+ test_name = "test-{}-add-manual-machine-ssh".format(
+ uuid.uuid4().hex[-4:]
+ )
+
+ # create profile w/cloud-init and juju ssh key
+ public_key = ""
+ with open(public_key_path, "r") as f:
+ public_key = f.readline()
+
+ profile = client.profiles.create(
+ test_name,
+ config={'user.user-data': '#cloud-config\nssh_authorized_keys:\n- {}'.format(public_key)},
+ devices={
+ 'root': {'path': '/', 'pool': 'default', 'type': 'disk'},
+ 'eth0': {
+ 'nictype': 'bridged',
+ 'parent': 'lxdbr0',
+ 'type': 'nic'
+ }
+ }
+ )
+
+ # create lxc machine
+ config = {
+ 'name': test_name,
+ 'source': {
+ 'type': 'image',
+ 'alias': 'xenial',
+ 'mode': 'pull',
+ 'protocol': 'simplestreams',
+ 'server': 'https://cloud-images.ubuntu.com/releases',
+ },
+ 'profiles': [test_name],
+ }
+ container = client.containers.create(config, wait=True)
+ container.start(wait=True)
+
+ def wait_for_network(container, timeout=30):
+ """Wait for eth0 to have an ipv4 address."""
+ starttime = time.time()
+ while(time.time() < starttime + timeout):
+ time.sleep(1)
+ if 'eth0' in container.state().network:
+ addresses = container.state().network['eth0']['addresses']
+ if len(addresses) > 0:
+ if addresses[0]['family'] == 'inet':
+ return addresses[0]
+ return None
+
+ host = wait_for_network(container)
+
+ # HACK: We need to give sshd a chance to bind to the interface,
+ # and pylxd's container.execute seems to be broken and fails and/or
+ # hangs trying to properly check if the service is up.
+ time.sleep(5)
+
+ if host:
+ # add a new manual machine
+ machine1 = await model.add_machine(spec='ssh:{}@{}:{}'.format(
+ "ubuntu",
+ host['address'],
+ private_key_path,
+ ))
+
+ assert len(model.machines) == 1
+
+ res = await machine1.destroy(force=True)
+
+ assert res is None
+ assert len(model.machines) == 0
+
+ container.stop(wait=True)
+ container.delete(wait=True)
+
+ profile.delete()
+
+
@base.bootstrapped
@pytest.mark.asyncio
async def test_relate(event_loop):
assert result['extra-info'].source == 'model'
assert result['extra-info'].value == 'booyah'
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_set_constraints(event_loop):
+ async with base.CleanModel() as model:
+ await model.set_constraints({'cpu-power': 1})
+ cons = await model.get_constraints()
+ assert cons['cpu_power'] == 1
+
# @base.bootstrapped
# @pytest.mark.asyncio
# async def test_grant(event_loop)