[SEBA-450] (part 2)
Add tox testing support on additional XOS library modules:
- xos-api
- xos-kafka (has no tests)
- xos-migrate (has no tests)
- xos-synchronizer
Change-Id: I98195bc9747971d3515882d517affe058dd86ac5
diff --git a/lib/xos-synchronizer/xos-synchronizer-tests/test_run.py b/lib/xos-synchronizer/xos-synchronizer-tests/test_run.py
index a9c6c19..f21428e 100644
--- a/lib/xos-synchronizer/xos-synchronizer-tests/test_run.py
+++ b/lib/xos-synchronizer/xos-synchronizer-tests/test_run.py
@@ -12,6 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+from __future__ import absolute_import
import json
import unittest
import mock
@@ -19,6 +20,12 @@
import os
import sys
+try:
+ # Python 2: "reload" is built-in
+ reload # pylint: disable=reload-builtin
+except NameError:
+ from importlib import reload
+
test_path = os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
sync_lib_dir = os.path.join(test_path, "..", "xossynchronizer")
xos_dir = os.path.join(test_path, "..", "..", "..", "xos")
@@ -47,13 +54,15 @@
Config.clear()
Config.init(config, "synchronizer-config-schema.yaml")
- from xossynchronizer.mock_modelaccessor_build import (
- build_mock_modelaccessor,
+ from xossynchronizer.mock_modelaccessor_build import build_mock_modelaccessor
+
+ build_mock_modelaccessor(
+ sync_lib_dir, xos_dir, services_dir=None, service_xprotos=[]
)
- build_mock_modelaccessor(sync_lib_dir, xos_dir, services_dir=None, service_xprotos=[])
-
- os.chdir(os.path.join(test_path, "..")) # config references xos-synchronizer-tests/model-deps
+ os.chdir(
+ os.path.join(test_path, "..")
+ ) # config references xos-synchronizer-tests/model-deps
import xossynchronizer.event_loop
@@ -72,7 +81,9 @@
b = xossynchronizer.backend.Backend(model_accessor=model_accessor)
steps_dir = Config.get("steps_dir")
self.steps = b.load_sync_step_modules(steps_dir)
- self.synchronizer = xossynchronizer.event_loop.XOSObserver(self.steps, model_accessor)
+ self.synchronizer = xossynchronizer.event_loop.XOSObserver(
+ self.steps, model_accessor
+ )
try:
os.remove("/tmp/sync_ports")
except OSError: