Add tests for Python websocket server

Author: carafesicing0z

python_websocket_server/README.md

@@ -70,3 +70,11 @@ DeepSpeech service. The websocket timeout on the ingress is set to 1 hour.
 ## Contributing
 
 Bug reports and merge requests are welcome.
+
+### Running tests
+
+Navigate to the [`python_websocket_server`](python_websocket_server) directory. Then, run tests by executing:
+
+```
+python -m pytest tests/test_app.py
+```

python_websocket_server/application.conf

@@ -11,4 +11,6 @@ server {
   threadpool {
     count = 5
   }
+
+  stt_endpoint = "/api/v1/stt"
 }

python_websocket_server/deepspeech_server/app.py

@@ -27,7 +27,7 @@ async def healthcheck(_):
     return response.text("Welcome to DeepSpeech Server!")
 
 
-@app.websocket("/api/v1/stt")
+@app.websocket(conf['server.stt_endpoint'])
 async def stt(request, ws):
     logger.debug(f"Received {request.method} request at {request.path}")
     try:

python_websocket_server/requirements.txt

@@ -4,4 +4,7 @@ black>=20.8b1
 ffmpeg-python==0.2.0
 sanic==20.3.0
 pyhocon==0.3.54
-numpy~=1.17.0
+numpy~=1.19.3
+pytest-sanic==1.6.2
+click~=7.1.2
+pytest~=6.1.2

python_websocket_server/tests/experience_proves_this.wav

@@ -0,0 +1,37 @@
+import ast
+
+import pytest
+from pyhocon import ConfigFactory
+from sanic.websocket import WebSocketProtocol
+
+from deepspeech_server.app import app
+
+# Load app configs and initialize DeepSpeech model
+conf = ConfigFactory.parse_file("application.conf")
+
+
+@pytest.fixture
+def sanic_server(loop, sanic_client):
+    return loop.run_until_complete(sanic_client(app, protocol=WebSocketProtocol))
+
+
+def test_index_returns_200():
+    request, response = app.test_client.get("/")
+    assert response.status == 200
+
+
+async def test_valid_audio(sanic_server):
+    ws_conn = await sanic_server.ws_connect(conf['server.stt_endpoint'])
+    with open("tests/experience_proves_this.wav", mode="rb") as file:
+        audio = file.read()
+        await ws_conn.send_bytes(audio)
+        result = await ws_conn.receive()
+        assert ast.literal_eval(result.data)["text"] == "experience proves this"
+
+
+async def test_invalid_audio(sanic_server):
+    ws_conn = await sanic_server.ws_connect(conf['server.stt_endpoint'])
+    audio = b"this ain't audio"
+    await ws_conn.send_bytes(audio)
+    result = await ws_conn.receive()
+    assert ast.literal_eval(result.data)["message"] == "Something went wrong"