[MZ] Finished inital app - first working version

Author: manuelzander

Dockerfile

@@ -1,6 +1,4 @@
-FROM python:3.7-alpine
-
-RUN apk update && apk add bash make automake gcc g++ subversion python3-dev
+FROM python:3.7.7-slim
 
 # We copy just the requirements.txt first to leverage cache
 COPY ./requirements.txt /app/requirements.txt
@@ -11,6 +9,6 @@ RUN python3 -m pip install -r requirements.txt
 
 COPY . /app
 
-EXPOSE 3001
+EXPOSE 3001/udp
 
-CMD ["python3", "src/app.py"]
+ENTRYPOINT ["python3", "-u", "app.py"]

Makefile

@@ -1,18 +1,24 @@
 all: sort-imports format type-check test
 
-.PHONY: sort-imports format type-check test
+.PHONY: sort-imports format type-check test build run
 
 SHELL=/bin/bash
 SHELLFLAGS=-euo pipefail -c
 
 sort-imports:
-	python3 -m isort --atomic app tests
+	python3 -m isort --atomic app.py test_app.py
 
 format:
-	python3 -m black app tests
+	python3 -m black app.py test_app.py
 
 type-check:
-	python3 -m mypy app
+	python3 -m mypy app.py
 
 test:
-	python3 -m pytest -o log_cli=true tests
+	python3 -m pytest -o log_cli=true test_app.py
+
+build: all
+	docker build --tag emoji-app:latest .
+
+run:
+	docker run --rm --name emoji-app -p 3001:3001/udp emoji-app:latest

README.md

@@ -23,26 +23,46 @@ Install the required Python packages with:
 
     pip3 install -r requirements.txt
 
-
 ## Run the app
 
-The `Dockerfile` can be used to build an image and deploy it locally:
-Make sure you have Docker installed:
+Make sure you have Docker installed locally:
 
     docker -v
 
-Build the image with:
+Build the image locally with:
 
     docker build --tag emoji-app:latest .
 
-Run the container locally:
+Or:
+
+    make build
+
+Run the container locally with:
 
-    docker run --name emoji-app:latest -p 3001:3001 app
+    docker run --rm --name emoji-app -p 3001:3001/udp emoji-app:latest
+
+Or:
+
+    make run
+
+Optionally, the following flags can be used: `--n`, `--r`, `--s` and `--h/-h`:
+
+* `--n` Multiply number of emojis by n (`int`, default: `1`)
+* `--r` Disable the translation from keyword to emoji (`bool`, default: `False`)
+* `--s` Separator between each emoji (`str`, default: `""`)
+* `--h/-h` See usage information
+
+For example:
+
+    docker run --rm --name emoji-app -p 3001:3001/udp emoji-app:latest --n 2 --r True --s "+"
 
-You can now trigger the endpoint (`localhost:3001`), for example using `nc`.
+In another shell session, you can now trigger the endpoint (`0.0.0.0:3001`), for example using `nc`:
+
+    nc -u 0.0.0.0 3001
+
+Send a message and hit enter:
 
-<!-- TODO -->
-<!-- Include usage info -->
+    2 :ok:
 
 ## Development and testing
 

app.py

@@ -0,0 +1,98 @@
+#!/usr/bin/env python
+
+import asyncio
+import logging
+from typing import Any, Union
+
+import coloredlogs
+
+from utils import create_parser
+
+logger = logging.getLogger(__name__)
+coloredlogs.install(level="INFO")
+
+
+# App settings
+APP_PORT = 3001
+TIME_SERVE_SECONDS = 3600  # Serve for 1 hour
+
+TRANSLATION_TABLE = {
+    ":thumbsup:": "👍",
+    ":thumbsdown:": "👎",
+    ":ok:": "👌",
+    ":crossed:": "🤞",
+}
+
+# Need these for static type checking with mypy
+multiplier: int
+separator: str
+translation_toggle: bool
+
+
+# For further reference see https://docs.python.org/3.7/library/asyncio-protocol.html#udp-echo-client
+class EchoServerProtocol(asyncio.BaseProtocol):
+    def __init__(self) -> None:
+        super().__init__()
+
+    def connection_made(self, transport) -> None:  # type: ignore
+        self.transport = transport
+
+    def datagram_received(self, data, addr) -> None:  # type: ignore
+        message = data.decode()
+        logger.debug(f"Received message: '{message}' from addr: {addr}")
+        response = generate_response(message)
+        logger.debug(f"Generated response: '{response}'")
+        print(response)
+
+
+def generate_response(message: str) -> Union[str, Any]:
+    logger.debug(f"Generating response for message: {message}")
+
+    try:
+        params = message.split()
+        int(params[0])
+    except Exception as e:
+        logger.error(f"Unknown command: {message}")
+        return f"Unknown command: {message}"
+
+    if params[1] not in TRANSLATION_TABLE:
+        logger.error(f"Unknown command: {message}")
+        return f"Unknown command: {message}"
+
+    number = int(params[0]) * multiplier
+    text = TRANSLATION_TABLE[params[1]] if not translation_toggle else params[1]
+    repsonse = number * [text]
+    return separator.join(repsonse)
+
+
+async def start_udp_server(host: str = "0.0.0.0", port: int = APP_PORT) -> None:
+    logger.info(f"Starting UDP server listening on: {host}:{port}")
+
+    loop = asyncio.get_running_loop()
+    transport, protocol = await loop.create_datagram_endpoint(
+        lambda: EchoServerProtocol(), local_addr=(host, port)
+    )
+
+    try:
+        await asyncio.sleep(TIME_SERVE_SECONDS)
+    finally:
+        transport.close()
+
+
+def main() -> None:
+    parser = create_parser()
+    args = parser.parse_args()
+    logger.debug(f"Running with args: n: {args.n}, s: {args.s}, r: {args.r}")
+
+    global multiplier, separator, translation_toggle
+    multiplier = args.n
+    separator = args.s
+    translation_toggle = args.r
+
+    asyncio.run(start_udp_server())
+
+
+if __name__ == "__main__":
+    logger.debug(f"Running {__file__}")
+    logger.debug(f"APP_PORT: {APP_PORT}")
+    main()

