Fix API unit tests (APIError)

This commit is contained in:
Emmanuel Viennet 2024-08-14 10:47:48 +02:00 committed by Iziram
parent 37dcdca65b
commit 6b8667522b
2 changed files with 16 additions and 9 deletions

View File

@ -52,19 +52,20 @@ print(f"API_USER={API_USER}")
class APIError(Exception): class APIError(Exception):
def __init__(self, message: str = "", payload=None): def __init__(self, message: str = "", payload=None, status_code=None):
self.message = message self.message = message
self.payload = payload or {} self.payload = payload or {}
self.status_code = status_code
def __str__(self): def __str__(self):
return f"APIError: {self.message} payload={self.payload}" return f"APIError: {self.message} payload={self.payload} status_code={self.status_code}"
def get_auth_headers(user, password) -> dict: def get_auth_headers(user, password) -> dict:
"Demande de jeton, dict à utiliser dans les en-têtes de requêtes http" "Demande de jeton, dict à utiliser dans les en-têtes de requêtes http"
ans = requests.post(API_URL + "/tokens", auth=(user, password), timeout=5) ans = requests.post(API_URL + "/tokens", auth=(user, password), timeout=5)
if ans.status_code != 200: if ans.status_code != 200:
raise APIError(f"Echec demande jeton par {user}") raise APIError(f"Echec demande jeton par {user}", status_code=ans.status_code)
token = ans.json()["token"] token = ans.json()["token"]
return {"Authorization": f"Bearer {token}"} return {"Authorization": f"Bearer {token}"}
@ -109,11 +110,12 @@ def GET(path: str, headers: dict = None, errmsg=None, dept=None, raw=False):
timeout=SCO_TEST_API_TIMEOUT, timeout=SCO_TEST_API_TIMEOUT,
) )
if reply.status_code != 200: if reply.status_code != 200:
print("url", SCODOC_URL)
print("url", url) print("url", url)
print("reply", reply.text) print("reply", reply.text)
raise APIError( raise APIError(
errmsg or f"""erreur status={reply.status_code} !""", reply.json() errmsg or f"""erreur get {url} !""",
reply.json(),
status_code=reply.status_code,
) )
if raw: if raw:
return reply return reply
@ -129,7 +131,10 @@ def GET(path: str, headers: dict = None, errmsg=None, dept=None, raw=False):
"Content-Disposition": reply.headers.get("Content-Disposition", None), "Content-Disposition": reply.headers.get("Content-Disposition", None),
} }
return retval return retval
raise APIError("Unknown returned content {r.headers.get('Content-Type', None} !\n") raise APIError(
"Unknown returned content {r.headers.get('Content-Type', None} !\n",
status_code=reply.status_code,
)
def POST( def POST(
@ -161,7 +166,9 @@ def POST(
except requests.exceptions.JSONDecodeError: except requests.exceptions.JSONDecodeError:
payload = r.text payload = r.text
raise APIError( raise APIError(
errmsg or f"erreur url={url} status={r.status_code} !", payload=payload errmsg or f"erreur url={url} status={r.status_code} !",
payload=payload,
status_code=r.status_code,
) )
return r if raw else r.json() # decode la reponse JSON return r if raw else r.json() # decode la reponse JSON

View File

@ -185,7 +185,7 @@ def test_modif_users_depts(api_admin_headers):
headers=admin_h, headers=admin_h,
) )
except APIError as exc: except APIError as exc:
if exc.args[1]["status"] == 400: if exc.status_code == 400:
ok = True ok = True
assert ok assert ok
# Un "vrai" mot de passe: # Un "vrai" mot de passe:
@ -234,7 +234,7 @@ def test_modif_users_depts(api_admin_headers):
dept=dept3["acronym"], dept=dept3["acronym"],
) )
except APIError as exc: except APIError as exc:
if exc.args[1]["status"] == 401: if exc.status_code == 401:
ok = True ok = True
assert ok assert ok
# Nettoyage: # Nettoyage: