ryu/build/lib.linux-armv7l-2.7/ryu/tests/unit/app/test_ws_topology.py

   1 # Copyright (C) 2013 Stratosphere Inc.
   2 #
   3 # Licensed under the Apache License, Version 2.0 (the "License");
   4 # you may not use this file except in compliance with the License.
   5 # You may obtain 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,
  11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
  12 # implied.
  13 # See the License for the specific language governing permissions and
  14 # limitations under the License.
  15
  16 # vim: tabstop=4 shiftwidth=4 softtabstop=4
  17
  18 import unittest
  19 from socket import error as SocketError
  20
  21 import mock
  22
  23 from ryu.app.ws_topology import WebSocketTopology
  24
  25
  26 class Test_ws_topology(unittest.TestCase):
  27
  28     def test_when_sock_error(self):
  29         args = {
  30             'wsgi': mock.Mock(),
  31         }
  32         app = WebSocketTopology(**args)
  33
  34         rpc_client_mock1 = mock.Mock()
  35         config = {
  36             'get_proxy.return_value.event_link_add.side_effect': SocketError,
  37         }
  38         rpc_client_mock1.configure_mock(**config)
  39
  40         rpc_client_mock2 = mock.Mock()
  41
  42         app.rpc_clients = [
  43             rpc_client_mock1,
  44             rpc_client_mock2,
  45         ]
  46
  47         ev_mock = mock.Mock()
  48         app._event_link_add_handler(ev_mock)
  49
  50         rpc_client_mock1.get_proxy.assert_called_once_with()
  51         rpc_client_mock2.get_proxy.assert_called_once_with()
  52
  53
  54 if __name__ == "__main__":
  55     unittest.main()