improving test

Author: sanderegg

packages/pytest-fixtures/docker_registry.py

@@ -5,6 +5,7 @@
 # pylint:disable=redefined-outer-name
 
 import json
+import os
 import time
 from typing import Dict
 
@@ -24,7 +25,7 @@ def docker_registry(keepdockerup: bool) -> str:
     container = None
     try:
         docker_client.login(registry=url, username="simcore")
-        container = docker_client.containers.list({"name":"pytest_registry"})[0]
+        container = docker_client.containers.list({"name": "pytest_registry"})[0]
     except Exception:
         print("Warning: docker registry is already up!")
         container = docker_client.containers.run(
@@ -35,7 +36,7 @@ def docker_registry(keepdockerup: bool) -> str:
             restart_policy={"Name": "always"},
             detach=True,
         )
-    
+
         # Wait until we can connect
         assert _wait_till_registry_is_responsive(url)
 
@@ -55,6 +56,11 @@ def docker_registry(keepdockerup: bool) -> str:
     private_image = docker_client.images.pull(repo)
     docker_client.images.remove(image=private_image.id)
 
+    # necessary for old school configs
+    os.environ["REGISTRY_URL"] = url
+    os.environ["REGISTRY_USER"] = "simcore"
+    os.environ["REGISTRY_PW"] = ""    
+
     yield url
     if not keepdockerup:
         container.stop()
@@ -88,9 +94,12 @@ def sleeper_service(docker_registry: str) -> Dict[str, str]:
     image_labels = image.labels
 
     yield {
-        key[len("io.simcore."):]: json.loads(value)[key[len("io.simcore."):]]
-        for key, value in image_labels.items()
-        if key.startswith("io.simcore.")
+        "schema": {
+            key[len("io.simcore.") :]: json.loads(value)[key[len("io.simcore.") :]]
+            for key, value in image_labels.items()
+            if key.startswith("io.simcore.")
+        },
+        "image": repo,
     }
 
 
@@ -120,7 +129,10 @@ def jupyter_service(docker_registry: str) -> Dict[str, str]:
     image_labels = image.labels
 
     yield {
-        key[len("io.simcore."):]: json.loads(value)[key[len("io.simcore."):]]
-        for key, value in image_labels.items()
-        if key.startswith("io.simcore.")
+        "schema": {
+            key[len("io.simcore.") :]: json.loads(value)[key[len("io.simcore.") :]]
+            for key, value in image_labels.items()
+            if key.startswith("io.simcore.")
+        },
+        "image": repo,
     }

services/sidecar/src/sidecar/core.py

@@ -299,15 +299,15 @@ async def initialize(self, task, user_id: str):
                   task.internal_id, user_id)
         self._task = task
         self._user_id = user_id
-
+        import pdb; pdb.set_trace()
         self._docker.image_name = self._docker.registry_name + \
-            "/" + task.image['name']
-        self._docker.image_tag = task.image['tag']
+            "/" + task.schema["key"]
+        self._docker.image_tag = task.schema["version"]
 
         # volume paths for side-car container
-        self._executor.in_dir = Path.home() / "input" / task.job_id
-        self._executor.out_dir = Path.home() / 'output' / task.job_id
-        self._executor.log_dir = Path.home() / 'log' / task.job_id
+        self._executor.in_dir = Path.home() / f"input/{task.job_id}"
+        self._executor.out_dir = Path.home() / f"output/{task.job_id}"
+        self._executor.log_dir = Path.home() / f"log/{task.job_id}"
 
         # volume paths for car container (w/o prefix)
         self._docker.env = [
@@ -324,6 +324,7 @@ async def initialize(self, task, user_id: str):
                   task.internal_id, user_id)
 
     async def preprocess(self):
+        import pdb; pdb.set_trace()
         log.debug('Pre-Processing Pipeline %s:node %s:internal id %s from container',
                   self._task.project_id, self._task.node_id, self._task.internal_id)
         await self._create_shared_folders()
@@ -333,6 +334,7 @@ async def preprocess(self):
                   self._task.project_id, self._task.node_id, self._task.internal_id)
 
     async def process(self):
