Squashed 'modules/libjuju/' content from commit c50c361
git-subtree-dir: modules/libjuju
git-subtree-split: c50c361a8b9a3bbf1a33f5659e492b481f065cd2
diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py
new file mode 100644
index 0000000..e69de29
--- /dev/null
+++ b/tests/integration/__init__.py
diff --git a/tests/integration/bundle/bundle.yaml b/tests/integration/bundle/bundle.yaml
new file mode 100644
index 0000000..d0245c5
--- /dev/null
+++ b/tests/integration/bundle/bundle.yaml
@@ -0,0 +1,12 @@
+series: xenial
+services:
+ ghost:
+ charm: "cs:ghost-19"
+ num_units: 1
+ mysql:
+ charm: "cs:trusty/mysql-57"
+ num_units: 1
+ test:
+ charm: "./tests/integration/charm"
+relations:
+ - ["ghost", "mysql"]
diff --git a/tests/integration/charm/metadata.yaml b/tests/integration/charm/metadata.yaml
new file mode 100644
index 0000000..74eab3d
--- /dev/null
+++ b/tests/integration/charm/metadata.yaml
@@ -0,0 +1,5 @@
+name: charm
+series: ["xenial"]
+summary: "test"
+description: "test"
+maintainers: ["test"]
diff --git a/tests/integration/test_application.py b/tests/integration/test_application.py
new file mode 100644
index 0000000..1a4fcaa
--- /dev/null
+++ b/tests/integration/test_application.py
@@ -0,0 +1,94 @@
+import pytest
+
+from .. import base
+
+MB = 1
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_action(event_loop):
+ async with base.CleanModel() as model:
+ ubuntu_app = await model.deploy(
+ 'mysql',
+ application_name='mysql',
+ series='trusty',
+ channel='stable',
+ config={
+ 'tuning-level': 'safest',
+ },
+ constraints={
+ 'mem': 256 * MB,
+ },
+ )
+
+ # update and check app config
+ await ubuntu_app.set_config({'tuning-level': 'fast'})
+ config = await ubuntu_app.get_config()
+ assert config['tuning-level']['value'] == 'fast'
+
+ # update and check app constraints
+ await ubuntu_app.set_constraints({'mem': 512 * MB})
+ constraints = await ubuntu_app.get_constraints()
+ assert constraints['mem'] == 512 * MB
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_add_units(event_loop):
+ from juju.unit import Unit
+
+ async with base.CleanModel() as model:
+ app = await model.deploy(
+ 'ubuntu-0',
+ application_name='ubuntu',
+ series='trusty',
+ channel='stable',
+ )
+ units = await app.add_units(count=2)
+
+ assert len(units) == 2
+ for unit in units:
+ assert isinstance(unit, Unit)
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_upgrade_charm(event_loop):
+ async with base.CleanModel() as model:
+ app = await model.deploy('ubuntu-0')
+ assert app.data['charm-url'] == 'cs:ubuntu-0'
+ await app.upgrade_charm()
+ assert app.data['charm-url'].startswith('cs:ubuntu-')
+ assert app.data['charm-url'] != 'cs:ubuntu-0'
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_upgrade_charm_channel(event_loop):
+ async with base.CleanModel() as model:
+ app = await model.deploy('ubuntu-0')
+ assert app.data['charm-url'] == 'cs:ubuntu-0'
+ await app.upgrade_charm(channel='stable')
+ assert app.data['charm-url'].startswith('cs:ubuntu-')
+ assert app.data['charm-url'] != 'cs:ubuntu-0'
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_upgrade_charm_revision(event_loop):
+ async with base.CleanModel() as model:
+ app = await model.deploy('ubuntu-0')
+ assert app.data['charm-url'] == 'cs:ubuntu-0'
+ await app.upgrade_charm(revision=8)
+ assert app.data['charm-url'] == 'cs:ubuntu-8'
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_upgrade_charm_switch(event_loop):
+ async with base.CleanModel() as model:
+ app = await model.deploy('ubuntu-0')
+ assert app.data['charm-url'] == 'cs:ubuntu-0'
+ await app.upgrade_charm(switch='ubuntu-8')
+ assert app.data['charm-url'] == 'cs:ubuntu-8'
diff --git a/tests/integration/test_client.py b/tests/integration/test_client.py
new file mode 100644
index 0000000..e4c9c92
--- /dev/null
+++ b/tests/integration/test_client.py
@@ -0,0 +1,21 @@
+import pytest
+
+from juju.client import client
+
+from .. import base
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_user_info(event_loop):
+ async with base.CleanModel() as model:
+ controller_conn = await model.connection.controller()
+
+ um = client.UserManagerFacade.from_connection(controller_conn)
+ result = await um.UserInfo(
+ [client.Entity('user-admin')], True)
+ await controller_conn.close()
+
+ assert isinstance(result, client.UserInfoResults)
+ for r in result.results:
+ assert isinstance(r, client.UserInfoResult)
diff --git a/tests/integration/test_connection.py b/tests/integration/test_connection.py
new file mode 100644
index 0000000..290203d
--- /dev/null
+++ b/tests/integration/test_connection.py
@@ -0,0 +1,83 @@
+import asyncio
+import pytest
+
+from juju.client.connection import Connection
+from juju.client import client
+from .. import base
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_connect_current(event_loop):
+ async with base.CleanModel():
+ conn = await Connection.connect_current()
+
+ assert isinstance(conn, Connection)
+ await conn.close()
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_monitor(event_loop):
+
+ async with base.CleanModel():
+ conn = await Connection.connect_current()
+
+ assert conn.monitor.status == 'connected'
+ await conn.close()
+
+ assert conn.monitor.status == 'disconnected'
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_monitor_catches_error(event_loop):
+
+ async with base.CleanModel():
+ conn = await Connection.connect_current()
+
+ assert conn.monitor.status == 'connected'
+ await conn.ws.close()
+
+ assert conn.monitor.status == 'error'
+
+ await conn.close()
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_full_status(event_loop):
+ async with base.CleanModel() as model:
+ await model.deploy(
+ 'ubuntu-0',
+ application_name='ubuntu',
+ series='trusty',
+ channel='stable',
+ )
+
+ c = client.ClientFacade.from_connection(model.connection)
+
+ await c.FullStatus(None)
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_reconnect(event_loop):
+ async with base.CleanModel() as model:
+ conn = await Connection.connect(
+ model.connection.endpoint,
+ model.connection.uuid,
+ model.connection.username,
+ model.connection.password,
+ model.connection.cacert,
+ model.connection.macaroons,
+ model.connection.loop,
+ model.connection.max_frame_size)
+ try:
+ await asyncio.sleep(0.1)
+ assert conn.is_open
+ await conn.ws.close()
+ assert not conn.is_open
+ await model.block_until(lambda: conn.is_open, timeout=3)
+ finally:
+ await conn.close()
diff --git a/tests/integration/test_controller.py b/tests/integration/test_controller.py
new file mode 100644
index 0000000..f3840cc
--- /dev/null
+++ b/tests/integration/test_controller.py
@@ -0,0 +1,77 @@
+import pytest
+import uuid
+
+from .. import base
+from juju.controller import Controller
+from juju.errors import JujuAPIError
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_add_user(event_loop):
+ async with base.CleanController() as controller:
+ username = 'test{}'.format(uuid.uuid4())
+ await controller.add_user(username)
+ result = await controller.get_user(username)
+ res_ser = result.serialize()['results'][0].serialize()
+ assert res_ser['result'] is not None
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_disable_enable_user(event_loop):
+ async with base.CleanController() as controller:
+ username = 'test-disable{}'.format(uuid.uuid4())
+ await controller.add_user(username)
+ await controller.disable_user(username)
+ result = await controller.get_user(username)
+ res_ser = result.serialize()['results'][0].serialize()
+ assert res_ser['result'].serialize()['disabled'] is True
+ await controller.enable_user(username)
+ result = await controller.get_user(username)
+ res_ser = result.serialize()['results'][0].serialize()
+ assert res_ser['result'].serialize()['disabled'] is False
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_change_user_password(event_loop):
+ async with base.CleanController() as controller:
+ username = 'test-password{}'.format(uuid.uuid4())
+ await controller.add_user(username)
+ await controller.change_user_password(username, 'password')
+ try:
+ new_controller = Controller()
+ await new_controller.connect(
+ controller.connection.endpoint, username, 'password')
+ result = True
+ await new_controller.disconnect()
+ except JujuAPIError:
+ result = False
+ assert result is True
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_grant(event_loop):
+ async with base.CleanController() as controller:
+ username = 'test-grant{}'.format(uuid.uuid4())
+ await controller.add_user(username)
+ await controller.grant(username, 'superuser')
+ result = await controller.get_user(username)
+ result = result.serialize()['results'][0].serialize()['result']\
+ .serialize()
+ assert result['access'] == 'superuser'
+ await controller.grant(username, 'login')
+ result = await controller.get_user(username)
+ result = result.serialize()['results'][0].serialize()['result']\
+ .serialize()
+ assert result['access'] == 'login'
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_get_models(event_loop):
+ async with base.CleanController() as controller:
+ result = await controller.get_models()
+ assert isinstance(result.serialize()['user-models'], list)
diff --git a/tests/integration/test_errors.py b/tests/integration/test_errors.py
new file mode 100644
index 0000000..06b3826
--- /dev/null
+++ b/tests/integration/test_errors.py
@@ -0,0 +1,68 @@
+import pytest
+
+from .. import base
+
+MB = 1
+GB = 1024
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_juju_api_error(event_loop):
+ '''
+ Verify that we raise a JujuAPIError for responses with an error in
+ a top level key (for completely invalid requests).
+
+ '''
+ from juju.errors import JujuAPIError
+
+ async with base.CleanModel() as model:
+ with pytest.raises(JujuAPIError):
+ await model.add_machine(constraints={'mem': 'foo'})
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_juju_error_in_results_list(event_loop):
+ '''
+ Replicate the code that caused
+ https://github.com/juju/python-libjuju/issues/67, and verify that
+ we get a JujuError instead of passing silently by the failure.
+
+ (We don't raise a JujuAPIError, because the request isn't
+ completely invalid -- it's just passing a tag that doesn't exist.)
+
+ This also verifies that we will raise a JujuError any time there
+ is an error in one of a list of results.
+
+ '''
+ from juju.errors import JujuError
+ from juju.client import client
+
+ async with base.CleanModel() as model:
+ ann_facade = client.AnnotationsFacade.from_connection(model.connection)
+
+ ann = client.EntityAnnotations(
+ entity='badtag',
+ annotations={'gui-x': '1', 'gui-y': '1'},
+ )
+ with pytest.raises(JujuError):
+ return await ann_facade.Set([ann])
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_juju_error_in_result(event_loop):
+ '''
+ Verify that we raise a JujuError when appropraite when we are
+ looking at a single result coming back.
+
+ '''
+ from juju.errors import JujuError
+ from juju.client import client
+
+ async with base.CleanModel() as model:
+ app_facade = client.ApplicationFacade.from_connection(model.connection)
+
+ with pytest.raises(JujuError):
+ return await app_facade.GetCharmURL('foo')
diff --git a/tests/integration/test_machine.py b/tests/integration/test_machine.py
new file mode 100644
index 0000000..60de035
--- /dev/null
+++ b/tests/integration/test_machine.py
@@ -0,0 +1,62 @@
+import asyncio
+import pytest
+
+from tempfile import NamedTemporaryFile
+
+from .. import base
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_status(event_loop):
+ async with base.CleanModel() as model:
+ await model.deploy(
+ 'ubuntu-0',
+ application_name='ubuntu',
+ series='trusty',
+ channel='stable',
+ )
+
+ await asyncio.wait_for(
+ model.block_until(lambda: len(model.machines)),
+ timeout=240)
+ machine = model.machines['0']
+
+ assert machine.status in ('allocating', 'pending')
+ assert machine.agent_status == 'pending'
+ assert not machine.agent_version
+
+ await asyncio.wait_for(
+ model.block_until(lambda: (machine.status == 'running' and
+ machine.agent_status == 'started')),
+ timeout=480)
+
+ assert machine.status == 'running'
+ # there is some inconsistency in the message case between providers
+ assert machine.status_message.lower() == 'running'
+ assert machine.agent_status == 'started'
+ assert machine.agent_version.major >= 2
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_scp(event_loop):
+ async with base.CleanModel() as model:
+ await model.add_machine()
+ await asyncio.wait_for(
+ model.block_until(lambda: model.machines),
+ timeout=240)
+ machine = model.machines['0']
+ await asyncio.wait_for(
+ model.block_until(lambda: (machine.status == 'running' and
+ machine.agent_status == 'started')),
+ timeout=480)
+
+ with NamedTemporaryFile() as f:
+ f.write(b'testcontents')
+ f.flush()
+ await machine.scp_to(f.name, 'testfile')
+
+ with NamedTemporaryFile() as f:
+ await machine.scp_from('testfile', f.name)
+ assert f.read() == b'testcontents'
diff --git a/tests/integration/test_model.py b/tests/integration/test_model.py
new file mode 100644
index 0000000..8506786
--- /dev/null
+++ b/tests/integration/test_model.py
@@ -0,0 +1,249 @@
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
+import pytest
+
+from .. import base
+from juju.model import Model
+from juju.client.client import ConfigValue
+
+MB = 1
+GB = 1024
+SSH_KEY = 'ssh-rsa AAAAB3NzaC1yc2EAAAADAQABAAABAQCsYMJGNGG74HAJha3n2CFmWYsOOaORnJK6VqNy86pj0MIpvRXBzFzVy09uPQ66GOQhTEoJHEqE77VMui7+62AcMXT+GG7cFHcnU8XVQsGM6UirCcNyWNysfiEMoAdZScJf/GvoY87tMEszhZIUV37z8PUBx6twIqMdr31W1J0IaPa+sV6FEDadeLaNTvancDcHK1zuKsL39jzAg7+LYjKJfEfrsQP+lj/EQcjtKqlhVS5kzsJVfx8ZEd0xhW5G7N6bCdKNalS8mKCMaBXJpijNQ82AiyqCIDCRrre2To0/i7pTjRiL0U9f9mV3S4NJaQaokR050w/ZLySFf6F7joJT mathijs@Qrama-Mathijs' # noqa
+
+
+@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'
+
+ async with base.CleanModel() as model:
+ await model.deploy(str(bundle_path))
+
+ for app in ('wordpress', 'mysql'):
+ assert app in model.applications
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_deploy_bundle(event_loop):
+ async with base.CleanModel() as model:
+ await model.deploy('bundle/wiki-simple')
+
+ for app in ('wiki', 'mysql'):
+ assert app in model.applications
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_deploy_channels_revs(event_loop):
+ async with base.CleanModel() as model:
+ charm = 'cs:~johnsca/libjuju-test'
+ stable = await model.deploy(charm, 'a1')
+ edge = await model.deploy(charm, 'a2', channel='edge')
+ rev = await model.deploy(charm+'-2', 'a3')
+
+ assert [a.charm_url for a in (stable, edge, rev)] == [
+ 'cs:~johnsca/libjuju-test-1',
+ 'cs:~johnsca/libjuju-test-2',
+ 'cs:~johnsca/libjuju-test-2',
+ ]
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_add_machine(event_loop):
+ from juju.machine import Machine
+
+ async with base.CleanModel() as model:
+ # add a new default machine
+ machine1 = await model.add_machine()
+
+ # add a machine with constraints, disks, and series
+ machine2 = await model.add_machine(
+ constraints={
+ 'mem': 256 * MB,
+ },
+ disks=[{
+ 'pool': 'rootfs',
+ 'size': 10 * GB,
+ 'count': 1,
+ }],
+ series='xenial',
+ )
+
+ # add a lxd container to machine2
+ machine3 = await model.add_machine(
+ 'lxd:{}'.format(machine2.id))
+
+ for m in (machine1, machine2, machine3):
+ assert isinstance(m, Machine)
+
+ assert len(model.machines) == 3
+
+ await machine3.destroy(force=True)
+ await machine2.destroy(force=True)
+ res = await machine1.destroy(force=True)
+
+ assert res is None
+ assert len(model.machines) == 0
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_relate(event_loop):
+ from juju.relation import Relation
+
+ async with base.CleanModel() as model:
+ await model.deploy(
+ 'ubuntu',
+ application_name='ubuntu',
+ series='trusty',
+ channel='stable',
+ )
+ await model.deploy(
+ 'nrpe',
+ application_name='nrpe',
+ series='trusty',
+ channel='stable',
+ # subordinates must be deployed without units
+ num_units=0,
+ )
+ my_relation = await model.add_relation(
+ 'ubuntu',
+ 'nrpe',
+ )
+
+ assert isinstance(my_relation, Relation)
+
+
+async def _deploy_in_loop(new_loop, model_name):
+ new_model = Model(new_loop)
+ await new_model.connect_model(model_name)
+ try:
+ await new_model.deploy('cs:xenial/ubuntu')
+ assert 'ubuntu' in new_model.applications
+ finally:
+ await new_model.disconnect()
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_explicit_loop(event_loop):
+ async with base.CleanModel() as model:
+ model_name = model.info.name
+ new_loop = asyncio.new_event_loop()
+ new_loop.run_until_complete(
+ _deploy_in_loop(new_loop, model_name))
+ await model._wait_for_new('application', 'ubuntu')
+ assert 'ubuntu' in model.applications
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_explicit_loop_threaded(event_loop):
+ async with base.CleanModel() as model:
+ model_name = model.info.name
+ new_loop = asyncio.new_event_loop()
+ with ThreadPoolExecutor(1) as executor:
+ f = executor.submit(
+ new_loop.run_until_complete,
+ _deploy_in_loop(new_loop, model_name))
+ f.result()
+ await model._wait_for_new('application', 'ubuntu')
+ assert 'ubuntu' in model.applications
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_store_resources_charm(event_loop):
+ async with base.CleanModel() as model:
+ ghost = await model.deploy('cs:ghost-19')
+ assert 'ghost' in model.applications
+ terminal_statuses = ('active', 'error', 'blocked')
+ await model.block_until(
+ lambda: (
+ len(ghost.units) > 0 and
+ ghost.units[0].workload_status in terminal_statuses)
+ )
+ # ghost will go in to blocked (or error, for older
+ # charm revs) if the resource is missing
+ assert ghost.units[0].workload_status == 'active'
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_store_resources_bundle(event_loop):
+ async with base.CleanModel() as model:
+ bundle = str(Path(__file__).parent / 'bundle')
+ await model.deploy(bundle)
+ assert 'ghost' in model.applications
+ ghost = model.applications['ghost']
+ terminal_statuses = ('active', 'error', 'blocked')
+ await model.block_until(
+ lambda: (
+ len(ghost.units) > 0 and
+ ghost.units[0].workload_status in terminal_statuses)
+ )
+ # ghost will go in to blocked (or error, for older
+ # charm revs) if the resource is missing
+ assert ghost.units[0].workload_status == 'active'
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_ssh_key(event_loop):
+ async with base.CleanModel() as model:
+ await model.add_ssh_key('admin', SSH_KEY)
+ result = await model.get_ssh_key(True)
+ result = result.serialize()['results'][0].serialize()['result']
+ assert SSH_KEY in result
+ await model.remove_ssh_key('admin', SSH_KEY)
+ result = await model.get_ssh_key(True)
+ result = result.serialize()['results'][0].serialize()['result']
+ assert result is None
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_get_machines(event_loop):
+ async with base.CleanModel() as model:
+ result = await model.get_machines()
+ assert isinstance(result, list)
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_watcher_reconnect(event_loop):
+ async with base.CleanModel() as model:
+ await model.connection.ws.close()
+ await asyncio.sleep(0.1)
+ assert model.connection.is_open
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_config(event_loop):
+ async with base.CleanModel() as model:
+ await model.set_config({
+ 'extra-info': 'booyah',
+ 'test-mode': ConfigValue(value=True),
+ })
+ result = await model.get_config()
+ assert 'extra-info' in result
+ assert result['extra-info'].source == 'model'
+ assert result['extra-info'].value == 'booyah'
+
+# @base.bootstrapped
+# @pytest.mark.asyncio
+# async def test_grant(event_loop)
+# async with base.CleanController() as controller:
+# await controller.add_user('test-model-grant')
+# await controller.grant('test-model-grant', 'superuser')
+# async with base.CleanModel() as model:
+# await model.grant('test-model-grant', 'admin')
+# assert model.get_user('test-model-grant')['access'] == 'admin'
+# await model.grant('test-model-grant', 'login')
+# assert model.get_user('test-model-grant')['access'] == 'login'
diff --git a/tests/integration/test_unit.py b/tests/integration/test_unit.py
new file mode 100644
index 0000000..1604c31
--- /dev/null
+++ b/tests/integration/test_unit.py
@@ -0,0 +1,78 @@
+import asyncio
+import pytest
+
+from tempfile import NamedTemporaryFile
+
+from .. import base
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_run(event_loop):
+ from juju.action import Action
+
+ async with base.CleanModel() as model:
+ app = await model.deploy(
+ 'ubuntu-0',
+ application_name='ubuntu',
+ series='trusty',
+ channel='stable',
+ )
+
+ for unit in app.units:
+ action = await unit.run('unit-get public-address')
+ assert isinstance(action, Action)
+ assert 'Stdout' in action.results
+ break
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_run_action(event_loop):
+ async def run_action(unit):
+ # unit.run() returns a juju.action.Action instance
+ action = await unit.run_action('add-repo', repo='myrepo')
+ # wait for the action to complete
+ return await action.wait()
+
+ async with base.CleanModel() as model:
+ app = await model.deploy(
+ 'git',
+ application_name='git',
+ series='trusty',
+ channel='stable',
+ )
+
+ for unit in app.units:
+ action = await run_action(unit)
+ assert action.results == {'dir': '/var/git/myrepo.git'}
+ break
+
+
+@base.bootstrapped
+@pytest.mark.asyncio
+async def test_scp(event_loop):
+ async with base.CleanModel() as model:
+ app = await model.deploy('ubuntu')
+
+ await asyncio.wait_for(
+ model.block_until(lambda: app.units),
+ timeout=60)
+ unit = app.units[0]
+ await asyncio.wait_for(
+ model.block_until(lambda: unit.machine),
+ timeout=60)
+ machine = unit.machine
+ await asyncio.wait_for(
+ model.block_until(lambda: (machine.status == 'running' and
+ machine.agent_status == 'started')),
+ timeout=480)
+
+ with NamedTemporaryFile() as f:
+ f.write(b'testcontents')
+ f.flush()
+ await unit.scp_to(f.name, 'testfile')
+
+ with NamedTemporaryFile() as f:
+ await unit.scp_from('testfile', f.name)
+ assert f.read() == b'testcontents'