1 files changed, 283 insertions, 0 deletions
diff --git a/jython-tosca-parser/src/main/resources/Lib/site-packages/tosca_parser-0.7.0-py2.7.egg/toscaparser/tests/test_topology_template.py b/jython-tosca-parser/src/main/resources/Lib/site-packages/tosca_parser-0.7.0-py2.7.egg/toscaparser/tests/test_topology_template.py
new file mode 100644
index 0000000..f9e2122
--- /dev/null
+++ b/jython-tosca-parser/src/main/resources/Lib/site-packages/tosca_parser-0.7.0-py2.7.egg/toscaparser/tests/test_topology_template.py
@@ -0,0 +1,283 @@
+#    Licensed under the Apache License, Version 2.0 (the "License"); you may
+#    not use this file except in compliance with the License. You may obtain
+#    a copy of the License at
+#
+#         http://www.apache.org/licenses/LICENSE-2.0
+#
+#    Unless required by applicable law or agreed to in writing, software
+#    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
+#    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
+#    License for the specific language governing permissions and limitations
+#    under the License.
+
+import os
+
+from toscaparser.common import exception
+from toscaparser.substitution_mappings import SubstitutionMappings
+from toscaparser.tests.base import TestCase
+from toscaparser.topology_template import TopologyTemplate
+from toscaparser.tosca_template import ToscaTemplate
+from toscaparser.utils.gettextutils import _
+import toscaparser.utils.yamlparser
+
+YAML_LOADER = toscaparser.utils.yamlparser.load_yaml
+
+
+class TopologyTemplateTest(TestCase):
+
+    def setUp(self):
+        TestCase.setUp(self)
+        '''TOSCA template.'''
+        self.tosca_tpl_path = os.path.join(
+            os.path.dirname(os.path.abspath(__file__)),
+            "data/topology_template/transactionsubsystem.yaml")
+        self.tpl = YAML_LOADER(self.tosca_tpl_path)
+        self.topo_tpl = self.tpl.get('topology_template')
+        self.imports = self.tpl.get('imports')
+        self.topo = TopologyTemplate(self.topo_tpl,
+                                     self._get_all_custom_def())
+
+    def _get_custom_def(self, type_definition):
+        custom_defs = {}
+        for definition in self.imports:
+            if os.path.isabs(definition):
+                def_file = definition
+            else:
+                tpl_dir = os.path.dirname(os.path.abspath(self.tosca_tpl_path))
+                def_file = os.path.join(tpl_dir, definition)
+            custom_type = YAML_LOADER(def_file)
+            custom_defs.update(custom_type.get(type_definition))
+        return custom_defs
+
+    def _get_all_custom_def(self):
+        custom_defs = {}
+        custom_defs.update(self._get_custom_def('node_types'))
+        custom_defs.update(self._get_custom_def('capability_types'))
+        return custom_defs
+
+    def _get_custom_types(self):
+        custom_types = {}
+        def_file = os.path.join(
+            os.path.dirname(os.path.abspath(__file__)),
+            "data/topology_template/definitions.yaml")
+        custom_type = YAML_LOADER(def_file)
+        node_types = custom_type['node_types']
+        for name in node_types:
+            defintion = node_types[name]
+            custom_types[name] = defintion
+        return custom_types
+
+    def test_description(self):
+        expected_desc = 'Template of a database including its hosting stack.'
+        self.assertEqual(expected_desc, self.topo.description)
+
+    def test_inputs(self):
+        self.assertEqual(
+            ['mq_server_ip', 'my_cpus', 'receiver_port'],
+            sorted([input.name for input in self.topo.inputs]))
+
+        input_name = "receiver_port"
+        expected_description = "Port to be used for receiving messages."
+        for input in self.topo.inputs:
+            if input.name == input_name:
+                self.assertEqual(expected_description, input.description)
+
+    def test_node_tpls(self):
+        '''Test nodetemplate names.'''
+        self.assertEqual(
+            ['app', 'server', 'websrv'],
+            sorted([tpl.name for tpl in self.topo.nodetemplates]))
+
+        tpl_name = "app"
+        expected_type = "example.SomeApp"
+        expected_properties = ['admin_user', 'pool_size']
+        expected_capabilities = ['feature', 'message_receiver']
+        expected_requirements = [{'host': {'node': 'websrv'}}]
+        expected_relationshp = ['tosca.relationships.HostedOn']
+        expected_host = ['websrv']
+        for tpl in self.topo.nodetemplates:
+            if tpl_name == tpl.name:
+                '''Test node type.'''
+                self.assertEqual(tpl.type, expected_type)
+
+                '''Test properties.'''
+                self.assertEqual(
+                    expected_properties,
+                    sorted(tpl.get_properties().keys()))
+
+                '''Test capabilities.'''
+                self.assertEqual(
+                    expected_capabilities,
+                    sorted(tpl.get_capabilities().keys()))
+
+                '''Test requirements.'''
+                self.assertEqual(
+                    expected_requirements, tpl.requirements)
+
+                '''Test relationship.'''
+                ''' TODO : skip tempororily. need to fix it
+                '''
+                self.assertEqual(
+                    expected_relationshp,
+                    [x.type for x in tpl.relationships.keys()])
+                self.assertEqual(
+                    expected_host,
+                    [y.name for y in tpl.relationships.values()])
+                '''Test interfaces.'''
+                # TODO(hurf) add interface test when new template is available
+
+            if tpl.name == 'server':
+                '''Test property value'''
+                props = tpl.get_properties()
+                if props and 'mem_size' in props.keys():
+                    self.assertEqual(props['mem_size'].value, '4096 MB')
+                '''Test capability'''
+                caps = tpl.get_capabilities()
+                self.assertIn('os', caps.keys())
+                os_props_objs = None
+                os_props = None
+                os_type_prop = None
+                if caps and 'os' in caps.keys():
+                    capability = caps['os']
+                    os_props_objs = capability.get_properties_objects()
+                    os_props = capability.get_properties()
+                    os_type_prop = capability.get_property_value('type')
+                    break
+                self.assertEqual(
+                    ['Linux'],
+                    [p.value for p in os_props_objs if p.name == 'type'])
+                self.assertEqual(
+                    'Linux',
+                    os_props['type'].value if 'type' in os_props else '')
+                self.assertEqual('Linux', os_props['type'].value)
+                self.assertEqual('Linux', os_type_prop)
+
+    def test_outputs(self):
+        self.assertEqual(
+            sorted(['receiver_ip', 'receiver_port']),
+            sorted([output.name for output in self.topo.outputs]))
+
+    def test_groups(self):
+        group = self.topo.groups[0]
+        self.assertEqual('webserver_group', group.name)
+        self.assertEqual(['websrv', 'server'], group.members)
+        for node in group.get_member_nodes():
+            if node.name == 'server':
+                '''Test property value'''
+                props = node.get_properties()
+                if props and 'mem_size' in props.keys():
+                    self.assertEqual(props['mem_size'].value, '4096 MB')
+
+    def test_system_template(self):
+        tpl_path = os.path.join(
+            os.path.dirname(os.path.abspath(__file__)),
+            "data/topology_template/system.yaml")
+        system_tosca_template = ToscaTemplate(tpl_path)
+        self.assertIsNotNone(system_tosca_template)
+        self.assertEqual(
+            len(system_tosca_template.
+                nested_tosca_templates_with_topology), 4)
+        self.assertTrue(system_tosca_template.has_nested_templates())
+
+    def test_invalid_keyname(self):
+        tpl_snippet = '''
+        substitution_mappings:
+          node_type: example.DatabaseSubsystem
+          capabilities:
+            database_endpoint: [ db_app, database_endpoint ]
+          requirements:
+            receiver1: [ tran_app, receiver1 ]
+          invalid_key: 123
+        '''
+        sub_mappings = (toscaparser.utils.yamlparser.
+                        simple_parse(tpl_snippet))['substitution_mappings']
+        expected_message = _(
+            'SubstitutionMappings contains unknown field '
+            '"invalid_key". Refer to the definition '
+            'to verify valid values.')
+        err = self.assertRaises(
+            exception.UnknownFieldError,
+            lambda: SubstitutionMappings(sub_mappings, None, None,
+                                         None, None, None))
+        self.assertEqual(expected_message, err.__str__())
+
+    def test_missing_required_keyname(self):
+        tpl_snippet = '''
+        substitution_mappings:
+          capabilities:
+            database_endpoint: [ db_app, database_endpoint ]
+          requirements:
+            receiver1: [ tran_app, receiver1 ]
+        '''
+        sub_mappings = (toscaparser.utils.yamlparser.
+                        simple_parse(tpl_snippet))['substitution_mappings']
+        expected_message = _('SubstitutionMappings used in topology_template '
+                             'is missing required field "node_type".')
+        err = self.assertRaises(
+            exception.MissingRequiredFieldError,
+            lambda: SubstitutionMappings(sub_mappings, None, None,
+                                         None, None, None))
+        self.assertEqual(expected_message, err.__str__())
+
+    def test_invalid_nodetype(self):
+        tpl_snippet = '''
+        substitution_mappings:
+          node_type: example.DatabaseSubsystem1
+          capabilities:
+            database_endpoint: [ db_app, database_endpoint ]
+          requirements:
+            receiver1: [ tran_app, receiver1 ]
+        '''
+        sub_mappings = (toscaparser.utils.yamlparser.
+                        simple_parse(tpl_snippet))['substitution_mappings']
+        custom_defs = self._get_custom_types()
+        expected_message = _('Node type "example.DatabaseSubsystem1" '
+                             'is not a valid type.')
+        err = self.assertRaises(
+            exception.InvalidNodeTypeError,
+            lambda: SubstitutionMappings(sub_mappings, None, None,
+                                         None, None, custom_defs))
+        self.assertEqual(expected_message, err.__str__())
+
+    def test_system_with_input_validation(self):
+        tpl_path0 = os.path.join(
+            os.path.dirname(os.path.abspath(__file__)),
+            "data/topology_template/validate/system_invalid_input.yaml")
+        tpl_path1 = os.path.join(
+            os.path.dirname(os.path.abspath(__file__)),
+            "data/topology_template/validate/"
+            "queuingsubsystem_invalid_input.yaml")
+        errormsg = _('SubstitutionMappings with node_type '
+                     'example.QueuingSubsystem is missing '
+                     'required input definition of input "server_port".')
+
+        # It's invalid in nested template.
+        self.assertRaises(exception.ValidationError,
+                          lambda: ToscaTemplate(tpl_path0))
+        exception.ExceptionCollector.assertExceptionMessage(
+            exception.MissingRequiredInputError, errormsg)
+
+        # Subtemplate deploy standaolone is also invalid.
+        self.assertRaises(exception.ValidationError,
+                          lambda: ToscaTemplate(tpl_path1))
+        exception.ExceptionCollector.assertExceptionMessage(
+            exception.MissingRequiredInputError, errormsg)
+
+    def test_substitution_mappings_valid_output(self):
+        tpl_path = os.path.join(
+            os.path.dirname(os.path.abspath(__file__)),
+            "data/topology_template/validate/"
+            "test_substitution_mappings_valid_output.yaml")
+        self.assertIsNotNone(ToscaTemplate(tpl_path))
+
+    def test_system_with_unknown_output_validation(self):
+        tpl_path = os.path.join(
+            os.path.dirname(os.path.abspath(__file__)),
+            "data/topology_template/validate/"
+            "test_substitution_mappings_invalid_output.yaml")
+        errormsg = _('\'Attribute "my_cpu_output" was not found in node '
+                     'template "substitute_app".\'')
+        self.assertRaises(exception.ValidationError,
+                          lambda: ToscaTemplate(tpl_path))
+        exception.ExceptionCollector.assertExceptionMessage(
+            KeyError, errormsg)