neutron/tests/unit/api/rpc/handlers/test_securitygroups_rpc.py

   1 # All Rights Reserved.
   2 #
   3 #    Licensed under the Apache License, Version 2.0 (the "License"); you may
   4 #    not use this file except in compliance with the License. You may obtain
   5 #    a copy of the License at
   6 #
   7 #         http://www.apache.org/licenses/LICENSE-2.0
   8 #
   9 #    Unless required by applicable law or agreed to in writing, software
  10 #    distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
  11 #    WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
  12 #    License for the specific language governing permissions and limitations
  13 #    under the License.
  14
  15 import mock
  16
  17 from neutron.api.rpc.handlers import securitygroups_rpc
  18 from neutron.tests import base
  19
  20
  21 class SecurityGroupServerRpcApiTestCase(base.BaseTestCase):
  22
  23     def test_security_group_rules_for_devices(self):
  24         rpcapi = securitygroups_rpc.SecurityGroupServerRpcApi('fake_topic')
  25
  26         with mock.patch.object(rpcapi.client, 'call') as rpc_mock,\
  27                 mock.patch.object(rpcapi.client, 'prepare') as prepare_mock:
  28             prepare_mock.return_value = rpcapi.client
  29             rpcapi.security_group_rules_for_devices('context', ['fake_device'])
  30
  31             rpc_mock.assert_called_once_with(
  32                     'context',
  33                     'security_group_rules_for_devices',
  34                     devices=['fake_device'])
  35
  36
  37 class SGAgentRpcCallBackMixinTestCase(base.BaseTestCase):
  38
  39     def setUp(self):
  40         super(SGAgentRpcCallBackMixinTestCase, self).setUp()
  41         self.rpc = securitygroups_rpc.SecurityGroupAgentRpcCallbackMixin()
  42         self.rpc.sg_agent = mock.Mock()
  43
  44     def test_security_groups_rule_updated(self):
  45         self.rpc.security_groups_rule_updated(None,
  46                                               security_groups=['fake_sgid'])
  47         self.rpc.sg_agent.assert_has_calls(
  48             [mock.call.security_groups_rule_updated(['fake_sgid'])])
  49
  50     def test_security_groups_member_updated(self):
  51         self.rpc.security_groups_member_updated(None,
  52                                                 security_groups=['fake_sgid'])
  53         self.rpc.sg_agent.assert_has_calls(
  54             [mock.call.security_groups_member_updated(['fake_sgid'])])
  55
  56     def test_security_groups_provider_updated(self):
  57         self.rpc.security_groups_provider_updated(None)
  58         self.rpc.sg_agent.assert_has_calls(
  59             [mock.call.security_groups_provider_updated(None)])