远程 HTTP 模拟

项目描述

jj

安装

pip3 install jj

用法

import jj

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

jj.serve()

文档

文档

匹配器

方法

匹配方法（`method`）

from jj.http.methods import GET

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

匹配方法（`methods`）

from jj.http.methods import PUT, PATCH

@jj.match_methods(PUT, PATCH)
async def handler(request):
    return jj.Response(body="Method: " + request.method)

小路

匹配路径（`path`）

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

细分市场

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

参数

匹配参数( `name`, `val`)

@jj.match_param("locale", "en_US")
async def handler(request):
    locales = request.params.getall('locale')
    return jj.Response(body="Locales: " + ",".join(locales))

匹配参数（`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}")

标头

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)

匹配标题（`headers`）

@jj.match_headers({
    "x-user-id": "1432",
    "x-client-id": "iphone",
})
async def handler(request):
    return jj.Response(body=f"Headers: {request.headers}")

组合匹配器

匹配任意（`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")

全部匹配（`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")

匹配（`method`，，，，）`path`_`paramsheaders`

@jj.match("*", "/", {"locale": "en_US"}, {"x-request-id": "0fefbf48"})
async def handler(request):
    return jj.Response(body="200 OK")

回应

JSON 响应

@jj.match("*")
async def handler(request):
    return jj.Response(json={"message": "200 OK"})

HTML 响应

@jj.match("*")
async def handler(request):
    return jj.Response(body="<p>text<p>", headers={"Content-Type": "text/html"})

二进制响应

@jj.match("*")
async def handler(request):
    return jj.Response(body=b"<binary>")

未找到响应

@jj.match("*")
async def handler(request):
    return jj.Response(status=404, reason="Not Found")

预定义的主体

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)

静态响应

内联内容

from jj.http import GET

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

可下载文件

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

中继响应`β`

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

应用

单一应用

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)

多个应用程序

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

应用继承

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

中间件

处理程序中间件

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

应用中间件

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

远程模拟

服务器端

启动远程模拟

import jj
from jj.mock import Mock

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

或通过码头工人

docker run -p 8080:80 nikitanovosibirsk/jj

客户端

import asyncio

import jj
from jj.mock import mocked


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

    async with mocked(matcher, response) as mock:
        # Request GET /users
        # Returns status=200 body=[]
    assert len(mock.history) == 1

asyncio.run(main())

使用jj-district42测试请求

低级 API

注册远程处理程序

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

注销远程处理程序

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

检索远程处理程序历史

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.fetch_history()
  print(history)

  await remote_handler.deregister()

asyncio.run(main())

历史：

[
    {
        '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']
    }
]

到期政策

import jj
from jj.mock import mocked
from jj.expiration_policy import ExpireAfterRequests
from httpx import AsyncClient

matcher = jj.match("GET", "/")
response = jj.Response(status=200)
policy = ExpireAfterRequests(1)

async with mocked(matcher, response, expiration_policy=policy):
    async with AsyncClient() as client:
        resp1 = await client.get("/")
        assert resp1.status_code == 200
        # expired
        resp2 = await client.get("/")
        assert resp2.status_code == 404

自定义记录器

导入 日志

 从jj.logs导入jj
从 jj.mock 导入 SimpleFormatter
导入Mock ，SystemLogFilter    


类 格式化程序（SimpleFormatter ）：
    def  format_request （self ， request ： jj.Request ，record ：logging.LogRecord ）- > str ：return f “ - > { request.method } { request.url.path_qs } { request.headers } ” _ _  _ _ _ _ _

jj 2.6.2

导航

项目链接

统计数据

Meta

Maintainers

分类

项目描述

jj

安装

用法

文档

匹配器

方法

匹配方法（method）

匹配方法（methods）

小路

匹配路径（path）

细分市场

参数

匹配参数( name, val)

匹配参数（params）

标头

match_header( name, val)

匹配标题（headers）

组合匹配器

匹配任意（matchers）

全部匹配（matchers）

匹配（method，，，，）path_paramsheaders

回应

回复

JSON 响应

HTML 响应

二进制响应

未找到响应

预定义的主体

静态响应

内联内容

可下载文件

中继响应β

应用

单一应用

多个应用程序

应用继承

中间件

处理程序中间件

应用中间件

远程模拟

服务器端

启动远程模拟

客户端

低级 API

注册远程处理程序

注销远程处理程序

检索远程处理程序历史

到期政策

自定义记录器

匹配方法（`method`）

匹配方法（`methods`）

匹配路径（`path`）

匹配参数( `name`, `val`)

匹配参数（`params`）

match_header( `name`, `val`)

匹配标题（`headers`）

匹配任意（`matchers`）

全部匹配（`matchers`）

匹配（`method`，，，，）`path`_`paramsheaders`

中继响应`β`