193 lines
7.4 KiB
Python
193 lines
7.4 KiB
Python
import unittest
|
|
import json
|
|
import httpx
|
|
from pwo import async_test
|
|
from kaya import BugisApp, HttpContext, HttpMethod
|
|
from typing import Sequence, List
|
|
|
|
|
|
class AsgiTest(unittest.TestCase):
|
|
app: BugisApp
|
|
|
|
def setUp(self):
|
|
self.app = BugisApp()
|
|
|
|
@self.app.GET('/hello')
|
|
@self.app.GET('/hello2')
|
|
@self.app.route('/hello3')
|
|
@self.app.GET('/hello/*')
|
|
async def handle_request(ctx: HttpContext) -> None:
|
|
async for chunk in ctx.request_body:
|
|
print(chunk)
|
|
await ctx.send_str(200, 'Hello World!')
|
|
|
|
@self.app.route(('/foo/bar',), HttpMethod.PUT, recursive=True)
|
|
async def handle_request(ctx: HttpContext) -> None:
|
|
async for chunk in ctx.request_body:
|
|
print(chunk)
|
|
await ctx.send_str(200, ctx.path)
|
|
|
|
@self.app.route(('/foo/*',), HttpMethod.PUT, recursive=True)
|
|
async def handle_request(ctx: HttpContext, path: Sequence[str]) -> None:
|
|
async for chunk in ctx.request_body:
|
|
print(chunk)
|
|
await ctx.send_str(200, json.dumps(path))
|
|
|
|
@self.app.route(('/foo/*',), recursive=True)
|
|
async def handle_request(ctx: HttpContext, path: Sequence[str]) -> None:
|
|
await ctx.send_str(500, f"Unable to handle request for {ctx.path}")
|
|
|
|
@self.app.GET('/employee/${employee_id}')
|
|
async def handle_request(ctx: HttpContext, employee_id: str) -> None:
|
|
async for chunk in ctx.request_body:
|
|
print(chunk)
|
|
await ctx.send_str(200, employee_id)
|
|
|
|
@self.app.GET('/square/${x:int}')
|
|
async def handle_request(ctx: HttpContext, x: int) -> None:
|
|
async for chunk in ctx.request_body:
|
|
print(chunk)
|
|
await ctx.send_str(200, str(x * x))
|
|
|
|
@self.app.GET('/department/${department_id:int}/employee/${employee_id:int}')
|
|
async def handle_request(ctx: HttpContext, department_id: int, employee_id: int) -> None:
|
|
async for chunk in ctx.request_body:
|
|
print(chunk)
|
|
await ctx.send_str(200, json.dumps({
|
|
'department_id': department_id,
|
|
'employee_id': employee_id
|
|
}))
|
|
|
|
@self.app.PUT('/hello/*', recursive=True)
|
|
async def handle_request(ctx: HttpContext, _: List[str]) -> None:
|
|
await ctx.stream_body(200, (chunk async for chunk in ctx.request_body))
|
|
|
|
@async_test
|
|
async def test_hello(self):
|
|
transport = httpx.ASGITransport(app=self.app)
|
|
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
|
|
r = await client.get("/hello")
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertEqual(r.text, "Hello World!")
|
|
|
|
r = await client.get("/hello2")
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertEqual(r.text, "Hello World!")
|
|
|
|
r = await client.post("/hello3")
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertEqual(r.text, "Hello World!")
|
|
|
|
r = await client.get("/hello4")
|
|
self.assertEqual(404, r.status_code)
|
|
self.assertTrue(len(r.text) == 0)
|
|
|
|
body = {'name': 'John', 'surname': 'Smith'}
|
|
r = await client.put("/hello/foo/bar", json=body)
|
|
self.assertEqual(200, r.status_code)
|
|
ans = json.loads(r.text)
|
|
self.assertTrue(body, ans)
|
|
|
|
@async_test
|
|
async def test_foo(self):
|
|
transport = httpx.ASGITransport(app=self.app)
|
|
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
|
|
r = await client.put("/foo/fizz/baz")
|
|
self.assertEqual(200, r.status_code)
|
|
response = json.loads(r.text)
|
|
self.assertEqual(['fizz', 'baz'], response)
|
|
|
|
r = await client.get("/foo/not-put")
|
|
self.assertEqual(500, r.status_code)
|
|
self.assertEqual("Unable to handle request for /foo/not-put", r.text)
|
|
|
|
@async_test
|
|
async def test_method_agnostic_fallback_order_independence(self):
|
|
app = BugisApp()
|
|
|
|
@app.route(('/foo/*',), recursive=True)
|
|
async def handle_request(ctx: HttpContext, path: Sequence[str]) -> None:
|
|
await ctx.send_str(500, f"Unable to handle request for {ctx.path}")
|
|
|
|
@app.route(('/foo/*',), HttpMethod.PUT, recursive=True)
|
|
async def handle_request(ctx: HttpContext, path: Sequence[str]) -> None:
|
|
await ctx.send_str(200, json.dumps(path))
|
|
|
|
transport = httpx.ASGITransport(app=app)
|
|
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
|
|
r = await client.put("/foo/fizz/baz")
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertEqual(['fizz', 'baz'], json.loads(r.text))
|
|
|
|
r = await client.get("/foo/not-put")
|
|
self.assertEqual(500, r.status_code)
|
|
self.assertEqual("Unable to handle request for /foo/not-put", r.text)
|
|
|
|
@async_test
|
|
async def test_disjoint_method_specific_matchers(self):
|
|
app = BugisApp()
|
|
|
|
@app.route(('/foo/*',), HttpMethod.PUT, recursive=True)
|
|
async def handle_request(ctx: HttpContext, path: Sequence[str]) -> None:
|
|
await ctx.send_str(200, "PUT")
|
|
|
|
@app.route(('/foo/*',), HttpMethod.GET, recursive=True)
|
|
async def handle_request(ctx: HttpContext, path: Sequence[str]) -> None:
|
|
await ctx.send_str(200, "GET")
|
|
|
|
transport = httpx.ASGITransport(app=app)
|
|
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
|
|
r = await client.put("/foo/bar")
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertEqual("PUT", r.text)
|
|
|
|
r = await client.get("/foo/bar")
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertEqual("GET", r.text)
|
|
|
|
@async_test
|
|
async def test_foo_bar(self):
|
|
transport = httpx.ASGITransport(app=self.app)
|
|
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
|
|
r = await client.put("/foo/bar/baz")
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertEqual('/foo/bar/baz', r.text)
|
|
|
|
@async_test
|
|
async def test_employee(self):
|
|
transport = httpx.ASGITransport(app=self.app)
|
|
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
|
|
r = await client.get("/employee/101325")
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertEqual(r.text, '101325')
|
|
|
|
@async_test
|
|
async def test_square(self):
|
|
transport = httpx.ASGITransport(app=self.app)
|
|
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
|
|
x = 30
|
|
r = await client.get(f"/square/{x}")
|
|
self.assertEqual(200, r.status_code)
|
|
self.assertEqual(r.text, str(x * x))
|
|
|
|
@async_test
|
|
async def test_department_employee(self):
|
|
transport = httpx.ASGITransport(app=self.app)
|
|
|
|
async with httpx.AsyncClient(transport=transport, base_url="http://127.0.0.1:80") as client:
|
|
r = await client.get("department/189350/employee/101325")
|
|
self.assertEqual(200, r.status_code)
|
|
response = json.loads(r.text)
|
|
self.assertEqual({
|
|
'department_id': 189350,
|
|
'employee_id': 101325
|
|
}, response)
|
|
|