mirror of
https://github.com/creyD/creyPY.git
synced 2026-04-12 19:30: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:
|
||||
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
|
||||
permissions:
|
||||
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