Primitive inputs (#169)

* move tests

* update pytest and python versions

* pytest

* support primitive classes on UI side

* support primitive types

* add pytests

* linter to tests folder

* fix tests

* fix tests

* pr fix

* update template config

.pre-commit-config.yaml

@@ -3,16 +3,24 @@ repos:
 -   repo: local
     hooks:
     - id: system
-      name: $ flake8
-      entry: flake8 plynx
+      name: $ flake8 plynx tests
+      entry: flake8 plynx tests
       pass_filenames: false
       language: system
 
 -   repo: local
     hooks:
     - id: system
-      name: $ isort plynx
-      entry: isort plynx
+      name: $ isort plynx tests
+      entry: isort plynx tests
+      pass_filenames: false
+      language: system
+
+-   repo: local
+    hooks:
+    - id: system
+      name: $ make pytest
+      entry: make pytest
       pass_filenames: false
       language: system
 

Makefile

@@ -11,6 +11,7 @@ build_frontend:
 build: build_backend build_frontend;
 
 run_tests:
+	mkdir -p $(CURDIR)/data/resources
 	@$(MAKE) -f $(THIS_FILE) build_backend
 	docker-compose -f $(DOCKER_COMPOSE_DEV_FILE) up --abort-on-container-exit --scale workers=5 --scale frontend=0 --scale test=1
 
@@ -33,8 +34,11 @@ dev:
 	mkdir -p ./data/resources
 	PLYNX_IMAGES="backend ui_dev" ./scripts/build_images.sh
 	python -m webbrowser "http://localhost:3001/"
-	docker-compose -f $(DOCKER_COMPOSE_DEV_FILE) up --abort-on-container-exit --scale api=1 --scale test=0 --scale frontend=0
+	docker-compose -f $(DOCKER_COMPOSE_DEV_FILE) up --abort-on-container-exit --scale api=1 --scale test=0 --scale frontend=0 --scale workers=0
 
 build_package:
 	python setup.py sdist
 	python setup.py bdist_wheel
+
+pytest:
+	python -m pytest

docker-compose-dev.yml

@@ -113,7 +113,7 @@ services:
             - ./data:/data
             - ./plynx:/app/plynx
             - ./config.yaml:/app/config.yaml
-        command: /app/tests/run_tests.sh
+        command: /app/run_tests.sh
 
     frontend:
         image: plynxteam/ui_dev:latest

docker/backend/Dockerfile

@@ -1,5 +1,5 @@
 # Use an official Python runtime as a parent image
-FROM python:3.7-slim
+FROM python:3.8-slim
 
 # Set the working directory to /app
 WORKDIR /app
@@ -19,7 +19,7 @@ ADD ./wsgi.py /app/wsgi.py
 ADD ./LICENSE /app
 
 # Copy test data
-ADD ./docker/backend/tests /app/tests
+ADD ./docker/backend/tests /app/integrated_tests
 
 # master / worker port
 EXPOSE 17011
@@ -34,12 +34,14 @@ RUN chmod 1777 /tmp
 
 # Add dev watch script
 COPY ./scripts/watch.sh /app/watch.sh
+COPY ./scripts/run_tests.sh /app/run_tests.sh
 
 # Build PLynx package
 ADD ./plynx /tmp/plynx
 ADD ./setup.py /tmp/setup.py
 ADD ./requirements.txt /tmp/requirements.txt
 ADD ./requirements-dev.txt /tmp/requirements-dev.txt
+ADD ./tests /app/tests
 RUN cd /tmp && python setup.py install
 
 # install extra docker requirements

docker/backend/requirements.txt

@@ -5,5 +5,5 @@ python-dateutil==2.8.1
 
 # test and development:
 coverage==5.0.4
-pytest==4.3.1
+pytest==6.2.2
 watchdog==2.2.0

docker/backend/tests/test_00_sum.py

@@ -1,11 +1,10 @@
 #!/usr/bin/env python
-from plynx.constants import NodeRunningStatus
 import plynx.db.node
-import plynx.utils.file_handler as file_handler
-import plynx.plugins.executors.local as local
 import plynx.plugins.executors.dag as dag
+import plynx.plugins.executors.local as local
 import plynx.plugins.resources.common as common
-
+import plynx.utils.file_handler as file_handler
+from plynx.constants import NodeRunningStatus
 
 SEQ_OUTPUT = 'seq_o'
 GREP_INPUT = 'grep_i'
@@ -36,7 +35,7 @@ def create_seq_operation(num):
         })
     )
 
