LOG = log.getLogger(__name__)
class OpenDaylightRestClient(object):
class OpenDaylightRestClientGlobal(object):
    def _check_opt(url):
    def create_client(cls, url=None):
    def __init__(self, url, username, password, timeout):
    def get_resource(self, resource_type, resource_id):
    def get(self, urlpath='', data=None):
    def request(self, method, urlpath='', data=None):
    def sendjson(self, method, urlpath, obj):
    def _check_response(self, response):
    def __init__(self):
    def get_client(self): ODL_NETWORK = 'network'
ODL_NETWORKS = 'networks'
ODL_SUBNET = 'subnet'
ODL_SUBNETS = 'subnets'
ODL_PORT = 'port'
ODL_PORTS = 'ports'
ODL_SG = 'security_group'
ODL_SGS = 'security_groups'
ODL_SG_RULE = 'security_group_rule'
ODL_SG_RULES = 'security_group_rules'
ODL_ROUTER = 'router'
ODL_ROUTERS = 'routers'
ODL_FLOATINGIP = 'floatingip'
ODL_FLOATINGIPS = 'floatingips'
ODL_QOS = 'qos'
ODL_QOS_POLICY = 'policy'
ODL_QOS_POLICIES = 'policies'
ODL_SFC = 'sfc'
ODL_SFC_FLOW_CLASSIFIER = 'flowclassifier'
ODL_SFC_FLOW_CLASSIFIERS = 'flowclassifiers'
ODL_SFC_PORT_PAIR = 'portpair'
ODL_SFC_PORT_PAIRS = 'portpairs'
ODL_SFC_PORT_PAIR_GROUP = 'portpairgroup'
ODL_SFC_PORT_PAIR_GROUPS = 'portpairgroups'
ODL_SFC_PORT_CHAIN = 'portchain'
ODL_SFC_PORT_CHAINS = 'portchains'
NETWORKING_SFC_FLOW_CLASSIFIER = 'flow_classifier'
NETWORKING_SFC_FLOW_CLASSIFIERS = 'flow_classifiers'
NETWORKING_SFC_PORT_PAIR = 'port_pair'
NETWORKING_SFC_PORT_PAIRS = 'port_pairs'
NETWORKING_SFC_PORT_PAIR_GROUP = 'port_pair_group'
NETWORKING_SFC_PORT_PAIR_GROUPS = 'port_pair_groups'
NETWORKING_SFC_PORT_CHAIN = 'port_chain'
NETWORKING_SFC_PORT_CHAINS = 'port_chains'
ODL_TRUNK = 'trunk'
ODL_TRUNKS = 'trunks'
ODL_L2GATEWAY = 'l2_gateway'
ODL_L2GATEWAYS = 'l2_gateways'
ODL_L2GATEWAY_CONNECTION = 'l2gateway_connection'
ODL_L2GATEWAY_CONNECTIONS = 'l2_gateway_connections'
ODL_BGPVPN = 'bgpvpn'
ODL_BGPVPNS = 'bgpvpns'
ODL_BGPVPN_NETWORK_ASSOCIATION = 'bgpvpn_network_association'
ODL_BGPVPN_NETWORK_ASSOCIATIONS = 'bgpvpn_network_associations'
ODL_BGPVPN_ROUTER_ASSOCIATION = 'bgpvpn_router_association'
ODL_BGPVPN_ROUTER_ASSOCIATIONS = 'bgpvpn_router_associations'
ODL_ML2_MECH_DRIVER_V2 = "opendaylight_v2"
ODL_DEVICE_ID_START = 'OpenDaylight'
ODL_CREATE = 'create'
ODL_UPDATE = 'update'
ODL_DELETE = 'delete'
PENDING = 'pending'
PROCESSING = 'processing'
FAILED = 'failed'
COMPLETED = 'completed'
BEFORE_COMPLETE = 'before_complete'
AFTER_DELETE = 'after_delete'
RESOURCE_URL_MAPPINGS = {
    ODL_QOS_POLICY: "%s/%s" % (ODL_QOS, ODL_QOS_POLICIES),
    ODL_SFC_FLOW_CLASSIFIER: "%s/%s" % (ODL_SFC, ODL_SFC_FLOW_CLASSIFIERS),
    ODL_SFC_PORT_CHAIN: "%s/%s" % (ODL_SFC, ODL_SFC_PORT_CHAINS),
    ODL_SFC_PORT_PAIR: "%s/%s" % (ODL_SFC, ODL_SFC_PORT_PAIRS),
    ODL_SFC_PORT_PAIR_GROUP: "%s/%s" % (ODL_SFC, ODL_SFC_PORT_PAIR_GROUPS)
} LOG = logging.getLogger(__name__)
def get_pending_or_processing_ops(context, object_uuid, operation=None):
def get_pending_delete_ops_with_parent(context, object_type, parent_id):
def get_all_db_rows(context):
def get_all_db_rows_by_state(context, state):
def get_oldest_pending_db_row_with_lock(context):
def delete_dependency(context, entry):
def update_db_row_state(context, row, state, flush=True):
def update_pending_db_row_retry(context, row, retry_count):
def delete_row(context, row=None, row_id=None, flush=True):
def create_pending_row(context, object_type, object_uuid,
                       operation, data, depending_on=None):
