lib/xos-genx/xos-genx-tests/test_policy.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.


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

"""
The tests below convert the policy logic expression
into Python, set up an appropriate environment and execute the Python.
"""


class XProtoPolicyTest(unittest.TestCase):
    def test_annotation(self):
        xproto = """
    policy true_policy < True >

    message always::true_policy {
        required int still = 9;
    }
"""

        target = XProtoTestHelpers.write_tmp_target("{{ proto.messages.0 }}")

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

        output = XOSProcessor.process(args)
        self.assertIn("true_policy", output)

    def test_constant(self):
        xproto = """
    policy true_policy < True >
"""

        target = XProtoTestHelpers.write_tmp_target("{{ proto.policies.true_policy }}")

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

        output = XOSProcessor.process(args).replace("t", "T")
        self.assertTrue(eval(output))

    def test_function_term(self):
        xproto = """
    policy slice_user < slice.user.compute_is_admin() >
"""

        target = XProtoTestHelpers.write_tmp_target("{{ proto.policies.slice_user }}")
        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = target

        output = XOSProcessor.process(args)

        slice = FakeObject()
        slice.user = FakeObject()
        slice.user.compute_is_admin = lambda: True

        expr = eval(output)
        self.assertTrue(expr)

    def test_term(self):
        xproto = """
    policy slice_user < slice.user.is_admin >
"""

        target = XProtoTestHelpers.write_tmp_target("{{ proto.policies.slice_user }}")
        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = target

        output = XOSProcessor.process(args)

        slice = FakeObject()
        slice.user = FakeObject()
        slice.user.is_admin = True

        expr = eval(output)
        self.assertTrue(expr)

    def test_num_constant(self):
        xproto = """
    policy slice_user < slice.user.age = 57 >
"""

        target = XProtoTestHelpers.write_tmp_target("{{ proto.policies.slice_user }}")
        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = target

        output = XOSProcessor.process(args)

        slice = FakeObject()
        slice.user = FakeObject()
        slice.user.is_admin = True

        expr = eval(output)
        self.assertTrue(expr)

    def test_string_constant(self):
        xproto = """
    policy slice_user < slice.user.email = "admin@opencord.org" >
"""

        target = XProtoTestHelpers.write_tmp_target("{{ proto.policies.slice_user }}")
        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = target

        output = XOSProcessor.process(args)

        slice = FakeObject()
        slice.user = FakeObject()
        slice.user.is_admin = True

        expr = eval(output)
        self.assertTrue(expr)

    def test_equal(self):
        xproto = """
    policy slice_user < slice.user = obj.user >
"""

        target = XProtoTestHelpers.write_tmp_target("{{ proto.policies.slice_user }}")
        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = target

        output = XOSProcessor.process(args)

        slice = FakeObject()
        slice.user = "twin"
        obj = FakeObject()
        obj.user = "twin"

        (op, operands), = eval(output).items()
        expr = op.join(operands).replace("=", "==")

        self.assertTrue(eval(expr))

    def test_bin(self):
        xproto = """
    policy slice_admin < slice.is_admin | obj.empty >
"""
        target = XProtoTestHelpers.write_tmp_target("{{ proto.policies.slice_admin }}")
        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = target

        output = XOSProcessor.process(args)

        slice = FakeObject()
        slice.is_admin = False
        obj = FakeObject()
        obj.empty = []

        (op, operands), = eval(output).items()
        expr = op.join(operands).replace("|", " or ")

        self.assertFalse(eval(expr))

    def test_implies(self):
        xproto = """
    policy implies < obj.name -> obj.creator >
"""
        target = XProtoTestHelpers.write_tmp_target("{{ proto.policies.implies }}")
        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = target

        output = XOSProcessor.process(args)

        slice = FakeObject()
        slice.is_admin = False
        obj = FakeObject()
        obj.name = "Thing 1"
        obj.creator = None

        (op, operands), = eval(output).items()
        expr = "not " + op.join(operands).replace("->", " or ")

        self.assertFalse(eval(expr))

    def test_exists(self):
        xproto = """
    policy privilege < exists Privilege: Privilege.object_id = obj.id >
"""

        target = XProtoTestHelpers.write_tmp_target("{{ proto.policies.privilege }} ")
        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = target

        output = XOSProcessor.process(args)

        Privilege = FakeObject()
        Privilege.object_id = 1
        obj = FakeObject()
        obj.id = 1

        (op, operands), = eval(output).items()
        (op2, operands2), = operands[1].items()
        expr = op2.join(operands2).replace("=", "==")

        self.assertTrue(eval(expr))

    def test_policy_function(self):
        xproto = """
    policy slice_policy < exists Privilege: Privilege.object_id = obj.id >
    policy network_slice_policy < *slice_policy(slice) >
"""

        target = XProtoTestHelpers.write_tmp_target(
            "{{ proto.policies.network_slice_policy }} "
        )
        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = target

        output = XOSProcessor.process(args)

        (op, operands), = eval(output).items()

        self.assertIn("slice_policy", operands)
        self.assertIn("slice", operands)

    def test_policy_missing_function(self):
        xproto = """
    policy slice_policy < exists Privilege: Privilege.object_id = obj.id >
    policy network_slice_policy < *slice_policyX(slice) >
"""

        target = XProtoTestHelpers.write_tmp_target(
            "{{ proto.policies.network_slice_policy }} "
        )
        args = XOSProcessorArgs()
        args.inputs = xproto
        args.target = target

        with self.assertRaises(Exception):
            output = XOSProcessor.process(args)

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

        target = XProtoTestHelpers.write_tmp_target("{{ proto.policies.instance }}")

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

        output = XOSProcessor.process(args)
        (op, operands), = eval(output).items()

        self.assertEqual(op, "forall")


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