Improved Primitive support and better testing

[osm/N2VC.git] / modules / libjuju / tests / integration / test_model.py

diff --git a/modules/libjuju/tests/integration/test_model.py b/modules/libjuju/tests/integration/test_model.py
index ba2da92..1cba79a 100644 (file)
--- a/modules/libjuju/tests/integration/test_model.py
+++ b/modules/libjuju/tests/integration/test_model.py
@@ -6,6 +6,12 @@ from pathlib import Path
 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
 
@@ -20,7 +26,6 @@ SSH_KEY = 'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCsYMJGNGG74HAJha3n2CFmWYsOOaORn
 @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'
@@ -33,6 +38,16 @@ async def test_deploy_local_bundle(event_loop):
             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):
@@ -110,6 +125,114 @@ async def test_add_machine(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):
@@ -269,6 +392,14 @@ async def test_config(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)