def delete_pending_rows(context, operations_to_delete):
def _update_periodic_task_state(context, expected_state, state, task):
def was_periodic_task_executed_recently(context, task, interval):
def lock_periodic_task(context, task):
def unlock_periodic_task(context, task):
def update_periodic_task(context, task, operation=None):
def delete_rows_by_state_and_time(context, state, time_delta):
def reset_processing_rows(context, max_timedelta): class OpenDaylightJournal(model_base.BASEV2):
class OpenDaylightPeriodicTask(model_base.BASEV2): LOG = logging.getLogger(__name__)
def delete_completed_rows(context):
def cleanup_processing_rows(context): def _get_delete_dependencies(context, object_type, object_uuid):
def _get_older_operations(context, object_ids):
def _generate_subnet_deps(data):
def _generate_port_deps(data):
def _generate_network_deps(data):
def _generate_sg_rule_deps(data):
def _generate_router_deps(data):
def _generate_floatingip_deps(data):
def _generate_trunk_deps(data):
def _generate_l2gateway_connection_deps(data):
def _generate_sfc_port_pair_deps(data):
def _generate_sfc_port_pair_group_deps(data):
def _generate_sfc_port_chain_deps(data):
def _generate_bgpvpn_deps(data):
def calculate(context, operation, object_type, object_uuid, data):
_CREATE_OR_UPDATE_DEP_GENERATOR = {
    odl_const.ODL_NETWORK: _generate_network_deps,
    odl_const.ODL_SUBNET: _generate_subnet_deps,
    odl_const.ODL_PORT: _generate_port_deps,
    # TODO(yamahata): dependency between SG and PORT
    odl_const.ODL_SG_RULE: _generate_sg_rule_deps,
    odl_const.ODL_ROUTER: _generate_router_deps,
    odl_const.ODL_FLOATINGIP: _generate_floatingip_deps,
    odl_const.ODL_TRUNK: _generate_trunk_deps,
    odl_const.ODL_L2GATEWAY_CONNECTION: _generate_l2gateway_connection_deps,
    odl_const.ODL_SFC_PORT_PAIR: _generate_sfc_port_pair_deps,
    odl_const.ODL_SFC_PORT_PAIR_GROUP: _generate_sfc_port_pair_group_deps,
    odl_const.ODL_SFC_PORT_CHAIN: _generate_sfc_port_chain_deps,
    odl_const.ODL_BGPVPN: _generate_bgpvpn_deps,
}
_DELETE_DEPENDENCIES = {
    odl_const.ODL_NETWORK: (odl_const.ODL_SUBNET, odl_const.ODL_PORT,
                            odl_const.ODL_ROUTER,
                            odl_const.ODL_L2GATEWAY_CONNECTION,
                            odl_const.ODL_BGPVPN),
    odl_const.ODL_SUBNET: (odl_const.ODL_PORT,),
    odl_const.ODL_ROUTER: (odl_const.ODL_PORT, odl_const.ODL_FLOATINGIP,
                           odl_const.ODL_BGPVPN),
    odl_const.ODL_PORT: (odl_const.ODL_TRUNK,),
    # TODO(yamahata): dependency between SG and PORT
    odl_const.ODL_SG: (odl_const.ODL_SG_RULE,),
    odl_const.ODL_L2GATEWAY: (odl_const.ODL_L2GATEWAY_CONNECTION,),
    odl_const.ODL_SFC_FLOW_CLASSIFIER: (odl_const.ODL_SFC_PORT_CHAIN,),
    odl_const.ODL_SFC_PORT_PAIR: (odl_const.ODL_SFC_PORT_PAIR_GROUP,),
    odl_const.ODL_SFC_PORT_PAIR_GROUP: (odl_const.ODL_SFC_PORT_CHAIN,),
    odl_const.ODL_QOS_POLICY: (odl_const.ODL_PORT, odl_const.ODL_NETWORK),
} @db_api.retry_if_session_inactive()
@db_api.CONTEXT_WRITER.savepoint
def full_sync(context):
    if not _full_sync_needed(context):
        return

    db.delete_pending_rows(context, _OPS_TO_DELETE_ON_SYNC)

    for resource_type in _ORDERED_ODL_RESOURCES:
        handler = FULL_SYNC_RESOURCES.get(resource_type)
        if handler:
            _sync_resources(context, resource_type, handler)

    journal.record(context, odl_const.ODL_NETWORK, _CANARY_NETWORK_ID,
                   odl_const.ODL_CREATE, _CANARY_NETWORK_DATA) LOG = logging.getLogger(__name__)
