dcae-policy/tests/mock_cloudify_ctx.py

   1 # ================================================================================
   2 # Copyright (c) 2017-2018 AT&T Intellectual Property. All rights reserved.
   3 # ================================================================================
   4 # Licensed under the Apache License, Version 2.0 (the "License");
   5 # you may not use this file except in compliance with the License.
   6 # You may obtain a copy of the License at
   7 #
   8 #      http://www.apache.org/licenses/LICENSE-2.0
   9 #
  10 # Unless required by applicable law or agreed to in writing, software
  11 # distributed under the License is distributed on an "AS IS" BASIS,
  12 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15 # ============LICENSE_END=========================================================
  16 #
  17 # ECOMP is a trademark and service mark of AT&T Intellectual Property.
  18
  19 """mock cloudify context with relationships and type_hierarchy"""
  20
  21 from cloudify.mocks import (MockCloudifyContext, MockNodeContext,
  22                             MockNodeInstanceContext)
  23
  24 TARGET_NODE_ID = "target_node_id"
  25 TARGET_NODE_NAME = "target_node_name"
  26
  27 class MockContextNode(MockNodeContext):
  28     """ctx.node with type and type_hierarchy"""
  29
  30     def __init__(self, id=None, properties=None, node_type=None, type_hierarchy=None):
  31         super(MockContextNode, self).__init__(id, properties or {})
  32         self._type = node_type
  33         self._type_hierarchy = type_hierarchy or [self._type]
  34         MockCloudifyContextFull.nodes[id] = self
  35
  36     @property
  37     def type(self):
  38         """node type"""
  39         return self._type
  40
  41     @property
  42     def type_hierarchy(self):
  43         """node type hierarchy is a list of types"""
  44         return self._type_hierarchy
  45
  46 class MockContextNodeInstance(MockNodeInstanceContext):
  47     """ctx.instance with relationships"""
  48
  49     def __init__(self, id=None, runtime_properties=None, relationships=None):
  50         super(MockContextNodeInstance, self).__init__(id, runtime_properties or {})
  51         self._relationships = []
  52         self.add_relationships(relationships)
  53         MockCloudifyContextFull.instances[id] = self
  54
  55     def add_relationships(self, relationships):
  56         """add more relationships to the node instance"""
  57         if not relationships:
  58             return
  59         if not self._relationships:
  60             self._relationships = []
  61         self._relationships.extend([
  62             MockContextRelationship(relationship)
  63             for relationship in (relationships or []) if TARGET_NODE_ID in relationship
  64         ])
  65
  66     @property
  67     def relationships(self):
  68         """list of relationships to other node instances"""
  69         return self._relationships
  70
  71 class MockContextRelationshipTarget(object):
  72     """target of relationship"""
  73     def __init__(self, relationship):
  74         target_node_name = relationship[TARGET_NODE_NAME]
  75         target_node_id = relationship[TARGET_NODE_ID]
  76
  77         self.node = MockCloudifyContextFull.nodes.get(target_node_name)
  78         self.instance = MockCloudifyContextFull.instances.get(target_node_id)
  79
  80         if not self.node:
  81             self.node = MockContextNode(target_node_name)
  82         if not self.instance:
  83             self.instance = MockContextNodeInstance(target_node_id)
  84
  85 class MockContextRelationship(object):
  86     """item of ctx.instance.relationships"""
  87
  88     def __init__(self, relationship):
  89         self.target = MockContextRelationshipTarget(relationship)
  90         self.type = relationship.get("type", "cloudify.relationships.depends_on")
  91         self.type_hierarchy = relationship.get("type_hierarchy") or [self.type]
  92
  93 class MockCloudifyContextFull(MockCloudifyContext):
  94     """
  95     ctx1 = MockCloudifyContextFull(node_id='node_1',
  96                                    node_name='my_1', properties={'foo': 'bar'})
  97     ctx2 = MockCloudifyContextFull(node_id='node_2',
  98                                    node_name='my_2',
  99                                    relationships=[{'target_node_id': 'node_1',
 100                                                    'target_node_name': 'my_1'}])
 101     """
 102     nodes = {}
 103     instances = {}
 104
 105     def __init__(self,
 106                  node_id=None,
 107                  node_name=None,
 108                  blueprint_id=None,
 109                  deployment_id=None,
 110                  execution_id=None,
 111                  properties=None, node_type=None, type_hierarchy=None,
 112                  runtime_properties=None,
 113                  capabilities=None,
 114                  related=None,
 115                  source=None,
 116                  target=None,
 117                  operation=None,
 118                  resources=None,
 119                  provider_context=None,
 120                  bootstrap_context=None,
 121                  relationships=None):
 122         super(MockCloudifyContextFull, self).__init__(
 123             node_id=node_id,
 124             node_name=node_name,
 125             blueprint_id=blueprint_id,
 126             deployment_id=deployment_id,
 127             execution_id=execution_id,
 128             properties=properties,
 129             capabilities=capabilities,
 130             related=related,
 131             source=source,
 132             target=target,
 133             operation=operation,
 134             resources=resources,
 135             provider_context=provider_context,
 136             bootstrap_context=bootstrap_context,
 137             runtime_properties=runtime_properties
 138         )
 139         self._node = MockContextNode(node_name, properties, node_type, type_hierarchy)
 140         self._instance = MockContextNodeInstance(node_id, runtime_properties, relationships)
 141
 142     @staticmethod
 143     def clear():
 144         """clean up the context links"""
 145         MockCloudifyContextFull.instances.clear()
 146         MockCloudifyContextFull.nodes.clear()