# Path: networking_odl/db/db.py
# 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):
# 
# Path: networking_odl/journal/cleanup.py
# LOG = logging.getLogger(__name__)
# def delete_completed_rows(context):
# def cleanup_processing_rows(context):
# 
# Path: networking_odl/tests/unit/test_base_db.py
# RETRIABLE_EXCEPTIONS = (db_exc.DBDeadlock, exc.StaleDataError,
#                         db_exc.DBConnectionError, db_exc.DBDuplicateEntry,
#                         db_exc.RetryRequest)
# RETRY_INTERVAL = 0.001
# RETRY_MAX = 2
#     UPDATE_ROW = [constants.ODL_NETWORK, 'id', constants.ODL_UPDATE,
#                   {'test': 'data'}]
# class _InnerException(Exception):
# class ODLBaseDbTestCase(SqlTestCaseLight):
#         class RetryTracker(object):
#     def setUp(self):
#     def _setup_retry_tracker_table(self):
#     def _db_cleanup(self):
#     def _test_db_exceptions_handled(self, method, mock_object, expect_retries):
#         def _assertRaises(exceptions, method, context, *args, **kwargs):
#             def increase_retry_counter_and_except(*args, **kwargs):
#     def _assertRetryCount(self, expected_count):
#     def _test_retry_exceptions(self, method, mock_object,
#                                assert_transaction=True):


from unittest import mock
from networking_odl.db import db
from networking_odl.journal import cleanup
from networking_odl.tests.unit import test_base_db
# Copyright (C) 2017 Red Hat, Inc.
#
#  Licensed under the Apache License, Version 2.0 (the "License"); you may
#  not use this file except in compliance with the License. You may obtain
#  a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#  License for the specific language governing permissions and limitations
#  under the License.
#




class JournalCleanupTestCase(test_base_db.ODLBaseDbTestCase):
    def test_delete_completed_rows_retries_exceptions(self):
        self.cfg.config(completed_rows_retention=1, group='ml2_odl')
        with mock.patch.object(db, 'delete_rows_by_state_and_time') as m:
            self._test_retry_exceptions(
                cleanup.delete_completed_rows, m)

    def test_cleanup_processsing_rows_retries_exceptions(self):
        with mock.patch.object(db, 'reset_processing_rows') as m:
            self._test_retry_exceptions(
                cleanup.cleanup_processing_rows, m)


# Path: networking_odl/db/db.py
# 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):
# 
# Path: networking_odl/journal/cleanup.py
# LOG = logging.getLogger(__name__)
# def delete_completed_rows(context):
# def cleanup_processing_rows(context):
# 
# Path: networking_odl/tests/unit/test_base_db.py
# RETRIABLE_EXCEPTIONS = (db_exc.DBDeadlock, exc.StaleDataError,
#                         db_exc.DBConnectionError, db_exc.DBDuplicateEntry,
#                         db_exc.RetryRequest)
# RETRY_INTERVAL = 0.001
# RETRY_MAX = 2
#     UPDATE_ROW = [constants.ODL_NETWORK, 'id', constants.ODL_UPDATE,
#                   {'test': 'data'}]
# class _InnerException(Exception):
# class ODLBaseDbTestCase(SqlTestCaseLight):
#         class RetryTracker(object):
#     def setUp(self):
#     def _setup_retry_tracker_table(self):
#     def _db_cleanup(self):
#     def _test_db_exceptions_handled(self, method, mock_object, expect_retries):
#         def _assertRaises(exceptions, method, context, *args, **kwargs):
#             def increase_retry_counter_and_except(*args, **kwargs):
#     def _assertRetryCount(self, expected_count):
#     def _test_retry_exceptions(self, method, mock_object,
#                                assert_transaction=True):


# Path: networking_odl/tests/unit/journal/test_cleanup.py
from unittest import mock
from networking_odl.db import db
from networking_odl.journal import cleanup
from networking_odl.tests.unit import test_base_db

# Copyright (C) 2017 Red Hat, Inc.
#
#  Licensed under the Apache License, Version 2.0 (the "License"); you may
#  not use this file except in compliance with the License. You may obtain
#  a copy of the License at
#
#       http://www.apache.org/licenses/LICENSE-2.0
#
#  Unless required by applicable law or agreed to in writing, software
#  distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
#  WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
#  License for the specific language governing permissions and limitations
#  under the License.
#




class JournalCleanupTestCase(test_base_db.ODLBaseDbTestCase):
    def test_delete_completed_rows_retries_exceptions(self):
        self.cfg.config(completed_rows_retention=1, group='ml2_odl')
        with mock.patch.object(db, 'delete_rows_by_state_and_time') as m:
            self._test_retry_exceptions(
                cleanup.delete_completed_rows, m)

    def test_cleanup_processsing_rows_retries_exceptions(self):
        with mock.patch.object(db, 'reset_processing_rows') as m:
            self._test_retry_exceptions(
                cleanup.cleanup_processing_rows, m)


    @mock.patch.object(db, 'delete_rows_by_state_and_time')