lib/xos-genx/xos-genx-tests/test_generator.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
import os
from helpers import OUTPUT_DIR
from xosgenx.generator import XOSProcessor, XOSProcessorArgs

TEST_EXPECTED_OUTPUT = """
    name: XOSModel
    fields:
            name:
                type: string
                description: "Help Name"
            files:
                type: string
                description: "Help Files"
"""

BASE_XPROTO = os.path.abspath(
    os.path.dirname(os.path.realpath(__file__)) + "/xproto/base.xproto"
)
TEST_XPROTO = os.path.abspath(
    os.path.dirname(os.path.realpath(__file__)) + "/xproto/test.xproto"
)
FIELDTEST_XPROTO = os.path.abspath(
    os.path.dirname(os.path.realpath(__file__)) + "/xproto/fieldtest.xproto"
)
REVERSEFIELDTEST_XPROTO = os.path.abspath(
    os.path.dirname(os.path.realpath(__file__)) + "/xproto/reversefieldtest.xproto"
)
FILTERTEST_XPROTO = os.path.abspath(
    os.path.dirname(os.path.realpath(__file__)) + "/xproto/filtertest.xproto"
)
CUSTOM_TEST1_XPROTO = os.path.abspath(
    os.path.dirname(os.path.realpath(__file__)) + "/xproto/custom_test1.xproto"
)
CUSTOM_TEST2_XPROTO = os.path.abspath(
    os.path.dirname(os.path.realpath(__file__)) + "/xproto/custom_test2.xproto"
)
SKIP_DJANGO_XPROTO = os.path.abspath(
    os.path.dirname(os.path.realpath(__file__)) + "/xproto/skip_django.xproto"
)
VROUTER_XPROTO = os.path.abspath(
    os.path.dirname(os.path.realpath(__file__)) + "/xproto/vrouterport.xproto"
)
TEST_TARGET = os.path.abspath(
    os.path.dirname(os.path.realpath(__file__)) + "/xtarget/test.xtarget"
)
FIELDTEST_TARGET = os.path.abspath(
    os.path.dirname(os.path.realpath(__file__)) + "/xtarget/fieldtest.xtarget"
)
FILTERTEST_TARGET = os.path.abspath(
    os.path.dirname(os.path.realpath(__file__)) + "/xtarget/filtertest.xtarget"
)
SPLIT_TARGET = os.path.abspath(
    os.path.dirname(os.path.realpath(__file__)) + "/xtarget/split.xtarget"
)

TEST_ATTICS = os.path.abspath(os.path.dirname(os.path.realpath(__file__)) + "/attics/")


