jj 2.0.1

Remote HTTP Mock

jj

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

• Documentation
  • Matchers
    • Method
      • match_method(method)
    • Path
      • match_path(path)
    • Segments
    • Params
      • match_param(name, val)
      • match_params(params)
    • Headers
      • match_header(name, val)
      • match_headers(headers)
    • Combining Matchers
      • match_any(matchers)
      • match_all(matchers)
      • match(method, path, params, headers)
  • Responses
    • Response
      • JSON Response
      • HTML Response
      • Binary Response
      • Not Found Response
      • Predefined Body
    • StaticResponse
      • Inline Content
      • Downloadable File
    • RelayResponse β
  • Apps
    • Single App
    • Multiple Apps
    • App Inheritance
  • Middlewares
    • Handler Middleware
    • App Middleware
  • Remote Mock
    • Server Side
      • Start Remote Mock
    • Client Side
      • Register Remote Handler
      • Deregister Remote Handler
      • Retrieve Remote Handler History
      • Register & Deregister Remote Handler (via context manager)

---

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())