MAKE_URL = {}
LOG_ENTRY_TEMPLATE = ("%(log_type)s (Entry ID: %(entry_id)s) - %(op)s "
                      "%(obj_type)s %(obj_id)s (Time stamp: %(timestamp)s)")
LOG_RECORDED = 'Recorded'
LOG_PROCESSING = 'Processing'
LOG_COMPLETED = 'Completed'
LOG_ERROR_PROCESSING = 'Error while processing'
    _RETRY_SLEEP_MIN = 0.1
    _RETRY_SLEEP_MAX = 60
def call_thread_on_end(func):
    def new_func(obj, *args, **kwargs):
def _enrich_port(plugin_context, ml2_context, object_type, operation, data):
def _log_entry(log_type, entry, log_level=logging.INFO, **kwargs):
def record(plugin_context, object_type, object_uuid, operation, data,
           ml2_context=None):
def entry_complete(context, entry):
def entry_reset(context, entry):
def entry_update_state_by_retry_count(context, entry, retry_count):
def _make_url(row):
def register_url_builder(object_type, method):
def _build_url(row):
    def __init__(self, start_thread=True):
    def _create_odl_sync_thread(self):
    def start(self):
    def stop(self, timeout=None):
    def set_sync_event(self):
    def _json_data(row):
    def run_sync_thread(self):
    def sync_pending_entries(self):
    def _retry_sleep(self):
    def _retry_reset(self):
    def _sync_entry(self, context, entry):
class OpenDaylightJournalThread(object): LOG = logging.getLogger(__name__)
class PeriodicTask(object):
    def __init__(self, task, interval):
    def start(self):
    def cleanup(self):
    def _set_operation(self, context, operation):
    def _execute_op(self, operation, context):
    def task_already_executed_recently(self, context):
    def _clear_and_unlock_task(self, context):
    def _lock_task(self, context):
    def execute_ops(self, forced=False):
    def register_operation(self, phase): _CLIENT = client.OpenDaylightRestClientGlobal()
LOG = logging.getLogger(__name__)
def journal_recovery(context):
def get_latest_resource(context, row):
def _get_latest_resource(context, row):
def _sync_resource_to_odl(context, row, operation_type, exists_on_odl):
def _handle_existing_resource(context, row):
def _handle_non_existing_resource(context, row): LOG = logging.getLogger(__name__)
class JournalPeriodicProcessor(worker.BaseWorker):
    def __init__(self):
    def _create_pidfile(self):
    def _delete_pidfile(self):
    def start(self):
    def stop(self):
    def wait(self):
    def reset(self):
    def _call_journal(self):
    def _start_maintenance_task(self): class DietTestCase(base.DietTestCase):
class OpenDaylightRestClientFixture(fixtures.Fixture):
class OpenDaylightRestClientGlobalFixture(fixtures.Fixture):
class OpenDaylightFeaturesFixture(fixtures.Fixture):
class OpenDaylightJournalThreadFixture(fixtures.Fixture):
class JournalWorkerPidFileFixture(fixtures.Fixture):
class OpenDaylightPeriodicTaskFixture(fixtures.Fixture):
class OpenDaylightPseudoAgentPrePopulateFixture(
        nl_fixture.CallbackRegistryFixture):
class OpenDaylightFullSyncFixture(fixtures.Fixture):
    def patch(self, target, name, *args, **kwargs):
    def _setUp(self):
    def __init__(self, global_client):
    def _setUp(self):
    def _setUp(self):
    def _setUp(self):
    def remock_atexit(self):
    def _setUp(self):
    def _setUp(self):
    def _setUp(self):
    def _restore(self):
    def _setUp(self): class OpenDaylightConfigBase(test_plugin.Ml2PluginV2TestCase,
                             test_base_db.ODLBaseDbTestCase):
