x/quark
Code Issues Proposed changes

Merge "Remove ip_block create/delete notifications"

Browse Source
This commit is contained in:
Jenkins 2016-07-06 20:35:06 +00:00 committed by Gerrit Code Review
commit 7a906e1406
2 changed files with 0 additions and 79 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

15
quark/plugin_modules/subnets.py
View File

@ -15,13 +15,11 @@
import netaddr
from neutron.common import config as neutron_cfg
from neutron.common import rpc as n_rpc
from neutron import quota
from neutron_lib import exceptions as n_exc
from oslo_config import cfg
from oslo_log import log as logging
from oslo_utils import importutils
from oslo_utils import timeutils
from quark import allocation_pool
from quark.db import api as db_api
@ -225,12 +223,6 @@ def create_subnet(context, subnet):
subnet_dict = v._make_subnet_dict(new_subnet)
subnet_dict["gateway_ip"] = gateway_ip
n_rpc.get_notifier("network").info(
context,
"ip_block.create",
dict(tenant_id=subnet_dict["tenant_id"],
ip_block_id=subnet_dict["id"],
created_at=new_subnet["created_at"]))
return subnet_dict
@ -473,15 +465,8 @@ def delete_subnet(context, id):
# existence.
raise n_exc.SubnetNotFound(subnet_id=id)
payload = dict(tenant_id=subnet["tenant_id"],
ip_block_id=subnet["id"],
created_at=subnet["created_at"],
deleted_at=timeutils.utcnow())
_delete_subnet(context, subnet)
n_rpc.get_notifier("network").info(context, "ip_block.delete", payload)
def diagnose_subnet(context, id, fields):
if not context.is_admin:

64
quark/tests/plugin_modules/test_subnets.py
View File

@ -17,7 +17,6 @@ import contextlib
import copy
from datetime import datetime
import json
import time
import uuid
import mock
@ -1387,69 +1386,6 @@ class TestSubnetsQuotas(test_quark_plugin.TestQuarkPlugin):
"QUOTAS")
class TestSubnetsNotification(test_quark_plugin.TestQuarkPlugin):
@contextlib.contextmanager
def _stubs(self, s, deleted_at=None):
class FakeContext(object):
def __enter__(*args, **kwargs):
pass
def __exit__(*args, **kwargs):
pass
self.context.session.begin = FakeContext
s["network"] = models.Network()
s["network"]["created_at"] = s["created_at"]
subnet = models.Subnet(**s)
with contextlib.nested(
mock.patch("quark.plugin_modules.subnets.get_subnets"),
mock.patch("quark.db.api.subnet_find"),
mock.patch("quark.db.api.network_find"),
mock.patch("quark.db.api.subnet_create"),
mock.patch("quark.db.api.subnet_delete"),
mock.patch("neutron.common.rpc.get_notifier"),
mock.patch("neutron.quota.QUOTAS"),
mock.patch("oslo_utils.timeutils.utcnow"),
mock.patch("quark.plugin_modules.subnets._validate_subnet_cidr")
) as (get_subnets, sub_find, net_find, sub_create, sub_del, notify,
quota_engine, time_func, sub_validate):
sub_create.return_value = subnet
get_subnets.return_value = []
sub_find.return_value = subnet
time_func.return_value = deleted_at
yield notify
def test_create_subnet_notification(self):
s = dict(network_id=1, cidr="192.168.10.0/24",
tenant_id=1, id=1, created_at="123")
with self._stubs(s) as notify:
admin_ctx = self.context.elevated()
self.plugin.create_subnet(admin_ctx, dict(subnet=s))
notify.assert_called_once_with("network")
notify.return_value.info.assert_called_once_with(
admin_ctx,
"ip_block.create",
dict(tenant_id=s["tenant_id"],
ip_block_id=s["id"],
created_at=s["created_at"]))
def test_delete_subnet_notification(self):
now = time.strftime('%Y-%m-%d %H:%M:%S')
later = time.strftime('%Y-%m-%d %H:%M:%S')
s = dict(tenant_id=1, id=1, created_at=now)
with self._stubs(s, deleted_at=later) as notify:
self.plugin.delete_subnet(self.context, 1)
notify.assert_called_once_with("network")
notify.return_value.info.assert_called_once_with(
self.context,
"ip_block.delete",
dict(tenant_id=s["tenant_id"],
created_at=s["created_at"],
ip_block_id=s["id"],
deleted_at=later))
class TestQuarkDiagnoseSubnets(test_quark_plugin.TestQuarkPlugin):
@contextlib.contextmanager
def _stubs(self, subnets=None, routes=None):