woggioni/bugis
1
0
Fork 0
Code Actions Packages Releases Activity

tmp

Browse Source
This commit is contained in:
Walter Oggioni
2024-11-12 07:22:45 +08:00
parent 544229b7a6
commit ee6e645cc1
10 changed files with 475 additions and 164 deletions
Download Patch File Download Diff File Expand all files Collapse all files

89
core/tests/test_asgi.py
View File

@@ -1,7 +1,9 @@
import unittest
import json
import httpx
from pwo import async_test
from bugis.core import BugisApp, HttpContext
from bugis.core import BugisApp, HttpContext, HttpMethod
from typing import Sequence
class AsgiTest(unittest.TestCase):
@@ -18,6 +20,39 @@ class AsgiTest(unittest.TestCase):
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.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
}))
@async_test
async def test_hello(self):
transport = httpx.ASGITransport(app=self.app)
@@ -38,3 +73,55 @@ class AsgiTest(unittest.TestCase):
r = await client.get("/hello4")
self.assertEqual(r.status_code, 404)
self.assertTrue(len(r.text) == 0)
@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(r.status_code, 200)
response = json.loads(r.text)
self.assertEqual(['fizz', 'baz'], response)
@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(r.status_code, 200)
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(r.status_code, 200)
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(r.status_code, 200)
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(r.status_code, 200)
response = json.loads(r.text)
self.assertEqual({
'department_id': 189350,
'employee_id': 101325
}, response)