class OpenDaylightTestCase(OpenDaylightConfigBase):
    def setUp(self):
    def run_journal_processing(self):
    def setUp(self):
    def check_sendjson(self, method, urlpath, obj): def in_session(fn):
    def wrapper(self, *args, **kwargs):
    def setUp(self):
    def _update_row(self, row):
    def _test_validate_updates(self, first_entry, second_entry, expected_deps,
                               state=None):
    def _test_retry_count(self, retry_num, max_retry,
                          expected_retry_count, expected_state):
    def _test_retry_wrapper(self, decorated_function):
    def _decorated_retry_if_session_inactive(self, context):
    def test_retry_if_session_inactive(self):
    def _test_update_row_state(self, from_state, to_state, dry_flush=False):
    def test_updates_same_object_uuid(self):
    def test_validate_updates_different_object_uuid(self):
    def test_validate_updates_different_object_type(self):
    def test_check_for_older_ops_processing(self):
    def test_get_oldest_pending_row_none_when_no_rows(self):
    def _test_get_oldest_pending_row_none(self, state):
    def test_get_oldest_pending_row_none_when_row_processing(self):
    def test_get_oldest_pending_row_none_when_row_failed(self):
    def test_get_oldest_pending_row_none_when_row_completed(self):
    def test_get_oldest_pending_row(self):
    def test_get_oldest_pending_row_order(self):
    def _test_get_oldest_pending_row_with_dep(self, dep_state):
    def test_get_oldest_pending_row_when_dep_completed(self):
    def test_get_oldest_pending_row_when_dep_failed(self):
    def test_get_oldest_pending_row_returns_parent_when_dep_pending(self):
    def test_get_oldest_pending_row_none_when_dep_processing(self):
    def test_get_oldest_pending_row_retries_exceptions(self):
    def _test_delete_row(self, by_row=False, by_row_id=False, dry_flush=False):
    def test_delete_row_by_row(self):
    def test_delete_row_by_row_id(self):
    def test_delete_row_by_row_without_flushing(self):
    def test_create_pending_row(self):
    def _test_delete_rows_by_state_and_time(self, last_retried, row_retention,
                                            state, expected_rows,
                                            dry_delete=False):
    def test_delete_completed_rows_no_new_rows(self):
    def test_delete_completed_rows_one_new_row(self):
    def test_delete_completed_rows_wrong_state(self):
    def test_delete_completed_rows_individually(self):
    def test_delete_completed_rows_without_flush(self, mock_delete_row):
    def _test_reset_processing_rows(self, last_retried, max_timedelta,
                                    quantity, dry_reset=False):
    def test_reset_processing_rows(self):
    def test_reset_processing_rows_no_new_rows(self):
    def test_reset_processing_rows_individually(self, mock_update_row):
    def test_valid_retry_count(self):
    def test_invalid_retry_count(self):
    def test_update_row_state_to_pending(self):
    def test_update_row_state_to_processing(self):
    def test_update_row_state_to_failed(self):
    def test_update_row_state_to_completed(self):
    def test_update_row_state_to_status_without_flush(self):
    def _test_periodic_task_lock_unlock(self, db_func, existing_state,
                                        expected_state, expected_result,
                                        task='test_task'):
    def test_lock_periodic_task(self):
    def test_lock_periodic_task_fails_when_processing(self):
    def test_unlock_periodic_task(self):
    def test_unlock_periodic_task_fails_when_pending(self):
    def test_multiple_row_tasks(self):
    def _add_tasks(self, tasks):
    def _perform_ops_on_all_rows(self, tasks, to_lock):
    def test_multiple_row_tasks_lock_unlock(self):
class DbTestCase(test_base_db.ODLBaseDbTestCase):
import os
import signal
import fixtures
from unittest import mock
from neutron.common import utils
from oslo_db import exception
from oslo_log import log as logging
from oslo_service.tests import test_service
from oslo_utils import uuidutils
from networking_odl.common import client
from networking_odl.common import constants as odl_const
from networking_odl.db import db
from networking_odl.db import models
from networking_odl.journal import cleanup
from networking_odl.journal import dependency_validations
from networking_odl.journal import full_sync
from networking_odl.journal import journal
from networking_odl.journal import periodic_task
from networking_odl.journal import recovery
from networking_odl.journal import worker
from networking_odl.tests import base
from networking_odl.tests.unit import base_v2
from networking_odl.tests.unit.db import test_db
        periodic_processor._delete_pidfile()

        self.assertFalse(os.path.isfile(pidfile))

        atexit_mock.assert_called_once_with(periodic_processor._delete_pidfile)

    def test_atexit_delete_pidfile_registered_only_once(self):
        atexit_mock = self.journal_thread_fixture.remock_atexit()
        periodic_processor = self._create_periodic_processor()

        for _ in range(0, 2):
            periodic_processor.start()
            periodic_processor.stop()

        atexit_mock.assert_called_once()