app/app.py

@@ -0,0 +1,125 @@
+#!/usr/bin/env python
+
+import unittest
+
+import app
+
+
+class AppTestCase(unittest.TestCase):
+    def test_base_case_thumbsup(self) -> None:
+        app.multiplier = 1
+        app.separator = ""
+        app.translation_toggle = False
+        self.assertEqual(app.generate_response("1 :thumbsup:"), "👍")
+
+    def test_base_case_thumbsdown(self) -> None:
+        app.multiplier = 1
+        app.separator = ""
+        app.translation_toggle = False
+        self.assertEqual(app.generate_response("2 :thumbsdown:"), "👎👎")
+
+    def test_base_case_ok(self) -> None:
+        app.multiplier = 1
+        app.separator = ""
+        app.translation_toggle = False
+        self.assertEqual(app.generate_response("5 :ok:"), "👌👌👌👌👌")
+
+    def test_base_case_crossed(self) -> None:
+        app.multiplier = 1
+        app.separator = ""
+        app.translation_toggle = False
+        self.assertEqual(app.generate_response("10 :crossed:"), "🤞🤞🤞🤞🤞🤞🤞🤞🤞🤞")
+
+    def test_separator_1_thumbsup(self) -> None:
+        app.multiplier = 1
+        app.separator = ","
+        app.translation_toggle = False
+        self.assertEqual(app.generate_response("1 :thumbsup:"), "👍")
+
+    def test_separator_2_thumbsup(self) -> None:
+        app.multiplier = 1
+        app.separator = ","
+        app.translation_toggle = False
+        self.assertEqual(app.generate_response("2 :thumbsup:"), "👍,👍")
+
+    def test_separator_3_thumbsup(self) -> None:
+        app.multiplier = 1
+        app.separator = ","
+        app.translation_toggle = False
+        self.assertEqual(app.generate_response("3 :thumbsup:"), "👍,👍,👍")
+
+    def test_no_translation_1_thumbsup(self) -> None:
+        app.multiplier = 1
+        app.separator = ""
+        app.translation_toggle = True
+        self.assertEqual(app.generate_response("1 :thumbsup:"), ":thumbsup:")
+
+    def test_no_translation_2_thumbsup(self) -> None:
+        app.multiplier = 1
+        app.separator = ""
+        app.translation_toggle = True
+        self.assertEqual(app.generate_response("2 :thumbsup:"), ":thumbsup::thumbsup:")
+
+    def test_no_translation_3_thumbsup(self) -> None:
+        app.multiplier = 1
+        app.separator = ""
+        app.translation_toggle = True
+        self.assertEqual(
+            app.generate_response("3 :thumbsup:"), ":thumbsup::thumbsup::thumbsup:"
+        )
+
+    def test_multiplier_2_thumbsup(self) -> None:
+        app.multiplier = 2
+        app.separator = ""
+        app.translation_toggle = False
+        self.assertEqual(app.generate_response("1 :thumbsup:"), "👍👍")
+
+    def test_multiplier_3_thumbsup(self) -> None:
+        app.multiplier = 3
+        app.separator = ""
+        app.translation_toggle = False
+        self.assertEqual(app.generate_response("1 :thumbsup:"), "👍👍👍")
+
+    def test_multiplier_5_thumbsup(self) -> None:
+        app.multiplier = 5
+        app.separator = ""
+        app.translation_toggle = False
+        self.assertEqual(app.generate_response("1 :thumbsup:"), "👍👍👍👍👍")
+
+    def test_complex_ok_1(self) -> None:
+        app.multiplier = 2
+        app.separator = "++"
+        app.translation_toggle = False
+        self.assertEqual(app.generate_response("2 :ok:"), "👌++👌++👌++👌")
+
+    def test_complex_ok_2(self) -> None:
+        app.multiplier = 3
+        app.separator = " "
+        app.translation_toggle = True
+        self.assertEqual(
+            app.generate_response("2 :ok:"), ":ok: :ok: :ok: :ok: :ok: :ok:"
+        )
+
+    def test_complex_ok_2(self) -> None:
+        app.multiplier = 3
+        app.separator = " "
+        app.translation_toggle = True
+        self.assertEqual(
+            app.generate_response("2 :ok:"), ":ok: :ok: :ok: :ok: :ok: :ok:"
+        )
+
+    def test_failure(self) -> None:
+        val = "x"
+        self.assertEqual(app.generate_response(val), f"Unknown command: {val}")
+
+    def test_failure(self) -> None:
+        val = "x :ok:"
+        self.assertEqual(app.generate_response(val), f"Unknown command: {val}")
+
+    def test_failure(self) -> None:
+        val = "1 :okokokokok:"
+        self.assertEqual(app.generate_response(val), f"Unknown command: {val}")
+
+
+if __name__ == "__main__":
+    unittest.main()