Include resources from store when deploying (#102)

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

diff --git a/tests/integration/test_model.py b/tests/integration/test_model.py
index 2fe97d0..96c786a 100644 (file)
--- a/tests/integration/test_model.py
+++ b/tests/integration/test_model.py
@@ -1,6 +1,10 @@
+import asyncio
+from concurrent.futures import ThreadPoolExecutor
+from pathlib import Path
 import pytest
 
 from .. import base
+from juju.model import Model
 
 MB = 1
 GB = 1024
@@ -95,3 +99,76 @@ async def test_relate(event_loop):
         )
 
         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-18')
+        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'