Add a development server with a simple CLI

README.rst

@@ -273,6 +273,49 @@ For the full documentation, please see https://schemathesis.readthedocs.io/en/la
 
 Or you can look at the ``docs/`` directory in the repository.
 
+Local development
+-----------------
+
+First, you need to prepare a virtual environment with `poetry`_.
+Install ``poetry`` (check out the `installation guide`_) and run this command inside the project root:
+
+.. code:: bash
+
+    poetry install
+
+For simpler local development Schemathesis includes a ``aiohttp``-based server with 3 endpoints in Swagger 2.0 schema:
+
+- ``/api/success`` - always returns ``{"success": true}``
+- ``/api/failure`` - always returns 500
+- ``/api/slow`` - always returns ``{"slow": true}`` after 250 ms delay
+
+To start the server:
+
+.. code:: bash
+
+    ./test_server.sh 8081
+
+It is possible to configure available endpoints via ``--endpoints`` option.
+The value is expected to be a comma separated string with endpoint names (``success``, ``failure`` or ``slow``):
+
+.. code:: bash
+
+    ./test_server.sh 8081 --endpoints=success,slow
+
+Then you could use CLI against this server:
+
+.. code:: bash
+
+    schemathesis run http://127.0.0.1:8081/swagger.yaml
+    Running schemathesis test cases ...
+
+    -------------------------------------------------------------
+    not_a_server_error            2 / 2 passed          PASSED
+    -------------------------------------------------------------
+
+    Tests succeeded.
+
+
 Contributing
 ------------
 
@@ -307,6 +350,8 @@ will be licensed under its MIT license.
 .. _hypothesis: https://hypothesis.works/
 .. _hypothesis_jsonschema: https://github.com/Zac-HD/hypothesis-jsonschema
 .. _pytest: http://pytest.org/en/latest/
+.. _poetry: https://github.com/sdispater/poetry
+.. _installation guide: https://github.com/sdispater/poetry#installation
 .. _here: https://hypothesis.readthedocs.io/en/latest/reproducing.html#providing-explicit-examples
 .. _CONTRIBUTING.rst: https://github.com/kiwicom/schemathesis/blob/contributing/CONTRIBUTING.rst
 .. _MIT license: https://opensource.org/licenses/MIT

docs/changelog.rst

@@ -11,7 +11,7 @@ Added
 
 - HTTP Digest Auth support. `#106`_
 - Support for Hypothesis settings in CLI & Runner. `#107`_
-
+- Local development server. `#126`_
 Removed
 ~~~~~~~
 
@@ -221,6 +221,7 @@ Fixed
 .. _0.2.0: https://github.com/kiwicom/schemathesis/compare/v0.1.0...v0.2.0
 
 .. _#130: https://github.com/kiwicom/schemathesis/issues/130
+.. _#126: https://github.com/kiwicom/schemathesis/issues/126
 .. _#121: https://github.com/kiwicom/schemathesis/issues/121
 .. _#118: https://github.com/kiwicom/schemathesis/issues/118
 .. _#107: https://github.com/kiwicom/schemathesis/issues/107

test/app.py

@@ -1,11 +1,36 @@
 import asyncio
+import logging
 import threading
+from enum import Enum
+from functools import wraps
 from time import sleep
 from typing import Dict, Tuple
 
+import click
 import yaml
 from aiohttp import web
 
+from schemathesis.cli import CSVOption
+
+
+async def success(request):
+    return web.json_response({"success": True})
+
+
+async def failure(request):
+    raise web.HTTPInternalServerError
+
+
+async def slow(request):
+    await asyncio.sleep(0.25)
+    return web.json_response({"slow": True})
+
+
+class Endpoint(Enum):
+    success = ("/api/success", success)
+    failure = ("/api/failure", failure)
+    slow = ("/api/slow", slow)
+
 
 def create_app(endpoints=("success", "failure")) -> web.Application:
     """Factory for aioHTTP app.
@@ -25,24 +50,18 @@ async def schema(request):
         content = yaml.dump(schema_data)
         return web.Response(body=content)
 
-    async def success(request):
-        incoming_requests.append(request)
-        return web.Response()
-
-    async def failure(request):
-        incoming_requests.append(request)
-        raise web.HTTPInternalServerError
+    def wrapper(handler):
+        @wraps(handler)
+        async def inner(request):
+            incoming_requests.append(request)
+            return await handler(request)
 
-    async def slow(request):
-        incoming_requests.append(request)
-        await asyncio.sleep(0.25)
-        return web.Response()
-
-    mapping = {"success": ("/api/success", success), "slow": ("/api/slow", slow), "failure": ("/api/failure", failure)}
+        return inner
 
     app = web.Application()
     app.add_routes(
-        [web.get("/swagger.yaml", schema)] + [web.get(*value) for key, value in mapping.items() if key in endpoints]
+        [web.get("/swagger.yaml", schema)]
+        + [web.get(item.value[0], wrapper(item.value[1])) for item in Endpoint if item.name in endpoints]
     )
     app["incoming_requests"] = incoming_requests
     return app
@@ -74,6 +93,7 @@ def _run_server(app: web.Application, port: int) -> None:
     """Run the given app on the given port.
 
     Intended to be called as a target for a separate thread.
+    NOTE. `aiohttp.web.run_app` works only in the main thread and can't be used here (or maybe can we some tuning)
     """
     # Set a loop for a new thread (there is no by default for non-main threads)
     loop = asyncio.new_event_loop()
@@ -91,3 +111,16 @@ def run_server(app: web.Application, port: int, timeout: float = 0.05) -> None:
     server_thread.daemon = True
     server_thread.start()
     sleep(timeout)
+
+
+@click.command()
+@click.argument("port", type=int)
+@click.option("--endpoints", type=CSVOption(Endpoint))
+def run_app(port, endpoints):
+    app = create_app(endpoints or ("success", "failure"))
+    web.run_app(app, port=port)
+
+
+if __name__ == "__main__":
+    logging.basicConfig(level=logging.DEBUG)
+    run_app()

test/test_app.py

@@ -0,0 +1,18 @@
+import sys
+from pathlib import Path
+
+import pytest
+from _pytest import pytester
+
+HERE = Path(__file__).absolute().parent
+
+
+def test_app(testdir, aiohttp_unused_port):
+    # When the testing app is run from CMD
+    port = aiohttp_unused_port()
+    with pytest.raises(pytester.Testdir.TimeoutExpired):
+        testdir.run(sys.executable, f"{HERE / 'app.py'}", str(port), timeout=0.75)
+    with testdir.tmpdir.join("stderr").open() as fd:
+        stderr = fd.read()
+    # Then it should start OK and emit debug logs
+    assert "DEBUG:asyncio:Using selector:" in stderr

test_server.sh

@@ -0,0 +1,2 @@
+# Run testing AioHTTP app on the given port
+PYTHONPATH=$(pwd)/src python test/app.py "$@"