lib/xos-genx/xos-genx-tests/test_xos_validation.py

# Copyright 2017-present Open Networking Foundation
#
# 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.

from __future__ import absolute_import
import unittest
from xosgenx.generator import XOSProcessor, XOSProcessorArgs
from helpers import FakeObject, XProtoTestHelpers


def policy_output_validator(x, y):
    """
    For eliminating warnings arising due to the missing
    policy_output_validator, which is generated and loaded dynamically.
    """
    raise Exception("Validator not generated. Test failed.")
    return False

class XProtoXOSModelValidationTest(unittest.TestCase):
    """
    Use the Python code target to generate Python validation policies, set up an
    appropriate environment and execute the Python.
    """

    def setUp(self):
        self.target = XProtoTestHelpers.write_tmp_target(
            "{{ xproto_fol_to_python_validator('output', proto.policies.test_policy, None, 'Necessary Failure') }}"
        )

    def test_instance_container(self):
        xproto = """
    policy test_policy < (obj.isolation = "container" | obj.isolation = "container_vm" ) -> (obj.image.kind = "container") >
"""
        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = self.target

        output = XOSProcessor.process(args)

        obj = FakeObject()
        obj.isolation = "container"
        obj.kind = "not a container"

        exec(output)  # This loads the generated function, which should look like this:

        """
        def policy_output_validator(obj, ctx):
            i4 = (obj.isolation == 'container')
            i5 = (self.isolation == 'container_vm')
            i2 = (i4 or i5)
            i3 = (obj.image.kind == 'container')
            i1 = (i2 or i3)
            return i1
        """

        with self.assertRaises(Exception):
            policy_output_validator(obj, {})

    def test_slice_name_validation(self):
        xproto = """
    policy test_policy < not obj.id -> {{ obj.name.startswith(obj.site.login_base) }} >
"""
        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = self.target

        output = XOSProcessor.process(args)

        obj = FakeObject()
        obj.isolation = "container"
        obj.kind = "not a container"

        exec(output)  # This loads the generated function, which should look like this:

        """
        def policy_output_validator(obj, ctx):
            i3 = obj.id
            i4 = obj.name.startswith(obj.site.login_base)
            i2 = ((not i3) or i4)
            i1 = (not i2)
            if (not i1):
                raise ValidationError('Necessary Failure')
        """

        with self.assertRaises(Exception):
            policy_output_validator(obj, {})

    def test_equal(self):
        xproto = """
    policy test_policy < not (ctx.user = obj.user) >
"""

        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = self.target

        output = XOSProcessor.process(args)

        exec(output)  # This loads the generated function, which should look like this:

        """
        def policy_output_validator(obj, ctx):
            i2 = (ctx.user == obj.user)
            i1 = (not i2)
            if (not i1):
                raise Exception('Necessary Failure')
        """

        obj = FakeObject()
        obj.user = 1
        ctx = FakeObject()
        ctx.user = 1

        with self.assertRaises(Exception):
            policy_output_validator(obj, ctx)

    def test_bin(self):
        xproto = """
    policy test_policy < not (ctx.is_admin = True | obj.empty = True) | False>
"""

        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = self.target

        output = XOSProcessor.process(args)
        exec(output)  # This loads the generated function, which should look like this:

        """
        def policy_output_validator(obj, ctx):
            i2 = (ctx.is_admin == True)
            i3 = (obj.empty == True)
            i1 = (i2 or i3)
            if (not i1):
                raise Exception('Necessary Failure')
        """

        obj = FakeObject()
        obj.empty = True

        ctx = FakeObject()
        ctx.is_admin = True

        with self.assertRaises(Exception):
            verdict = policy_output_validator(obj, ctx)

    def test_exists(self):
        xproto = """
    policy test_policy < exists Privilege: Privilege.object_id = obj.id >
"""
        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = self.target

        output = XOSProcessor.process(args)
        exec(output)  # This loads the generated function, which should look like this:

        """
        def policy_output_validator(obj, ctx):
            i1 = Privilege.objects.filter(Q(object_id=obj.id))[0]
            if (not i1):
                raise Exception('Necessary Failure')
        """

        self.assertTrue(policy_output_validator is not None)

    def test_python(self):
        xproto = """
    policy test_policy < {{ "jack" in ["the", "box"] }} = True >
"""
        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = self.target
        output = XOSProcessor.process(args)
        exec(output)  # This loads the generated function, which should look like this:

        """
        def policy_output_validator(obj, ctx):
            i2 = ('jack' in ['the', 'box'])
            i1 = (i2 == True)
            if (not i1):
                raise Exception('Necessary Failure')
        """

        with self.assertRaises(Exception):
            self.assertTrue(policy_output_validator({}, {}) is True)

    def test_forall(self):
        # This one we only parse
        xproto = """
    policy test_policy < forall Credential: Credential.obj_id = obj_id >
"""

        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = self.target

        output = XOSProcessor.process(args)

        """
        def policy_output_enforcer(obj, ctx):
            i2 = Credential.objects.filter((~ Q(obj_id=obj_id)))[0]
            i1 = (not i2)
            return i1
        """

        self.assertIn("policy_output_validator", output)


if __name__ == "__main__":
    unittest.main()