mirror of
https://github.com/creyD/creyPY.git
synced 2026-04-14 04:10:30 +02:00
Compare commits
4 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
8b037fbeb5 | ||
| b86b58f3e4 | |||
|
|
17f96c920d | ||
|
|
523241ac4b |
2
.github/workflows/ci.yml
vendored
2
.github/workflows/ci.yml
vendored
@@ -45,7 +45,7 @@ jobs:
|
|||||||
|
|
||||||
tag_and_publish:
|
tag_and_publish:
|
||||||
runs-on: ubuntu-latest
|
runs-on: ubuntu-latest
|
||||||
if: github.ref_name == 'master' || github.ref_name == 'dev'
|
if: (github.ref_name == 'master' || github.ref_name == 'dev') && github.event_name == 'push'
|
||||||
needs: test
|
needs: test
|
||||||
permissions:
|
permissions:
|
||||||
id-token: write # IMPORTANT: this permission is mandatory for trusted publishing
|
id-token: write # IMPORTANT: this permission is mandatory for trusted publishing
|
||||||
|
|||||||
139
creyPY/fastapi/testing_async.py
Normal file
139
creyPY/fastapi/testing_async.py
Normal file
@@ -0,0 +1,139 @@
|
|||||||
|
import json
|
||||||
|
from httpx import AsyncClient
|
||||||
|
|
||||||
|
|
||||||
|
class AsyncGenericClient:
|
||||||
|
def __init__(self, app):
|
||||||
|
self.c = AsyncClient(app=app, base_url="http://testserver", follow_redirects=True)
|
||||||
|
self.default_headers = {}
|
||||||
|
|
||||||
|
async def get(self, url: str, r_code: int = 200, parse_json=True):
|
||||||
|
re = await self.c.get(url, headers=self.default_headers)
|
||||||
|
if re.status_code != r_code:
|
||||||
|
print(re.content)
|
||||||
|
assert r_code == re.status_code
|
||||||
|
return re.json() if parse_json else re.content
|
||||||
|
|
||||||
|
async def delete(self, url: str, r_code: int = 204):
|
||||||
|
re = await self.c.delete(url, headers=self.default_headers)
|
||||||
|
if re.status_code != r_code:
|
||||||
|
print(re.content)
|
||||||
|
assert r_code == re.status_code
|
||||||
|
return re.json() if r_code != 204 else None
|
||||||
|
|
||||||
|
async def post(
|
||||||
|
self, url: str, obj: dict | str = {}, r_code: int = 201, raw_response=False, *args, **kwargs
|
||||||
|
):
|
||||||
|
re = await self.c.post(
|
||||||
|
url,
|
||||||
|
data=json.dumps(obj) if isinstance(obj, dict) else obj,
|
||||||
|
headers=self.default_headers | {"Content-Type": "application/json"},
|
||||||
|
*args,
|
||||||
|
**kwargs,
|
||||||
|
)
|
||||||
|
if re.status_code != r_code:
|
||||||
|
print(re.content)
|
||||||
|
assert r_code == re.status_code
|
||||||
|
return re.json() if not raw_response else re
|
||||||
|
|
||||||
|
async def post_file(
|
||||||
|
self, url: str, file, r_code: int = 201, raw_response=False, *args, **kwargs
|
||||||
|
):
|
||||||
|
re = await self.c.post(
|
||||||
|
url,
|
||||||
|
files={"file": file},
|
||||||
|
headers=self.default_headers | {"Content-Type": "application/json"},
|
||||||
|
*args,
|
||||||
|
**kwargs,
|
||||||
|
)
|
||||||
|
if re.status_code != r_code:
|
||||||
|
print(re.content)
|
||||||
|
assert r_code == re.status_code
|
||||||
|
return re.json() if not raw_response else re
|
||||||
|
|
||||||
|
async def patch(
|
||||||
|
self, url: str, obj: dict | str = {}, r_code: int = 200, raw_response=False, *args, **kwargs
|
||||||
|
):
|
||||||
|
re = await self.c.patch(
|
||||||
|
url,
|
||||||
|
data=json.dumps(obj) if isinstance(obj, dict) else obj,
|
||||||
|
headers=self.default_headers | {"Content-Type": "application/json"},
|
||||||
|
*args,
|
||||||
|
**kwargs,
|
||||||
|
)
|
||||||
|
if re.status_code != r_code:
|
||||||
|
print(re.content)
|
||||||
|
assert r_code == re.status_code
|
||||||
|
return re.json() if not raw_response else re
|
||||||
|
|
||||||
|
async def put(
|
||||||
|
self, url: str, obj: dict | str = {}, r_code: int = 200, raw_response=False, *args, **kwargs
|
||||||
|
):
|
||||||
|
re = await self.c.put(
|
||||||
|
url,
|
||||||
|
data=json.dumps(obj) if isinstance(obj, dict) else obj,
|
||||||
|
headers=self.default_headers
|
||||||
|
| {
|
||||||
|
"Content-Type": "application/json",
|
||||||
|
"accept": "application/json",
|
||||||
|
},
|
||||||
|
*args,
|
||||||
|
**kwargs,
|
||||||
|
)
|
||||||
|
if re.status_code != r_code:
|
||||||
|
print(re.content)
|
||||||
|
assert r_code == re.status_code
|
||||||
|
return re.json() if not raw_response else re
|
||||||
|
|
||||||
|
async def obj_lifecycle(
|
||||||
|
self,
|
||||||
|
input_obj: dict,
|
||||||
|
url: str,
|
||||||
|
pagination: bool = True,
|
||||||
|
id_field: str = "id",
|
||||||
|
created_at_check: bool = True,
|
||||||
|
):
|
||||||
|
# GET LIST
|
||||||
|
re = await self.get(url)
|
||||||
|
if pagination:
|
||||||
|
assert re["total"] == 0
|
||||||
|
assert len(re["results"]) == 0
|
||||||
|
else:
|
||||||
|
assert len(re) == 0
|
||||||
|
|
||||||
|
# CREATE
|
||||||
|
re = await self.post(url, obj=input_obj)
|
||||||
|
assert id_field in re
|
||||||
|
assert re[id_field] is not None
|
||||||
|
|
||||||
|
if created_at_check:
|
||||||
|
assert "created_at" in re
|
||||||
|
assert re["created_at"] is not None
|
||||||
|
|
||||||
|
obj_id = str(re[id_field])
|
||||||
|
|
||||||
|
# GET
|
||||||
|
re = await self.get(f"{url}{obj_id}/")
|
||||||
|
assert re[id_field] == obj_id
|
||||||
|
|
||||||
|
# GET LIST
|
||||||
|
re = await self.get(url)
|
||||||
|
if pagination:
|
||||||
|
assert re["total"] == 1
|
||||||
|
assert len(re["results"]) == 1
|
||||||
|
else:
|
||||||
|
assert len(re) == 1
|
||||||
|
|
||||||
|
# DELETE
|
||||||
|
await self.delete(f"{url}{obj_id}")
|
||||||
|
|
||||||
|
# GET LIST
|
||||||
|
re = await self.get(url)
|
||||||
|
if pagination:
|
||||||
|
assert re["total"] == 0
|
||||||
|
assert len(re["results"]) == 0
|
||||||
|
else:
|
||||||
|
assert len(re) == 0
|
||||||
|
|
||||||
|
# GET
|
||||||
|
await self.get(f"{url}{obj_id}", parse_json=False, r_code=404)
|
||||||
7
renovate.json
Normal file
7
renovate.json
Normal file
@@ -0,0 +1,7 @@
|
|||||||
|
{
|
||||||
|
"$schema": "https://docs.renovatebot.com/renovate-schema.json",
|
||||||
|
"extends": [
|
||||||
|
"config:recommended",
|
||||||
|
":semanticCommitTypeAll(feat)"
|
||||||
|
]
|
||||||
|
}
|
||||||
Reference in New Issue
Block a user