class XOSProcessorTest(unittest.TestCase):
    """
    Testing the XOS Generative Toolchain
    """

    def setUp(self):
        os.chdir(
            os.path.join(
                os.path.abspath(os.path.dirname(os.path.realpath(__file__))), ".."
            )
        )
        filesToRemove = [f for f in os.listdir(OUTPUT_DIR)]
        for f in filesToRemove:
            if not f.startswith("."):
                os.remove(OUTPUT_DIR + "/" + f)

    def test_generator_custom_target_from_file(self):
        """
        [XOS-GenX] Generate output from base.xproto
        """
        args = XOSProcessorArgs(files=[TEST_XPROTO], target=TEST_TARGET)
        output = XOSProcessor.process(args)
        self.assertEqual(output, TEST_EXPECTED_OUTPUT)

    def test_generator_custom_target_from_inputs(self):
        """
        [XOS-GenX] Generate output from base.xproto
        """
        args = XOSProcessorArgs(inputs=open(TEST_XPROTO).read(), target=TEST_TARGET)
        output = XOSProcessor.process(args)
        self.assertEqual(output, TEST_EXPECTED_OUTPUT)

    def test_django_with_attic(self):
        """
        [XOS-GenX] Generate django output from test.xproto
        """
        args = XOSProcessorArgs(
            files=[TEST_XPROTO, VROUTER_XPROTO],
            target="django.xtarget",
            attic=TEST_ATTICS,
            output=OUTPUT_DIR,
            dest_extension="py",
            write_to_file="model",
        )
        output = XOSProcessor.process(args)

        # xosmodel has custom header attic
        self.assertIn("from core.models.xosbase import *", output["XOSModel"])
        self.assertIn("class XOSModel_decl(XOSBase):", output["XOSModel"])

        # vrouter port use the default header
        self.assertIn("from core.models.xosbase import *", output["VRouterPort"])
        self.assertIn("class VRouterPort_decl(XOSBase):", output["VRouterPort"])

        # verify files
        xosmodel = OUTPUT_DIR + "/xosmodel.py"
        self.assertTrue(os.path.isfile(xosmodel))
        xmf = open(xosmodel).read()
        self.assertIn("from core.models.xosbase import *", xmf)
        self.assertIn("class XOSModel_decl(XOSBase):", xmf)

        vrouterport = OUTPUT_DIR + "/vrouterport.py"
        self.assertTrue(os.path.isfile(vrouterport))
        vrpf = open(vrouterport).read()
        self.assertIn("from core.models.xosbase import *", vrpf)
        self.assertIn("class VRouterPort_decl(XOSBase):", vrpf)

    def test_django_with_base(self):
        args = XOSProcessorArgs(
            files=[TEST_XPROTO, BASE_XPROTO],
            target="django.xtarget",
            attic=TEST_ATTICS,
            output=OUTPUT_DIR,
            dest_extension="py",
            write_to_file="model",
        )
        output = XOSProcessor.process(args)

        # verify files
        xosmodel = OUTPUT_DIR + "/xosmodel.py"
        self.assertTrue(os.path.isfile(xosmodel))
        xmf = open(xosmodel).read()
        self.assertIn("from core.models.xosbase import *", xmf)
        self.assertIn("class XOSModel_decl(XOSBase):", xmf)

        xosbase = OUTPUT_DIR + "/xosbase.py"
        self.assertTrue(os.path.isfile(xosbase))
        xbf = open(xosbase).read()
        self.assertIn("from core.models.xosbase import *", xbf)
        self.assertIn("class XOSBase_decl(models.Model, PlModelMixIn):", xbf)

    def test_write_multiple_files(self):
        """
        [XOS-GenX] read multiple models as input, print one file per model
        """
        args = XOSProcessorArgs(
            files=[TEST_XPROTO, VROUTER_XPROTO],
            target=TEST_TARGET,
            output=OUTPUT_DIR,
            dest_extension="txt",
            write_to_file="model",
        )
        XOSProcessor.process(args)

        generated_files = [f for f in os.listdir(OUTPUT_DIR) if not f.startswith(".")]
        self.assertEqual(len(generated_files), 2)

        xosmodel = open(os.path.join(OUTPUT_DIR, "xosmodel.txt"), "r").read()
        vrouterport = open(os.path.join(OUTPUT_DIR, "vrouterport.txt"), "r").read()

        self.assertIn("name: XOSModel", xosmodel)
        self.assertIn("name: VRouterPort", vrouterport)

    def test_write_multiple_files_from_xtarget(self):
        """
        [XOS-GenX] read multiple models as input, print separate files based on +++
        """
        args = XOSProcessorArgs(
            files=[TEST_XPROTO, VROUTER_XPROTO],
            target=SPLIT_TARGET,
            output=OUTPUT_DIR,
            write_to_file="target",
        )
        XOSProcessor.process(args)

        generated_files = [f for f in os.listdir(OUTPUT_DIR) if not f.startswith(".")]
        self.assertEqual(len(generated_files), 2)

        xosmodel = open(os.path.join(OUTPUT_DIR, "xosmodel.txt"), "r").read()
        vrouterport = open(os.path.join(OUTPUT_DIR, "vrouterport.txt"), "r").read()

        self.assertIn("name: XOSModel", xosmodel)
        self.assertIn("name: VRouterPort", vrouterport)

    def test_skip_django(self):
        args = XOSProcessorArgs(
            files=[SKIP_DJANGO_XPROTO],
            target="django.xtarget",
            output=OUTPUT_DIR,
            dest_extension="py",
            write_to_file="model",
        )
        output = XOSProcessor.process(args)

        # should not print a file if options.skip_django = True
        file = OUTPUT_DIR + "/user.py"
        self.assertFalse(os.path.isfile(file))

    def test_service_order(self):
        args = XOSProcessorArgs(
            files=[BASE_XPROTO, TEST_XPROTO, VROUTER_XPROTO],
            target="service.xtarget",
            output=OUTPUT_DIR,
            write_to_file="target",
        )
        output = XOSProcessor.process(args)

        model = OUTPUT_DIR + "/models.py"
        self.assertTrue(os.path.isfile(model))
        line_num = 0

        for line in open(model).readlines():
            line_num += 1
            if line.find("class XOSBase(models.Model, PlModelMixIn):") >= 0:
                base_line = line_num
            if line.find("XOSModel(XOSBase):") >= 0:
                xosmodel_line = line_num
            if line.find("class VRouterPort(XOSBase):") >= 0:
                vrouter_line = line_num
        self.assertLess(base_line, xosmodel_line)
        self.assertLess(xosmodel_line, vrouter_line)

    def test_field_numbers(self):
        args = XOSProcessorArgs(files=[FIELDTEST_XPROTO], target=FIELDTEST_TARGET)
        output = XOSProcessor.process(args)

        def _assert_field(modelname, fieldname, id):
            self.assertIn("%s,%s,%s" % (modelname, fieldname, id), output)

        _assert_field("Site", "id", 1)
        _assert_field("Site", "base_field", 2)
        _assert_field("Site", "base_field2", 3)
        _assert_field("Site", "otherstuff_field", 102)
        _assert_field("Site", "slice_field", 201)
        _assert_field("Site", "slices_ids", 1002)

        _assert_field("Slice", "id", 1)
        _assert_field("Slice", "base_field", 2)
        _assert_field("Slice", "base_field2", 3)
        _assert_field("Slice", "slice_field", 101)
        _assert_field("Slice", "site", 102)

    def test_field_numbers(self):
        args = XOSProcessorArgs(
            files=[REVERSEFIELDTEST_XPROTO], target=FIELDTEST_TARGET
        )
        output = XOSProcessor.process(args)

        def _assert_field(modelname, fieldname, id):
            self.assertIn("%s,%s,%s" % (modelname, fieldname, id), output)

        # rel_int1s_ids is the reverse link from RelatedToIntermediate1. It gets the related id with no offset, so it
        # will be assigned 1001. rel_leaf1as_ids inherits from Intermediate1, so its reverse links will all be offset
        # by 100
        _assert_field("Leaf1a", "rel_int1s_ids", 1001)
        _assert_field("Leaf1a", "rel_leaf1as_ids", 1101)

        # rel_int2s_ids is the reverse link from RelatedToIntermediate1. It gets the related id with no offset, so it
        # will be assigned 1001. rel_leaf1bs_ids inherits from Intermediate1, so its reverse links will all be offset
        # by 100
        _assert_field("Leaf1b", "rel_int1s_ids", 1001)
        _assert_field("Leaf1b", "rel_leaf1bs_ids", 1101)

        # There are no reverse numbers specified for Intermediate2 or Leaf2, so xproto will fall back to automatic
        # numbering starting at 1900.
        _assert_field("Leaf2", "rel_int2s_ids", 1900)
        _assert_field("Leaf2", "rel_leaf2s_ids", 1901)

    def test_unfiltered(self):
        """ With no include_* args, should get all models """
        args = XOSProcessorArgs(files=[FILTERTEST_XPROTO], target=FILTERTEST_TARGET)
        output = XOSProcessor.process(args)

        self.assertEqual(output, "Model1,Model2,Model3,")

    def test_filter_models(self):
        """ Should only get models specified by include_models """
        args = XOSProcessorArgs(
            files=[FILTERTEST_XPROTO],
            target=FILTERTEST_TARGET,
            include_models=["Model1", "Model3"],
        )
        output = XOSProcessor.process(args)

        self.assertEqual(output, "Model1,Model3,")

    def test_filter_apps(self):
        """ Should only get models whose apps are specified by include_apps """
        args = XOSProcessorArgs(
            files=[FILTERTEST_XPROTO], target=FILTERTEST_TARGET, include_apps=["core"]
        )
        output = XOSProcessor.process(args)

        self.assertEqual(output, "Model1,Model2,")

    def test_django_custom_test1(self):
        args = XOSProcessorArgs(
            files=[CUSTOM_TEST1_XPROTO, BASE_XPROTO],
            target="django.xtarget",
            attic=TEST_ATTICS,
            output=OUTPUT_DIR,
            dest_extension="py",
            write_to_file="model",
        )
        output = XOSProcessor.process(args)

        # verify files
        xosmodel = OUTPUT_DIR + "/xosmodel_decl.py"
        self.assertTrue(os.path.isfile(xosmodel))
        xmf = open(xosmodel).read()
        self.assertIn("class XOSModel_decl(XOSBase):", xmf)

        xosmodel = OUTPUT_DIR + "/xosmodel2_decl.py"
        self.assertTrue(os.path.isfile(xosmodel))
        xmf = open(xosmodel).read()
        self.assertIn("class XOSModel2_decl(XOSBase):", xmf)

    def test_django_custom_test2(self):
        args = XOSProcessorArgs(
            files=[CUSTOM_TEST2_XPROTO, BASE_XPROTO],
            target="django.xtarget",
            attic=TEST_ATTICS,
            output=OUTPUT_DIR,
            dest_extension="py",
            write_to_file="model",
        )
        output = XOSProcessor.process(args)

        # verify files
        xosmodel = OUTPUT_DIR + "/xosmodel_decl.py"
        self.assertTrue(os.path.isfile(xosmodel))
        xmf = open(xosmodel).read()
        self.assertIn("class XOSModel_decl(XOSBase):", xmf)
        self.assertNotIn("class XOSModel(XOSModel_decl):", xmf)

        xosmodel = OUTPUT_DIR + "/xosmodel2.py"
        self.assertTrue(os.path.isfile(xosmodel))
        xmf = open(xosmodel).read()
        self.assertIn("class XOSModel2_decl(XOSBase):", xmf)
        self.assertIn("class XOSModel2(XOSModel2_decl):", xmf)

    def test_service_custom_test1(self):
        args = XOSProcessorArgs(
            files=[CUSTOM_TEST1_XPROTO, BASE_XPROTO],
            target="service.xtarget",
            attic=TEST_ATTICS,
            output=OUTPUT_DIR,
            dest_extension="py",
            write_to_file="target",
        )
        output = XOSProcessor.process(args)

        # verify files
        xosmodel = OUTPUT_DIR + "/models_decl.py"
        self.assertTrue(os.path.isfile(xosmodel))
        xmf = open(xosmodel).read()
        self.assertIn("class XOSModel_decl(XOSBase_decl):", xmf)

        xosmodel = OUTPUT_DIR + "/models_decl.py"
        self.assertTrue(os.path.isfile(xosmodel))
        xmf = open(xosmodel).read()
        self.assertIn("class XOSModel2_decl(XOSBase_decl):", xmf)


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