forked from ScoDoc/ScoDoc
Fix API unit tests (APIError)
This commit is contained in:
parent
07c2f00d0d
commit
cb0c9d8f53
@ -52,19 +52,20 @@ print(f"API_USER={API_USER}")
|
||||
|
||||
|
||||
class APIError(Exception):
|
||||
def __init__(self, message: str = "", payload=None):
|
||||
def __init__(self, message: str = "", payload=None, status_code=None):
|
||||
self.message = message
|
||||
self.payload = payload or {}
|
||||
self.status_code = status_code
|
||||
|
||||
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:
|
||||
"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)
|
||||
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"]
|
||||
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,
|
||||
)
|
||||
if reply.status_code != 200:
|
||||
print("url", SCODOC_URL)
|
||||
print("url", url)
|
||||
print("reply", reply.text)
|
||||
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:
|
||||
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),
|
||||
}
|
||||
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(
|
||||
@ -161,7 +166,9 @@ def POST(
|
||||
except requests.exceptions.JSONDecodeError:
|
||||
payload = r.text
|
||||
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
|
||||
|
||||
|
@ -185,7 +185,7 @@ def test_modif_users_depts(api_admin_headers):
|
||||
headers=admin_h,
|
||||
)
|
||||
except APIError as exc:
|
||||
if exc.args[1]["status"] == 400:
|
||||
if exc.status_code == 400:
|
||||
ok = True
|
||||
assert ok
|
||||
# Un "vrai" mot de passe:
|
||||
@ -234,7 +234,7 @@ def test_modif_users_depts(api_admin_headers):
|
||||
dept=dept3["acronym"],
|
||||
)
|
||||
except APIError as exc:
|
||||
if exc.args[1]["status"] == 401:
|
||||
if exc.status_code == 401:
|
||||
ok = True
|
||||
assert ok
|
||||
# Nettoyage:
|
||||
|
Loading…
Reference in New Issue
Block a user