aboutsummaryrefslogtreecommitdiffstats
path: root/src/resource_inventory/forms.py
blob: fb8c102dded74ff64c9cd9f73fd36dc5c49809ca (plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
##############################################################################
# Copyright (c) 2020 Sawyer Bergeron, Sean Smith, and others.
#
# All rights reserved. This program and the accompanying materials
# are made available under the terms of the Apache License, Version 2.0
# which accompanies this distribution, and is available at
# http://www.apache.org/licenses/LICENSE-2.0
##############################################################################

from django.core.exceptions import ValidationError
from django import forms

from resource_inventory.models import Network, InterfaceConfiguration


class InterfaceConfigurationForm(forms.ModelForm):
    class Meta:
        model = InterfaceConfiguration
        fields = ['profile', 'resource_config', 'connections']

    def clean(self):
        connections = self.cleaned_data.get('connections')
        resource_config = self.cleaned_data.get('resource_config')

        valid_nets = set(Network.objects.filter(bundle=resource_config.template))
        curr_nets = set([conn.network for conn in connections])

        if not curr_nets.issubset(valid_nets):
            raise ValidationError("Cannot have network connection to network outside pod")

        return self.cleaned_data
a id='n335' href='#n335'>335 336 337 338 339 340 341 342 343 344 345 346 347 348 349 350 351 352 353 354 355 356 357 358 359 360 361 362 363 364 365 366 367 368 369 370 371 372 373
#    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
import six
from toscaparser.common import exception
from toscaparser import functions
from toscaparser.tests.base import TestCase
from toscaparser.tosca_template import ToscaTemplate
from toscaparser.utils.gettextutils import _


class IntrinsicFunctionsTest(TestCase):

    tosca_tpl = os.path.join(
        os.path.dirname(os.path.abspath(__file__)),
        "data/tosca_single_instance_wordpress.yaml")
    params = {'db_name': 'my_wordpress', 'db_user': 'my_db_user',
              'db_root_pwd': '12345678'}
    tosca = ToscaTemplate(tosca_tpl, parsed_params=params)

    def _get_node(self, node_name, tosca=None):
        if tosca is None:
            tosca = self.tosca
        return [
            node for node in tosca.nodetemplates
            if node.name == node_name][0]

    def _get_operation(self, interfaces, operation):
        return [
            interface for interface in interfaces
            if interface.name == operation][0]

    def _get_property(self, node_template, property_name):
        return [prop.value for prop in node_template.get_properties_objects()
                if prop.name == property_name][0]

    def _get_inputs_dict(self):
        inputs = {}
        for input in self.tosca.inputs:
            inputs[input.name] = input.default
        return inputs

    def _get_input(self, name):
        self._get_inputs_dict()[name]

    def test_get_property(self):
        wordpress = self._get_node('wordpress')
        operation = self._get_operation(wordpress.interfaces, 'configure')
        wp_db_password = operation.inputs['wp_db_password']
        self.assertIsInstance(wp_db_password, functions.GetProperty)
        result = wp_db_password.result()
        self.assertEqual('wp_pass', result)

    def test_get_property_with_input_param(self):
        wordpress = self._get_node('wordpress')
        operation = self._get_operation(wordpress.interfaces, 'configure')
        wp_db_user = operation.inputs['wp_db_user']
        self.assertIsInstance(wp_db_user, functions.GetProperty)
        result = wp_db_user.result()
        self.assertEqual('my_db_user', result)

    def test_unknown_capability_property(self):
        self.assertRaises(exception.ValidationError, self._load_template,
                          'functions/test_unknown_capability_property.yaml')
        exception.ExceptionCollector.assertExceptionMessage(
            KeyError,
            _('\'Property "unknown" was not found in capability '
              '"database_endpoint" of node template "database" referenced '
              'from node template "database".\''))

    def test_get_input_in_properties(self):
        mysql_dbms = self._get_node('mysql_dbms')
        expected_inputs = ['db_root_pwd', 'db_port']
        props = mysql_dbms.get_properties()
        for key in props.keys():
            prop = props[key]
            self.assertIsInstance(prop.value, functions.GetInput)
            expected_inputs.remove(prop.value.input_name)
        self.assertListEqual(expected_inputs, [])

    def test_get_input_validation(self):
        self.assertRaises(
            exception.ValidationError, self._load_template,
            'functions/test_unknown_input_in_property.yaml')
        exception.ExceptionCollector.assertExceptionMessage(
            exception.UnknownInputError,
            _('Unknown input "objectstore_name".'))

        self.assertRaises(
            exception.ValidationError, self._load_template,
            'functions/test_unknown_input_in_interface.yaml')
        exception.ExceptionCollector.assertExceptionMessage(
            exception.UnknownInputError,
            _('Unknown input "image_id".'))

        self.assertRaises(
            exception.ValidationError, self._load_template,
            'functions/test_invalid_function_signature.yaml')
        exception.ExceptionCollector.assertExceptionMessage(
            ValueError,
            _('Expected one argument for function "get_input" but received '
              '"[\'cpus\', \'cpus\']".'))

    def test_get_input_default_value_result(self):
        mysql_dbms = self._get_node('mysql_dbms')
        dbms_port = self._get_property(mysql_dbms, 'port')
        self.assertEqual(3306, dbms_port.result())
        dbms_root_password = self._get_property(mysql_dbms,
                                                'root_password')
        self.assertEqual(dbms_root_password.result(), '12345678')

    def test_get_property_with_host(self):
        tosca_tpl = os.path.join(
            os.path.dirname(os.path.abspath(__file__)),
            "data/functions/test_get_property_with_host.yaml")
        mysql_database = self._get_node('mysql_database',
                                        ToscaTemplate(tosca_tpl,
                                                      parsed_params={
                                                          'db_root_pwd': '123'
                                                      }))
        operation = self._get_operation(mysql_database.interfaces, 'configure')
        db_port = operation.inputs['db_port']
        self.assertIsInstance(db_port, functions.GetProperty)
        result = db_port.result()
        self.assertEqual(3306, result)
        test = operation.inputs['test']
        self.assertIsInstance(test, functions.GetProperty)
        result = test.result()
        self.assertEqual(1, result)

    def test_get_property_with_nested_params(self):
        tosca_tpl = os.path.join(
            os.path.dirname(os.path.abspath(__file__)),
            "data/functions/tosca_nested_property_names_indexes.yaml")
        webserver = self._get_node('wordpress',
                                   ToscaTemplate(tosca_tpl,
                                                 parsed_params={
                                                     'db_root_pwd': '1234'}))
        operation = self._get_operation(webserver.interfaces, 'configure')
        wp_endpoint_prot = operation.inputs['wp_endpoint_protocol']
        self.assertIsInstance(wp_endpoint_prot, functions.GetProperty)
        self.assertEqual('tcp', wp_endpoint_prot.result())
        wp_list_prop = operation.inputs['wp_list_prop']
        self.assertIsInstance(wp_list_prop, functions.GetProperty)
        self.assertEqual(3, wp_list_prop.result())

    def test_get_property_with_capabilties_inheritance(self):
        tosca_tpl = os.path.join(
            os.path.dirname(os.path.abspath(__file__)),
            "data/functions/test_capabilties_inheritance.yaml")
        some_node = self._get_node('some_node',
                                   ToscaTemplate(tosca_tpl,
                                                 parsed_params={
                                                     'db_root_pwd': '1234'}))
        operation = self._get_operation(some_node.interfaces, 'configure')
        some_input = operation.inputs['some_input']
        self.assertIsInstance(some_input, functions.GetProperty)
        self.assertEqual('someval', some_input.result())

    def test_get_property_source_target_keywords(self):
        tosca_tpl = os.path.join(
            os.path.dirname(os.path.abspath(__file__)),
            "data/functions/test_get_property_source_target_keywords.yaml")
        tosca = ToscaTemplate(tosca_tpl,
                              parsed_params={'db_root_pwd': '1234'})

        for node in tosca.nodetemplates:
            for relationship, trgt in node.relationships.items():
                rel_template = trgt.get_relationship_template()[0]
                break

        operation = self._get_operation(rel_template.interfaces,
                                        'pre_configure_source')
        target_test = operation.inputs['target_test']
        self.assertIsInstance(target_test, functions.GetProperty)
        self.assertEqual(1, target_test.result())
        source_port = operation.inputs['source_port']
        self.assertIsInstance(source_port, functions.GetProperty)
        self.assertEqual(3306, source_port.result())


class GetAttributeTest(TestCase):

    def _load_template(self, filename):
        return ToscaTemplate(os.path.join(
            os.path.dirname(os.path.abspath(__file__)),
            'data',
            filename),
            parsed_params={'db_root_pwd': '1234'})

    def _get_operation(self, interfaces, operation):
        return [
            interface for interface in interfaces
            if interface.name == operation][0]

    def test_get_attribute_in_outputs(self):
        tpl = self._load_template('tosca_single_instance_wordpress.yaml')
        website_url_output = [
            x for x in tpl.outputs if x.name == 'website_url'][0]
        self.assertIsInstance(website_url_output.value, functions.GetAttribute)
        self.assertEqual('server', website_url_output.value.node_template_name)
        self.assertEqual('private_address',
                         website_url_output.value.attribute_name)

    def test_get_attribute_invalid_args(self):
        expected_msg = _('Illegal arguments for function "get_attribute".'
                         ' Expected arguments: "node-template-name", '
                         '"req-or-cap"(optional), "property name"')
        err = self.assertRaises(ValueError,
                                functions.get_function, None, None,
                                {'get_attribute': []})
        self.assertIn(expected_msg, six.text_type(err))
        err = self.assertRaises(ValueError,
                                functions.get_function, None, None,
                                {'get_attribute': ['x']})
        self.assertIn(expected_msg, six.text_type(err))

    def test_get_attribute_unknown_node_template_name(self):
        self.assertRaises(
            exception.ValidationError, self._load_template,
            'functions/test_get_attribute_unknown_node_template_name.yaml')
        exception.ExceptionCollector.assertExceptionMessage(
            KeyError,
            _('\'Node template "unknown_node_template" was not found.\''))

    def test_get_attribute_unknown_attribute(self):
        self.assertRaises(
            exception.ValidationError, self._load_template,
            'functions/test_get_attribute_unknown_attribute_name.yaml')
        exception.ExceptionCollector.assertExceptionMessage(
            KeyError,
            _('\'Attribute "unknown_attribute" was not found in node template '
              '"server".\''))

    def test_get_attribute_host_keyword(self):
        tpl = self._load_template(
            'functions/test_get_attribute_host_keyword.yaml')

        def assert_get_attribute_host_functionality(node_template_name):
            node = [x for x in tpl.nodetemplates
                    if x.name == node_template_name][0]
            configure_op = [
                x for x in node.interfaces if x.name == 'configure'][0]
            ip_addr_input = configure_op.inputs['ip_address']
            self.assertIsInstance(ip_addr_input, functions.GetAttribute)
            self.assertEqual('server',
                             ip_addr_input.get_referenced_node_template().name)

        assert_get_attribute_host_functionality('dbms')
        assert_get_attribute_host_functionality('database')

    def test_get_attribute_host_not_found(self):
        self.assertRaises(
            exception.ValidationError, self._load_template,
            'functions/test_get_attribute_host_not_found.yaml')
        exception.ExceptionCollector.assertExceptionMessage(
            ValueError,
            _('"get_attribute: [ HOST, ... ]" was used in node template '
              '"server" but "tosca.relationships.HostedOn" was not found in '
              'the relationship chain.'))

    def test_get_attribute_illegal_host_in_outputs(self):
        self.assertRaises(
            exception.ValidationError, self._load_template,
            'functions/test_get_attribute_illegal_host_in_outputs.yaml')
        exception.ExceptionCollector.assertExceptionMessage(
            ValueError,
            _('"get_attribute: [ HOST, ... ]" is not allowed in "outputs" '
              'section of the TOSCA template.'))

    def test_get_attribute_with_index(self):
        self._load_template(
            'functions/test_get_attribute_with_index.yaml')

    def test_get_attribute_with_index_error(self):
        self.assertRaises(
            exception.ValidationError, self._load_template,
            'functions/test_get_attribute_with_index_error.yaml')
        exception.ExceptionCollector.assertExceptionMessage(
            ValueError,
            _('Illegal arguments for function "get_attribute". '
              'Unexpected attribute/index value "0"'))

    def test_get_attribute_source_target_keywords(self):
        tosca_tpl = os.path.join(
            os.path.dirname(os.path.abspath(__file__)),
            "data/functions/test_get_attribute_source_target_keywords.yaml")
        tosca = ToscaTemplate(tosca_tpl,
                              parsed_params={'db_root_pwd': '12345678'})

        for node in tosca.nodetemplates:
            for relationship, trgt in node.relationships.items():
                rel_template = trgt.get_relationship_template()[0]
                break

        operation = self._get_operation(rel_template.interfaces,
                                        'pre_configure_source')
        target_test = operation.inputs['target_test']
        self.assertIsInstance(target_test, functions.GetAttribute)
        source_port = operation.inputs['source_port']
        self.assertIsInstance(source_port, functions.GetAttribute)

    def test_get_attribute_with_nested_params(self):
        self._load_template(
            'functions/test_get_attribute_with_nested_params.yaml')

    def test_implicit_attribute(self):
        self.assertIsNotNone(self._load_template(
            'functions/test_get_implicit_attribute.yaml'))


class ConcatTest(TestCase):

    def _load_template(self, filename):
        return ToscaTemplate(os.path.join(
            os.path.dirname(os.path.abspath(__file__)),
            filename))

    def test_validate_concat(self):
        tosca = self._load_template("data/functions/test_concat.yaml")
        server_url_output = [
            output for output in tosca.outputs if output.name == 'url'][0]
        func = functions.get_function(self, tosca.outputs,
                                      server_url_output.value)
        self.assertIsInstance(func, functions.Concat)

        self.assertRaises(exception.ValidationError, self._load_template,
                          'data/functions/test_concat_invalid.yaml')
        exception.ExceptionCollector.assertExceptionMessage(
            ValueError,
            _('Invalid arguments for function "concat". Expected at least '
              'one arguments.'))


class TokenTest(TestCase):

    def _load_template(self, filename):
        return ToscaTemplate(os.path.join(
            os.path.dirname(os.path.abspath(__file__)),
            filename))

    def test_validate_token(self):
        tosca = self._load_template("data/functions/test_token.yaml")
        server_url_output = [
            output for output in tosca.outputs if output.name == 'url'][0]
        func = functions.get_function(self, tosca.outputs,
                                      server_url_output.value)
        self.assertIsInstance(func, functions.Token)

        self.assertRaises(exception.ValidationError, self._load_template,
                          'data/functions/test_token_invalid.yaml')
        exception.ExceptionCollector.assertExceptionMessage(
            ValueError,
            _('Invalid arguments for function "token". Expected at least '
              'three arguments.'))
        exception.ExceptionCollector.assertExceptionMessage(
            ValueError,
            _('Invalid arguments for function "token". Expected '
              'integer value as third argument.'))
        exception.ExceptionCollector.assertExceptionMessage(
            ValueError,
            _('Invalid arguments for function "token". Expected '
              'single char value as second argument.'))