dcae-policy/tests/mock_cloudify_ctx.py

   1 # ================================================================================
   2 # Copyright (c) 2017-2020 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
  18 """mock cloudify context with relationships and type_hierarchy"""
  19
  20 from cloudify.mocks import (MockCloudifyContext, MockNodeContext,
  21                             MockNodeInstanceContext)
  22
  23 TARGET_NODE_ID = "target_node_id"
  24 TARGET_NODE_NAME = "target_node_name"
  25
  26 class MockContextNode(MockNodeContext):
  27     """ctx.node with type and type_hierarchy"""
  28
  29     def __init__(self, id=None, properties=None, node_type=None, type_hierarchy=None):
  30         super(MockContextNode, self).__init__(id, properties or {})
  31         self._type = node_type
  32         self._type_hierarchy = type_hierarchy or [self._type]
  33         MockCloudifyContextFull.nodes[id] = self
  34
  35     @property
  36     def type(self):
  37         """node type"""
  38         return self._type
  39
  40     @property
  41     def type_hierarchy(self):
  42         """node type hierarchy is a list of types"""
  43         return self._type_hierarchy
  44
  45 class MockContextNodeInstance(MockNodeInstanceContext):
  46     """ctx.instance with relationships"""
  47
  48     def __init__(self, id=None, runtime_properties=None, relationships=None):
  49         super(MockContextNodeInstance, self).__init__(id, runtime_properties or {})
  50         self._relationships = []
  51         self.add_relationships(relationships)
  52         MockCloudifyContextFull.instances[id] = self
  53
  54     def add_relationships(self, relationships):
  55         """add more relationships to the node instance"""
  56         if not relationships:
  57             return
  58         if not self._relationships:
  59             self._relationships = []
  60         self._relationships.extend([
  61             MockContextRelationship(relationship)
  62             for relationship in (relationships or []) if TARGET_NODE_ID in relationship
  63         ])
  64
  65     @property
  66     def relationships(self):
  67         """list of relationships to other node instances"""
  68         return self._relationships
  69
  70 class MockContextRelationshipTarget(object):
  71     """target of relationship"""
  72     def __init__(self, relationship):
  73         target_node_name = relationship[TARGET_NODE_NAME]
  74         target_node_id = relationship[TARGET_NODE_ID]
  75
  76         self.node = MockCloudifyContextFull.nodes.get(target_node_name)
  77         self.instance = MockCloudifyContextFull.instances.get(target_node_id)
  78
  79         if not self.node:
  80             self.node = MockContextNode(target_node_name)
  81         if not self.instance:
  82             self.instance = MockContextNodeInstance(target_node_id)
  83
  84 class MockContextRelationship(object):
  85     """item of ctx.instance.relationships"""
  86
  87     def __init__(self, relationship):
  88         self.target = MockContextRelationshipTarget(relationship)
  89         self.type = relationship.get("type", "cloudify.relationships.depends_on")
  90         self.type_hierarchy = relationship.get("type_hierarchy") or [self.type]
  91
  92 class MockCloudifyContextFull(MockCloudifyContext):
  93     """
  94     ctx1 = MockCloudifyContextFull(node_id='node_1',
  95                                    node_name='my_1', properties={'foo': 'bar'})
  96     ctx2 = MockCloudifyContextFull(node_id='node_2',
  97                                    node_name='my_2',
  98                                    relationships=[{'target_node_id': 'node_1',
  99                                                    'target_node_name': 'my_1'}])
 100     """
 101     nodes = {}
 102     instances = {}
 103
 104     def __init__(self,
 105                  node_id=None,
 106                  node_name=None,
 107                  blueprint_id=None,
 108                  deployment_id=None,
 109                  execution_id=None,
 110                  properties=None, node_type=None, type_hierarchy=None,
 111                  runtime_properties=None,
 112                  capabilities=None,
 113                  related=None,
 114                  source=None,
 115                  target=None,
 116                  operation=None,
 117                  resources=None,
 118                  provider_context=None,
 119                  bootstrap_context=None,
 120                  relationships=None):
 121         super(MockCloudifyContextFull, self).__init__(
 122             node_id=node_id,
 123             node_name=node_name,
 124             blueprint_id=blueprint_id,
 125             deployment_id=deployment_id,
 126             execution_id=execution_id,
 127             properties=properties,
 128             capabilities=capabilities,
 129             related=related,
 130             source=source,
 131             target=target,
 132             operation=operation,
 133             resources=resources,
 134             provider_context=provider_context,
 135             bootstrap_context=bootstrap_context,
 136             runtime_properties=runtime_properties
 137         )
 138         self._node = MockContextNode(node_name, properties, node_type, type_hierarchy)
 139         self._instance = MockContextNodeInstance(node_id, runtime_properties, relationships)
 140
 141     @staticmethod
 142     def clear():
 143         """clean up the context links"""
 144         MockCloudifyContextFull.instances.clear()
 145         MockCloudifyContextFull.nodes.clear()