| # 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 json |
| import os |
| import shutil |
| import tempfile |
| import unittest |
| from mock import patch |
| |
| from xosconfig import Config |
| |
| |
| class DynamicLoadItem: |
| def __init__(self, **kwargs): |
| for (k, v) in kwargs.items(): |
| setattr(self, k, v) |
| |
| |
| class DynamicLoadRequest: |
| def __init__(self, **kwargs): |
| self.xprotos = [] |
| self.decls = [] |
| self.attics = [] |
| for (k, v) in kwargs.items(): |
| setattr(self, k, v) |
| |
| |
| class DynamicUnloadRequest: |
| def __init__(self, **kwargs): |
| for (k, v) in kwargs.items(): |
| setattr(self, k, v) |
| |
| |
| class TestDynamicBuild(unittest.TestCase): |
| def setUp(self): |
| global dynamicbuild |
| |
| config = os.path.abspath( |
| os.path.dirname(os.path.realpath(__file__)) + "/test_config.yaml" |
| ) |
| Config.clear() # in case left unclean by a previous test case |
| Config.init(config) |
| |
| import dynamicbuild |
| |
| self.base_dir = tempfile.mkdtemp() |
| self.example_xproto = """option app_label = "exampleservice"; |
| option name = "exampleservice"; |
| |
| message ExampleService (Service){ |
| option verbose_name = "Example Service"; |
| required string service_message = 1 [help_text = "Service Message to Display", |
| max_length = 254, null = False, db_index = False, blank = False]; |
| } |
| |
| message Color (XOSBase){ |
| option verbose_name = "Color"; |
| required string name = 1 [help_text = "Name for this color", db_index = False, |
| max_length = 256, null = False, blank = False]; |
| required string html_code = 2 [help_text = "Code for this color", db_index = False, |
| max_length = 256, null = False, blank = False]; |
| } |
| |
| message ExampleServiceInstance (TenantWithContainer){ |
| option verbose_name = "Example Service Instance"; |
| required string tenant_message = 1 [help_text = "Tenant Message to Display", |
| max_length = 254, null = False, db_index = False, blank = False]; |
| optional manytoone foreground_color->Color:serviceinstance_foreground_colors = 3 [db_index = True, |
| null = True, blank = True]; |
| optional manytoone background_color->Color:serviceinstance_background_colors = 3 [db_index = True, |
| null = True, blank = True]; |
| } |
| |
| message EmbeddedImage (XOSBase){ |
| option verbose_name = "Embedded Image"; |
| required string name = 1 [help_text = "Name for this image", db_index = False, |
| max_length = 256, null = False, blank = False]; |
| required string url = 2 [help_text = "URL for this image", db_index = False, |
| max_length = 256, null = False, blank = False]; |
| optional manytoone serviceinstance->ExampleServiceInstance:embedded_images = 3 [db_index = True, |
| null = True, blank = True]; |
| } |
| """ |
| |
| self.example_xproto_item = DynamicLoadItem( |
| filename="exampleservice.xproto", contents=self.example_xproto |
| ) |
| |
| self.example_request = DynamicLoadRequest( |
| name="exampleservice", |
| version="1", |
| xprotos=[self.example_xproto_item], |
| convenience_methods=[], |
| ) |
| |
| self.example_unload_request = DynamicUnloadRequest( |
| name="exampleservice", version="1" |
| ) |
| |
| self.builder = dynamicbuild.DynamicBuilder(base_dir=self.base_dir) |
| |
| def tearDown(self): |
| if os.path.abspath(self.base_dir).startswith( |
| "/tmp" |
| ): # be paranoid about recursive deletes |
| shutil.rmtree(self.base_dir) |
| |
| def test_pre_validate_file(self): |
| self.builder.pre_validate_file(self.example_xproto_item) |
| |
| def test_pre_validate_models(self): |
| self.builder.pre_validate_models(self.example_request) |
| |
| def test_generate_request_hash(self): |
| hash = self.builder.generate_request_hash(self.example_request, state="load") |
| self.assertEqual(hash, "44951ff370c565c796f05f7c3fce67f9a4d4d3f6") |
| |
| def test_handle_loadmodels_request(self): |
| with patch.object( |
| dynamicbuild.DynamicBuilder, "save_models", wraps=self.builder.save_models |
| ) as save_models, patch.object( |
| dynamicbuild.DynamicBuilder, |
| "run_xosgenx_service", |
| wraps=self.builder.run_xosgenx_service, |
| ) as run_xosgenx_service, patch.object( |
| dynamicbuild.DynamicBuilder, |
| "remove_service", |
| wraps=self.builder.remove_service, |
| ) as remove_service: |
| result = self.builder.handle_loadmodels_request(self.example_request) |
| |
| save_models.assert_called() |
| run_xosgenx_service.assert_called() |
| remove_service.assert_not_called() |
| |
| self.assertEqual(result, self.builder.SOMETHING_CHANGED) |
| |
| self.assertTrue(os.path.exists(self.builder.manifest_dir)) |
| self.assertTrue( |
| os.path.exists( |
| os.path.join(self.builder.manifest_dir, "exampleservice.json") |
| ) |
| ) |
| |
| service_dir = os.path.join(self.base_dir, "services", "exampleservice") |
| |
| self.assertTrue(os.path.exists(service_dir)) |
| self.assertTrue(os.path.exists(os.path.join(service_dir, "__init__.py"))) |
| self.assertTrue(os.path.exists(os.path.join(service_dir, "models.py"))) |
| self.assertTrue(os.path.exists(os.path.join(service_dir, "security.py"))) |
| |
| manifest = json.loads( |
| open( |
| os.path.join(self.builder.manifest_dir, "exampleservice.json"), "r" |
| ).read() |
| ) |
| self.assertEqual(manifest.get("state"), "load") |
| |
| def test_handle_unloadmodels_request(self): |
| with patch.object( |
| dynamicbuild.DynamicBuilder, "save_models", wraps=self.builder.save_models |
| ) as save_models, patch.object( |
| dynamicbuild.DynamicBuilder, |
| "run_xosgenx_service", |
| wraps=self.builder.run_xosgenx_service, |
| ) as run_xosgenx_service, patch.object( |
| dynamicbuild.DynamicBuilder, |
| "remove_service", |
| wraps=self.builder.remove_service, |
| ) as remove_service: |
| result = self.builder.handle_unloadmodels_request( |
| self.example_unload_request |
| ) |
| |
| save_models.assert_called() |
| run_xosgenx_service.assert_not_called() |
| remove_service.assert_called() |
| |
| self.assertEqual(result, self.builder.SOMETHING_CHANGED) |
| |
| self.assertTrue(os.path.exists(self.builder.manifest_dir)) |
| self.assertTrue( |
| os.path.exists( |
| os.path.join(self.builder.manifest_dir, "exampleservice.json") |
| ) |
| ) |
| |
| manifest = json.loads( |
| open( |
| os.path.join(self.builder.manifest_dir, "exampleservice.json"), "r" |
| ).read() |
| ) |
| self.assertEqual(manifest.get("state"), "unload") |
| |
| def test_handle_loadmodels_request_twice(self): |
| result = self.builder.handle_loadmodels_request(self.example_request) |
| self.assertEqual(result, self.builder.SOMETHING_CHANGED) |
| |
| result = self.builder.handle_loadmodels_request(self.example_request) |
| self.assertEqual(result, self.builder.NOTHING_TO_DO) |
| |
| def test_save_models(self): |
| manifest = self.builder.save_models(self.example_request, state="load") |
| |
| dynamic_dir = os.path.join(self.base_dir, "dynamic_services", "exampleservice") |
| service_dir = os.path.join(self.base_dir, "services", "exampleservice") |
| |
| self.assertEqual(manifest["name"], self.example_request.name) |
| self.assertEqual(manifest["version"], self.example_request.version) |
| self.assertEqual(manifest["hash"], "44951ff370c565c796f05f7c3fce67f9a4d4d3f6") |
| self.assertEqual(manifest["dir"], dynamic_dir) |
| self.assertEqual(manifest["dest_dir"], service_dir) |
| self.assertEqual(len(manifest["xprotos"]), 1) |
| |
| def test_save_models_precomputed_hash(self): |
| manifest = self.builder.save_models( |
| self.example_request, state="load", hash="1234" |
| ) |
| |
| dynamic_dir = os.path.join(self.base_dir, "dynamic_services", "exampleservice") |
| service_dir = os.path.join(self.base_dir, "services", "exampleservice") |
| |
| self.assertEqual(manifest["name"], self.example_request.name) |
| self.assertEqual(manifest["version"], self.example_request.version) |
| self.assertEqual(manifest["hash"], "1234") |
| self.assertEqual(manifest["dir"], dynamic_dir) |
| self.assertEqual(manifest["dest_dir"], service_dir) |
| self.assertEqual(len(manifest["xprotos"]), 1) |
| |
| def test_pre_validate_python_good(self): |
| good_python = """ |
| import foo |
| |
| x=1 |
| y="abc" |
| """ |
| python_item = DynamicLoadItem(filename="somefile.py", contents=good_python) |
| |
| self.builder.pre_validate_python(python_item) |
| |
| def test_pre_validate_python_bad(self): |
| bad_python = """ |
| import foo |
| |
| this is not valid code |
| y="abc" |
| """ |
| python_item = DynamicLoadItem(filename="somefile.py", contents=bad_python) |
| |
| with self.assertRaises(Exception) as e: |
| self.builder.pre_validate_python(python_item) |
| |
| self.assertEqual( |
| e.exception.message, "python file somefile.py failed compile test" |
| ) |
| |
| |
| def main(): |
| unittest.main() |
| |
| |
| if __name__ == "__main__": |
| main() |