+        import pdb; pdb.set_trace()
         log.debug('Processing Pipeline %s:node %s:internal id %s from container',
                   self._task.project_id, self._task.node_id, self._task.internal_id)
 
@@ -396,6 +398,7 @@ async def process(self):
                   self._task.project_id, self._task.node_id, self._task.internal_id)
 
     async def run(self):
+        import pdb; pdb.set_trace()
         log.debug('Running Pipeline %s:node %s:internal id %s from container',
                   self._task.project_id, self._task.node_id, self._task.internal_id)
         # NOTE: the rabbit has a timeout of 60seconds so blocking this channel for more is a no go.
@@ -448,7 +451,7 @@ async def inspect(self, celery_task, user_id: str, project_id: str, node_id: str
                   user_id, project_id, node_id)
 
         next_task_nodes = []
-
+        import pdb; pdb.set_trace()
         with session_scope(self._db.Session) as _session:
             _pipeline = _session.query(ComputationalPipeline).filter_by(
                 project_id=project_id).one()
@@ -501,7 +504,7 @@ async def inspect(self, celery_task, user_id: str, project_id: str, node_id: str
             _session.add(task)
             _session.commit()
 
-            self.initialize(task, user_id)
+            await self.initialize(task, user_id)
 
         # now proceed actually running the task (we do that after the db session has been closed)
         # try to run the task, return empyt list of next nodes if anything goes wrong

services/sidecar/tests/integration/test_sidecar.py

@@ -1,15 +1,84 @@
-#pylint: disable=unused-argument
+# pylint: disable=unused-argument
+# pylint: disable=redefined-outer-name
 
-# import pytest
+import os
+from typing import Dict
+from uuid import uuid4
+
+import pytest
+import sqlalchemy as sa
+
+from simcore_sdk.models.pipeline_models import (
+    ComputationalPipeline,
+    ComputationalTask,
+    comp_pipeline,
+    comp_tasks,
+)
 
 # Selection of core and tool services started in this swarm fixture (integration)
 core_services = ["storage", "postgres", "rabbit"]
 
 ops_services = ["minio", "adminer"]
 
 
+@pytest.fixture
+def project_id() -> str:
+    return str(uuid4())
+
+
+@pytest.fixture
+def user_id() -> int:
+    return 1
+
+
+@pytest.fixture
+def node_uuid() -> str:
+    return str(uuid4())
+
+
+@pytest.fixture
+def pipeline_db(
+    postgres_session: sa.orm.session.Session, project_id: str, node_uuid
+) -> ComputationalPipeline:
+    pipeline = ComputationalPipeline(project_id=project_id, dag_adjacency_list={node_uuid:[]})
+    postgres_session.add(pipeline)
+    postgres_session.commit()
+    yield pipeline
+
+@pytest.fixture
+def task_db(
+    postgres_session: sa.orm.session.Session,
+    sleeper_service: Dict[str, str],
+    pipeline_db: ComputationalPipeline,
+    node_uuid: str,
+) -> ComputationalTask:
+    comp_task = ComputationalTask(
+        project_id=pipeline_db.project_id,
+        node_id=node_uuid,
+        schema=sleeper_service["schema"],
+        image=sleeper_service["image"],
+        inputs={},
+        outputs={},
+    )
+    postgres_session.add(comp_task)
+    postgres_session.commit()
+
+    yield comp_task
+
+async def test_run_sleepers(
+    loop,
+    docker_stack: Dict,
+    postgres_session: sa.orm.session.Session,
+    sleeper_service: Dict[str, str],
+    task_db: ComputationalTask,
+    user_id: int,
+    mocker,
+):
+    celery_task = mocker.MagicMock()
+    celery_task.request.id = 1
 
-async def test_run_sleepers(loop, docker_stack, postgres_session, sleeper_service):
-    import pdb
-    pdb.set_trace()
-    # next_task_nodes = await SIDECAR.inspect(task, user_id, pipeline_id, node_id)
+    # Note this must happen here since DB is set already at that time
+    from sidecar.core import SIDECAR
+    next_task_nodes = await SIDECAR.inspect(
+        celery_task, user_id, task_db.project_id, task_db.node_id
+    )