-    cmd_param = node.get_parameter_by_name('_cmd', throw=True)
+    cmd_param = node.get_parameter_by_name('_cmd')
     cmd_param.value.value = 'seq {{{{ params.N }}}} > {{{{ outputs.{0} }}}}\n'.format(SEQ_OUTPUT)
 
     return node
@@ -72,7 +71,7 @@ def create_grep_operation(input_reference, template):
         })
     )
 
-    cmd_param = node.get_parameter_by_name('_cmd', throw=True)
+    cmd_param = node.get_parameter_by_name('_cmd')
     cmd_param.value.value = 'cat {{{{ inputs.{0} }}}} | grep {{{{ params.template }}}} > {{{{ outputs.{1} }}}}\n'.format(
         GREP_INPUT,
         GREP_OUTPUT
@@ -103,7 +102,7 @@ def create_sum_operation(input_references):
         })
     )
 
-    cmd_param = node.get_parameter_by_name('_cmd', throw=True)
+    cmd_param = node.get_parameter_by_name('_cmd')
     CMD = """
 res = 0
 for filename in inputs['{inputs}']:
@@ -139,7 +138,7 @@ def create_dag_executor(N):
     dag_node.title = 'Test sum DAG'
     dag_node.kind = 'basic-dag-workflow'
     dag_node.node_running_status = 'READY'
-    nodes = dag_node.get_parameter_by_name('_nodes', throw=True).value.value
+    nodes = dag_node.get_sub_nodes()
 
     seq_operation = create_seq_operation(N)
     nodes.append(seq_operation)
@@ -178,7 +177,7 @@ def test_dag():
 
     executor.run()
 
-    output_ids = executor.node.get_parameter_by_name('_nodes').value.value[-1].get_output_by_name(SUM_OUTPUT).values
+    output_ids = executor.node.get_sub_nodes()[-1].get_output_by_name(SUM_OUTPUT).values
     assert len(output_ids) == 1, 'Unexpected number of outputs: `{}`'.format(len(output_ids))
 
     res = int(file_handler.get_file_stream(output_ids[0]).read())

plynx/base/executor.py

@@ -3,7 +3,7 @@
 from dataclasses import dataclass
 from typing import Optional, Union
 
-from plynx.constants import NodeStatus, SpecialNodeId, ValidationCode, ValidationTargetType
+from plynx.constants import PRIMITIVE_TYPES, NodeStatus, SpecialNodeId, ValidationCode, ValidationTargetType
 from plynx.db.node import Node, NodeRunningStatus, Parameter, ParameterListOfNodes, ParameterTypes
 from plynx.db.validation_error import ValidationError
 
@@ -137,11 +137,11 @@ def validate(self, ignore_inputs: bool = True) -> Union[ValidationError, None]:
                     validation_code=ValidationCode.MISSING_PARAMETER
                 ))
 
-        # Meaning the node is in the graph. Otherwise souldn't be in validation step
+        # Meaning the node is in the graph. Otherwise souldn't be in the validation step
         if not ignore_inputs:
             for input in self.node.inputs:  # pylint: disable=redefined-builtin
                 min_count = input.min_count if input.is_array else 1
-                if len(input.input_references) < min_count:
+                if len(input.input_references) < min_count and input.file_type not in PRIMITIVE_TYPES:
                     violations.append(
                         ValidationError(
                             target=ValidationTargetType.INPUT,

plynx/constants/__init__.py

@@ -3,7 +3,7 @@
 from plynx.constants.collections import Collections
 from plynx.constants.node_enums import (IGNORED_CACHE_PARAMETERS, NodeClonePolicy, NodeOrigin, NodePostAction, NodePostStatus, NodeRunningStatus, NodeStatus,
                                         NodeVirtualCollection, SpecialNodeId)
-from plynx.constants.parameter_types import ParameterTypes
+from plynx.constants.parameter_types import PRIMITIVE_TYPES, ParameterTypes
 from plynx.constants.resource_enums import HubSearchParams, NodeResources
 from plynx.constants.users import IAMPolicies, RegisterUserExceptionCode, TokenType, UserPostAction
 from plynx.constants.validation_enums import ValidationCode, ValidationTargetType

plynx/constants/parameter_types.py

@@ -13,3 +13,11 @@ class ParameterTypes:
     LIST_INT: str = 'list_int'
     LIST_NODE: str = 'list_node'
     CODE: str = 'code'
+
+
+PRIMITIVE_TYPES = {
+    ParameterTypes.BOOL,
+    ParameterTypes.FLOAT,
+    ParameterTypes.INT,
+    ParameterTypes.STR,
+}

plynx/db/demo_user_manager.py

@@ -47,7 +47,7 @@ def create_demo_templates(user):
         for template_id in DemoUserManager.demo_config.template_ids:
             node_id = to_object_id(template_id)     # may raise bson.objectid.InvalidId
             node_dict = template_collection_manager.get_db_node(node_id, user._id)
-            node: Node = get_class(node_dict['_type']).from_dict(node_dict).clone(NodeClonePolicy.NODE_TO_NODE)
+            node: Node = get_class(node_dict['_type']).from_dict(node_dict).clone(NodeClonePolicy.NODE_TO_NODE)     # type: ignore
             node.author = user._id
             node.latest_run_id = None
             node_utils.reset_nodes(node)

plynx/db/node.py

@@ -5,7 +5,7 @@
 from dataclasses_json import dataclass_json
 from past.builtins import basestring
 
-from plynx.constants import Collections, NodeClonePolicy, NodeOrigin, NodeRunningStatus, NodeStatus, ParameterTypes
+from plynx.constants import PRIMITIVE_TYPES, Collections, NodeClonePolicy, NodeOrigin, NodeRunningStatus, NodeStatus, ParameterTypes
 from plynx.db.db_object import DBObject
 from plynx.plugins.resources.common import FILE_KIND
 from plynx.utils.common import ObjectId
@@ -42,7 +42,7 @@ def _clone_update_in_place(node: "Node", node_clone_policy: int, override_finish
         node.node_running_status = NodeRunningStatus.READY
     node.node_status = NodeStatus.CREATED
 
-    sub_nodes = node.get_parameter_by_name('_nodes', throw=False)
+    sub_nodes = node.get_parameter_by_name_safe('_nodes')
     if sub_nodes:
         object_id_mapping = {}
         for sub_node in sub_nodes.value.value:
@@ -59,7 +59,7 @@ def _clone_update_in_place(node: "Node", node_clone_policy: int, override_finish
             for parameter in sub_node.parameters:
                 if not parameter.reference:
                     continue
-                parameter.value = node.get_parameter_by_name(parameter.reference, throw=True).value
+                parameter.value = node.get_parameter_by_name(parameter.reference).value
 
             if sub_node.node_running_status == NodeRunningStatus.STATIC:
                 # do not copy the rest of the elements because they don't change
@@ -101,17 +101,26 @@ class Output(_BaseResource):
 class InputReference(DBObject):
     """Basic Value of the Input structure."""
 
-    node_id: str = ""
+    node_id: ObjectId = field(default_factory=ObjectId)
     output_id: str = ""
 
 
 @dataclass_json
 @dataclass
 class Input(_BaseResource):
     """Basic Input structure."""
-
+    primitive_override: Any = None
     input_references: List[InputReference] = field(default_factory=list)
 
+    def __post_init__(self):
+        """Set default primitive_override if it is not set and file_type is primitive"""
+        if self.primitive_override is None and self.file_type in PRIMITIVE_TYPES:
+            self.primitive_override = _get_default_by_type(self.file_type)
+
+    def add_input_reference(self, node_id: ObjectId, output_id: str):
+        """Add input reference to the input"""
+        self.input_references.append(InputReference(node_id=node_id, output_id=output_id))
+
 
 @dataclass_json
 @dataclass
@@ -229,25 +238,29 @@ def _get_custom_element(
             return arr[-1]
         return None
 
-    def get_input_by_name(self, name: str, throw: bool = True):
+    def get_input_by_name(self, name: str) -> Input:
         """Find Input object"""
-        return self._get_custom_element(self.inputs, name, throw)
+        return self._get_custom_element(self.inputs, name, throw=True)
+
+    def get_parameter_by_name(self, name: str) -> "Parameter":
+        """Find Parameter object"""
+        return self._get_custom_element(self.parameters, name, throw=True)
 
-    def get_parameter_by_name(self, name: str, throw: bool = True):
+    def get_parameter_by_name_safe(self, name: str) -> Optional["Parameter"]:
         """Find Parameter object"""
-        return self._get_custom_element(self.parameters, name, throw)
+        return self._get_custom_element(self.parameters, name, throw=False)
 
-    def get_output_by_name(self, name: str, throw: bool = True):
+    def get_output_by_name(self, name: str) -> Output:
         """Find Output object"""
-        return self._get_custom_element(self.outputs, name, throw)
+        return self._get_custom_element(self.outputs, name, throw=True)
 
-    def get_log_by_name(self, name: str, throw: bool = False):
+    def get_log_by_name(self, name: str) -> Output:
         """Find Log object"""
-        return self._get_custom_element(self.logs, name, throw, default=Node._default_log)
+        return self._get_custom_element(self.logs, name, throw=False, default=Node._default_log)
 
-    def get_sub_nodes(self):
+    def get_sub_nodes(self) -> List["Node"]:
         """Get a list of subnodes"""
-        sub_nodes_parameter = self.get_parameter_by_name('_nodes', throw=False)
+        sub_nodes_parameter = self.get_parameter_by_name_safe('_nodes')
         if not sub_nodes_parameter:
             raise Exception("Subnodes not found")
         return sub_nodes_parameter.value.value

plynx/db/node_collection_manager.py

@@ -240,8 +240,8 @@ def upgrade_sub_nodes(self, main_node: Node) -> int:
             (int):  Number of upgraded Nodes
         """
         assert self.collection == Collections.TEMPLATES
