Skip to main content

Remote HTTP Mock

Project description

jj

Codecov PyPI PyPI - Downloads Python Version

Installation

pip3 install jj

Usage

import jj

@jj.match("*")
async def handler(request: jj.Request) -> jj.Response:
    return jj.Response(body="200 OK")

jj.serve()

Documentation


Matchers

Method

match_method(method)
from jj.http.methods import ANY, GET, POST

@jj.match_method(GET)
async def handler(request):
    return jj.Response(body="Method: " + request.method)

Path

match_path(path)
@jj.match_path("/users")
async def handler(request):
    return jj.Response(body="Path: " + request.path)

Segments

@jj.match_path("/users/{users_id}")
async def handler(request):
    return jj.Response(body=f"Segments: {request.segments}")

More information available here https://docs.aiohttp.org/en/stable/web_quickstart.html#variable-resources

Params

match_param(name, val)
@jj.match_param("locale", "en_US")
async def handler(request):
    locales = request.params.getall('locale')
    return jj.Response(body="Locales: " + ",".join(locales))
match_params(params)
@jj.match_params({"locale": "en_US", "timezone": "UTC"})
async def handler(request):
    # Literal String Interpolation (PEP 498)
    return jj.Response(body=f"Params: {request.params}")

Headers

match_header(name, val)
@jj.match_header("X-Forwarded-Proto", "https")
async def handler(request):
    proto = request.headers.getone("X-Forwarded-Proto")
    return jj.Response(body="Proto: " + proto)
match_headers(headers)
@jj.match_headers({
    "x-user-id": "1432",
    "x-client-id": "iphone",
})
async def handler(request):
    return jj.Response(body=f"Headers: {request.headers}")

Combining Matchers

match_any(matchers)
from jj.http import PATCH, PUT

@jj.match_any([
    jj.match_method(PUT),
    jj.match_method(PATCH),
])
async def handler(request):
    return jj.Response(body="200 OK")
match_all(matchers)
@jj.match_all([
    jj.match_method("*"),
    jj.match_path("/"),
    jj.match_params({"locale": "en_US"}),
    jj.match_headers({"x-request-id": "0fefbf48"}),
])
async def handler(request):
    return jj.Response(body="200 OK")
match(method, path, params, headers)
@jj.match("*", "/", {"locale": "en_US"}, {"x-request-id": "0fefbf48"})
async def handler(request):
    return jj.Response(body="200 OK")

Responses

Response

JSON Response
@jj.match("*")
async def handler(request):
    return jj.Response(json={"message": "200 OK"})
HTML Response
@jj.match("*")
async def handler(request):
    return jj.Response(body="<p>text<p>", headers={"Content-Type": "text/html"})
Binary Response
@jj.match("*")
async def handler(request):
    return jj.Response(body=b"<binary>")
Not Found Response
@jj.match("*")
async def handler(request):
    return jj.Response(status=404, reason="Not Found")
Predefined Body
from jj.http import GET

@jj.match(GET, "/users")
async def handler(request):
    return jj.Response(body=open("responses/users.json", "rb"))
from jj.http import POST, CREATED

@jj.match(POST, "/users")
async def handler(request):
    return jj.Response(body=open("responses/created.json", "rb"), status=CREATED)

StaticResponse

Inline Content
from jj.http import GET

@jj.match(GET, "/image")
async def handler(request):
    return jj.StaticResponse("public/image.jpg")
Downloadable File
from jj.http import GET

@jj.match(GET, "/report")
async def handler(request):
    return jj.StaticResponse("public/report.csv", attachment=True)
from jj.http import GET

@jj.match(GET, "/")
async def handler(request):
    return jj.StaticResponse("public/report.csv", attachment="report.csv")

For more information visit https://developer.mozilla.org/en-US/docs/Web/HTTP/Headers/Content-Disposition

RelayResponse β

@jj.match("*")
async def handler(request):
    return jj.RelayResponse(target="https://httpbin.org/")

Apps

Single App

import jj
from jj.http.methods import GET, ANY
from jj.http.codes import OK, NOT_FOUND

class App(jj.App):
    @jj.match(GET, "/")
    async def root_handler(self, request: jj.Request) -> jj.Response:
        return jj.Response(status=OK, json={"message": "200 OK"})

    @jj.match(ANY)
    async def default_handler(self, request: jj.Request) -> jj.Response:
        return jj.Response(status=NOT_FOUND, json={"message": "Not Found"})

jj.serve(App(), port=5000)

Multiple Apps

import jj

class App(jj.App):
    @jj.match("*")
    async def handler(self, request: jj.Request) -> jj.Response:
        return jj.Response(body="App")

class AnotherApp(jj.App):
    @jj.match("*")
    async def handler(self, request: jj.Request) -> jj.Response:
        return jj.Response(body="AnotherApp")

