[MZ] Refactoring done

Author: manuelzander

README.md

@@ -23,21 +23,31 @@ Install the required Python packages with:
 
     pip3 install -r requirements.txt
 
-## Run the app
+## Run the app without Docker
+
+Quickstart (see below for flags, or use `-h`):
+
+    python3 app.py
+
+## Run the app with Docker
 
 Make sure you have Docker installed locally:
 
     docker -v
 
-Build the image locally with:
+Quickstart:
+
+    make build && make run
+
+Alternatively, build the image locally with:
 
     docker build --tag emoji-app:latest .
 
 Or:
 
     make build
 
-Run the container locally with:
+Then, run the container locally with:
 
     docker run --rm --name emoji-app -p 3001:3001/udp emoji-app:latest
 
@@ -48,7 +58,7 @@ Or:
 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`)
+* `--r` Disable translation from keyword to emoji (`bool`, default: `False`)
 * `--s` Separator between each emoji (`str`, default: `""`)
 * `--h/-h` See usage information
 
@@ -66,7 +76,7 @@ Send a message and hit enter:
 
 ## Development and testing
 
-Use the `Makefile` you can run `make <cmd>` where `<cmd>` is one of:
+Using the `Makefile` you can run `make <cmd>` where `<cmd>` is one of:
 
 * `sort-imports` to ensure Python imports are in the correct PEP format
 * `format` to format Python files using black
@@ -78,4 +88,4 @@ or just type `make`/`make all`, which will run all of the above.
 
 ## Authors
 
-* Manuel Zander
+* Manuel Zander

app.py

@@ -11,10 +11,9 @@
 logger = logging.getLogger(__name__)
 coloredlogs.install(level="INFO")
 
-
 # App settings
+APP_HOST = "0.0.0.0"
 APP_PORT = 3001
-TIME_SERVE_SECONDS = 3600  # Serve for 1 hour
 
 TRANSLATION_TABLE = {
     ":thumbsup:": "👍",
@@ -23,73 +22,80 @@
     ":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:
+class MessagePrinterServerProtocol(asyncio.BaseProtocol):
+    def __init__(self, n: int, s: str, r: bool) -> None:
         super().__init__()
+        self.multiplier = n
+        self.separator = s
+        self.translation_toggle = r
 
     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)
 
+        formatted_message = format_message(
+            message, self.multiplier, self.separator, self.translation_toggle
+        )
 
-def generate_response(message: str) -> Union[str, Any]:
-    logger.debug(f"Generating response for message: {message}")
+        # Print main output of application to stdout
+        print(formatted_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}"
+def handle_error(message: str) -> Union[str, Any]:
+    logger.warning(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)
 
+def format_message(
+    message: str, multiplier: int, separator: str, translation_toggle: bool
+) -> Union[str, Any]:
+    logger.debug(f"Formatting message: {message}")
 
-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}")
+    params = message.split()
 
-    loop = asyncio.get_running_loop()
-    transport, protocol = await loop.create_datagram_endpoint(
-        lambda: EchoServerProtocol(), local_addr=(host, port)
-    )
+    if len(params) > 2:
+        return handle_error(message)
 
     try:
-        await asyncio.sleep(TIME_SERVE_SECONDS)
-    finally:
-        transport.close()
+        nb_repetitions, command = int(params[0]), params[1]
+        text = TRANSLATION_TABLE[command]
+    except (ValueError, KeyError, IndexError) as e:
+        return handle_error(message)
+
+    response = (
+        nb_repetitions * multiplier * [text if not translation_toggle else command]
+    )
+
+    formatted_message = separator.join(response)
+    logger.debug(f"Formatted message: '{formatted_message}'")
+    return formatted_message
 
 
 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}")
+    logger.info(f"Starting UDP server listening on: {APP_HOST}:{APP_PORT}")
+
+    loop = asyncio.get_event_loop()
+    transport, protocol = loop.run_until_complete(
+        loop.create_datagram_endpoint(
+            lambda: MessagePrinterServerProtocol(args.n, args.s, args.r),
+            local_addr=(APP_HOST, APP_PORT),
+        )
+    )
 
-    global multiplier, separator, translation_toggle
-    multiplier = args.n
-    separator = args.s
-    translation_toggle = args.r
-
-    asyncio.run(start_udp_server())
+    try:
+        loop.run_forever()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        transport.close()
+        loop.close()
 
 
 if __name__ == "__main__":

test_app.py

@@ -6,119 +6,95 @@
 
 
 class AppTestCase(unittest.TestCase):
+    def test_handle_error(self) -> None:
+        self.assertEqual(app.handle_error("message"), "Unknown command: message")
+
     def test_base_case_thumbsup(self) -> None:
-        app.multiplier = 1
-        app.separator = ""
-        app.translation_toggle = False
-        self.assertEqual(app.generate_response("1 :thumbsup:"), "👍")
+        self.assertEqual(app.format_message("1 :thumbsup:", 1, "", False), "👍")
 
     def test_base_case_thumbsdown(self) -> None:
-        app.multiplier = 1
-        app.separator = ""
-        app.translation_toggle = False
-        self.assertEqual(app.generate_response("2 :thumbsdown:"), "👎👎")
+        self.assertEqual(app.format_message("2 :thumbsdown:", 1, "", False), "👎👎")
 
     def test_base_case_ok(self) -> None:
-        app.multiplier = 1
-        app.separator = ""
-        app.translation_toggle = False
-        self.assertEqual(app.generate_response("5 :ok:"), "👌👌👌👌👌")
+        self.assertEqual(app.format_message("5 :ok:", 1, "", False), "👌👌👌👌👌")
 
     def test_base_case_crossed(self) -> None:
-        app.multiplier = 1
-        app.separator = ""
-        app.translation_toggle = False
-        self.assertEqual(app.generate_response("10 :crossed:"), "🤞🤞🤞🤞🤞🤞🤞🤞🤞🤞")
+        self.assertEqual(app.format_message("10 :crossed:", 1, "", False), "🤞🤞🤞🤞🤞🤞🤞🤞🤞🤞")
 
     def test_separator_1_thumbsup(self) -> None:
-        app.multiplier = 1
-        app.separator = ","
-        app.translation_toggle = False
-        self.assertEqual(app.generate_response("1 :thumbsup:"), "👍")
+        self.assertEqual(app.format_message("1 :thumbsup:", 1, ",", False), "👍")
 
     def test_separator_2_thumbsup(self) -> None:
-        app.multiplier = 1
-        app.separator = ","
-        app.translation_toggle = False
-        self.assertEqual(app.generate_response("2 :thumbsup:"), "👍,👍")
+        self.assertEqual(app.format_message("2 :thumbsup:", 1, ",", False), "👍,👍")
 
     def test_separator_3_thumbsup(self) -> None:
-        app.multiplier = 1
-        app.separator = ","
-        app.translation_toggle = False
-        self.assertEqual(app.generate_response("3 :thumbsup:"), "👍,👍,👍")
+        self.assertEqual(app.format_message("3 :thumbsup:", 1, ",", False), "👍,👍,👍")
 
     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:")
+        self.assertEqual(app.format_message("1 :thumbsup:", 1, "", True), ":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:")
+        self.assertEqual(
+            app.format_message("2 :thumbsup:", 1, "", True), ":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:"
+            app.format_message("3 :thumbsup:", 1, "", True),
+            ":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:"), "👍👍")
+        self.assertEqual(app.format_message("1 :thumbsup:", 2, "", False), "👍👍")
 
     def test_multiplier_3_thumbsup(self) -> None:
-        app.multiplier = 3
-        app.separator = ""
-        app.translation_toggle = False
-        self.assertEqual(app.generate_response("1 :thumbsup:"), "👍👍👍")
+        self.assertEqual(app.format_message("1 :thumbsup:", 3, "", False), "👍👍👍")
 
     def test_multiplier_5_thumbsup(self) -> None:
-        app.multiplier = 5
-        app.separator = ""
-        app.translation_toggle = False
-        self.assertEqual(app.generate_response("1 :thumbsup:"), "👍👍👍👍👍")
+        self.assertEqual(app.format_message("1 :thumbsup:", 5, "", False), "👍👍👍👍👍")
 
     def test_complex_ok_1(self) -> None:
-        app.multiplier = 2
-        app.separator = "++"
-        app.translation_toggle = False
-        self.assertEqual(app.generate_response("2 :ok:"), "👌++👌++👌++👌")
+        self.assertEqual(app.format_message("2 :ok:", 2, "++", False), "👌++👌++👌++👌")
 
     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:"
+            app.format_message("2 :ok:", 3, " ", True), ":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:"
+            app.format_message("2 :ok:", 3, " ", True), ":ok: :ok: :ok: :ok: :ok: :ok:"
         )
 
     def test_failure(self) -> None:
+        val = ""
+        self.assertEqual(
+            app.format_message(val, 1, "", False), f"Unknown command: {val}"
+        )
+
+    def test_failure_1(self) -> None:
         val = "x"
-        self.assertEqual(app.generate_response(val), f"Unknown command: {val}")
+        self.assertEqual(
+            app.format_message(val, 1, "", False), f"Unknown command: {val}"
+        )
 
-    def test_failure(self) -> None:
+    def test_failure_2(self) -> None:
         val = "x :ok:"
-        self.assertEqual(app.generate_response(val), f"Unknown command: {val}")
+        self.assertEqual(
+            app.format_message(val, 1, "", False), f"Unknown command: {val}"
+        )
 
-    def test_failure(self) -> None:
+    def test_failure_3(self) -> None:
         val = "1 :okokokokok:"
-        self.assertEqual(app.generate_response(val), f"Unknown command: {val}")
+        self.assertEqual(
+            app.format_message(val, 1, "", False), f"Unknown command: {val}"
+        )
+
+    def test_failure_4(self) -> None:
+        val = "1 :ok: :ok:"
+        self.assertEqual(
+            app.format_message(val, 1, "", False), f"Unknown command: {val}"
+        )
 
 
 if __name__ == "__main__":

utils.py

@@ -6,23 +6,24 @@
 def create_parser() -> ArgumentParser:
     parser = ArgumentParser(description="Emoji app")
     parser.add_argument(
-        "--n", default=1, 
-        type=int, 
-        required=False, 
-        help="Multiply number of emojis by n",
+        "--n",
+        default=1,
+        type=int,
+        required=False,
+        help="Multiply number of emojis by n (int, default: 1)",
     )
     parser.add_argument(
         "--s",
         default="",
         type=str,
         required=False,
-        help="Separator between each emoji",
+        help="Separator between each emoji (str, default: " ")",
     )
     parser.add_argument(
         "--r",
         default=False,
         type=bool,
         required=False,
-        help="Disable the translation from keyword to emoji",
+        help="Disable translation from keyword to emoji (bool, default: False)",
     )
     return parser