SEBA-450 Update Att-Workflow-Driver to use synchronizer library
Change-Id: I86740c694950e86f2d7afea6221d33e2aad8aeef
diff --git a/xos/synchronizer/model_policies/test_model_policy_att_workflow_driver_serviceinstance.py b/xos/synchronizer/model_policies/test_model_policy_att_workflow_driver_serviceinstance.py
index 1abc7c8..19bcaba 100644
--- a/xos/synchronizer/model_policies/test_model_policy_att_workflow_driver_serviceinstance.py
+++ b/xos/synchronizer/model_policies/test_model_policy_att_workflow_driver_serviceinstance.py
@@ -20,39 +20,31 @@
import os, sys
test_path=os.path.abspath(os.path.dirname(os.path.realpath(__file__)))
-service_dir=os.path.join(test_path, "../../../..")
-xos_dir=os.path.join(test_path, "../../..")
-if not os.path.exists(os.path.join(test_path, "new_base")):
- xos_dir=os.path.join(test_path, "../../../../../../orchestration/xos/xos")
- services_dir=os.path.join(xos_dir, "../../xos_services")
-def get_models_fn(service_name, xproto_name):
- name = os.path.join(service_name, "xos", "synchronizer", "models", xproto_name)
- if os.path.exists(os.path.join(services_dir, name)):
- return name
- raise Exception("Unable to find service=%s xproto=%s" % (service_name, xproto_name))
class TestModelPolicyAttWorkflowDriverServiceInstance(unittest.TestCase):
def setUp(self):
self.sys_path_save = sys.path
- sys.path.append(xos_dir)
- sys.path.append(os.path.join(xos_dir, 'synchronizers', 'new_base'))
config = os.path.join(test_path, "../test_config.yaml")
from xosconfig import Config
Config.clear()
Config.init(config, 'synchronizer-config-schema.yaml')
- from synchronizers.new_base.mock_modelaccessor_build import build_mock_modelaccessor
- build_mock_modelaccessor(xos_dir, services_dir, [
- get_models_fn("att-workflow-driver", "att-workflow-driver.xproto"),
- get_models_fn("olt-service", "volt.xproto"),
- get_models_fn("../profiles/rcord", "rcord.xproto")
- ])
+ from xossynchronizer.mock_modelaccessor_build import mock_modelaccessor_config
+ mock_modelaccessor_config(test_path, [("att-workflow-driver", "att-workflow-driver.xproto"),
+ ("olt-service", "volt.xproto"),
+ ("../profiles/rcord", "rcord.xproto")])
- import synchronizers.new_base.modelaccessor
- from model_policy_att_workflow_driver_serviceinstance import AttWorkflowDriverServiceInstancePolicy, model_accessor
+ import xossynchronizer.modelaccessor
+ import mock_modelaccessor
+ reload(mock_modelaccessor) # in case nose2 loaded it in a previous test
+ reload(xossynchronizer.modelaccessor) # in case nose2 loaded it in a previous test
+
+ from xossynchronizer.modelaccessor import model_accessor
+ from model_policy_att_workflow_driver_serviceinstance import AttWorkflowDriverServiceInstancePolicy, AttHelpers
+ self.AttHelpers = AttHelpers
from mock_modelaccessor import MockObjectList
@@ -65,7 +57,7 @@
model_accessor.reset_all_object_stores()
- self.policy = AttWorkflowDriverServiceInstancePolicy()
+ self.policy = AttWorkflowDriverServiceInstancePolicy(model_accessor=model_accessor)
self.si = AttWorkflowDriverServiceInstance()
self.si.owner = AttWorkflowDriverService()
self.si.serial_number = "BRCM1234"
@@ -92,8 +84,7 @@
def test_enable_onu(self):
- from helpers import AttHelpers
- with patch.object(AttHelpers, "validate_onu") as validate_onu, \
+ with patch.object(self.AttHelpers, "validate_onu") as validate_onu, \
patch.object(self.policy, "update_onu") as update_onu, \
patch.object(self.si, "save") as save_si:
validate_onu.return_value = [True, "valid onu"]
@@ -106,8 +97,7 @@
self.assertIn("valid onu", self.si.status_message)
def test_disable_onu(self):
- from helpers import AttHelpers
- with patch.object(AttHelpers, "validate_onu") as validate_onu, \
+ with patch.object(self.AttHelpers, "validate_onu") as validate_onu, \
patch.object(self.policy, "update_onu") as update_onu, \
patch.object(self.si, "save") as save_si:
validate_onu.return_value = [False, "invalid onu"]