jj.start(App(), port=5001)
jj.start(AnotherApp(), port=5002)

jj.wait_for([KeyboardInterrupt])

App Inheritance

import jj

class UsersApp(jj.App):
    @jj.match("*", path="/users")
    async def handler(self, request: jj.Request) -> jj.Response:
        return jj.Response(body="Users")

class GroupsApp(jj.App):
    @jj.match("*", path="/groups")
    async def handler(self, request: jj.Request) -> jj.Response:
        return jj.Response(body="Groups")

class App(UsersApp, GroupsApp):
    pass

jj.serve(App())

Middlewares

Handler Middleware

import jj
from jj.http.codes import OK, FORBIDDEN

class Middleware(jj.Middleware):
    async def do(self, request, handler, app):
        if request.headers.get("x-secret-key") != "<SECRET_KEY>":
            return jj.Response(status=FORBIDDEN, body="Forbidden")
        return await handler(request)

class App(jj.App):
    @Middleware()
    @jj.match("*")
    async def handler(self, request: jj.Request) -> jj.Response:
        return jj.Response(status=OK, body="Ok")

jj.serve(App())

App Middleware

import jj
from jj.http.codes import OK, FORBIDDEN

class ReusableMiddleware(jj.Middleware):
    def __init__(self, secret_key):
        super().__init__()
        self._secret_key = secret_key

    async def do(self, request, handler, app):
        if request.headers.get("x-secret-key") != self._secret_key:
            return jj.Response(status=FORBIDDEN, body="Forbidden")
        return await handler(request)

private = ReusableMiddleware("<SECRET_KEY>")

@private
class App(jj.App):
    @jj.match("*")
    async def handler(self, request: jj.Request) -> jj.Response:
        return jj.Response(status=OK, body="Ok")

jj.serve(App())

Remote Mock

Server Side

Start Remote Mock
import jj
from jj.mock import Mock

jj.serve(Mock(), port=8080)

Client Side

Register Remote Handler
import asyncio

import jj
from jj.mock import RemoteMock


async def main():
    remote_mock = RemoteMock("http://localhost:8080")

    matcher = jj.match("GET", "/users")
    response = jj.Response(status=200, json=[])
    remote_handler = remote_mock.create_handler(matcher, response)
    await remote_handler.register()

    # Request GET /users
    # Returns status=200 body=[]

asyncio.run(main())
Deregister Remote Handler
import asyncio

import jj
from jj.mock import RemoteMock


async def main():
    remote_mock = RemoteMock("http://localhost:8080")

    matcher = jj.match("GET", "/users")
    response = jj.Response(status=200, json=[])
    remote_handler = remote_mock.create_handler(matcher, response)
    await remote_handler.register()

    # Request GET /users
    # Returns status=200 body=[]

    await remote_handler.deregister()

asyncio.run(main())
Retrieve Remote Handler History
import asyncio

import jj
from jj.mock import RemoteMock


async def main():
    remote_mock = RemoteMock("http://localhost:8080")

    matcher = jj.match("GET", "/users")
    response = jj.Response(status=200, json=[])
    remote_handler = remote_mock.create_handler(matcher, response)
    await remote_handler.register()

    # Request GET /users
    # Returns status=200 body=[]

    history = await remote_handler.history()
    print(history)

    await remote_handler.deregister()

asyncio.run(main())

History:

[
    {
        'request': HistoryRequest(
            method='GET',
            path='/users',
            params=<MultiDictProxy()>,
            headers=<CIMultiDictProxy('Host': 'localhost:8080',
                                      'Accept': '*/*',
                                      'Accept-Encoding': 'gzip, deflate',
                                      'User-Agent': 'Python/3.8 aiohttp/3.7.3')>,
            body=b'',
        ),
        'response': HistoryResponse(
            status=200,
            reason='OK',
            headers=<CIMultiDictProxy('Content-Type': 'application/json',
                                      'Server': 'jj via aiohttp/3.7.3',
                                      'Content-Length': '2',
                                      'Date': 'Sun, 09 May 2021 08:08:19 GMT')>,
            body=b'[]',
        ),
        'tags': ['f75c2ab7-f68d-4b4a-85e0-1f38bb0abe9a']
    }
]
Register & Deregister Remote Handler (via context manager)
import asyncio

import jj
from jj.mock import RemoteMock


async def main():
    remote_mock = RemoteMock("http://localhost:8080")

    matcher = jj.match("GET", "/users")
    response = jj.Response(status=200, json=[])

    async with remote_mock.create_handler(matcher, response):
        # Request GET /users
        # Returns status=200 body=[]

asyncio.run(main())

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

jj-2.0.1.tar.gz (50.9 kB view hashes)

Uploaded Source

Built Distribution

jj-2.0.1-py3-none-any.whl (90.0 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page