Update websocket-client-1.2.1

lib/websocket/tests/test_app.py

@@ -0,0 +1,178 @@
+# -*- coding: utf-8 -*-
+#
+"""
+test_app.py
+websocket - WebSocket client library for Python
+
+Copyright 2021 engn33r
+
+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.
+"""
+
+import os
+import os.path
+import websocket as ws
+import sys
+import ssl
+import unittest
+sys.path[0:0] = [""]
+
+# Skip test to access the internet.
+TEST_WITH_INTERNET = os.environ.get('TEST_WITH_INTERNET', '0') == '1'
+TRACEABLE = True
+
+
+class WebSocketAppTest(unittest.TestCase):
+
+    class NotSetYet(object):
+        """ A marker class for signalling that a value hasn't been set yet.
+        """
+
+    def setUp(self):
+        ws.enableTrace(TRACEABLE)
+
+        WebSocketAppTest.keep_running_open = WebSocketAppTest.NotSetYet()
+        WebSocketAppTest.keep_running_close = WebSocketAppTest.NotSetYet()
+        WebSocketAppTest.get_mask_key_id = WebSocketAppTest.NotSetYet()
+
+    def tearDown(self):
+        WebSocketAppTest.keep_running_open = WebSocketAppTest.NotSetYet()
+        WebSocketAppTest.keep_running_close = WebSocketAppTest.NotSetYet()
+        WebSocketAppTest.get_mask_key_id = WebSocketAppTest.NotSetYet()
+
+    @unittest.skipUnless(TEST_WITH_INTERNET, "Internet-requiring tests are disabled")
+    def testKeepRunning(self):
+        """ A WebSocketApp should keep running as long as its self.keep_running
+        is not False (in the boolean context).
+        """
+
+        def on_open(self, *args, **kwargs):
+            """ Set the keep_running flag for later inspection and immediately
+            close the connection.
+            """
+            self.send("hello!")
+            WebSocketAppTest.keep_running_open = self.keep_running
+            self.keep_running = False
+
+        def on_message(wsapp, message):
+            print(message)
+            self.close()
+
+        def on_close(self, *args, **kwargs):
+            """ Set the keep_running flag for the test to use.
+            """
+            WebSocketAppTest.keep_running_close = self.keep_running
+
+        app = ws.WebSocketApp('ws://127.0.0.1:8765', on_open=on_open, on_close=on_close, on_message=on_message)
+        app.run_forever()
+
+    @unittest.skipUnless(TEST_WITH_INTERNET, "Internet-requiring tests are disabled")
+    def testSockMaskKey(self):
+        """ A WebSocketApp should forward the received mask_key function down
+        to the actual socket.
+        """
+
+        def my_mask_key_func():
+            return "\x00\x00\x00\x00"
+
+        app = ws.WebSocketApp('wss://stream.meetup.com/2/rsvps', get_mask_key=my_mask_key_func)
+
+        # if numpy is installed, this assertion fail
+        # Note: We can't use 'is' for comparing the functions directly, need to use 'id'.
+        self.assertEqual(id(app.get_mask_key), id(my_mask_key_func))
+
+    @unittest.skipUnless(TEST_WITH_INTERNET, "Internet-requiring tests are disabled")
+    def testInvalidPingIntervalPingTimeout(self):
+        """ Test exception handling if ping_interval < ping_timeout
+        """
+
+        def on_ping(app, msg):
+            print("Got a ping!")
+            app.close()
+
+        def on_pong(app, msg):
+            print("Got a pong! No need to respond")
+            app.close()
+
+        app = ws.WebSocketApp('wss://api-pub.bitfinex.com/ws/1', on_ping=on_ping, on_pong=on_pong)
+        self.assertRaises(ws.WebSocketException, app.run_forever, ping_interval=1, ping_timeout=2, sslopt={"cert_reqs": ssl.CERT_NONE})
+
+    @unittest.skipUnless(TEST_WITH_INTERNET, "Internet-requiring tests are disabled")
+    def testPingInterval(self):
+        """ Test WebSocketApp proper ping functionality
+        """
+
+        def on_ping(app, msg):
+            print("Got a ping!")
+            app.close()
+
+        def on_pong(app, msg):
+            print("Got a pong! No need to respond")
+            app.close()
+
+        app = ws.WebSocketApp('wss://api-pub.bitfinex.com/ws/1', on_ping=on_ping, on_pong=on_pong)
+        app.run_forever(ping_interval=2, ping_timeout=1, sslopt={"cert_reqs": ssl.CERT_NONE})
+
+    @unittest.skipUnless(TEST_WITH_INTERNET, "Internet-requiring tests are disabled")
+    def testOpcodeClose(self):
+        """ Test WebSocketApp close opcode
+        """
+
+        app = ws.WebSocketApp('wss://tsock.us1.twilio.com/v3/wsconnect')
+        app.run_forever(ping_interval=2, ping_timeout=1, ping_payload="Ping payload")
+
+    @unittest.skipUnless(TEST_WITH_INTERNET, "Internet-requiring tests are disabled")
+    def testOpcodeBinary(self):
+        """ Test WebSocketApp binary opcode
+        """
+
+        app = ws.WebSocketApp('streaming.vn.teslamotors.com/streaming/')
+        app.run_forever(ping_interval=2, ping_timeout=1, ping_payload="Ping payload")
+
+    @unittest.skipUnless(TEST_WITH_INTERNET, "Internet-requiring tests are disabled")
+    def testBadPingInterval(self):
+        """ A WebSocketApp handling of negative ping_interval
+        """
+        app = ws.WebSocketApp('wss://api-pub.bitfinex.com/ws/1')
+        self.assertRaises(ws.WebSocketException, app.run_forever, ping_interval=-5, sslopt={"cert_reqs": ssl.CERT_NONE})
+
+    @unittest.skipUnless(TEST_WITH_INTERNET, "Internet-requiring tests are disabled")
+    def testBadPingTimeout(self):
+        """ A WebSocketApp handling of negative ping_timeout
+        """
+        app = ws.WebSocketApp('wss://api-pub.bitfinex.com/ws/1')
+        self.assertRaises(ws.WebSocketException, app.run_forever, ping_timeout=-3, sslopt={"cert_reqs": ssl.CERT_NONE})
+
+    @unittest.skipUnless(TEST_WITH_INTERNET, "Internet-requiring tests are disabled")
+    def testCloseStatusCode(self):
+        """ Test extraction of close frame status code and close reason in WebSocketApp
+        """
+        def on_close(wsapp, close_status_code, close_msg):
+            print("on_close reached")
+
+        app = ws.WebSocketApp('wss://tsock.us1.twilio.com/v3/wsconnect', on_close=on_close)
+        closeframe = ws.ABNF(opcode=ws.ABNF.OPCODE_CLOSE, data=b'\x03\xe8no-init-from-client')
+        self.assertEqual([1000, 'no-init-from-client'], app._get_close_args(closeframe))
+
+        closeframe = ws.ABNF(opcode=ws.ABNF.OPCODE_CLOSE, data=b'')
+        self.assertEqual([None, None], app._get_close_args(closeframe))
+
+        app2 = ws.WebSocketApp('wss://tsock.us1.twilio.com/v3/wsconnect')
+        closeframe = ws.ABNF(opcode=ws.ABNF.OPCODE_CLOSE, data=b'')
+        self.assertEqual([None, None], app2._get_close_args(closeframe))
+
+        self.assertRaises(ws.WebSocketConnectionClosedException, app.send, data="test if connection is closed")
+
+
+if __name__ == "__main__":
+    unittest.main()