-        sub_nodes = main_node.get_parameter_by_name('_nodes').value.value
-        node_ids = [node.original_node_id for node in sub_nodes]
+        sub_nodes = main_node.get_sub_nodes()
+        node_ids = [node.original_node_id for node in sub_nodes if node.original_node_id]
         db_nodes = self.get_db_objects_by_ids(node_ids)
         new_node_db_mapping = {}
 
@@ -270,9 +270,12 @@ def upgrade_sub_nodes(self, main_node: Node) -> int:
         # Update nodes from the hub
         # -------------------------
         node_locations = list(set(map(lambda node: node.code_function_location, new_nodes)))
+        node_locations_non_empty: List[str] = [
+            node_location for node_location in node_locations if node_location is not None
+        ]
         hub_nodes_mapping = {
             node.code_function_location: node
-            for node in registry.find_nodes(node_locations)
+            for node in registry.find_nodes(node_locations_non_empty)
         }
         new_nodes = [
             NodeCollectionManager._transplant_node(

plynx/plugins/executors/dag.py

@@ -48,7 +48,7 @@ def __init__(self, node: Node):
         super().__init__(node)
         assert self.node, "Attribute `node` is not defined"
 
-        self.subnodes: List[Node] = self.node.get_parameter_by_name('_nodes', throw=True).value.value
+        self.subnodes: List[Node] = self.node.get_sub_nodes()
 
         self.node_id_to_node: Dict[ObjectId, Node] = {
             node._id: node for node in self.subnodes
@@ -308,7 +308,7 @@ def validate(self, ignore_inputs: bool = True) -> Optional[ValidationError]:
             return validation_error
 
         violations = []
-        sub_nodes = self.node.get_parameter_by_name('_nodes').value.value
+        sub_nodes = self.node.get_sub_nodes()
 
         if len(sub_nodes) == 0:
             violations.append(

plynx/plugins/executors/python/dag.py

@@ -184,6 +184,9 @@ def run(self, preview: bool = False) -> str:
         if preview:
             raise Exception("`preview` is not supported for the DAG")
 
+        if self.worker_pool is None:
+            self.init_executor()
+
         for sub_node in node_utils.traverse_in_order(self.node):
             if NodeRunningStatus.is_finished(sub_node.node_running_status):
                 continue