websocket test

.editorconfig

@@ -0,0 +1,16 @@
+# EditorConfig <https://EditorConfig.org>
+root = true
+
+# elementary defaults
+[*]
+charset = utf-8
+end_of_line = lf
+indent_size = tab
+indent_style = space
+insert_final_newline = false
+max_line_length = 80
+tab_width = 4
+
+# Markup files
+[{*.html,*.xml,*.yaml,*.yml,*.json}]
+tab_width = 2

.github/workflows/python-package.yml

@@ -28,7 +28,8 @@ jobs:
       run: |
         python -m pip install --upgrade pip
         pip install -U flake8 setuptools
-        pip install -U pytest pytest-doctestplus pytest-pylint pytest-mypy requests openapi-core uwsgi simplejson
+        pip install -U openapi-core uwsgi simplejson WSocket
+        pip install -U pytest pytest-doctestplus pytest-pylint pytest-mypy requests websocket-client
         pip install -U types-simplejson types-requests types-PyYAML
     - name: Lint with flake8
       run: |

.gitignore

@@ -0,0 +1,5 @@
+PoorWSGI.egg-info/
+dist/
+build/
+.eggs/
+tags

tests_integrity/test_websocket.py

@@ -0,0 +1,64 @@
+"""HTTP Digest example test."""
+from os import environ
+from os.path import dirname, join, pardir
+from uuid import uuid1
+from base64 import encodebytes
+
+from pytest import fixture
+# websocket-client
+from websocket import WebSocket  # type: ignore
+
+from . support import start_server, check_url
+
+# pylint: disable=inconsistent-return-statements
+# pylint: disable=missing-function-docstring
+# pylint: disable=redefined-outer-name
+# pylint: disable=no-self-use
+
+
+@fixture(scope="module")
+def server(request):
+    value = environ.get("TEST_SIMPLE_URL", "").strip('/')
+    if value:
+        return value
+
+    process = start_server(
+        request,
+        join(dirname(__file__), pardir, 'examples/websocket.py'))
+
+    yield "localhost:8080"  # server is running
+    process.kill()
+    process.wait()
+
+
+@fixture(scope="module")
+def http_url(server):
+    return f"http://{server}"
+
+
+@fixture(scope="module")
+def ws_url(server):
+    return f"ws://{server}"
+
+
+class TestWebSocket:
+    """Test for WebSocket example."""
+
+    def test_upgrade(self, http_url):
+        uuid = uuid1().bytes
+        check_url(http_url+"/ws", status_code=101,
+                  headers={"Connection": "Upgrade",
+                           "Upgrade": "websocket",
+                           "Sec-WebSocket-Version": "13",
+                           "Sec-WebSocket-Key":
+                           encodebytes(uuid).decode().strip()})
+
+    def test_websocket(self, ws_url):
+        wsck = WebSocket()
+        wsck.connect(ws_url+"/ws")
+        msg = wsck.recv()
+        assert msg.endswith("Hello")
+        wsck.send("Test")
+        msg = wsck.recv()
+        assert msg.endswith("<b>Test</b>")
+        wsck.close()