Fix tests by creating simulated mailbox server

Since the re-write of the mailboxer server to rust, the tests of this repository cannot run.
Created a simulation server for tests instead

Author: ayalash

pyproject.toml

@@ -28,7 +28,7 @@ authors = [{ name = "Rotem Yaari", email = "vmalloc@gmail.com" }]
 "Homepage" = "https://github.com/getslash/mailboxer-python"
 
 [project.optional-dependencies]
-testing = ["Flask-Loopback", "pylint", "pytest"]
+testing = ["dataclasses_json", "Flask-Loopback", "pylint", "pytest"]
 
 [tool.hatch.build.targets.wheel]
 packages = ["mailboxer"]

tests/conftest.py

@@ -1,8 +1,3 @@
-import os
-import shutil
-import subprocess
-import sys
-import tempfile
 import uuid
 
 from urlobject import URLObject as URL
@@ -11,44 +6,21 @@
 import pytest
 from mailboxer import Mailboxer
 
-sys.path.insert(0, os.path.join(
-    os.path.abspath(os.path.dirname(__file__)),
-    "..", ".env", "mailboxer"))
-from flask_app.app import create_app
-from flask_app import models
+from .flask_app import create_app, app_initializations
 
-def pytest_addoption(parser):
-    parser.addoption("--setup-db", action="store_true", default=False)
 
 @pytest.fixture(scope="session")
-def db_engine(request):
-    if request.config.getoption("--setup-db"):
-        tmpdir = tempfile.mkdtemp()
-        subprocess.check_call("pg_ctl init -D {0} -w".format(tmpdir), shell=True)
-        subprocess.check_call("pg_ctl start -D {0} -w".format(tmpdir), shell=True)
-
-        @request.addfinalizer
-        def finalize():
-            subprocess.check_call("pg_ctl stop -D {0} -w -m immediate".format(tmpdir), shell=True)
-            shutil.rmtree(tmpdir)
-
-        subprocess.check_call("createdb mailboxer", shell=True)
-
-    with create_app().app_context():
-        models.db.session.close()
-        models.db.drop_all()
-        models.db.create_all()
-
-
-@pytest.fixture(scope="session")
-def mailboxer_url(request, db_engine):
-    loopback = FlaskLoopback(create_app())
+def mailboxer_url(request):
+    app = create_app()
+    loopback = FlaskLoopback(app)
     hostname = str(uuid.uuid1())
-    loopback.activate_address((hostname, 80))
-    @request.addfinalizer
-    def close():
-        loopback.deactivate_address((hostname, 80))
-    return URL("http://{0}".format(hostname))
+    port = 80
+    loopback.activate_address((hostname, port))
+    with app.app_context():
+        app_initializations()
+        yield URL(f"http://{hostname}:{port}")
+        loopback.deactivate_address((hostname, port))
+
 
 @pytest.fixture
 def mailboxer(mailboxer_url):

tests/flask_app.py

@@ -0,0 +1,86 @@
+import datetime
+import dataclasses
+from dataclasses_json import DataClassJsonMixin
+from itertools import count
+from typing import Any, Dict
+from flask import Flask, g, request, jsonify
+
+
+@dataclasses.dataclass
+class Mailbox(DataClassJsonMixin):
+    id: int
+    address: str
+    last_activity: str
+
+
+@dataclasses.dataclass
+class Email(DataClassJsonMixin):
+    id: int
+    mailbox_id: int
+    fromaddr: str
+    message: str
+    timestamp: str
+    sent_via_ssl: bool
+    read: bool
+
+
+def app_initializations() -> None:
+    g.mailboxes = {}
+    g.emails = {}
+    g.mailboxes_count = count(1)
+
+
+def create_app() -> Flask:
+    app = Flask("MailboxerSimulation")
+
+    def _get_success_response() -> Dict[str, str]:
+        return {"result": "ok"}
+
+    def _get_paginated_response(objects: list[Any]) -> Dict[str, Any]:
+        page = request.args.get("page", default=1, type=int)
+        page_size = request.args.get("page_size", default=1000, type=int)
+        result_objects = objects[(page - 1) * page_size : page_size * page]
+        return {
+            "metadata": {"total_num_objects": len(objects)},
+            "result": result_objects,
+        }
+
+    def now() -> str:
+        return datetime.datetime.now(tz=datetime.timezone.utc).isoformat()
+
+    @app.route("/v2/mailboxes", methods=["GET"])
+    def query_mailboxes():
+        sorted_mailboxes = sorted(
+            g.mailboxes.values(), key=lambda mailbox: mailbox.address
+        )
+        return _get_paginated_response(sorted_mailboxes)
+
+    @app.route("/v2/mailboxes", methods=["POST"])
+    def create_mailbox():
+        data = request.get_json(silent=True)
+        if not isinstance(data, dict):
+            return jsonify({"error": "Not JSON body"}), 400
+        address = data.get("address")
+        if not isinstance(address, str):
+            return jsonify({"error": "Invalid address {address!r}"}), 400
+
+        g.mailboxes[address] = Mailbox(
+            id=next(g.mailboxes_count), address=address, last_activity=now()
+        )
+        return _get_success_response()
+
+    @app.route("/v2/mailboxes/<address>", methods=["DELETE"])
+    def delete_mailbox(address: str):
+        g.mailboxes.pop(address, None)
+        return _get_success_response()
+
+    @app.route("/v2/mailboxes/<address>/emails", methods=["GET"])
+    def query_mailbox_emails(address: str):
+        emails = g.emails.get(address, [])
+        return _get_paginated_response(emails)
+
+    @app.route("/v2/vacuum", methods=["POST"])
+    def vacuum():
+        return _get_success_response()
+
+    return app