blob: fdbff3cf3c8bc021156a35c77879f64f1636d0f0 [file] [log] [blame]
Scott Bakerbba67b62019-01-28 17:38:21 -08001# Copyright 2017-present Open Networking Foundation
2#
3# Licensed under the Apache License, Version 2.0 (the "License");
4# you may not use this file except in compliance with the License.
5# You may obtain a copy of the License at
6#
7# http://www.apache.org/licenses/LICENSE-2.0
8#
9# Unless required by applicable law or agreed to in writing, software
10# distributed under the License is distributed on an "AS IS" BASIS,
11# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12# See the License for the specific language governing permissions and
13# limitations under the License.
14
15import json
16import unittest
Scott Bakerbba67b62019-01-28 17:38:21 -080017import mock
Scott Bakerbba67b62019-01-28 17:38:21 -080018
19import os
20import sys
21
22test_path = os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
23sync_lib_dir = os.path.join(test_path, "..", "xossynchronizer")
24xos_dir = os.path.join(test_path, "..", "..", "..", "xos")
25
26ANSIBLE_FILE = "/tmp/payload_test"
27
28
29def run_fake_ansible_template(*args, **kwargs):
30 opts = args[1]
31 open(ANSIBLE_FILE, "w").write(json.dumps(opts))
32
33
34def get_ansible_output():
35 ansible_str = open(ANSIBLE_FILE).read()
36 return json.loads(ansible_str)
37
38
39class TestRun(unittest.TestCase):
40 def setUp(self):
41 self.sys_path_save = sys.path
42 self.cwd_save = os.getcwd()
43
44 config = os.path.join(test_path, "test_config.yaml")
45 from xosconfig import Config
46
47 Config.clear()
48 Config.init(config, "synchronizer-config-schema.yaml")
49
50 from xossynchronizer.mock_modelaccessor_build import (
51 build_mock_modelaccessor,
52 )
53
54 build_mock_modelaccessor(sync_lib_dir, xos_dir, services_dir=None, service_xprotos=[])
55
Scott Bakerf0d7e5c2019-02-05 08:35:31 -080056 os.chdir(os.path.join(test_path, "..")) # config references xos-synchronizer-tests/model-deps
Scott Bakerbba67b62019-01-28 17:38:21 -080057
58 import xossynchronizer.event_loop
59
60 reload(xossynchronizer.event_loop)
61 import xossynchronizer.backend
62
63 reload(xossynchronizer.backend)
64 from xossynchronizer.modelaccessor import model_accessor
65
66 # import all class names to globals
67 for (k, v) in model_accessor.all_model_classes.items():
68 globals()[k] = v
69
Scott Bakerc2fddaa2019-01-30 15:45:03 -080070 from xossynchronizer.modelaccessor import model_accessor
71
72 b = xossynchronizer.backend.Backend(model_accessor=model_accessor)
Scott Bakerbba67b62019-01-28 17:38:21 -080073 steps_dir = Config.get("steps_dir")
74 self.steps = b.load_sync_step_modules(steps_dir)
Scott Bakerc2fddaa2019-01-30 15:45:03 -080075 self.synchronizer = xossynchronizer.event_loop.XOSObserver(self.steps, model_accessor)
Scott Bakerbba67b62019-01-28 17:38:21 -080076 try:
77 os.remove("/tmp/sync_ports")
78 except OSError:
79 pass
80 try:
81 os.remove("/tmp/delete_ports")
82 except OSError:
83 pass
84
85 def tearDown(self):
86 sys.path = self.sys_path_save
87 os.chdir(self.cwd_save)
88
89 @mock.patch(
Scott Bakerf0d7e5c2019-02-05 08:35:31 -080090 "test_steps.sync_instances.ansiblesyncstep.run_template",
Scott Bakerbba67b62019-01-28 17:38:21 -080091 side_effect=run_fake_ansible_template,
92 )
Scott Bakerc2fddaa2019-01-30 15:45:03 -080093 def test_run_once(self, mock_run_template):
Scott Bakerbba67b62019-01-28 17:38:21 -080094 pending_objects, pending_steps = self.synchronizer.fetch_pending()
95 pending_objects2 = list(pending_objects)
96
97 any_cs = next(
98 obj for obj in pending_objects if obj.leaf_model_name == "ControllerSlice"
99 )
100 any_instance = next(
101 obj for obj in pending_objects2 if obj.leaf_model_name == "Instance"
102 )
103
104 slice = Slice()
105 any_instance.slice = slice
106 any_cs.slice = slice
107
108 self.synchronizer.run_once()
109
110 sync_ports = open("/tmp/sync_ports").read()
111 delete_ports = open("/tmp/delete_ports").read()
112
113 self.assertIn("successful", sync_ports)
114 self.assertIn("successful", delete_ports)
115
116
117if __name__ == "__main__":
118 unittest.main()