Repository: xlab-si/xopera-opera (language: python)

Issue:
Throw an error for invalid and undefined interface operations
## Description
The #144 shows that opera doesn't issue any warning if wrong interface operation name is used. In these cases opera proceeds with the deployment and doesn't execute the wrong or undefined interface operations. I think that the user should be at least warned about that.

[TOSCA section 3.6.17 -  Operation definition](https://docs.oasis-open.org/tosca/TOSCA-Simple-Profile-YAML/v1.3/os/TOSCA-Simple-Profile-YAML-v1.3-os.html#DEFN_ELEMENT_INTERFACE_DEF) says that `operation_name` which represents the required symbolic name of the operation as a string can be any string. That also applies for `<interface_definition_name>` ([TOSCA section 3.6.20 - Interface definition](https://docs.oasis-open.org/tosca/TOSCA-Simple-Profile-YAML/v1.3/os/TOSCA-Simple-Profile-YAML-v1.3-os.html#DEFN_ELEMENT_INTERFACE_DEF)). However the operation name cannot be anything within already defined TOSCA interface types like `Standard` and `Configure` and when using the wrong names here the orchestrator should throw an error.

So I propose the following behavior for opera:

- throw an error when wrong operation names are used within `Standard` and `Configure` interfaces
- allow defining user's own interfaces and operations (within node/relationship types or individually) and run their implementations (e.g. Ansible playbooks) when they are referenced within TOSCA triggers or TOSCA Workflows (not available yet - tracked with issue https://github.com/xlab-si/xopera-opera/issues/77)
- throw an error when wrong operation names are used within already defined user's interface types

## Steps
We should add some logic to the opera parser and executor to resolve this.

## Current behaviour
Right now opera allows user to define his own operation and interface types but skips them later during the deployment. If wrong operation names is used in `Standard` and `Configure` interfaces, then opera proceeds without any errors.

## Expected results
To establish better management for TOSCA operation and interface types and definitions.

Store orchestration states in one place only (as enum class)
## Description
The `opera` orchestrator changes the state of nodes during the orchestration process (deployment, undeployment, update, etc.). These states are already explicitly defined in TOSCA as [normative values](https://docs.oasis-open.org/tosca/TOSCA-Simple-Profile-YAML/v1.3/os/TOSCA-Simple-Profile-YAML-v1.3-os.html#_Toc26969445). Opera sticks to that specification as we are using the same names for the states everywhere.

However, when programming new functionalities or updating the old ones, one can easily make a mistake. So, I believe that we should create an enum-like python class for storing these states and call them throughout the code. With 

## Steps
Create an [enum-like](https://docs.python.org/3/library/enum.html) python class (maybe `state.py`) for all possible node's states (we can also include the one and only state for relationships) and update the accessed orchestration states. 

## Current behaviour
Opera doesn't specify the TOSCA node's states in one place but we only use states' names as strings throughout the code.

## Expected results
To use pre-defined orchestration states (as enum values) in opera's source code.



Discussion / hints:

Fix patch:
diff --git a/Pipfile b/Pipfile
index a572faff..fdaf83ef 100644
--- a/Pipfile
+++ b/Pipfile
@@ -4,23 +4,23 @@ url = "https://pypi.org/simple"
 verify_ssl = true
 
 [dev-packages]
-bandit = "*"
-doc8 = "*"
-flake8 = "*"
-flake8-import-order = "*"
-mypy = "*"
-pydocstyle = "*"
-pylint = "*"
-pytest = "*"
-selinux = "*"
+bandit = "==1.7.0"
+doc8 = "==0.8.1"
+flake8 = "==3.8.4"
+flake8-import-order = "==0.18.1"
+mypy = "==0.790"
+pydocstyle = "==5.1.1"
+pylint = "==2.6.0"
+pytest = "==6.2.1"
+selinux = "==0.2.1"
 setuptools = "*"
 wheel = "*"
 twine = ">=3.0.0"
-sphinx = "*"
-sphinx-rtd-theme = "*"
-sphinx-tabs = "*"
-sphinx-argparse = "*"
-sphinx-copybutton = "*"
+sphinx = "==3.3.1"
+sphinx-rtd-theme = "==0.5.0"
+sphinx-tabs = "==2.1.0"
+sphinx-argparse = "==0.2.5"
+sphinx-copybutton = "==0.3.1"
 
 [packages]
-opera = {editable = true,extras = ["openstack"],path = "."}
+opera = {editable = true, extras = ["openstack"], path = "."}
diff --git a/examples/policy_triggers/playbooks/delete.yaml b/examples/policy_triggers/playbooks/delete.yaml
new file mode 100644
index 00000000..b7e1c3e3
--- /dev/null
+++ b/examples/policy_triggers/playbooks/delete.yaml
@@ -0,0 +1,8 @@
+---
+- hosts: all
+  gather_facts: false
+
+  tasks:
+    - name: OpenStack tasks
+      debug:
+        msg: Insert OpenStack tasks
diff --git a/examples/policy_triggers/service.yaml b/examples/policy_triggers/service.yaml
index 1f65653f..858e3294 100644
--- a/examples/policy_triggers/service.yaml
+++ b/examples/policy_triggers/service.yaml
@@ -25,8 +25,14 @@ node_types:
       Standard:
         type: tosca.interfaces.node.lifecycle.Standard
         operations:
-          create:
-            implementation: playbooks/create.yaml
+          create: playbooks/create.yaml
+          delete: playbooks/delete.yaml
+      scaling_up:
+        type: radon.interfaces.scaling.ScaleUp
+      scaling_down:
+        type: radon.interfaces.scaling.ScaleDown
+      autoscaling:
+        type: radon.interfaces.scaling.AutoScale
 
 interface_types:
   radon.interfaces.scaling.ScaleDown:
@@ -61,8 +67,8 @@ policy_types:
   radon.policies.scaling.ScaleDown:
     derived_from: tosca.policies.Scaling
     properties:
-      cpu_upper_bound:
-        description: The upper bound for the CPU
+      cpu_lower_bound:
+        description: The lower bound for the CPU
         type: float
         required: false
         constraints:
@@ -75,19 +81,20 @@ policy_types:
           - less_or_equal: -1
     targets: [ radon.nodes.OpenStack.VM ]
     triggers:
-      radon.triggers.scaling:
+      radon.triggers.scaling.ScaleDown:
         description: A trigger for scaling down
         event: scale_down_trigger
         target_filter:
           node: radon.nodes.OpenStack.VM
         condition:
-          - not:
-              - and:
-                  - available_instances: [ { greater_than: 42 } ]
-                  - available_space: [ { greater_than: 1000 } ]
+          constraint:
+            - not:
+                - and:
+                    - available_instances: [ { greater_than: 42 } ]
+                    - available_space: [ { greater_than: 1000 } ]
         action:
           - call_operation:
-              operation: radon.interfaces.scaling.ScaleDown.scale_down
+              operation: scaling_down.scale_down
               inputs:
                 adjustment: { get_property: [ SELF, adjustment ] }
 
@@ -108,19 +115,20 @@ policy_types:
           - greater_or_equal: 1
     targets: [ radon.nodes.OpenStack.VM ]
     triggers:
-      radon.triggers.scaling:
+      radon.triggers.scaling.ScaleUp:
         description: A trigger for scaling up
         event: scale_up_trigger
         target_filter:
           node: radon.nodes.OpenStack.VM
         condition:
-          - not:
-              - and:
-                  - available_instances: [ { greater_than: 42 } ]
-                  - available_space: [ { greater_than: 1000 } ]
+          constraint:
+            - not:
+                - and:
+                    - available_instances: [ { greater_than: 42 } ]
+                    - available_space: [ { greater_than: 1000 } ]
         action:
           - call_operation:
-              operation: radon.interfaces.scaling.ScaleUp.scale_up
+              operation: scaling_up.scale_up
               inputs:
                 adjustment: { get_property: [ SELF, adjustment ] }
 
@@ -165,7 +173,7 @@ topology_template:
     - scale_down:
         type: radon.policies.scaling.ScaleDown
         properties:
-          cpu_upper_bound: 90
+          cpu_lower_bound: 10
           adjustment: 1
 
     - scale_up:
@@ -181,7 +189,7 @@ topology_template:
           max_size: 7
         targets: [ openstack_vm ]
         triggers:
-          radon.triggers.scaling:
+          radon.triggers.scaling.AutoScale:
             description: A trigger for autoscaling
             event: auto_scale_trigger
             schedule:
@@ -201,5 +209,5 @@ topology_template:
               evaluations: 2
               method: average
             action:
-              - call_operation: radon.interfaces.scaling.AutoScale.retrieve_info
-              - call_operation: radon.interfaces.scaling.AutoScale.autoscale
+              - call_operation: autoscaling.retrieve_info
+              - call_operation: autoscaling.autoscale
diff --git a/examples/runme.sh b/examples/runme.sh
index a301c117..5740a282 100755
--- a/examples/runme.sh
+++ b/examples/runme.sh
@@ -47,6 +47,7 @@ $opera_executable outputs -p intrinsic_functions/.opera
 echo "Testing an example from ./policy_triggers ..."
 mkdir policy_triggers/.opera
 $opera_executable deploy -p policy_triggers/.opera policy_triggers/service.yaml
+$opera_executable undeploy -p policy_triggers/.opera
 
 # test an example from ./relationship_outputs
 echo "Testing an example from ./relationship_outputs ..."
diff --git a/src/opera/compare/instance_comparer.py b/src/opera/compare/instance_comparer.py
index 492f5d64..468a82de 100644
--- a/src/opera/compare/instance_comparer.py
+++ b/src/opera/compare/instance_comparer.py
@@ -1,5 +1,6 @@
 from typing import Dict
 
+from opera.constants import NodeState
 from .comparisons import MapComparison
 from .diff import Diff
 
@@ -28,12 +29,12 @@ def prepare_update(self, topology_template1, topology_template2, template_diff):
         for node1 in topology_template1.nodes.values():
             node_name = self._get_template_name(node1)
             if not nodes_diff.find_key(node_name):
-                node1.set_state("initial", write=False)
+                node1.set_state(NodeState.INITIAL, write=False)
 
         for node2 in topology_template2.nodes.values():
             node_name = self._get_template_name(node2)
             if not nodes_diff.find_key(node_name):
-                node2.set_state("started", write=False)
+                node2.set_state(NodeState.STARTED, write=False)
 
     def _check_dependencies(self, nodes, changed_nodes, parent_name, parent_changed):
         dependency_changes: Dict[str, set] = {}
@@ -81,5 +82,5 @@ def _get_source(relationship):
     def _compare_state(node1, node2, context):
         # comparing states we assume that all nodes in new instance model
         # would be in "started" state after deployment
-        state1 = node1.attributes["state"].data
-        return state1 == "started", [state1, "started"]
+        state1 = node1.state
+        return state1 == NodeState.STARTED, [state1, NodeState.STARTED]
diff --git a/src/opera/constants.py b/src/opera/constants.py
new file mode 100644
index 00000000..c7db1383
--- /dev/null
+++ b/src/opera/constants.py
@@ -0,0 +1,66 @@
+from enum import Enum
+
+
+# enum for tosca.interfaces.node.lifecycle.Standard operations
+class StandardInterfaceOperation(Enum):
+    CREATE = "create"
+    CONFIGURE = "configure"
+    START = "start"
+    STOP = "stop"
+    DELETE = "delete"
+
+    @classmethod
+    def shorthand_name(cls):
+        return "Standard"
+
+    @classmethod
+    def type_uri(cls):
+        return "tosca.interfaces.node.lifecycle.Standard"
+
+
+# enum for tosca.interfaces.relationship.Configure operations
+class ConfigureInterfaceOperation(Enum):
+    PRE_CONFIGURE_SOURCE = "pre_configure_source"
+    PRE_CONFIGURE_TARGET = "pre_configure_target"
+    POST_CONFIGURE_SOURCE = "post_configure_source"
+    POST_CONFIGURE_TARGET = "post_configure_target"
+    ADD_TARGET = "add_target"
+    ADD_SOURCE = "add_source"
+    TARGET_CHANGED = "target_changed"
+    REMOVE_TARGET = "remove_target"
+
+    @classmethod
+    def shorthand_name(cls):
+        return "Configure"
+
+    @classmethod
+    def type_uri(cls):
+        return "tosca.interfaces.relationship.Configure"
+
+
+# enum for TOSCA node orchestration states
+class NodeState(Enum):
+    INITIAL = "initial"
+    CREATING = "creating"
+    CREATED = "created"
+    CONFIGURING = "configuring"
+    CONFIGURED = "configured"
+    STARTING = "starting"
+    STARTED = "started"
+    STOPPING = "stopping"
+    DELETING = "deleting"
+    ERROR = "error"
+
+
+# enum for TOSCA relationship orchestration states
+class RelationshipState(Enum):
+    INITIAL = "initial"
+
+
+# enum for TOSCA operation hosts and reserved TOSCA functions keywords
+class OperationHost(Enum):
+    SELF = "SELF"
+    SOURCE = "SOURCE"
+    TARGET = "TARGET"
+    HOST = "HOST"
+    ORCHESTRATOR = "ORCHESTRATOR"
diff --git a/src/opera/error.py b/src/opera/error.py
index 63325668..5782fd5d 100644
--- a/src/opera/error.py
+++ b/src/opera/error.py
@@ -16,3 +16,7 @@ class DataError(OperaError):
 
 class OperationError(OperaError):
     """Raised on failed operation executions."""
+
+
+class ToscaDeviationError(OperaError):
+    """Raised when something is compatible with TOSCA standard, but not acceptable for the orchestrator."""
diff --git a/src/opera/instance/base.py b/src/opera/instance/base.py
index 328975bf..fa2fa15b 100644
--- a/src/opera/instance/base.py
+++ b/src/opera/instance/base.py
@@ -1,5 +1,7 @@
 import itertools
+from typing import Union
 
+from opera.constants import NodeState, OperationHost, StandardInterfaceOperation, ConfigureInterfaceOperation
 from opera.error import DataError, OperationError
 from opera.instance.topology import Topology
 from opera.value import Value
@@ -16,7 +18,7 @@ def __init__(self, template, instance_id):
         self.attributes = dict(
             tosca_name=Value(None, True, template.name),
             tosca_id=Value(None, True, instance_id),
-            state=Value(None, True, "initial"),
+            state=Value(None, True, NodeState.INITIAL.value),
         )
 
         self.reset_attributes()
@@ -57,15 +59,25 @@ def tosca_id(self):
 
     @property
     def state(self):
-        return self.attributes["state"].data
+        state_value = self.attributes["state"].data
+        try:
+            return next(s for s in NodeState if s.value == state_value)
+        except StopIteration as e:
+            raise DataError("Could not find state {} in {}".format(state_value, list(NodeState))) from e
 
-    def set_state(self, state, write=True):
-        self.set_attribute("state", state)
+    def set_state(self, state: NodeState, write=True):
+        self.set_attribute("state", state.value)
         if write:
             self.write()
 
-    def run_operation(self, host, interface, operation, verbose, workdir):
-        success, outputs, attributes = self.template.run_operation(host, interface, operation, self, verbose, workdir)
+    def run_operation(self,
+                      host: OperationHost,
+                      interface: str,
+                      operation_type: Union[StandardInterfaceOperation, ConfigureInterfaceOperation],
+                      verbose: bool,
+                      workdir: str):
+        success, outputs, attributes = \
+            self.template.run_operation(host, interface, operation_type, self, verbose, workdir)
 
         if not success:
             raise OperationError("Failed")
@@ -82,5 +94,7 @@ def update_attributes(self, outputs):
     def set_attribute(self, name, value):
         # TODO(@tadeboro): Add type validation.
         if name not in self.attributes:
-            raise DataError("Instance has no '{}' attribute".format(name))
+            raise DataError(
+                "Instance has no '{}' attribute. Available attributes: "
+                "{}".format(name, ", ".join(self.attributes.keys())))
         self.attributes[name].set(value)
diff --git a/src/opera/instance/node.py b/src/opera/instance/node.py
index 1eb0348a..845ab0d8 100644
--- a/src/opera/instance/node.py
+++ b/src/opera/instance/node.py
@@ -1,5 +1,11 @@
+from typing import Optional
+
+from opera.constants import StandardInterfaceOperation, ConfigureInterfaceOperation, NodeState, OperationHost
 from opera.error import DataError
+from opera.error import ToscaDeviationError
+from opera.template.trigger import Trigger
 from opera.threading import utils as thread_utils
+from opera.value import Value
 from .base import Base
 
 
@@ -9,6 +15,7 @@ def __init__(self, template, instance_id):
 
         self.in_edges = {}  # This gets filled by other instances for us.
         self.out_edges = {}  # This is what we fill during the linking phase.
+        self.notified = False  # This indicates whether the node has been notified.
 
     def instantiate_relationships(self):
         if not self.topology:
@@ -31,11 +38,11 @@ def instantiate_relationships(self):
 
     @property
     def deployed(self):
-        return self.state == "started"
+        return self.state == NodeState.STARTED
 
     @property
     def undeployed(self):
-        return self.state == "initial"
+        return self.state == NodeState.INITIAL
 
     @property
     def ready_for_deploy(self):
@@ -58,36 +65,47 @@ def get_host(self):
         # localhost if nothing suitable is found.
         return self.template.get_host() or "localhost"
 
-    def deploy(self, verbose, workdir):
-        thread_utils.print_thread("  Deploying {}".format(self.tosca_id))
-
-        self.set_state("creating")
-        self.run_operation("HOST", "Standard", "create", verbose, workdir)
-        self.set_state("created")
-        self.set_state("configuring")
-
+    def _configure_requirements(self,
+                                source_operation: ConfigureInterfaceOperation,
+                                target_operation: ConfigureInterfaceOperation,
+                                verbose: bool, workdir: str):
         for requirement in set(r.name for r in self.template.requirements):
             for relationship in self.out_edges[requirement].values():
-                relationship.run_operation("SOURCE", "Configure", "pre_configure_source", verbose, workdir)
+                relationship.run_operation(OperationHost.SOURCE, ConfigureInterfaceOperation.shorthand_name(),
+                                           source_operation, verbose, workdir)
 
         for requirement_dependants in self.in_edges.values():
             for relationship in requirement_dependants.values():
-                relationship.run_operation("TARGET", "Configure", "pre_configure_target", verbose, workdir)
+                relationship.run_operation(OperationHost.TARGET, ConfigureInterfaceOperation.shorthand_name(),
+                                           target_operation, verbose, workdir)
+
+    def deploy(self, verbose, workdir):
+        thread_utils.print_thread("  Deploying {}".format(self.tosca_id))
 
-        self.run_operation("HOST", "Standard", "configure", verbose, workdir)
+        self.set_state(NodeState.CREATING)
+        self.run_operation(OperationHost.HOST, StandardInterfaceOperation.shorthand_name(),
+                           StandardInterfaceOperation.CREATE, verbose, workdir)
+        self.set_state(NodeState.CREATED)
+        self.set_state(NodeState.CONFIGURING)
 
-        for requirement in set(r.name for r in self.template.requirements):
-            for relationship in self.out_edges[requirement].values():
-                relationship.run_operation("SOURCE", "Configure", "post_configure_source", verbose, workdir)
+        self._configure_requirements(ConfigureInterfaceOperation.PRE_CONFIGURE_SOURCE,
+                                     ConfigureInterfaceOperation.PRE_CONFIGURE_TARGET,
+                                     verbose, workdir)
 
-        for requirement_dependants in self.in_edges.values():
-            for relationship in requirement_dependants.values():
-                relationship.run_operation("TARGET", "Configure", "post_configure_target", verbose, workdir)
+        self.run_operation(OperationHost.HOST,
+                           StandardInterfaceOperation.shorthand_name(),
+                           StandardInterfaceOperation.CONFIGURE,
+                           verbose, workdir)
+
+        self._configure_requirements(ConfigureInterfaceOperation.POST_CONFIGURE_SOURCE,
+                                     ConfigureInterfaceOperation.POST_CONFIGURE_TARGET,
+                                     verbose, workdir)
 
-        self.set_state("configured")
-        self.set_state("starting")
-        self.run_operation("HOST", "Standard", "start", verbose, workdir)
-        self.set_state("started")
+        self.set_state(NodeState.CONFIGURED)
+        self.set_state(NodeState.STARTING)
+        self.run_operation(OperationHost.HOST, StandardInterfaceOperation.shorthand_name(),
+                           StandardInterfaceOperation.START, verbose, workdir)
+        self.set_state(NodeState.STARTED)
 
         # TODO(@tadeboro): Execute various add hooks
         thread_utils.print_thread("  Deployment of {} complete".format(self.tosca_id))
@@ -95,13 +113,15 @@ def deploy(self, verbose, workdir):
     def undeploy(self, verbose, workdir):
         thread_utils.print_thread("  Undeploying {}".format(self.tosca_id))
 
-        self.set_state("stopping")
-        self.run_operation("HOST", "Standard", "stop", verbose, workdir)
-        self.set_state("configured")
+        self.set_state(NodeState.STOPPING)
+        self.run_operation(OperationHost.HOST, StandardInterfaceOperation.shorthand_name(),
+                           StandardInterfaceOperation.STOP, verbose, workdir)
+        self.set_state(NodeState.CONFIGURED)
 
-        self.set_state("deleting")
-        self.run_operation("HOST", "Standard", "delete", verbose, workdir)
-        self.set_state("initial")
+        self.set_state(NodeState.DELETING)
+        self.run_operation(OperationHost.HOST, StandardInterfaceOperation.shorthand_name(),
+                           StandardInterfaceOperation.DELETE, verbose, workdir)
+        self.set_state(NodeState.INITIAL)
 
         # TODO(@tadeboro): Execute various remove hooks
 
@@ -110,18 +130,63 @@ def undeploy(self, verbose, workdir):
 
         thread_utils.print_thread("  Undeployment of {} complete".format(self.tosca_id))
 
+    def invoke_trigger(self, verbose: bool, workdir: str, trigger: Trigger,
+                       notification_file_contents: Optional[str]):
+        for interface_name, operation_name, _ in trigger.action:
+            # get node's interface and interface operations that are referenced in a trigger action
+            interface = self.template.interfaces.get(interface_name, None)
+            operation = interface.operations.get(operation_name, None) if interface else None
+
+            if interface is not None and operation is not None:
+                if notification_file_contents:
+                    # add notification file contents to operation inputs that are exposed in the executor
+                    if "notification" in operation.inputs:
+                        raise ToscaDeviationError("The input name: notification within the node: {} cannot be "
+                                                  "used. It it reserved and holds the contents of notification "
+                                                  "file. Please rename it.".format(self.template.name))
+
+                    notification_value = Value(None, True, notification_file_contents)
+                    operation.inputs.update({"notification": notification_value})
+
+                self.run_operation(OperationHost.HOST, interface_name, operation_name, verbose, workdir)
+
+    def notify(self, verbose: bool, workdir: str, trigger_name_or_event: Optional[str],
+               notification_file_contents: Optional[str]):
+        thread_utils.print_thread("  Notifying {}".format(self.tosca_id))
+
+        for policy in self.template.policies:
+            for trigger_name, trigger in policy.triggers.items():
+                # break here if trigger is not targeting this node
+                if trigger.target_filter and trigger.target_filter[0] != self.template.name:
+                    break
+
+                invoke_trigger = True
+                if trigger_name_or_event is not None and trigger_name_or_event not in (trigger_name,
+                                                                                       trigger.event.data):
+                    invoke_trigger = False
+
+                # invoke the chosen trigger only if it contains any actions
+                if invoke_trigger and trigger.action:
+                    thread_utils.print_thread("   Calling trigger {} with event {}".format(trigger_name, trigger.event))
+                    self.invoke_trigger(verbose, workdir, trigger, notification_file_contents)
+                    thread_utils.print_thread("   Calling trigger actions with event {} complete".format(trigger.event))
+
+        self.notified = True
+        thread_utils.print_thread("  Notification on {} complete".format(self.tosca_id))
+
     #
     # TOSCA functions
     #
     def get_attribute(self, params):
         host, attr, *rest = params
 
-        if host == "HOST":
-            raise DataError("HOST is not yet supported in opera.")
-        if host != "SELF":
-            raise DataError("Attribute host should be set to 'SELF' which is the only valid value. "
+        if host == OperationHost.HOST.value:
+            raise DataError(
+                "{} as the attribute's 'host' is not yet supported in opera.".format(OperationHost.HOST.value))
+        if host != OperationHost.SELF.value:
+            raise DataError("The attribute's 'host' should be set to '{}' which is the only valid value. "
                             "This is needed to indicate that the attribute is referenced locally from something "
-                            "in the node itself.")
+                            "in the node itself. Was: {}".format(OperationHost.SELF.value, host))
 
         # TODO(@tadeboro): Add support for nested attribute values once we
         # have data type support.
@@ -143,10 +208,10 @@ def get_attribute(self, params):
         # If we have no attribute, try searching for requirement.
         relationships = self.out_edges.get(attr, {})
         if len(relationships) == 0:
-            raise DataError("Cannot find attribute '{}'.".format(attr))
+            raise DataError("Cannot find attribute '{}' among {}.".format(attr, ", ".join(self.out_edges.keys())))
         if len(relationships) > 1:
             raise DataError("Targeting more than one instance via '{}'.".format(attr))
-        return next(iter(relationships.values())).target.get_attribute(["SELF"] + rest)
+        return next(iter(relationships.values())).target.get_attribute([OperationHost.SELF.value] + rest)
 
     def get_property(self, params):
         return self.template.get_property(params)
@@ -157,17 +222,18 @@ def get_input(self, params):
     def map_attribute(self, params, value):
         host, attr, *_ = params
 
-        if host == "HOST":
-            raise DataError("HOST is not yet supported in opera.")
-        if host != "SELF":
-            raise DataError("Attribute host should be set to 'SELF' which is the only valid value. "
+        if host == OperationHost.HOST.value:
+            raise DataError(
+                "{} as the attribute's 'host' is not yet supported in opera.".format(OperationHost.HOST.value))
+        if host != OperationHost.SELF.value:
+            raise DataError("The attribute's 'host' should be set to '{}' which is the only valid value. "
                             "This is needed to indicate that the attribute is referenced locally from something "
-                            "in the node itself.")
+                            "in the node itself. Was: {}".format(OperationHost.SELF.value, host))
 
         # TODO(@tadeboro): Add support for nested attribute values once we
         # have data type support.
         if attr not in self.attributes:
-            raise DataError("Cannot find attribute '{}'.".format(attr))
+            raise DataError("Cannot find attribute '{}' among {}.".format(attr, ", ".join(self.attributes.keys())))
 
         self.set_attribute(attr, value)
 
diff --git a/src/opera/instance/relationship.py b/src/opera/instance/relationship.py
index 6d637ae2..cb68a961 100644
--- a/src/opera/instance/relationship.py
+++ b/src/opera/instance/relationship.py
@@ -1,3 +1,4 @@
+from opera.constants import OperationHost
 from opera.error import DataError
 from opera.instance.base import Base
 
@@ -15,30 +16,33 @@ def __init__(self, template, instance_id, source, target):
     def get_attribute(self, params):
         host, attr, *rest = params
 
-        if host not in ("SELF", "SOURCE", "TARGET"):
-            raise DataError("Attribute host should be set to 'SELF', 'SOURCE' or 'TARGET'.")
+        valid_hosts = [i.value for i in OperationHost]
+        if host not in valid_hosts:
+            raise DataError("The attribute's 'host' should be set to one of {}.".format(", ".join(valid_hosts)))
 
-        if host == "SOURCE":
-            return self.source.get_attribute(["SELF", attr] + rest)
-        elif host == "TARGET":
-            return self.target.get_attribute(["SELF", attr] + rest)
+        if host == OperationHost.SOURCE.value:
+            return self.source.get_attribute([OperationHost.SELF.value, attr] + rest)
+        elif host == OperationHost.TARGET.value:
+            return self.target.get_attribute([OperationHost.SELF.value, attr] + rest)
 
         # TODO(@tadeboro): Add support for nested attribute values once we
         # have data type support.
         if attr not in self.attributes:
-            raise DataError("Instance has no '{}' attribute".format(attr))
+            raise DataError(
+                "Instance has no '{}' attribute. Available attributes: {}".format(attr, ", ".join(self.attributes)))
         return self.attributes[attr].eval(self, attr)
 
     def get_property(self, params):
         host, prop, *rest = params
 
-        if host not in ("SELF", "SOURCE", "TARGET"):
-            raise DataError("Property host should be set to 'SELF', 'SOURCE' or 'TARGET'.")
+        valid_hosts = [i.value for i in OperationHost]
+        if host not in valid_hosts:
+            raise DataError("Property host should be set to one of {}. Was: {}".format(", ".join(valid_hosts), host))
 
-        if host == "SOURCE":
-            return self.source.get_property(["SELF", prop] + rest)
-        if host == "TARGET":
-            return self.target.get_property(["SELF", prop] + rest)
+        if host == OperationHost.SOURCE.value:
+            return self.source.get_property([OperationHost.SELF.value, prop] + rest)
+        if host == OperationHost.TARGET.value:
+            return self.target.get_property([OperationHost.SELF.value, prop] + rest)
         return self.template.get_property(params)
 
     def get_input(self, params):
@@ -47,26 +51,32 @@ def get_input(self, params):
     def map_attribute(self, params, value):
         host, attr, *rest = params
 
-        if host not in ("SELF", "SOURCE", "TARGET"):
-            raise DataError("Attribute host should be set to 'SELF', 'SOURCE' or 'TARGET'.")
-
-        if host == "SOURCE":
-            self.source.map_attribute(["SELF", attr] + rest, value)
-        elif host == "TARGET":
-            self.target.map_attribute(["SELF", attr] + rest, value)
+        valid_hosts = [i.value for i in OperationHost]
+        if host == OperationHost.HOST.value:
+            raise DataError(
+                "{} as the attribute's 'host' is not yet supported in opera.".format(OperationHost.HOST.value))
+        if host not in valid_hosts:
+            raise DataError("The attribute's 'host' should be set to one of {}. Was: "
+                            "{}".format(", ".join(valid_hosts), host))
+
+        if host == OperationHost.SOURCE.value:
+            self.source.map_attribute([OperationHost.SELF.value, attr] + rest, value)
+        elif host == OperationHost.TARGET.value:
+            self.target.map_attribute([OperationHost.SELF.value, attr] + rest, value)
         else:
             self.set_attribute(attr, value)
 
     def get_artifact(self, params):
         host, prop, *rest = params
 
-        if host not in ("SELF", "SOURCE", "TARGET"):
-            raise DataError("Artifact host should be set to 'SELF', 'SOURCE' or 'TARGET'.")
+        valid_hosts = [i.value for i in OperationHost]
+        if host not in valid_hosts:
+            raise DataError("Artifact host should be set to one of {}. Was: {}".format(", ".join(valid_hosts), host))
 
-        if host == "SOURCE":
-            return self.source.get_property(["SELF", prop] + rest)
-        if host == "TARGET":
-            return self.target.get_property(["SELF", prop] + rest)
+        if host == OperationHost.SOURCE.value:
+            return self.source.get_property([OperationHost.SELF.value, prop] + rest)
+        if host == OperationHost.TARGET.value:
+            return self.target.get_property([OperationHost.SELF.value, prop] + rest)
         return self.template.get_artifact(params)
 
     def concat(self, params):
diff --git a/src/opera/instance/topology.py b/src/opera/instance/topology.py
index 0eefeb3a..d8a3766c 100644
--- a/src/opera/instance/topology.py
+++ b/src/opera/instance/topology.py
@@ -1,3 +1,5 @@
+from typing import Optional
+
 from opera.threading import NodeExecutor
 
 
@@ -34,7 +36,7 @@ def deploy(self, verbose, workdir, num_workers=None):
                 do_deploy = executor.wait_results()
 
     def undeploy(self, verbose, workdir, num_workers=None):
-        # Currently, we are running a really stupid O(n^3) algorithm, but  unless we get to the templates with
+        # Currently, we are running a really stupid O(n^3) algorithm, but unless we get to the templates with
         # millions of node instances, we should be fine.
         with NodeExecutor(num_workers) as executor:
             do_undeploy = True
@@ -46,6 +48,18 @@ def undeploy(self, verbose, workdir, num_workers=None):
                         executor.submit_operation(node.undeploy, node.tosca_id, verbose, workdir)
                 do_undeploy = executor.wait_results()
 
+    def notify(self, verbose: bool, workdir: str, trigger_name_or_event: Optional[str],
+               notification_file_contents: Optional[str], num_workers=1):
+        # This will run selected interface operations on triggers from policies that have been applied to nodes.
+        with NodeExecutor(num_workers) as executor:
+            do_notify = True
+            while do_notify:
+                for node in self.nodes.values():
+                    if not node.notified and executor.can_submit(node.tosca_id):
+                        executor.submit_operation(node.notify, node.tosca_id, verbose, workdir, trigger_name_or_event,
+                                                  notification_file_contents)
+                do_notify = executor.wait_results()
+
     def write(self, data, instance_id):
         self.storage.write_json(data, "instances", instance_id)
 
diff --git a/src/opera/parser/tosca/v_1_3/collector_mixin.py b/src/opera/parser/tosca/v_1_3/collector_mixin.py
index 9ae6c620..d79890ce 100644
--- a/src/opera/parser/tosca/v_1_3/collector_mixin.py
+++ b/src/opera/parser/tosca/v_1_3/collector_mixin.py
@@ -1,5 +1,6 @@
 # type: ignore
-
+from opera.constants import OperationHost
+from opera.error import DataError
 from opera.template.capability import Capability
 from opera.template.interface import Interface
 from opera.template.operation import Operation
@@ -61,7 +62,7 @@ def collect_interfaces(self, service_ast):  # pylint: disable=too-many-locals
 
         undeclared_interfaces = set(assignments.keys()) - definitions.keys()
         if undeclared_interfaces:
-            self.abort("Invalid interfaces: {}.".format(", ".join(undeclared_interfaces)), self.loc)
+            self.abort("Undeclared interfaces: {}.".format(", ".join(undeclared_interfaces)), self.loc)
 
         # Next section is nasty. You have been warned.
         interfaces = {}
@@ -110,8 +111,8 @@ def collect_interfaces(self, service_ast):  # pylint: disable=too-many-locals
                     ), self.loc)
 
                 # Outputs, which define the attribute mapping, come from:
-                #  1. inteface operation definition,
-                #  2. inteface operation assignment in template section
+                #  1. interface operation definition,
+                #  2. interface operation assignment in template section
                 outputs = {
                     k: [s.data for s in v.data]
                     for k, v in op_definition.get("outputs", {}).items()
@@ -123,17 +124,22 @@ def collect_interfaces(self, service_ast):  # pylint: disable=too-many-locals
 
                 # Operation implementation details
                 impl = op_assignment.get("implementation") or op_definition.get("implementation")
-                # TODO(@tadeboro): impl can be None here. Fix this.
+                # TODO: when impl is None we also pass that forward to operation objects. Fix this if needed.
                 timeout, operation_host = 0, None
-                if "timeout" in impl:
+                if impl and "timeout" in impl:
                     timeout = impl.timeout.data
-                if "operation_host" in impl:
-                    operation_host = impl.operation_host.data
+                if impl and "operation_host" in impl:
+                    operation_host_value = impl.operation_host.data
+                    try:
+                        operation_host = next(oh for oh in OperationHost if oh.value == operation_host_value)
+                    except StopIteration as e:
+                        raise DataError("Could not find operation host {} in "
+                                        "{}".format(operation_host_value, list(OperationHost))) from e
 
                 operations[op_name] = Operation(
                     op_name,
-                    primary=impl.primary.file.data,
-                    dependencies=[d.file.data for d in impl.get("dependencies", [])],
+                    primary=impl.primary.file.data if impl else None,
+                    dependencies=[d.file.data for d in impl.get("dependencies", [])] if impl else [],
                     artifacts=[a.data for a in self.collect_artifacts(service_ast).values()],
                     inputs=inputs,
                     outputs=outputs,
diff --git a/src/opera/parser/tosca/v_1_3/definition_collector_mixin.py b/src/opera/parser/tosca/v_1_3/definition_collector_mixin.py
index cd96caa8..b945a888 100644
--- a/src/opera/parser/tosca/v_1_3/definition_collector_mixin.py
+++ b/src/opera/parser/tosca/v_1_3/definition_collector_mixin.py
@@ -2,6 +2,8 @@
 
 import collections
 
+from opera.constants import StandardInterfaceOperation, ConfigureInterfaceOperation
+
 
 class DefinitionCollectorMixin:
     def collect_types(self, service_ast):
@@ -41,6 +43,33 @@ def collect_interface_definitions(self, service_ast):
                 defs[name]["operations"].update(definition["operations"])
 
         for name, definition in self.get("interfaces", {}).items():
+            standard_interface_names = [StandardInterfaceOperation.shorthand_name(),
+                                        StandardInterfaceOperation.type_uri()]
+            configure_interface_names = [ConfigureInterfaceOperation.shorthand_name(),
+                                         ConfigureInterfaceOperation.type_uri()]
+
+            if name in standard_interface_names:
+                valid_standard_interface_operation_names = [i.value for i in StandardInterfaceOperation]
+                for operation in definition.get("operations", {}):
+                    if operation not in valid_standard_interface_operation_names:
+                        self.abort("Invalid operation for {} interface: {}. Valid operation names are: {}."
+                                   .format(name, operation, valid_standard_interface_operation_names), self.loc)
+
+            if name in configure_interface_names:
+                valid_configure_interface_operation_names = [i.value for i in ConfigureInterfaceOperation]
+                for operation in definition.get("operations", {}):
+                    if operation not in valid_configure_interface_operation_names:
+                        self.abort("Invalid operation for {} interface: {}. Valid operation names are: {}."
+                                   .format(name, operation, valid_configure_interface_operation_names), self.loc)
+
+            # collect operations and inputs from linked interface_types
+            # do this only when type is specified and interface name is not Standard or Configure
+            if definition.get("type", None) and name not in standard_interface_names + configure_interface_names:
+                parent_interface_type = definition.type.resolve_reference(service_ast)
+                if parent_interface_type:
+                    defs[name]["inputs"].update(parent_interface_type.get("inputs", {}))
+                    defs[name]["operations"].update(parent_interface_type.get("operations", {}))
+
             defs[name]["inputs"].update(definition.get("inputs", {}))
             defs[name]["operations"].update(definition.get("operations", {}))
 
@@ -48,3 +77,9 @@ def collect_interface_definitions(self, service_ast):
 
     def collect_artifact_definitions(self, service_ast):
         return self._collect_definitions("artifacts", service_ast)
+
+    def collect_target_definitions(self, service_ast):
+        return {target.data: target for target in self.get("targets", [])}
+
+    def collect_trigger_definitions(self, service_ast):
+        return self._collect_definitions("triggers", service_ast)
diff --git a/src/opera/parser/tosca/v_1_3/operation_host.py b/src/opera/parser/tosca/v_1_3/operation_host.py
index 9cfd1a98..01f5b32d 100644
--- a/src/opera/parser/tosca/v_1_3/operation_host.py
+++ b/src/opera/parser/tosca/v_1_3/operation_host.py
@@ -1,8 +1,9 @@
+from opera.constants import OperationHost as OperationHostConstant
 from ..string import String
 
 
 class OperationHost(String):
-    VALID_HOSTS = {"SELF", "HOST", "SOURCE", "TARGET", "ORCHESTRATOR"}
+    VALID_HOSTS = [h.value for h in OperationHostConstant]
 
     @classmethod
     def validate(cls, yaml_node):
diff --git a/src/opera/parser/tosca/v_1_3/policy_definition.py b/src/opera/parser/tosca/v_1_3/policy_definition.py
index 4b0568d6..606f49b8 100644
--- a/src/opera/parser/tosca/v_1_3/policy_definition.py
+++ b/src/opera/parser/tosca/v_1_3/policy_definition.py
@@ -1,3 +1,10 @@
+from typing import Optional, Dict, Any, Tuple, List as TypingList
+
+from opera.template.node import Node
+from opera.template.operation import Operation
+from opera.template.policy import Policy
+from opera.template.trigger import Trigger
+from .collector_mixin import CollectorMixin  # type: ignore
 from .trigger_definition import TriggerDefinition
 from ..entity import Entity
 from ..list import List
@@ -7,7 +14,7 @@
 from ..void import Void
 
 
-class PolicyDefinition(Entity):
+class PolicyDefinition(CollectorMixin, Entity):
     ATTRS = dict(
         type=Reference("policy_types"),
         description=String,
@@ -17,3 +24,185 @@ class PolicyDefinition(Entity):
         triggers=Map(TriggerDefinition),
     )
     REQUIRED = {"type"}
+
+    def get_template(self, name: str, service_ast: Dict[str, Any], nodes: Dict[str, Node]):
+        # targets will be used also for collecting triggers so retrieve them here only once
+        targets = self.collect_targets(service_ast)
+
+        policy = Policy(
+            name=name,
+            types=self.collect_types(service_ast),
+            properties=self.collect_properties(service_ast),
+            targets=targets,
+            triggers=self.collect_triggers(service_ast, targets, nodes)
+        )
+
+        policy.resolve_targets(nodes)
+        return policy
+
+    # the next function is not part of the CollectorMixin because targets are policy only thing
+    def collect_targets(self, service_ast: Dict[str, Any]):
+        typ = self.type.resolve_reference(service_ast)
+        definitions = typ.collect_target_definitions(service_ast)
+        assignments = {target.data: target for target in self.get("targets", [])}
+
+        duplicate_targets = set(assignments.keys()).intersection(definitions.keys())
+        if duplicate_targets:
+            for duplicate in duplicate_targets:
+                definitions.pop(duplicate)
+
+        definitions.update(assignments)
+
+        return {
+            name: (assignments.get(name) or definition).resolve_reference(service_ast)
+            for name, definition in definitions.items()
+        }
+
+    # the next function is not part of the CollectorMixin because triggers are policy only thing
+    def collect_trigger_action_from_interfaces(self, targeted_nodes: Dict[str, Node],
+                                               call_operation_name: Optional[str],
+                                               action: Dict[str, Any]) -> Optional[Tuple[str, str, Operation]]:
+        collected_action = None
+        actions_found = 0
+        for _, target_node in targeted_nodes.items():
+            # loop through interfaces from targeted nodes
+            for interface_name, interface in target_node.interfaces.items():
+                # loop through interface operations from targeted nodes
+                for operation_name, operation in interface.operations.items():
+                    # find the corresponding node's interface operation
+                    if str(interface_name) + "." + str(operation_name) == str(call_operation_name):
+                        actions_found += 1
+
+                        # override the operation inputs with inputs from trigger's activity definition
+                        operation.inputs = {
+                            k: [s.data for s in v.data]
+                            for k, v in action.get("inputs", {}).items()
+                        }
+
+                        collected_action = (interface_name, operation_name, operation)
+                        break
+
+        if actions_found == 0:
+            self.abort("Trigger action: {} from call_operation does not belong to any node interface. Make "
+                       "sure that you have referenced it correctly (as "
+                       "<interface_sub_name>.<operation_sub_name>, where interface_sub_name is the "
+                       "interface name and the operation_sub_name is the name of the operation within this "
+                       "interface). The node that you're targeting with interface operation also has to be "
+                       "used in topology_template/node_templates section."
+                       .format(call_operation_name), self.loc)
+        elif actions_found > 1:
+            self.abort("Found duplicated trigger actions: {} from call_operation. It seems that the "
+                       "operation with the same name belongs to two different node types/templates. "
+                       "".format(call_operation_name), self.loc)
+
+        return collected_action
+
+    # the next function is not part of the CollectorMixin because triggers are policy only thing
+    def collect_trigger_target_nodes(self, target_filter: Optional[Tuple[str, Any]], nodes: Dict[str, Node],
+                                     policy_targets: Dict[str, Any]) -> Dict[str, Node]:
+        # pylint: disable=no-self-use
+        targeted_nodes = dict()
+        if target_filter:
+            # if target node filter is applied collect just one targeted node from it
+            for node_name, node in nodes.items():
+                if node_name == target_filter[0] or node.types[0] == target_filter[0]:
+                    targeted_nodes[node_name] = node
+                    break
+        elif policy_targets:
+            # if target_filter is not present collect target nodes from policy's targets
+            for node_name, node in nodes.items():
+                for policy_target_name, _ in policy_targets.items():
+                    if policy_target_name in (node_name, node.types[0]):
+                        targeted_nodes[node_name] = node
+        else:
+            # if we don't have any target node limits take all template's nodes into account
+            targeted_nodes = nodes
+
+        return targeted_nodes
+
+    # the next function is not part of the CollectorMixin because triggers are policy only thing
+    def collect_trigger_actions(self, definition: Dict[str, Any],
+                                target_filter: Optional[Tuple[str, Any]], nodes: Dict[str, Node],
+                                policy_targets: Dict[str, Any]) -> TypingList[Tuple[str, str, Operation]]:
+        actions = []
+        action_definitions = definition.get("action", [])
+        for action in action_definitions:
+            # TODO: implement support for other types of trigger activity definitions.
+            if list(action)[0] != "call_operation":
+                self.abort("Unsupported trigger activity definitions: {}. Only call_operation is supported."
+                           .format(list(action)[0]), self.loc)
+            else:
+                # collect connected node interface operations
+                call_operation = action.get("call_operation", None)
+                call_operation_name = None
+                # handle short call_operation notation
+                if isinstance(call_operation.data, str):
+                    call_operation_name = call_operation.data
+                # handle extended call_operation notation
+                elif isinstance(call_operation.data, dict):
+                    call_operation_name = call_operation.data.get("operation", None)
+                else:
+                    self.abort("Invalid call operation activity definition type: {}."
+                               .format(type(call_operation.data, )), self.loc)
+
+                # having no operation name should never happen but to be completely sure we can also check here
+                if not call_operation_name:
+                    self.abort("Missing required name for call_operation activity definition.", self.loc)
+
+                # find Node objects that we are targeting with trigger action
+                targeted_nodes = self.collect_trigger_target_nodes(target_filter, nodes, policy_targets)
+
+                # collect actions (interface operations) from targeted nodes
+                collected_action = self.collect_trigger_action_from_interfaces(targeted_nodes, call_operation_name,
+                                                                               action)
+                if collected_action:
+                    actions.append(collected_action)
+
+        return actions
+
+    # the next function is not part of the CollectorMixin because triggers are policy only thing
+    def collect_triggers(self, service_ast: Dict[str, Any], policy_targets: Dict[str, Any], nodes: Dict[str, Node]):
+        # pylint: disable=too-many-locals
+        typ = self.type.resolve_reference(service_ast)
+        definitions = typ.collect_trigger_definitions(service_ast)
+        assignments = self.get("triggers", {})
+
+        duplicate_triggers = set(assignments.keys()).intersection(definitions.keys())
+        if duplicate_triggers:
+            for duplicate in duplicate_triggers:
+                definitions.pop(duplicate)
+
+        definitions.update(assignments)
+
+        # TODO: optimize this code which is now nasty with a lot of parsing, looping and everything else.
+        triggers = dict()
+        for name, definition in definitions.items():
+            # collect and resolve target filter definition
+            target_filter = None
+            target_filter_definitions = definition.get("target_filter", None)
+            if target_filter_definitions:
+                target_node = target_filter_definitions.get("node", None)
+                if target_node:
+                    # resolve node reference and set target_filter
+                    target_filter = (target_node.data, target_node.resolve_reference(service_ast))
+                else:
+                    self.abort("Cannot obtain node from target_filter.", self.loc)
+
+            # check if target_filter also matches one node reference from policy's targets
+            if target_filter and policy_targets and target_filter[0] not in list(policy_targets):
+                self.abort("The node reference: {} from policy trigger's target_filter should be also present in "
+                           "policy's targets.".format(target_filter[0]), self.loc)
+
+            # collect action definitions
+            actions = self.collect_trigger_actions(definition, target_filter, nodes, policy_targets)
+
+            trigger = Trigger(name=name,
+                              event=definition.get("event", None),
+                              target_filter=target_filter,
+                              condition=definition.get("condition", None),
+                              action=actions)
+
+            trigger.resolve_event_filter(nodes)
+            triggers[name] = trigger
+
+        return triggers
diff --git a/src/opera/parser/tosca/v_1_3/policy_type.py b/src/opera/parser/tosca/v_1_3/policy_type.py
index b66570df..b9c7c13f 100644
--- a/src/opera/parser/tosca/v_1_3/policy_type.py
+++ b/src/opera/parser/tosca/v_1_3/policy_type.py
@@ -1,3 +1,4 @@
+from .definition_collector_mixin import DefinitionCollectorMixin  # type: ignore
 from .property_definition import PropertyDefinition
 from .trigger_definition import TriggerDefinition
 from ..entity import TypeEntity
@@ -6,7 +7,7 @@
 from ..reference import Reference, ReferenceXOR
 
 
-class PolicyType(TypeEntity):
+class PolicyType(DefinitionCollectorMixin, TypeEntity):
     REFERENCE = Reference("policy_types")
     ATTRS = dict(
         properties=Map(PropertyDefinition),
diff --git a/src/opera/parser/tosca/v_1_3/topology_template.py b/src/opera/parser/tosca/v_1_3/topology_template.py
index 70e471ea..459f3e68 100644
--- a/src/opera/parser/tosca/v_1_3/topology_template.py
+++ b/src/opera/parser/tosca/v_1_3/topology_template.py
@@ -24,19 +24,25 @@ class TopologyTemplate(Entity):
     )
 
     def get_template(self, inputs, service_ast):
+        # nodes will be used also for collecting policies so retrieve them here only once
+        nodes = {name: node_ast.get_template(name, service_ast)
+                 for name, node_ast in self.get("node_templates", {}).items()}
+
         topology = Topology(
             inputs=self.collect_inputs(inputs, service_ast),
             outputs=self.collect_outputs(service_ast),
-            nodes={
-                name: node_ast.get_template(name, service_ast)
-                for name, node_ast in self.get("node_templates", {}).items()
-            },
+            nodes=nodes,
             relationships={
                 name: rel_ast.get_template(name, service_ast)
                 for name, rel_ast in self.get("relationship_templates", {}).items()
             },
+            policies=[
+                next(pol_ast.get_template(name, service_ast, nodes) for name, pol_ast in pol.items())
+                for pol in self.get("policies", [])
+            ]
         )
         topology.resolve_requirements()
+        topology.resolve_policies()
         return topology
 
     def collect_inputs(self, inputs, service_ast):
diff --git a/src/opera/template/node.py b/src/opera/template/node.py
index b223a354..7c6f1fdc 100644
--- a/src/opera/template/node.py
+++ b/src/opera/template/node.py
@@ -1,9 +1,12 @@
 from collections import Counter
 from pathlib import Path
-from typing import Dict
+from typing import Dict, List, Union
 
+from opera.constants import OperationHost, StandardInterfaceOperation, ConfigureInterfaceOperation
 from opera.error import DataError
+from opera.instance.base import Base
 from opera.instance.node import Node as Instance
+from opera.template.policy import Policy
 from opera.template.topology import Topology
 
 
@@ -28,7 +31,10 @@ def __init__(
         self.interfaces = interfaces
         self.artifacts = artifacts
 
-        # This will be set when the node is inserted into a topology
+        # This will be set when the connected policies are resolved in topology.
+        self.policies: List[Policy] = []
+
+        # This will be set when the node is inserted into a topology.
         self.topology: Topology = None
 
         # This will be set at instantiation time.
@@ -87,9 +93,14 @@ def get_host(self):
 
         return host or "localhost"
 
-    def run_operation(self, host, interface, operation, instance, verbose,
-                      workdir):
-        operation = self.interfaces[interface].operations.get(operation)
+    def run_operation(self,
+                      host: OperationHost,
+                      interface: str,
+                      operation_type: Union[StandardInterfaceOperation, ConfigureInterfaceOperation],
+                      instance: Base,
+                      verbose: bool,
+                      workdir: str):
+        operation = self.interfaces[interface].operations.get(operation_type.value)
         if operation:
             return operation.run(host, instance, verbose, workdir)
         return True, {}, {}
@@ -100,12 +111,21 @@ def run_operation(self, host, interface, operation, instance, verbose,
     def get_property(self, params):
         host, prop, *rest = params
 
-        if host == "HOST":
-            raise DataError("HOST is not yet supported in opera.")
-        if host != "SELF":
+        if host == OperationHost.HOST.value:
+            raise DataError("{} is not yet supported in opera.".format(OperationHost.HOST.value))
+        if host != OperationHost.SELF.value:
+            # try to find the property within connected TOSCA polices
+            for policy in self.policies:
+                if host == policy.name or host in policy.types:
+                    # TODO: Add support for nested property values.
+                    if prop in policy.properties:
+                        return policy.properties[prop].eval(self, prop)
+
             raise DataError(
-                "Property host should be set to 'SELF' which is the only valid value. This is needed to indicate that "
-                "the property is referenced locally from something in the node itself."
+                "Property host should be set to '{}' which is the only valid value. This is needed to indicate that "
+                "the property is referenced locally from something in the node itself. The only valid entity to get "
+                "properties from are currently TOSCA policies, but we were unable to find the policy: {}.".format(
+                    OperationHost.SELF.value, host)
             )
 
         # TODO(@tadeboro): Add support for nested property values.
@@ -131,18 +151,20 @@ def get_property(self, params):
             raise DataError("Cannot find property '{}'.".format(prop))
         if len(requirements) > 1:
             raise DataError("More than one requirement is named '{}'.".format(prop))
-        return requirements[0].target.get_property(["SELF"] + rest)
+        return requirements[0].target.get_property([OperationHost.SELF.value] + rest)
 
     def get_attribute(self, params):
         host, *_ = params
 
-        if host == "HOST":
-            raise DataError("HOST is not yet supported in opera.")
-        if host != "SELF":
+        if host == OperationHost.HOST.value:
+            raise DataError("{} as the attribute's 'host' is not yet supported in opera.".format(host))
+        if host != OperationHost.SELF.value:
             raise DataError(
-                "Attribute host should be set to 'SELF' which is the only valid value. This is needed to indicate that "
-                "the attribute is referenced locally from something in the node itself."
+                "The attribute's 'host' should be set to 'SELF' which is the only valid value. "
+                "This is needed to indicate that the attribute is referenced locally from something in the node itself."
+                " Was: {}".format(host)
             )
+
         if len(self.instances) != 1:
             raise DataError("Cannot get an attribute from multiple instances")
 
@@ -154,12 +176,13 @@ def get_input(self, params):
     def map_attribute(self, params, value):
         host, *_ = params
 
-        if host == "HOST":
-            raise DataError("HOST is not yet supported in opera.")
-        if host != "SELF":
+        if host == OperationHost.HOST.value:
+            raise DataError("{} as the attribute's 'host' is not yet supported in opera.".format(host))
+        if host != OperationHost.SELF.value:
             raise DataError(
-                "Attribute host should be set to 'SELF' which is the only valid value. This is needed to indicate that "
-                "the attribute is referenced locally from something in the node itself."
+                "The attribute's 'host' should be set to 'SELF' which is the only valid value. "
+                "This is needed to indicate that the attribute is referenced locally from something in the node itself."
+                " Was: {}".format(host)
             )
 
         if len(self.instances) != 1:
@@ -179,12 +202,13 @@ def get_artifact(self, params):
         if len(rest) == 2:
             location, remove = rest
 
-        if host == "HOST":
+        if host == OperationHost.HOST.value:
             raise DataError("HOST is not yet supported in opera.")
-        if host != "SELF":
+        if host != OperationHost.SELF.value:
             raise DataError(
-                "Artifact host should be set to 'SELF' which is the only valid value. This is needed to indicate that "
-                "the artifact is referenced locally from something in the node itself."
+                "Artifact host should be set to '{}' which is the only valid value. This is needed to indicate that "
+                "the artifact is referenced locally from something in the node itself. "
+                "Was: {}".format(OperationHost.SELF.value, host)
             )
 
         if location == "LOCAL_FILE":
diff --git a/src/opera/template/operation.py b/src/opera/template/operation.py
index 19a0169d..04eaeeb5 100644
--- a/src/opera/template/operation.py
+++ b/src/opera/template/operation.py
@@ -1,10 +1,11 @@
+from opera.constants import OperationHost
 from opera.error import DataError
 from opera.executor import ansible
 from opera.threading import utils as thread_utils
 
 
 class Operation:
-    def __init__(self, name, primary, dependencies, artifacts, inputs, outputs, timeout, host):
+    def __init__(self, name, primary, dependencies, artifacts, inputs, outputs, timeout, host: OperationHost):
         self.name = name
         self.primary = primary
         self.dependencies = dependencies
@@ -14,7 +15,7 @@ def __init__(self, name, primary, dependencies, artifacts, inputs, outputs, time
         self.timeout = timeout
         self.host = host
 
-    def run(self, host, instance, verbose, workdir):
+    def run(self, host: OperationHost, instance, verbose, workdir):
         thread_utils.print_thread("    Executing {} on {}".format(self.name, instance.tosca_id))
 
         # TODO(@tadeboro): Respect the timeout option.
@@ -22,17 +23,21 @@ def run(self, host, instance, verbose, workdir):
         # TODO(@tadeboro): Properly handle SELF - not even sure what this
         # proper way would be at this time.
         host = self.host or host
-        if host in ("SELF", "HOST"):
+        if host in (OperationHost.SELF, OperationHost.HOST):
             actual_host = instance.get_host()
-        elif host == "SOURCE":
+        elif host == OperationHost.SOURCE:
             actual_host = instance.source.get_host()
-        elif host == "TARGET":
+        elif host == OperationHost.TARGET:
             actual_host = instance.target.get_host()
         else:  # ORCHESTRATOR
             actual_host = "localhost"
 
         operation_inputs = {k: v.eval(instance, k) for k, v in self.inputs.items()}
 
+        # TODO: Currently when primary is None we skip running the operation. Fix this if needed.
+        if not self.primary:
+            return True, {}, {}
+
         # TODO(@tadeboro): Generalize executors.
         success, ansible_outputs = ansible.run(
             actual_host, str(self.primary),
diff --git a/src/opera/template/policy.py b/src/opera/template/policy.py
new file mode 100644
index 00000000..58e80e95
--- /dev/null
+++ b/src/opera/template/policy.py
@@ -0,0 +1,29 @@
+class Policy:
+    def __init__(self, name, types, properties, targets, triggers):
+        self.types = types
+        self.name = name
+        self.properties = properties
+        self.targets = targets
+        self.triggers = triggers
+
+    # this will link targets to their corresponding node objects
+    def resolve_targets(self, nodes):
+        if self.targets:
+            for target_name in self.targets.copy().keys():
+                resolved = False
+                for node_name, node in nodes.items():
+                    if node_name == target_name:
+                        # just assign node object when target name is the same as the name of node object/template
+                        self.targets[node_name] = node
+                        resolved = True
+                    elif node.types[0] == target_name:
+                        # when target name matches node object's type we have to remove node type's key and replace it
+                        # with the name of node object/template and then we assign node object to this new key
+                        self.targets.pop(target_name, None)
+                        self.targets[node_name] = node
+                        resolved = True
+
+                # if we haven't found targeted node object (this can happen for example when node type is defined
+                # but is not used in anywhere in topology within node_templates section) then remove this target
+                if not resolved:
+                    self.targets.pop(target_name, None)
diff --git a/src/opera/template/relationship.py b/src/opera/template/relationship.py
index f387ece4..05247287 100644
--- a/src/opera/template/relationship.py
+++ b/src/opera/template/relationship.py
@@ -1,6 +1,8 @@
-from typing import Dict
+from typing import Dict, Union
 
+from opera.constants import OperationHost, StandardInterfaceOperation, ConfigureInterfaceOperation
 from opera.error import DataError
+from opera.instance.base import Base
 from opera.instance.relationship import Relationship as Instance
 from opera.template.topology import Topology
 
@@ -27,8 +29,14 @@ def instantiate(self, source, target):
         self.instances = {relationship_id: Instance(self, relationship_id, source, target)}
         return self.instances.values()
 
-    def run_operation(self, host, interface, operation, instance, verbose, workdir):
-        operation = self.interfaces[interface].operations.get(operation)
+    def run_operation(self,
+                      host: OperationHost,
+                      interface: str,
+                      operation_type: Union[StandardInterfaceOperation, ConfigureInterfaceOperation],
+                      instance: Base,
+                      verbose: bool,
+                      workdir: str):
+        operation = self.interfaces[interface].operations.get(operation_type.value)
         if operation:
             return operation.run(host, instance, verbose, workdir)
         return True, {}, {}
@@ -39,12 +47,13 @@ def run_operation(self, host, interface, operation, instance, verbose, workdir):
     def get_property(self, params):
         host, prop, *_ = params
 
-        if host == "HOST":
-            raise DataError("HOST is not yet supported in opera.")
-        if host != "SELF":
+        if host == OperationHost.TARGET.value:
+            raise DataError("{} is not yet supported in opera.".format(OperationHost.HOST.value))
+        if host != OperationHost.SELF.value:
             raise DataError(
-                "Property host should be set to 'SELF' which is the only valid value. This is needed to indicate that "
-                "the property is referenced locally from something in the node itself."
+                "Property host should be set to '{}' which is the only valid value. This is needed to indicate that "
+                "the property is referenced locally from something in the node itself. "
+                "Was: {}".format(OperationHost.SELF.value, host)
             )
 
         # TODO(@tadeboro): Add support for nested property values once we
@@ -57,12 +66,13 @@ def get_property(self, params):
     def get_attribute(self, params):
         host, attr, *_ = params
 
-        if host == "HOST":
-            raise DataError("HOST is not yet supported in opera.")
-        if host != "SELF":
+        if host == OperationHost.HOST.value:
+            raise DataError("{} is not yet supported in opera.".format(OperationHost.HOST.value))
+        if host != OperationHost.SELF.value:
             raise DataError(
-                "Attribute host should be set to 'SELF' which is the only valid value. This is needed to indicate that "
-                "the attribute is referenced locally from something in the node itself."
+                "The attribute's 'host' should be set to '{}' which is the only valid value."
+                "This is needed to indicate that the attribute is referenced locally from something in the node itself."
+                " Was: {}".format(OperationHost.SELF.value, host)
             )
 
         if attr not in self.attributes:
@@ -84,8 +94,9 @@ def get_input(self, params):
     def map_attribute(self, params, value):
         host, *_ = params
 
-        if host not in ("SELF", "SOURCE", "TARGET"):
-            raise DataError("Attribute host should be set to 'SELF', 'SOURCE' or 'TARGET'.")
+        valid_hosts = [i.value for i in OperationHost]
+        if host not in valid_hosts:
+            raise DataError("The attribute's 'host' should be set to one of {}.".format(", ".join(valid_hosts)))
 
         if len(self.instances) != 1:
             raise DataError("Mapping an attribute for multiple instances not supported")
diff --git a/src/opera/template/topology.py b/src/opera/template/topology.py
index 86131517..f0d696e6 100644
--- a/src/opera/template/topology.py
+++ b/src/opera/template/topology.py
@@ -1,15 +1,17 @@
 import itertools
 
+from opera.constants import OperationHost
 from opera.error import DataError
 from opera.instance.topology import Topology as Instance
 
 
 class Topology:
-    def __init__(self, inputs, outputs, nodes, relationships):
+    def __init__(self, inputs, outputs, nodes, relationships, policies):
         self.inputs = inputs
         self.outputs = outputs
         self.nodes = nodes
         self.relationships = relationships
+        self.policies = policies
 
         for node in self.nodes.values():
             node.topology = self
@@ -29,6 +31,17 @@ def resolve_relationships(self):
                 if relationship.name in self.relationships.keys():
                     self.relationships[relationship.name] = relationship
 
+    def resolve_policies(self):
+        for node in self.nodes.values():
+            for policy in self.policies:
+                if policy.targets:
+                    # apply policy to node if node name matches the target names that policy is limited to
+                    if node.name in policy.targets.keys():
+                        node.policies.append(policy)
+                else:
+                    # if we don't have target limits or filters apply policy to every node
+                    node.policies.append(policy)
+
     def instantiate(self, storage):
         return Instance(storage, itertools.chain.from_iterable(node.instantiate() for node in self.nodes.values()))
 
@@ -86,16 +99,16 @@ def get_input(self, params):
     def get_property(self, params):
         entity_name, *rest = params
 
-        return self.find_node_or_relationship(entity_name).get_property(["SELF"] + rest)
+        return self.find_node_or_relationship(entity_name).get_property([OperationHost.SELF.value] + rest)
 
     def get_attribute(self, params):
         entity_name, *rest = params
-        return self.find_node_or_relationship(entity_name).get_attribute(["SELF"] + rest)
+        return self.find_node_or_relationship(entity_name).get_attribute([OperationHost.SELF.value] + rest)
 
     def get_artifact(self, params):
         entity_name, *rest = params
 
-        return self.find_node_or_relationship(entity_name).get_artifact(["SELF"] + rest)
+        return self.find_node_or_relationship(entity_name).get_artifact([OperationHost.SELF.value] + rest)
 
     def concat(self, params, node=None):
         if not isinstance(params, list):
diff --git a/src/opera/template/trigger.py b/src/opera/template/trigger.py
new file mode 100644
index 00000000..2b25106c
--- /dev/null
+++ b/src/opera/template/trigger.py
@@ -0,0 +1,22 @@
+class Trigger:
+    def __init__(self, name, event, target_filter, condition, action):
+        self.name = name
+        self.event = event
+        self.target_filter = target_filter
+        self.condition = condition
+        self.action = action
+
+    # this will link trigger's target filter to targeted node object
+    def resolve_event_filter(self, nodes):
+        if self.target_filter:
+            resolved = False
+            for node_name, node in nodes.items():
+                if node_name == self.target_filter[0] or node.types[0] == self.target_filter[0]:
+                    self.target_filter = node_name, node
+                    resolved = True
+                    break
+
+            # if we haven't found targeted node object (this can happen for example when node type is defined but is
+            # not used in anywhere in topology within node_templates section) then unset target_filter
+            if not resolved:
+                self.target_filter = None
diff --git a/src/opera/threading/node_executor.py b/src/opera/threading/node_executor.py
index 45f3f0c4..c1d1349b 100644
--- a/src/opera/threading/node_executor.py
+++ b/src/opera/threading/node_executor.py
@@ -19,9 +19,9 @@ def __init__(self, num_workers=None):
     def can_submit(self, node_id):
         return len(self.processed_nodes) < self.num_workers and node_id not in self.processed_nodes
 
-    def submit_operation(self, operation, node_id, verbose, workdir):
+    def submit_operation(self, operation, node_id, verbose, workdir, *args):
         self.processed_nodes.add(node_id)
-        self.futures[self.submit(operation, verbose, workdir)] = node_id
+        self.futures[self.submit(operation, verbose, workdir, *args)] = node_id
 
     def wait_results(self):
         proceed = bool(self.futures)


Test patch:
diff --git a/tests/unit/opera/compare/test_instance_compare.py b/tests/unit/opera/compare/test_instance_compare.py
index a60b8fd6..b2d009b2 100644
--- a/tests/unit/opera/compare/test_instance_compare.py
+++ b/tests/unit/opera/compare/test_instance_compare.py
@@ -2,6 +2,7 @@
 
 from opera.compare.instance_comparer import InstanceComparer
 from opera.compare.template_comparer import TemplateComparer, TemplateContext
+from opera.constants import NodeState
 
 
 class TestInstanceCompare:
@@ -56,12 +57,12 @@ def test_dependency_comparison(self, service_template1,
     def test_prepare_update(self, service_template1, service_template2, template_diff):
         comparer = InstanceComparer()
         for node1 in service_template1[1].nodes.values():
-            node1.set_state("started")
-        assert service_template2[1].nodes["my-workstation_0"].state == "initial"
+            node1.set_state(NodeState.STARTED)
+        assert service_template2[1].nodes["my-workstation_0"].state == NodeState.INITIAL
 
         equal, diff = comparer.compare_topology_template(service_template1[1], service_template2[1], template_diff)
-        assert equal is False
+        assert not equal
 
         comparer.prepare_update(service_template1[1], service_template2[1], diff)
-        assert service_template1[1].nodes["my-workstation_0"].state == "initial"
-        assert service_template2[1].nodes["my-workstation_0"].state == "started"
+        assert service_template1[1].nodes["my-workstation_0"].state == NodeState.INITIAL
+        assert service_template2[1].nodes["my-workstation_0"].state == NodeState.STARTED
diff --git a/tests/unit/opera/instance/test_attribute_mapping.py b/tests/unit/opera/instance/test_attribute_mapping.py
index 64e7dda1..26756c4c 100644
--- a/tests/unit/opera/instance/test_attribute_mapping.py
+++ b/tests/unit/opera/instance/test_attribute_mapping.py
@@ -61,7 +61,7 @@ def test_map_attribute_node(self, service_template):
     @pytest.mark.parametrize("host", ["SOURCE", "TARGET", "HOST" "my_collector", "other"])
     def test_map_attribute_node_bad_host(self, service_template, host):
         node = service_template.find_node("my_node")
-        with pytest.raises(DataError, match="Attribute host should be set to 'SELF'"):
+        with pytest.raises(DataError, match="The attribute's 'host' should be set to 'SELF'"):
             node.map_attribute([host, "colour"], "black")
 
     def test_map_attribute_node_bad_attribute(self, service_template):
@@ -97,20 +97,24 @@ def test_map_attribute_relationship_self(self, service_template):
 
         assert "steampunk" == relationship_instance.get_attribute(["SELF", "colour"])
 
-    @pytest.mark.parametrize("host",
-                             ["HOST", "my_node", "my_collector", "other"])
+    @pytest.mark.parametrize("host", [
+        ("HOST", "HOST as the attribute's 'host' is not yet supported"),
+        ("my_node", "The attribute's 'host' should be set to one of"),
+        ("my_collector", "The attribute's 'host' should be set to one of"),
+        ("other", "The attribute's 'host' should be set to one of")
+    ])
     def test_map_attribute_relationship_bad_host(self, service_template, host):
         node_source = service_template.find_node("my_collector")
         node_source_instance = next(iter(node_source.instances.values()))
         relationship_instance = next(iter(node_source_instance.out_edges["my_target"].values()))
 
-        with pytest.raises(DataError, match="Attribute host should be set to 'SELF'"):
+        with pytest.raises(DataError, match="The attribute's 'host' should be set to one of"):
             relationship_instance.map_attribute([host, "colour"], "ochre")
 
     @pytest.mark.parametrize("host, pattern", [
         ("SELF", "Instance has no 'volume' attribute"),
-        ("SOURCE", "Cannot find attribute 'volume'."),
-        ("TARGET", "Cannot find attribute 'volume'.")
+        ("SOURCE", "Cannot find attribute 'volume' among"),
+        ("TARGET", "Cannot find attribute 'volume' among")
     ])
     def test_map_attr_rel_bad_attr(self, service_template, host, pattern):
         node_source = service_template.find_node("my_collector")
diff --git a/tests/unit/opera/instance/test_node_policies.py b/tests/unit/opera/instance/test_node_policies.py
new file mode 100644
index 00000000..ec0654bb
--- /dev/null
+++ b/tests/unit/opera/instance/test_node_policies.py
@@ -0,0 +1,343 @@
+import pathlib
+
+import pytest
+
+from opera.parser import tosca
+from opera.storage import Storage
+
+
+class TestNodePolicies:
+    @pytest.fixture
+    def service_template(self, tmp_path, yaml_text):
+        name = pathlib.PurePath("service.yaml")
+        (tmp_path / name).write_text(yaml_text(
+            # language=yaml
+            """
+            tosca_definitions_version: tosca_simple_yaml_1_3
+
+            node_types:
+              steampunk.nodes.VM:
+                derived_from: tosca.nodes.Compute
+                interfaces:
+                  Standard:
+                    type: tosca.interfaces.node.lifecycle.Standard
+                    operations:
+                      create: playbooks/create.yaml
+                      delete: playbooks/delete.yaml
+                  scaling_up:
+                    type: steampunk.interfaces.scaling.ScaleUp
+                  scaling_down:
+                    type: steampunk.interfaces.scaling.ScaleDown
+                  autoscaling:
+                    operations:
+                      retrieve_info:
+                        description: Operation for autoscaling.
+                        implementation: playbooks/retrieve_info.yaml
+                      autoscale:
+                        description: Operation for autoscaling.
+                        implementation: playbooks/auto_scale.yaml
+                        inputs:
+                          min_size:
+                            type: float
+                            default: { get_property: [ autoscale, min_size ] }
+                          max_size:
+                            type: float
+                            default: { get_property: [ autoscale, max_size ] }
+
+              steampunk.nodes.ConfigureMonitoring:
+                derived_from: tosca.nodes.SoftwareComponent
+                interfaces:
+                  Standard:
+                    type: tosca.interfaces.node.lifecycle.Standard
+                    operations:
+                      configure:
+                        implementation: playbooks/configure.yaml
+                        inputs:
+                          cpu_lower_bound:
+                            type: float
+                            default: { get_property: [ steampunk.policies.scaling.ScaleDown, cpu_lower_bound ] }
+                          cpu_upper_bound:
+                            type: float
+                            default: { get_property: [ steampunk.policies.scaling.ScaleUp, cpu_upper_bound ] }
+
+            interface_types:
+              steampunk.interfaces.scaling.ScaleDown:
+                derived_from: tosca.interfaces.Root
+                operations:
+                  scale_down:
+                    inputs:
+                      adjustment:
+                        type: float
+                        default: { get_property: [ steampunk.policies.scaling.ScaleDown, adjustment ] }
+                    description: Operation for scaling down.
+                    implementation: playbooks/scale_down.yaml
+
+              steampunk.interfaces.scaling.ScaleUp:
+                derived_from: tosca.interfaces.Root
+                operations:
+                  scale_up:
+                    inputs:
+                      adjustment:
+                        type: float
+                        default: { get_property: [ steampunk.policies.scaling.ScaleUp, adjustment ] }
+                    description: Operation for scaling up.
+                    implementation: playbooks/scale_up.yaml
+
+            policy_types:
+              steampunk.policies.scaling.ScaleDown:
+                derived_from: tosca.policies.Scaling
+                properties:
+                  cpu_lower_bound:
+                    description: The lower bound for the CPU
+                    type: float
+                    required: false
+                    constraints:
+                      - less_or_equal: 20.0
+                  adjustment:
+                    description: The amount by which to scale
+                    type: integer
+                    required: false
+                    constraints:
+                      - less_or_equal: -1
+                targets: [ steampunk.nodes.VM, steampunk.nodes.ConfigureMonitoring ]
+                triggers:
+                  steampunk.triggers.scaling.ScaleDown:
+                    description: A trigger for scaling down
+                    event: scale_down_trigger
+                    target_filter:
+                      node: steampunk.nodes.VM
+                    condition:
+                      constraint:
+                        - not:
+                            - and:
+                                - available_instances: [ { greater_than: 42 } ]
+                                - available_space: [ { greater_than: 1000 } ]
+                    action:
+                      - call_operation:
+                          operation: scaling_down.scale_down
+                          inputs:
+                            adjustment: { get_property: [ SELF, adjustment ] }
+
+              steampunk.policies.scaling.ScaleUp:
+                derived_from: tosca.policies.Scaling
+                properties:
+                  cpu_upper_bound:
+                    description: The upper bound for the CPU
+                    type: float
+                    required: false
+                    constraints:
+                      - greater_or_equal: 80.0
+                  adjustment:
+                    description: The amount by which to scale
+                    type: integer
+                    required: false
+                    constraints:
+                      - greater_or_equal: 1
+                targets: [ steampunk.nodes.VM, steampunk.nodes.ConfigureMonitoring ]
+                triggers:
+                  steampunk.triggers.scaling.ScaleUp:
+                    description: A trigger for scaling up
+                    event: scale_up_trigger
+                    target_filter:
+                      node: steampunk.nodes.VM
+                    condition:
+                      constraint:
+                        - not:
+                            - and:
+                                - available_instances: [ { greater_than: 42 } ]
+                                - available_space: [ { greater_than: 1000 } ]
+                    action:
+                      - call_operation:
+                          operation: scaling_up.scale_up
+                          inputs:
+                            adjustment: { get_property: [ SELF, adjustment ] }
+
+              steampunk.policies.scaling.AutoScale:
+                derived_from: tosca.policies.Scaling
+                properties:
+                  min_size:
+                    type: integer
+                    description: The minimum number of instances
+                    required: true
+                    status: supported
+                    constraints:
+                      - greater_or_equal: 1
+                  max_size:
+                    type: integer
+                    description: The maximum number of instances
+                    required: true
+                    status: supported
+                    constraints:
+                      - greater_or_equal: 10
+
+            topology_template:
+              node_templates:
+                VM:
+                  type: steampunk.nodes.VM
+
+                ConfigureMonitoring:
+                  type: steampunk.nodes.ConfigureMonitoring
+
+              policies:
+                - scale_down:
+                    type: steampunk.policies.scaling.ScaleDown
+                    properties:
+                      cpu_lower_bound: 10
+                      adjustment: 1
+
+                - scale_up:
+                    type: steampunk.policies.scaling.ScaleUp
+                    properties:
+                      cpu_upper_bound: 90
+                      adjustment: 5
+
+                - autoscale:
+                    type: steampunk.policies.scaling.AutoScale
+                    properties:
+                      min_size: 3
+                      max_size: 7
+                    targets: [ VM ]
+                    triggers:
+                      steampunk.triggers.scaling.AutoScale:
+                        description: A trigger for autoscaling
+                        event: auto_scale_trigger
+                        schedule:
+                          start_time: 2020-04-08T21:59:43.10-06:00
+                          end_time: 2022-04-08T21:59:43.10-06:00
+                        target_filter:
+                          node: VM
+                          requirement: workstation
+                          capability: host_capability
+                        condition:
+                          constraint:
+                            - not:
+                                - and:
+                                    - available_instances: [ { greater_than: 42 } ]
+                                    - available_space: [ { greater_than: 1000 } ]
+                          period: 60 sec
+                          evaluations: 2
+                          method: average
+                        action:
+                          - call_operation: autoscaling.retrieve_info
+                          - call_operation: autoscaling.autoscale
+            """
+        ))
+        # language=yaml
+        playbook = \
+            """
+            - hosts: all
+              tasks:
+                - name: Debug
+                  debug:
+                    msg: "Just testing."
+            """
+        pathlib.Path.mkdir(tmp_path / "playbooks")
+        (tmp_path / "playbooks" / "create.yaml").write_text(yaml_text(playbook))
+        (tmp_path / "playbooks" / "delete.yaml").write_text(yaml_text(playbook))
+        (tmp_path / "playbooks" / "configure.yaml").write_text(yaml_text(playbook))
+        (tmp_path / "playbooks" / "scale_up.yaml").write_text(yaml_text(playbook))
+        (tmp_path / "playbooks" / "scale_down.yaml").write_text(yaml_text(playbook))
+        (tmp_path / "playbooks" / "retrieve_info.yaml").write_text(yaml_text(playbook))
+        (tmp_path / "playbooks" / "auto_scale.yaml").write_text(yaml_text(playbook))
+
+        storage = Storage(tmp_path / pathlib.Path(".opera"))
+        storage.write("service.yaml", "root_file")
+        ast = tosca.load(tmp_path, name)
+        template = ast.get_template({})
+        template.instantiate(storage)
+        yield template
+
+    def test_count_policies_for_service_template(self, service_template):
+        assert len(service_template.policies) == 3
+
+    def test_count_policies_for_node(self, service_template):
+        node_vm = service_template.find_node("VM")
+        assert len(node_vm.policies) == 3
+
+        node_monitoring = service_template.find_node("ConfigureMonitoring")
+        assert len(node_monitoring.policies) == 2
+
+    def test_find_policies_for_node(self, service_template):
+        node_vm = service_template.find_node("VM")
+        node_vm_policies = [policy.name for policy in node_vm.policies]
+
+        assert "scale_down" in node_vm_policies
+        assert "scale_up" in node_vm_policies
+        assert "autoscale" in node_vm_policies
+
+        node_monitoring = service_template.find_node("ConfigureMonitoring")
+        node_monitoring_policies = [policy.name for policy in node_monitoring.policies]
+
+        assert "scale_down" in node_monitoring_policies
+        assert "scale_up" in node_monitoring_policies
+
+    def test_find_policy_targets(self, service_template):
+        node_vm = service_template.find_node("VM")
+        node_vm_policy_targets = [policy.targets for policy in node_vm.policies]
+
+        assert "VM" in node_vm_policy_targets[0]
+        assert "VM" in node_vm_policy_targets[1]
+        assert "VM" in node_vm_policy_targets[2]
+
+    def test_find_policy_triggers(self, service_template):
+        node_vm = service_template.find_node("VM")
+        node_vm_policy_triggers = [policy.triggers for policy in node_vm.policies]
+
+        assert "steampunk.triggers.scaling.ScaleDown" in node_vm_policy_triggers[0]
+        assert "steampunk.triggers.scaling.ScaleUp" in node_vm_policy_triggers[1]
+        assert "steampunk.triggers.scaling.AutoScale" in node_vm_policy_triggers[2]
+
+    def test_find_policy_trigger_events(self, service_template):
+        node_vm = service_template.find_node("VM")
+        node_vm_policy_triggers = [policy.triggers for policy in node_vm.policies]
+
+        assert "scale_down_trigger" == node_vm_policy_triggers[0]["steampunk.triggers.scaling.ScaleDown"].event.data
+        assert "scale_up_trigger" == node_vm_policy_triggers[1]["steampunk.triggers.scaling.ScaleUp"].event.data
+        assert "auto_scale_trigger" == node_vm_policy_triggers[2]["steampunk.triggers.scaling.AutoScale"].event.data
+
+    def test_find_policy_trigger_target_filter(self, service_template):
+        node_vm = service_template.find_node("VM")
+        node_vm_policy_triggers = [policy.triggers for policy in node_vm.policies]
+
+        assert "VM" == node_vm_policy_triggers[0]["steampunk.triggers.scaling.ScaleDown"].target_filter[0]
+        assert "VM" == node_vm_policy_triggers[1]["steampunk.triggers.scaling.ScaleUp"].target_filter[0]
+        assert "VM" == node_vm_policy_triggers[2]["steampunk.triggers.scaling.AutoScale"].target_filter[0]
+
+    def test_find_policy_trigger_action(self, service_template):
+        node_vm = service_template.find_node("VM")
+        node_vm_policy_triggers = [policy.triggers for policy in node_vm.policies]
+
+        interface1, operation1, _ = node_vm_policy_triggers[0]["steampunk.triggers.scaling.ScaleDown"].action[0]
+        assert ("scaling_down", "scale_down") == (interface1, operation1)
+
+        interface2, operation2, _ = node_vm_policy_triggers[1]["steampunk.triggers.scaling.ScaleUp"].action[0]
+        assert ("scaling_up", "scale_up") == (interface2, operation2)
+
+        interface3, operation3, _ = node_vm_policy_triggers[2]["steampunk.triggers.scaling.AutoScale"].action[0]
+        assert ("autoscaling", "retrieve_info") == (interface3, operation3)
+
+        interface3, operation3, _ = node_vm_policy_triggers[2]["steampunk.triggers.scaling.AutoScale"].action[1]
+        assert ("autoscaling", "autoscale") == (interface3, operation3)
+
+    def test_find_policy_properties(self, service_template):
+        node_vm = service_template.find_node("VM")
+        node_vm_policy_properties = [policy.properties for policy in node_vm.policies]
+
+        assert "cpu_lower_bound" in node_vm_policy_properties[0]
+        assert "adjustment" in node_vm_policy_properties[0]
+        assert "cpu_upper_bound" in node_vm_policy_properties[1]
+        assert "adjustment" in node_vm_policy_properties[1]
+        assert "min_size" in node_vm_policy_properties[2]
+        assert "max_size" in node_vm_policy_properties[2]
+
+    def test_get_policy_properties(self, service_template):
+        node_vm = service_template.find_node("VM")
+
+        assert node_vm.get_property(("scale_down", "cpu_lower_bound")) == 10.0
+        assert node_vm.get_property(("scale_down", "adjustment")) == 1
+
+        assert node_vm.get_property(("scale_up", "cpu_upper_bound")) == 90.0
+        assert node_vm.get_property(("scale_up", "adjustment")) == 5
+
+        assert node_vm.get_property(("autoscale", "min_size")) == 3
+        assert node_vm.get_property(("autoscale", "max_size")) == 7