# All rights reserved.
#
#    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.
#

from neutron_lib import context as nctx
from neutron_lib.db import api as db_api
from neutron_lib.plugins import constants
from neutron_lib.plugins import directory
from oslo_config import cfg
from oslo_db import exception as db_exc
from oslo_utils import uuidutils
from sqlalchemy.orm import session as se
from webob import exc

from neutron.common import utils
from neutron.db import models_v2
from neutron.objects import ports as port_obj
from neutron.tests.unit.plugins.ml2 import test_plugin


class TestRevisionPlugin(test_plugin.Ml2PluginV2TestCase):

    l3_plugin = ('neutron.tests.unit.extensions.test_extraroute.'
                 'TestExtraRouteL3NatServicePlugin')

    _extension_drivers = ['qos']

    def get_additional_service_plugins(self):
        p = super(TestRevisionPlugin, self).get_additional_service_plugins()
        p.update({'revision_plugin_name': 'revisions',
                  'qos_plugin_name': 'qos',
                  'tag_name': 'tag'})
        return p

    def setUp(self):
        cfg.CONF.set_override('extension_drivers',
                              self._extension_drivers,
                              group='ml2')
        super(TestRevisionPlugin, self).setUp()
        self.cp = directory.get_plugin()
        self.l3p = directory.get_plugin(constants.L3)
        self._ctx = nctx.get_admin_context()
        self._tenant_id = uuidutils.generate_uuid()

    @property
    def ctx(self):
        # TODO(kevinbenton): return ctx without expire_all after switch to
        # enginefacade complete. We expire_all here because the switch to
        # the new engine facade is resulting in changes being spread over
        # other sessions so we can end up getting stale reads in the parent
        # session if objects remain in the identity map.
        if not utils.is_session_active(self._ctx.session):
            self._ctx.session.expire_all()
        return self._ctx

    def test_handle_expired_object(self):
        rp = directory.get_plugin('revision_plugin')
        with self.port():
            with db_api.CONTEXT_WRITER.using(self.ctx):
                ipal_objs = port_obj.IPAllocation.get_objects(self.ctx)
                if not ipal_objs:
                    raise Exception("No IP allocations available.")
                ipal_obj = ipal_objs[0]
                # load port into our session
                port = self.ctx.session.query(models_v2.Port).one()
                # simulate concurrent delete in another session
                other_ctx = nctx.get_admin_context()
                other_ctx.session.delete(
                    other_ctx.session.query(models_v2.Port).first()
                )
                other_ctx.session.flush()

                # ensure no attribute lookups are attempted on an
                # object deleted from the session when doing related
                # bumps
                self.ctx.session.expire(port)

                collected = rp._collect_related_tobump(
                    self.ctx.session, [ipal_obj], set())
                rp._bump_obj_revisions(
                    self.ctx.session, collected, version_check=False)

    def test_shared_network_create(self):
        # this test intends to run db_base_plugin_v2 -> create_network_db,
        # which in turn creates a Network and then a NetworkRBAC object.
        # An issue was observed with the revision_plugin which would interfere
        # with the flush process that occurs with these two connected objects,
        # creating two copies of the Network object in the Session and putting
        # it into an invalid state.
        with self.network(shared=True):