test: add test with local flow and remote pods (#1583)

* fix: parse config source used to upload files in jinad

* test: add test simple local remote to daemon tests

jina/jaml/helper.py

@@ -103,7 +103,7 @@ def parse_config_source(path: Union[str, TextIO, Dict],
         # already a readable stream
         return path, None
     elif allow_yaml_file and (path.endswith('.yml') or path.endswith('.yaml')):
-        comp_path = _complete_path(path)
+        comp_path = complete_path(path)
         return open(comp_path, encoding='utf8'), comp_path
     elif allow_builtin_resource and path.startswith('_') and os.path.exists(
             resource_filename('jina', '/'.join(('resources', f'executors.{path}.yml')))):
@@ -138,7 +138,7 @@ def parse_config_source(path: Union[str, TextIO, Dict],
                               ' or a valid file path, or a supported class name.')
 
 
-def _complete_path(path: str, extra_search_paths: Optional[Tuple[str]] = None) -> str:
+def complete_path(path: str, extra_search_paths: Optional[Tuple[str]] = None) -> str:
     _p = None
     if os.path.exists(path):
         # this checks both abs and relative paths already
@@ -190,5 +190,5 @@ def _finditem(obj, key='py_modules'):
 
     _finditem(d)
     if mod:
-        mod = [_complete_path(m, extra_search_paths) for m in mod]
+        mod = [complete_path(m, extra_search_paths) for m in mod]
         PathImporter.add_modules(*mod)

jina/peapods/runtimes/jinad/api.py

@@ -4,6 +4,7 @@
 from pathlib import Path
 from typing import Dict, Tuple, Set, List, Optional
 
+from ....jaml.helper import complete_path
 from ....enums import RemotePeapodType
 from ....importer import ImportExtensions
 from ....jaml import JAML
@@ -12,8 +13,9 @@
 
 def _add_file_to_list(_file: str, _file_list: Set, logger: 'JinaLogger'):
     if _file and _file.endswith(('yml', 'yaml', 'py')):
-        if Path(_file).is_file():
-            _file_list.add(_file)
+        real_file = complete_path(_file)
+        if Path(real_file).is_file():
+            _file_list.add(real_file)
             logger.debug(f'adding file {_file} to be uploaded to remote context')
         else:
             logger.warning(f'file {_file} doesn\'t exist in the disk')

tests/integration/daemon/distributed/test_simple_local_remote/.env

@@ -0,0 +1 @@
+JINA_WORKSPACE=workspace

tests/integration/daemon/distributed/test_simple_local_remote/docker-compose.yml

@@ -0,0 +1,35 @@
+version: "3.3"
+services:
+  a:
+    image: test_simple_local_remote
+    build:
+      context: .
+      dockerfile: Dockerfiles/debianx.Dockerfile
+    ports:
+      - "8000:8000"
+      - "45000:45000"
+      - "45001:45001"
+      - "45002:45002"
+    env_file:
+      - tests/integration/daemon/distributed/test_simple_local_remote/.env
+    expose:
+      - 10000-60000
+    networks:
+      testing_net:
+        ipv4_address: 172.28.1.1
+  b:
+    image: test_simple_local_remote
+    expose:
+      - 8000
+      - 10000-60000
+    env_file:
+      - tests/integration/daemon/distributed/test_simple_local_remote/.env
+    networks:
+      testing_net:
+        ipv4_address: 172.28.1.2
+networks:
+    testing_net:
+        ipam:
+            driver: default
+            config:
+                - subnet: 172.28.0.0/16

tests/integration/daemon/distributed/test_simple_local_remote/dummy-encoder.py

@@ -0,0 +1,18 @@
+__copyright__ = "Copyright (c) 2020 Jina AI Limited. All rights reserved."
+__license__ = "Apache-2.0"
+
+import numpy as np
+
+from jina.executors.encoders import BaseEncoder
+from jina.executors.decorators import batching
+
+
+class DummyEncoder(BaseEncoder):
+    def __init__(self,
+                 *args,
+                 **kwargs):
+        super().__init__(*args, **kwargs)
+
+    @batching(batch_size=1)
+    def encode(self, data: 'np.ndarray', *args, **kwargs) -> 'np.ndarray':
+        return np.random.random((1, 3))

tests/integration/daemon/distributed/test_simple_local_remote/encode.yml

@@ -0,0 +1,5 @@
+!DummyEncoder
+with:
+  {}
+metas:
+  py_modules: dummy-encoder.py

tests/integration/daemon/distributed/test_simple_local_remote/flow.yml

@@ -0,0 +1,20 @@
+!Flow
+with:
+  read_only: true
+pods:
+  encoder:
+    # for now all the uploaded files are stored in the same folder, so no subpaths allowed in remote
+    uses: encode.yml
+    host: $JINA_ENCODER_HOST
+    port_in: 45000
+    port_ctrl: 45002
+    port_expose: 8000
+  indexer:
+    # for now all the uploaded files are stored in the same folder, so no subpaths allowed in remote
+    uses: index.yml
+    read_only: False
+    host: $JINA_INDEXER_HOST
+    port_in: 45000
+    port_out: 45001
+    port_ctrl: 45002
+    port_expose: 8000

tests/integration/daemon/distributed/test_simple_local_remote/index.yml

@@ -0,0 +1,18 @@
+!CompoundIndexer
+components:
+  - !NumpyIndexer
+    with:
+      index_filename: vec.gz
+      metric: euclidean
+    metas:
+      name: vecidx
+      workspace: $JINA_WORKSPACE
+  - !BinaryPbIndexer
+    with:
+      index_filename: doc.gz
+    metas:
+      name: docidx
+      workspace: $JINA_WORKSPACE
+metas:
+  name: indexer
+  workspace: $JINA_WORKSPACE

tests/integration/daemon/distributed/test_simple_local_remote/test_integration.py

@@ -0,0 +1,32 @@
+import os
+
+import pytest
+
+from jina import Document
+from jina.flow import Flow
+
+cur_dir = os.path.dirname(os.path.abspath(__file__))
+compose_yml = os.path.join(cur_dir, 'docker-compose.yml')
+flow_yml = os.path.join(cur_dir, 'flow.yml')
+
+
+@pytest.mark.parametrize('docker_compose', [compose_yml], indirect=['docker_compose'])
+def test_flow(docker_compose, mocker):
+    text = 'cats rules'
+    m = mocker.Mock()
+
+    def validate_output(resp):
+        m()
+        assert len(resp.index.docs) == 1
+        assert resp.index.docs[0].text == text
+
+    os.environ['JINA_ENCODER_HOST'] = '172.28.1.1'
+    os.environ['JINA_INDEXER_HOST'] = '172.28.1.2'
+
+    with Document() as doc:
+        doc.content = text
+
+    with Flow.load_config(flow_yml) as f:
+        f.index([doc], on_done=validate_output)
+
+    m.assert_called_once()

tests/unit/peapods/runtimes/remote/jinad/test_api.py

@@ -5,8 +5,7 @@
 from jina.peapods.runtimes.jinad.api import JinadAPI, PodJinadAPI, PeaJinadAPI, fetch_files_from_yaml
 
 logger = JinaLogger(context='test-remote')
-yaml_path = os.path.dirname(os.path.realpath(__file__))
-relative_pymodules_path = 'tests/unit/peapods/runtimes/remote/jinad'
+yaml_path = os.path.dirname(os.path.abspath(__file__))
 jinad_api = JinadAPI(host='0.0.0.0', port=8000, logger=logger)
 pod_api = PodJinadAPI(host='0.0.0.0', port=8000, logger=logger)
 pea_api = PeaJinadAPI(host='0.0.0.0', port=8000, logger=logger)
@@ -36,7 +35,7 @@ def test_fetch_files_from_yaml_pods():
     _uses_files, _pymodule_files = fetch_files_from_yaml(pea_args, logger)
     assert _uses_files == {f'{yaml_path}/yamls/encoder.yml',
                            f'{yaml_path}/yamls/indexer.yml'}
-    assert _pymodule_files == {f'{relative_pymodules_path}/yamls/dummy.py'}
+    assert _pymodule_files == {f'{yaml_path}/yamls/dummy.py'}
 
 
 def test_fetch_files_from_yaml_pea():
@@ -49,7 +48,7 @@ def test_fetch_files_from_yaml_pea():
     }
     _uses_files, _pymodule_files = fetch_files_from_yaml(pea_args, logger)
     assert _uses_files == {f'{yaml_path}/yamls/encoder.yml'}
-    assert _pymodule_files == {f'{relative_pymodules_path}/yamls/dummy.py'}
+    assert _pymodule_files == {f'{yaml_path}/yamls/dummy.py'}
 
 
 @mock.patch('requests.get')

tests/unit/test_helper.py

@@ -10,7 +10,7 @@
 from jina.drivers.querylang.queryset.dunderkey import dunder_get
 from jina.excepts import BadClientCallback
 from jina.helper import cached_property, convert_tuple_to_list
-from jina.jaml.helper import _complete_path
+from jina.jaml.helper import complete_path
 from jina.logging import default_logger
 from jina.logging.profile import TimeContext
 from jina.proto import jina_pb2
@@ -130,18 +130,15 @@ def test_pprint_routes(capfd):
     rr.routes.extend(result)
     pprint_routes(rr)
     out, err = capfd.readouterr()
-    assert out == '''+-----+------+------------+
-| \x1b[1mPod\x1b[0m | \x1b[1mTime\x1b[0m | \x1b[1mException\x1b[0m  |
-+-----+------+------------+
-| 🔴  | 0ms  | r1         |
-|     |      | line1r2    |
-|     |      | line2      |
-+-----+------+------------+
-| ⚪  | 0ms  | line1line2 |
-+-----+------+------------+
-| 🟢  | 0ms  |            |
-+-----+------+------------+
-'''
+    assert '⚪' in out
+    assert '🟢' in out
+    assert 'Pod' in out
+    assert 'Time' in out
+    assert 'Exception' in out
+    assert 'r1' in out
+    assert 'line1r2' in out
+    assert 'line2' in out
+    assert 'line1line2' in out
 
 
 def test_convert_tuple_to_list():
@@ -187,11 +184,11 @@ def test_random_docs():
 
 
 def test_complete_path_success():
-    assert _complete_path('test_helper.py')
-    assert _complete_path('helper.py')
-    assert _complete_path('bash')
+    assert complete_path('test_helper.py')
+    assert complete_path('helper.py')
+    assert complete_path('bash')
 
 
 def test_complete_path_not_found():
     with pytest.raises(FileNotFoundError):
-        assert _complete_path('unknown.yaml')
+        assert complete_path('unknown.yaml')