class OpenDaylightJournalThreadTest(base_v2.OpenDaylightTestCase):
    def setUp(self):
        super(OpenDaylightJournalThreadTest, self).setUp()
        self.journal = journal.OpenDaylightJournalThread()
        self.addCleanup(self.cleanup)

    @staticmethod
    def cleanup():
        journal.MAKE_URL.clear()

    def test_json_data(self):
        object_type = 'testobject'
        data = 'testdata'
        row = models.OpenDaylightJournal(object_type=object_type,
                                         object_uuid=uuidutils.generate_uuid(),
                                         operation=odl_const.ODL_CREATE,
                                         data=data)

        self.assertEqual("%ss" % object_type, self.journal._json_data(row)[1])

    def test_json_data_customized_url(self):
        object_type = 'randomtestobject'
        data = 'testdata'
        journal.register_url_builder(object_type, lambda row: row.object_type)
        row = models.OpenDaylightJournal(object_type=object_type,
                                         object_uuid=uuidutils.generate_uuid(),
                                         operation=odl_const.ODL_CREATE,
                                         data=data)

        url_param = self.journal._json_data(row)
        self.assertEqual(object_type, url_param[1])

    def test_entry_reset_retries_exceptions(self):
        with mock.patch.object(db, 'update_db_row_state') as m:
            self._test_retry_exceptions(journal.entry_reset, m)

    @test_db.in_session
    @mock.patch.object(client.OpenDaylightRestClient, 'sendjson',
                       mock.Mock(side_effect=Exception))
    def test__sync_entry_update_state_by_retry_count_on_exception(self):
        entry = db.create_pending_row(self.db_context, *self.UPDATE_ROW)
        self.journal._max_retry_count = 1
        self.assertEqual(entry.retry_count, 0)
        self.journal._sync_entry(self.db_context, entry)
        self.assertEqual(entry.retry_count, 1)
        self.assertEqual(entry.state, odl_const.PENDING)
        self.journal._sync_entry(self.db_context, entry)
        self.assertEqual(entry.retry_count, 1)
        self.assertEqual(entry.state, odl_const.FAILED)

    def _test__sync_entry_logs(self, log_type):
        entry = db.create_pending_row(self.db_context, *self.UPDATE_ROW)
        logger = self.useFixture(fixtures.FakeLogger())

        self.journal._sync_entry(self.db_context, entry)

        self.assertIn(log_type, logger.output)

    def test__sync_entry_logs_processing(self):
        self._test__sync_entry_logs(journal.LOG_PROCESSING)

    def test__sync_entry_logs_completed(self):
        self._test__sync_entry_logs(journal.LOG_COMPLETED)

    @mock.patch.object(client.OpenDaylightRestClient, 'sendjson',
                       mock.Mock(side_effect=Exception))
    def test__sync_entry_logs_failed(self):
        self._test__sync_entry_logs(journal.LOG_ERROR_PROCESSING)

    @mock.patch.object(journal.OpenDaylightJournalThread,
                       'sync_pending_entries')
    def test_terminate_journal_thread_correctly(self, mock_journal):
        self.journal_thread_fixture.journal_thread_mock.stop()
        self.addCleanup(self.journal_thread_fixture.journal_thread_mock.start)

        journal_thread = journal.OpenDaylightJournalThread(start_thread=True)

        journal_thread.stop(5)
        self.assertTrue(not journal_thread._odl_sync_thread.is_alive())
        mock_journal.assert_called_once()

    @mock.patch.object(journal.OpenDaylightJournalThread,
                       'sync_pending_entries')
    def test_allow_multiple_starts_gracefully(self, mock_journal):
        self.journal_thread_fixture.journal_thread_mock.stop()
        self.addCleanup(self.journal_thread_fixture.journal_thread_mock.start)
        journal_thread = journal.OpenDaylightJournalThread(start_thread=False)
        self.addCleanup(journal_thread.stop)
        journal_thread.start()

        try:
            journal_thread.start()
        except RuntimeError:
            self.fail('OpenDaylightJournalThread started twice')


def _raise_DBReferenceError(*args, **kwargs):
    args = [mock.Mock(unsafe=True)] * 4
    e = exception.DBReferenceError(*args)
    raise e


class JournalTest(base_v2.OpenDaylightTestCase):

    @mock.patch.object(dependency_validations, 'calculate')