[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/xossynchronizer/__init__.py b/lib/xos-synchronizer/xossynchronizer/__init__.py
index 18ab956..ea109c8 100644
--- a/lib/xos-synchronizer/xossynchronizer/__init__.py
+++ b/lib/xos-synchronizer/xossynchronizer/__init__.py
@@ -12,6 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+from __future__ import absolute_import
from .synchronizer import Synchronizer
__all__ = ["Synchronizer"]
diff --git a/lib/xos-synchronizer/xossynchronizer/ansible_helper.py b/lib/xos-synchronizer/xossynchronizer/ansible_helper.py
index c607607..846aeb9 100644
--- a/lib/xos-synchronizer/xossynchronizer/ansible_helper.py
+++ b/lib/xos-synchronizer/xossynchronizer/ansible_helper.py
@@ -14,24 +14,20 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from __future__ import print_function
-import jinja2
-import tempfile
-import os
-import json
-import pickle
-import pdb
-import string
-import random
-import re
-import traceback
-import subprocess
-import threading
+from __future__ import absolute_import, print_function
-from multiprocessing import Process, Queue
-from xosconfig import Config
+import json
+import os
+import pickle
+import random
+import string
+import tempfile
+
+import jinja2
from multistructlog import create_logger
+from xosconfig import Config
+from six.moves import range
log = create_logger(Config().get("logging"))
@@ -108,7 +104,7 @@
stats = result.get("stats", None)
aresults = result.get("aresults", None)
- except Exception as e:
+ except BaseException:
log.exception("Exception running ansible_main")
stats = None
aresults = None
@@ -220,7 +216,7 @@
if t["failures"] > 0:
raise ValueError("Ansible playbook reported failures for host %s" % h)
- except ValueError as e:
+ except ValueError:
if error_msg:
try:
error = " // ".join(error_msg)
@@ -230,7 +226,7 @@
else:
raise
- processed_results = map(lambda x: x._result, ok_results)
+ processed_results = [x._result for x in ok_results]
return processed_results[1:] # 0 is setup
diff --git a/lib/xos-synchronizer/xossynchronizer/ansible_main.py b/lib/xos-synchronizer/xossynchronizer/ansible_main.py
index 08283a4..6a8c711 100644
--- a/lib/xos-synchronizer/xossynchronizer/ansible_main.py
+++ b/lib/xos-synchronizer/xossynchronizer/ansible_main.py
@@ -12,15 +12,21 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+from __future__ import absolute_import
import os
import pickle
import sys
-
-# import json
import traceback
+
from xosconfig import Config
+try:
+ # Python 2: "reload" is built-in
+ reload # pylint: disable=reload-builtin
+except NameError:
+ from importlib import reload
+
sys.path.append("/opt/xos")
@@ -42,7 +48,7 @@
except KeyError:
pass
- import ansible_runner
+ from . import ansible_runner
reload(ansible_runner)
@@ -52,7 +58,7 @@
)
stats, aresults = runner.run()
- except Exception as e:
+ except Exception:
return {"stats": None, "aresults": None, "exception": traceback.format_exc()}
return {"stats": stats, "aresults": aresults}
diff --git a/lib/xos-synchronizer/xossynchronizer/ansible_runner.py b/lib/xos-synchronizer/xossynchronizer/ansible_runner.py
index d20feb5..2615346 100644
--- a/lib/xos-synchronizer/xossynchronizer/ansible_runner.py
+++ b/lib/xos-synchronizer/xossynchronizer/ansible_runner.py
@@ -14,22 +14,30 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from multistructlog import create_logger
-from xosconfig import Config
-from ansible.plugins.callback import CallbackBase
-from ansible.utils.display import Display
-from ansible.executor import playbook_executor
-from ansible.parsing.dataloader import DataLoader
-from ansible.vars.manager import VariableManager
-from ansible.inventory.manager import InventoryManager
-from tempfile import NamedTemporaryFile
-import os
-import sys
-import pdb
+from __future__ import absolute_import
+
import json
+import os
import uuid
from ansible import constants
+from ansible.executor import playbook_executor
+from ansible.inventory.manager import InventoryManager
+from ansible.parsing.dataloader import DataLoader
+from ansible.plugins.callback import CallbackBase
+from ansible.utils.display import Display
+from ansible.vars.manager import VariableManager
+
+from multistructlog import create_logger
+from xosconfig import Config
+
+try:
+ # Python 2: "reload" is built-in
+ # pylint: disable=W1626
+ reload
+except NameError:
+ # Python 3: "reload" is part of importlib
+ from importlib import reload
constants = reload(constants)
diff --git a/lib/xos-synchronizer/xossynchronizer/apiaccessor.py b/lib/xos-synchronizer/xossynchronizer/apiaccessor.py
index a56381b..b4e3b13 100644
--- a/lib/xos-synchronizer/xossynchronizer/apiaccessor.py
+++ b/lib/xos-synchronizer/xossynchronizer/apiaccessor.py
@@ -13,9 +13,11 @@
# limitations under the License.
-from modelaccessor import ModelAccessor
+from __future__ import absolute_import
+
import datetime
-import time
+
+from .modelaccessor import ModelAccessor
class CoreApiModelAccessor(ModelAccessor):
diff --git a/lib/xos-synchronizer/xossynchronizer/backend.py b/lib/xos-synchronizer/xossynchronizer/backend.py
index 55977b2..96e412f 100644
--- a/lib/xos-synchronizer/xossynchronizer/backend.py
+++ b/lib/xos-synchronizer/xossynchronizer/backend.py
@@ -12,22 +12,22 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from __future__ import print_function
-import os
-import inspect
+from __future__ import absolute_import, print_function
+
import imp
+import inspect
+import os
import sys
import threading
import time
-from xossynchronizer.steps.syncstep import SyncStep
+
+from multistructlog import create_logger
+from xosconfig import Config
+from xossynchronizer.event_engine import XOSEventEngine
from xossynchronizer.event_loop import XOSObserver
from xossynchronizer.model_policy_loop import XOSPolicyEngine
-from xossynchronizer.event_engine import XOSEventEngine
from xossynchronizer.pull_step_engine import XOSPullStepEngine
-from xosconfig import Config
-from multistructlog import create_logger
-
log = create_logger(Config().get("logging"))
@@ -100,7 +100,11 @@
# if we have at least one sync_step
if len(sync_steps) > 0:
# start the observer
- self.log.info("Starting XOSObserver", sync_steps=sync_steps, model_accessor=self.model_accessor)
+ self.log.info(
+ "Starting XOSObserver",
+ sync_steps=sync_steps,
+ model_accessor=self.model_accessor,
+ )
observer = XOSObserver(sync_steps, self.model_accessor, self.log)
observer_thread = threading.Thread(
target=observer.run, name="synchronizer"
@@ -131,14 +135,20 @@
self.log.info("Skipping event engine due to synchronizer unloading.")
else:
self.log.info("Starting XOSEventEngine", event_steps_dir=event_steps_dir)
- event_engine = XOSEventEngine(model_accessor=self.model_accessor, log=self.log)
+ event_engine = XOSEventEngine(
+ model_accessor=self.model_accessor, log=self.log
+ )
event_engine.load_event_step_modules(event_steps_dir)
event_engine.start()
# start model policies thread
policies_dir = Config.get("model_policies_dir")
if policies_dir:
- policy_engine = XOSPolicyEngine(policies_dir=policies_dir, model_accessor=self.model_accessor, log=self.log)
+ policy_engine = XOSPolicyEngine(
+ policies_dir=policies_dir,
+ model_accessor=self.model_accessor,
+ log=self.log,
+ )
model_policy_thread = threading.Thread(
target=policy_engine.run, name="policy_engine"
)
diff --git a/lib/xos-synchronizer/xossynchronizer/backend_modelpolicy.py b/lib/xos-synchronizer/xossynchronizer/backend_modelpolicy.py
index a8e826b..82eb7b8 100644
--- a/lib/xos-synchronizer/xossynchronizer/backend_modelpolicy.py
+++ b/lib/xos-synchronizer/xossynchronizer/backend_modelpolicy.py
@@ -12,18 +12,14 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-from __future__ import print_function
-import os
-import inspect
-import imp
+from __future__ import absolute_import, print_function
+
import sys
import threading
import time
-from syncstep import SyncStep
-from synchronizers.new_base.event_loop import XOSObserver
-from xosconfig import Config
from multistructlog import create_logger
+from xosconfig import Config
log = create_logger(Config().get("logging"))
diff --git a/lib/xos-synchronizer/xossynchronizer/dependency_walker_new.py b/lib/xos-synchronizer/xossynchronizer/dependency_walker_new.py
index 138c26d..8c9c8f9 100644
--- a/lib/xos-synchronizer/xossynchronizer/dependency_walker_new.py
+++ b/lib/xos-synchronizer/xossynchronizer/dependency_walker_new.py
@@ -12,26 +12,16 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-
-#!/usr/bin/env python
-
# TODO: Moved this into the synchronizer, as it appeared to require model
# access. Verify whether or not that's true and reconcile with
# generate/dependency_walker.py
-from __future__ import print_function
-import os
-import imp
-import inspect
-import time
-import traceback
-import commands
-import threading
-from xosconfig import Config
+from __future__ import absolute_import, print_function
+
import json
-from xosconfig import Config
from multistructlog import create_logger
+from xosconfig import Config
log = create_logger(Config().get("logging"))
diff --git a/lib/xos-synchronizer/xossynchronizer/event_engine.py b/lib/xos-synchronizer/xossynchronizer/event_engine.py
index 694a1a8..0455f0d 100644
--- a/lib/xos-synchronizer/xossynchronizer/event_engine.py
+++ b/lib/xos-synchronizer/xossynchronizer/event_engine.py
@@ -12,12 +12,16 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-import confluent_kafka
+from __future__ import absolute_import
+
import imp
import inspect
import os
import threading
import time
+
+import confluent_kafka
+
from xosconfig import Config
@@ -130,7 +134,9 @@
)
try:
- self.step(model_accessor=self.model_accessor, log=self.log).process_event(event_msg)
+ self.step(
+ model_accessor=self.model_accessor, log=self.log
+ ).process_event(event_msg)
except BaseException:
self.log.exception(
@@ -207,7 +213,9 @@
for step in self.event_steps:
if step.technology == "kafka":
- thread = XOSKafkaThread(step, [eventbus_endpoint], self.model_accessor, self.log)
+ thread = XOSKafkaThread(
+ step, [eventbus_endpoint], self.model_accessor, self.log
+ )
thread.start()
self.threads.append(thread)
else:
diff --git a/lib/xos-synchronizer/xossynchronizer/event_loop.py b/lib/xos-synchronizer/xossynchronizer/event_loop.py
index 5cbfda9..15f005c 100644
--- a/lib/xos-synchronizer/xossynchronizer/event_loop.py
+++ b/lib/xos-synchronizer/xossynchronizer/event_loop.py
@@ -16,23 +16,28 @@
# Add unit tests:
# - 2 sets of Instance, ControllerSlice, ControllerNetworks - delete and create case
-import time
-import threading
-import json
+from __future__ import absolute_import
+import json
+import threading
+import time
from collections import defaultdict
+
+from multistructlog import create_logger
from networkx import (
DiGraph,
- weakly_connected_component_subgraphs,
- all_shortest_paths,
NetworkXNoPath,
+ all_shortest_paths,
+ weakly_connected_component_subgraphs,
)
from networkx.algorithms.dag import topological_sort
-
-from xossynchronizer.steps.syncstep import InnocuousException, DeferredException, SyncStep
-
from xosconfig import Config
-from multistructlog import create_logger
+from xossynchronizer.steps.syncstep import (
+ DeferredException,
+ InnocuousException,
+ SyncStep,
+)
+from six.moves import range
log = create_logger(Config().get("logging"))
@@ -113,10 +118,12 @@
# src_port is the field that accesses Model2 from Model1
# dst_port is the field that accesses Model1 from Model2
static_dependencies = json.loads(dep_graph_str)
- dynamic_dependencies = [] # Dropped Service and ServiceInstance dynamic dependencies
+ dynamic_dependencies = (
+ []
+ ) # Dropped Service and ServiceInstance dynamic dependencies
joint_dependencies = dict(
- static_dependencies.items() + dynamic_dependencies
+ list(static_dependencies.items()) + dynamic_dependencies
)
model_dependency_graph = DiGraph()
@@ -267,7 +274,6 @@
def handle_sync_exception(self, o, e):
self.log.exception("sync step failed!", e=e, **o.tologdict())
- current_code = o.backend_code
if hasattr(e, "message"):
status = str(e.message)
@@ -359,7 +365,6 @@
sc_log = self.log.new(thread_id=threading.current_thread().ident)
try:
- start_time = time.time()
sc_log.debug("Starting to work on cohort", cohort=cohort, deletion=deletion)
cohort_emptied = False
@@ -417,14 +422,13 @@
self.run_once()
def fetch_pending(self, deletion=False):
- unique_model_list = list(set(self.model_to_step.keys()))
pending_objects = []
pending_steps = []
- step_list = self.step_lookup.values()
+ step_list = list(self.step_lookup.values())
for e in self.external_dependencies:
s = SyncStep
- if isinstance(e,str):
+ if isinstance(e, str):
# external dependency is a string that names a model class
s.observes = self.model_accessor.get_model_class(e)
else:
@@ -439,7 +443,9 @@
if not hasattr(step, "call"):
pending = step.fetch_pending(deletion)
for obj in pending:
- step = step_class(driver=self.driver, model_accessor=self.model_accessor)
+ step = step_class(
+ driver=self.driver, model_accessor=self.model_accessor
+ )
step.log = self.log.new(step=step)
obj.synchronizer_step = step
@@ -460,7 +466,7 @@
if o is None:
return [], None
try:
- o_lst = [o for o in o.all()]
+ o_lst = [oa for oa in o.all()]
edge_type = PROXY_EDGE
except (AttributeError, TypeError):
o_lst = [o]
@@ -606,10 +612,8 @@
"""
def compute_dependent_cohorts(self, objects, deletion):
- model_map = defaultdict(list)
n = len(objects)
- r = range(n)
- indexed_objects = zip(r, objects)
+ r = list(range(n))
oG = DiGraph()
@@ -652,8 +656,6 @@
# Why are we checking the DB connection here?
self.model_accessor.check_db_connection_okay()
- loop_start = time.time()
-
# Two passes. One for sync, the other for deletion.
for deletion in (False, True):
objects_to_process = []
@@ -693,8 +695,6 @@
except Exception as e:
self.log.exception("Legacy step failed", step=step, e=e)
- loop_end = time.time()
-
except Exception as e:
self.log.exception(
"Core error. This seems like a misconfiguration or bug. This error will not be relayed to the user!",
diff --git a/lib/xos-synchronizer/xossynchronizer/loadmodels.py b/lib/xos-synchronizer/xossynchronizer/loadmodels.py
index 1c1f8bd..43b429e 100644
--- a/lib/xos-synchronizer/xossynchronizer/loadmodels.py
+++ b/lib/xos-synchronizer/xossynchronizer/loadmodels.py
@@ -12,9 +12,12 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+from __future__ import absolute_import
+
import os
-from xosconfig import Config
+
from multistructlog import create_logger
+from xosconfig import Config
log = create_logger(Config().get("logging"))
@@ -89,4 +92,3 @@
result = self.api.dynamicload.UnloadModels(request)
return result
-
diff --git a/lib/xos-synchronizer/xossynchronizer/mock_modelaccessor_build.py b/lib/xos-synchronizer/xossynchronizer/mock_modelaccessor_build.py
index 99b2d46..1db6f26 100644
--- a/lib/xos-synchronizer/xossynchronizer/mock_modelaccessor_build.py
+++ b/lib/xos-synchronizer/xossynchronizer/mock_modelaccessor_build.py
@@ -12,10 +12,19 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+from __future__ import absolute_import
+
import os
-import cPickle
import subprocess
+try:
+ # Python 2 has separate pickle and cPickle
+ # pylint: disable=W1648
+ import cPickle
+except ImportError:
+ # Python 3 will use cPickle by dfault
+ import pickle as cPickle
+
"""
Support for autogenerating mock_modelaccessor.
@@ -31,7 +40,7 @@
# TODO: deprecate the dest_dir argument
# force modelaccessor to be found in /tmp
- dest_dir="/tmp/mock_modelaccessor"
+ dest_dir = "/tmp/mock_modelaccessor"
if not os.path.exists(dest_dir):
os.makedirs(dest_dir)
dest_fn = os.path.join(dest_dir, "mock_modelaccessor.py")
@@ -44,13 +53,13 @@
# Check to see if we've already run xosgenx. If so, don't run it again.
context_fn = dest_fn + ".context"
this_context = (xos_dir, services_dir, service_xprotos, target)
- need_xosgenx = True
+
if os.path.exists(context_fn):
try:
- context = cPickle.loads(open(context_fn).read())
+ context = cPickle.loads(open(context_fn, 'rb').read())
if context == this_context:
return
- except (cPickle.UnpicklingError, EOFError):
+ except (cPickle.UnpicklingError, EOFError, ValueError):
# Something went wrong with the file read or depickling
pass
@@ -74,7 +83,8 @@
)
# Save the context of this invocation of xosgenx
- open(context_fn, "w").write(cPickle.dumps(this_context))
+ open(context_fn, "wb").write(cPickle.dumps(this_context))
+
# generate model from xproto
def get_models_fn(services_dir, service_name, xproto_name):
@@ -86,8 +96,11 @@
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))
+
+
# END generate model from xproto
+
def mock_modelaccessor_config(test_dir, services):
""" Automatically configure the mock modelaccessor.
@@ -99,13 +112,13 @@
while not orchestration_dir.endswith("orchestration"):
# back up a level
orchestration_dir = os.path.dirname(orchestration_dir)
- if len(orchestration_dir)<10:
+ if len(orchestration_dir) < 10:
raise Exception("Failed to autodiscovery repository tree")
xos_dir = os.path.join(orchestration_dir, "xos", "xos")
services_dir = os.path.join(orchestration_dir, "xos_services")
- service_xprotos=[]
+ service_xprotos = []
for (service_name, xproto_name) in services:
service_xprotos.append(get_models_fn(services_dir, service_name, xproto_name))
diff --git a/lib/xos-synchronizer/xossynchronizer/model_policies/policy.py b/lib/xos-synchronizer/xossynchronizer/model_policies/policy.py
index 5877279..b33a634 100644
--- a/lib/xos-synchronizer/xossynchronizer/model_policies/policy.py
+++ b/lib/xos-synchronizer/xossynchronizer/model_policies/policy.py
@@ -18,8 +18,10 @@
Base Classes for Model Policies
"""
-from xosconfig import Config
+from __future__ import absolute_import
+
from multistructlog import create_logger
+from xosconfig import Config
log = create_logger(Config().get("logging"))
diff --git a/lib/xos-synchronizer/xossynchronizer/model_policy_loop.py b/lib/xos-synchronizer/xossynchronizer/model_policy_loop.py
index 20144a5..5259778 100644
--- a/lib/xos-synchronizer/xossynchronizer/model_policy_loop.py
+++ b/lib/xos-synchronizer/xossynchronizer/model_policy_loop.py
@@ -12,11 +12,9 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+from __future__ import absolute_import, print_function
-from __future__ import print_function
-from xossynchronizer.dependency_walker_new import *
-from xossynchronizer.model_policies.policy import Policy
-
+import os
import imp
import inspect
import time
@@ -25,11 +23,13 @@
class XOSPolicyEngine(object):
def __init__(self, policies_dir, model_accessor, log):
+
+ self.log = log # has to come before self.load_model_policies(), which logs
+
self.model_accessor = model_accessor
self.model_policies = self.load_model_policies(policies_dir)
self.policies_by_name = {}
self.policies_by_class = {}
- self.log = log
for policy in self.model_policies:
if policy.model_name not in self.policies_by_name:
@@ -53,7 +53,7 @@
def update_dep(self, d, o):
try:
- print("Trying to update %s" % d)
+ self.log.info("Trying to update %s", d)
save_fields = []
if d.updated < o.updated:
save_fields = ["updated"]
@@ -61,15 +61,15 @@
if save_fields:
d.save(update_fields=save_fields)
except AttributeError as e:
- log.exception("AttributeError in update_dep", e=e)
+ self.log.exception("AttributeError in update_dep", e=e)
raise e
except Exception as e:
- log.exception("Exception in update_dep", e=e)
+ self.log.exception("Exception in update_dep", e=e)
def delete_if_inactive(self, d, o):
try:
d.delete()
- print("Deleted %s (%s)" % (d, d.__class__.__name__))
+ self.log.info("Deleted %s (%s)" % (d, d.__class__.__name__))
except BaseException:
pass
return
@@ -103,13 +103,13 @@
and (c not in policies)
):
if not c.model_name:
- log.info(
+ self.log.info(
"load_model_policies: skipping model policy",
classname=classname,
)
continue
if not self.model_accessor.has_model_class(c.model_name):
- log.error(
+ self.log.error(
"load_model_policies: unable to find model policy",
classname=classname,
model=c.model_name,
@@ -117,12 +117,12 @@
c.model = self.model_accessor.get_model_class(c.model_name)
policies.append(c)
- log.info("Loaded model policies", policies=policies)
+ self.log.info("Loaded model policies", policies=policies)
return policies
def execute_model_policy(self, instance, action):
# These are the models whose children get deleted when they are
- delete_policy_models = ["Slice", "Instance", "Network"]
+ # delete_policy_models = ["Slice", "Instance", "Network"]
sender_name = getattr(instance, "model_name", instance.__class__.__name__)
# if (action != "deleted"):
@@ -136,15 +136,17 @@
method_name = "handle_%s" % action
if hasattr(policy, method_name):
try:
- log.debug(
+ self.log.debug(
"MODEL POLICY: calling handler",
sender_name=sender_name,
instance=instance,
policy=policy.__name__,
method=method_name,
)
- getattr(policy(model_accessor=self.model_accessor), method_name)(instance)
- log.debug(
+ getattr(policy(model_accessor=self.model_accessor), method_name)(
+ instance
+ )
+ self.log.debug(
"MODEL POLICY: completed handler",
sender_name=sender_name,
instance=instance,
@@ -152,7 +154,7 @@
method=method_name,
)
except Exception as e:
- log.exception("MODEL POLICY: Exception when running handler", e=e)
+ self.log.exception("MODEL POLICY: Exception when running handler", e=e)
policies_failed = True
try:
@@ -160,7 +162,7 @@
instance.policy_code = 2
instance.save(update_fields=["policy_status", "policy_code"])
except Exception as e:
- log.exception(
+ self.log.exception(
"MODEL_POLICY: Exception when storing policy_status", e=e
)
@@ -173,11 +175,13 @@
instance.save(update_fields=["policed", "policy_status", "policy_code"])
if hasattr(policy, "after_policy_save"):
- policy(model_accessor=self.model_accessor).after_policy_save(instance)
+ policy(model_accessor=self.model_accessor).after_policy_save(
+ instance
+ )
- log.info("MODEL_POLICY: Saved", o=instance)
+ self.log.info("MODEL_POLICY: Saved", o=instance)
except BaseException:
- log.exception(
+ self.log.exception(
"MODEL POLICY: Object failed to update policed timestamp",
instance=instance,
)
@@ -191,7 +195,7 @@
try:
self.run_policy_once()
except Exception as e:
- log.exception("MODEL_POLICY: Exception in run()", e=e)
+ self.log.exception("MODEL_POLICY: Exception in run()", e=e)
if time.time() - start < 5:
time.sleep(5)
@@ -199,7 +203,7 @@
# ways to combine them.
def run_policy_once(self):
- models = self.policies_by_class.keys()
+ models = list(self.policies_by_class.keys())
self.model_accessor.check_db_connection_okay()
@@ -222,4 +226,4 @@
self.model_accessor.reset_queries()
except Exception as e:
# this shouldn't happen, but in case it does, catch it...
- log.exception("MODEL POLICY: exception in reset_queries", e)
+ self.log.exception("MODEL POLICY: exception in reset_queries", e)
diff --git a/lib/xos-synchronizer/xossynchronizer/modelaccessor.py b/lib/xos-synchronizer/xossynchronizer/modelaccessor.py
index 72fc3c5..b4b26f3 100644
--- a/lib/xos-synchronizer/xossynchronizer/modelaccessor.py
+++ b/lib/xos-synchronizer/xossynchronizer/modelaccessor.py
@@ -12,7 +12,6 @@
# See the License for the specific language governing permissions and
# limitations under the License.
-
""" ModelAccessor
A class for abstracting access to models. Used to get any djangoisms out
@@ -23,24 +22,28 @@
models into the calling module's scope.
"""
+from __future__ import absolute_import
+
import functools
import importlib
import os
import signal
import sys
from threading import Timer
-from loadmodels import ModelLoadClient
from xosconfig import Config
-from multistructlog import create_logger
from xosutil.autodiscover_version import autodiscover_version_of_main
+from .loadmodels import ModelLoadClient
+
+from multistructlog import create_logger
log = create_logger(Config().get("logging"))
after_reactor_exit_code = None
orig_sigint = None
model_accessor = None
+
class ModelAccessor(object):
def __init__(self):
self.all_model_classes = self.get_all_model_classes()
@@ -161,6 +164,7 @@
reactor.callLater(1, functools.partial(keep_trying, client, reactor))
+
def unload_models(client, reactor, version):
# This function is called by a timer until it succeeds.
log.info("unload_models initiated by timer")
@@ -180,13 +184,14 @@
if result.status == result.TRYAGAIN:
log.info("TRYAGAIN received. Expect to try again in 30 seconds.")
- except Exception as e:
+ except Exception:
# If the synchronizer is operational, then assume the ORM's restart_on_disconnect will deal with the
# connection being lost.
log.exception("Error while unloading. Expect to try again in 30 seconds.")
Timer(30, functools.partial(unload_models, client, reactor, version)).start()
+
def exit_while_inside_reactor(reactor, code):
""" Calling sys.exit() while inside reactor ends up trapped by reactor.
@@ -261,7 +266,6 @@
exit_while_inside_reactor(reactor, 1)
return
-
log.exception("failed to onboard models")
# If it's some other error, then we don't need to force a reconnect. Just try the LoadModels() again.
reactor.callLater(10, functools.partial(grpcapi_reconnect, client, reactor))
@@ -279,7 +283,7 @@
client.xos_orm.restart_on_disconnect = True
- from apiaccessor import CoreApiModelAccessor
+ from .apiaccessor import CoreApiModelAccessor
model_accessor = CoreApiModelAccessor(orm=client.xos_orm)
@@ -370,7 +374,7 @@
global model_accessor
# the mock model accessor always gets built to a temporary location
- if not "/tmp/mock_modelaccessor" in sys.path:
+ if "/tmp/mock_modelaccessor" not in sys.path:
sys.path.append("/tmp/mock_modelaccessor")
from mock_modelaccessor import model_accessor as mock_model_accessor
diff --git a/lib/xos-synchronizer/xossynchronizer/pull_step_engine.py b/lib/xos-synchronizer/xossynchronizer/pull_step_engine.py
index a58bfa2..c3a2557 100644
--- a/lib/xos-synchronizer/xossynchronizer/pull_step_engine.py
+++ b/lib/xos-synchronizer/xossynchronizer/pull_step_engine.py
@@ -12,13 +12,16 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+from __future__ import absolute_import
+
import imp
import inspect
import os
import threading
import time
-from xosconfig import Config
+
from multistructlog import create_logger
+from xosconfig import Config
log = create_logger(Config().get("logging"))
@@ -44,7 +47,10 @@
threads = []
for step in self.steps:
- thread = threading.Thread(target=step(model_accessor=self.model_accessor).pull_records, name="pull_step")
+ thread = threading.Thread(
+ target=step(model_accessor=self.model_accessor).pull_records,
+ name="pull_step",
+ )
threads.append(thread)
for t in threads:
@@ -100,5 +106,7 @@
log.info("Starting pull steps engine", steps=self.pull_steps)
for step in self.pull_steps:
- sched = XOSPullStepScheduler(steps=self.pull_steps, model_accessor=self.model_accessor)
+ sched = XOSPullStepScheduler(
+ steps=self.pull_steps, model_accessor=self.model_accessor
+ )
sched.run()
diff --git a/lib/xos-synchronizer/xossynchronizer/steps/ansiblesyncstep.py b/lib/xos-synchronizer/xossynchronizer/steps/ansiblesyncstep.py
index 116f8c2..3781e1f 100644
--- a/lib/xos-synchronizer/xossynchronizer/steps/ansiblesyncstep.py
+++ b/lib/xos-synchronizer/xossynchronizer/steps/ansiblesyncstep.py
@@ -12,8 +12,12 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+from __future__ import absolute_import
+
from xossynchronizer.ansible_helper import run_template
-from syncstep import SyncStep
+
+from .syncstep import SyncStep
+
class AnsibleSyncStep(SyncStep):
def sync_record(self, o):
@@ -57,4 +61,4 @@
if res[0].get("rc", 0) != 0:
raise Exception("Nonzero rc from Ansible during delete_record")
- self.log.debug("Finished default delete record", **o.tologdict())
\ No newline at end of file
+ self.log.debug("Finished default delete record", **o.tologdict())
diff --git a/lib/xos-synchronizer/xossynchronizer/steps/syncstep.py b/lib/xos-synchronizer/xossynchronizer/steps/syncstep.py
index 7644822..08637ed 100644
--- a/lib/xos-synchronizer/xossynchronizer/steps/syncstep.py
+++ b/lib/xos-synchronizer/xossynchronizer/steps/syncstep.py
@@ -13,9 +13,12 @@
# limitations under the License.
-from xosconfig import Config
+from __future__ import absolute_import
+
from functools import reduce
+from xosconfig import Config
+
def f7(seq):
seen = set()
@@ -113,7 +116,6 @@
result.append(class_or_name)
return result
-
def fetch_pending(self, deletion=False):
# This is the most common implementation of fetch_pending
# Steps should override it if they have their own logic
@@ -121,12 +123,10 @@
return self.model_accessor.fetch_pending(self.observes_classes, deletion)
-
def sync_record(self, o):
self.log.debug("In abstract sync record", **o.tologdict())
# This method should be overridden by the service
-
def delete_record(self, o):
self.log.debug("In abstract delete record", **o.tologdict())
# This method should be overridden by the service
diff --git a/lib/xos-synchronizer/xossynchronizer/synchronizer.py b/lib/xos-synchronizer/xossynchronizer/synchronizer.py
index f204785..42344e9 100644
--- a/lib/xos-synchronizer/xossynchronizer/synchronizer.py
+++ b/lib/xos-synchronizer/xossynchronizer/synchronizer.py
@@ -1,5 +1,3 @@
-#!/usr/bin/env python
-
# Copyright 2017-present Open Networking Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License");
@@ -14,17 +12,20 @@
# See the License for the specific language governing permissions and
# limitations under the License.
+from __future__ import absolute_import
+
import time
-from xosconfig import Config
from multistructlog import create_logger
+from xosconfig import Config
+
class Synchronizer(object):
def __init__(self):
self.log = create_logger(Config().get("logging"))
def create_model_accessor(self):
- from modelaccessor import model_accessor
+ from .modelaccessor import model_accessor
self.model_accessor = model_accessor
@@ -33,7 +34,8 @@
wait = False
while not models_active:
try:
- _i = self.model_accessor.Site.objects.first()
+ # variable is unused
+ _i = self.model_accessor.Site.objects.first() # noqa: F841
models_active = True
except Exception as e:
self.log.info("Exception", e=e)
@@ -54,12 +56,8 @@
# use `from xossynchronizer.modelaccessor import ...` and require the model accessor to be initialized before
# their code can be imported.
- from backend import Backend
+ from .backend import Backend
log_closure = self.log.bind(synchronizer_name=Config().get("name"))
backend = Backend(log=log_closure, model_accessor=self.model_accessor)
backend.run()
-
-
-if __name__ == "__main__":
- main()