From 547040bb93a1e58d37ee8dc6e8867cf9661011ed Mon Sep 17 00:00:00 2001 From: iziram Date: Wed, 1 Feb 2023 20:00:14 +0100 Subject: [PATCH] =?UTF-8?q?justificatifs=20:=20=E2=9C=85archivage+test=20a?= =?UTF-8?q?pi=20(sauf=20export)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- app/api/justificatif.py | 186 ++++++++++++++++++++--- app/scodoc/sco_archives_justificatifs.py | 60 +++++--- tests/api/test_api_justificatif.txt | 1 + tests/api/test_api_justificatif2.txt | 1 + tests/api/test_api_justificatifs.py | 123 ++++++++++++++- 5 files changed, 332 insertions(+), 39 deletions(-) create mode 100644 tests/api/test_api_justificatif.txt create mode 100644 tests/api/test_api_justificatif2.txt diff --git a/app/api/justificatif.py b/app/api/justificatif.py index 66e58b9e4..c51b0459a 100644 --- a/app/api/justificatif.py +++ b/app/api/justificatif.py @@ -24,22 +24,6 @@ from flask_login import login_required from app.scodoc.sco_utils import json_error -# @bp.route("/justificatif/import") -# @api_web_bp.route("/justificatif/import") -# @scodoc -# def justificatif(): -# """ """ -# archiver: JustificatifArchiver = JustificatifArchiver() - -# filename: str = "lol.txt" -# data: bytes = "test".encode("utf-8") - -# archiver.save_justificatif( -# etudid=1, filename=filename, data=data, archive_id="2023-02-01-10-29-20" -# ) -# return jsonify([filename, "done"]) - - # @bp.route("/justificatif/remove") # @api_web_bp.route("/justificatif/remove") # @scodoc @@ -133,8 +117,6 @@ def justificatifs(etudid: int = None, with_query: bool = False): # TODO: justificatif-create - - @bp.route("/justificatif//create", methods=["POST"]) @api_web_bp.route("/justificatif//create", methods=["POST"]) @scodoc @@ -336,9 +318,177 @@ def _delete_singular(justif_id: int, database): # Partie archivage # TODO: justificatif-import +@bp.route("/justificatif/import/", methods=["POST"]) +@api_web_bp.route("/justificatif/import/", methods=["POST"]) +@scodoc +@login_required +@permission_required(Permission.ScoView) +# @permission_required(Permission.ScoAssiduiteChange) +def justif_import(justif_id: int = None): + """ + Importation d'un fichier (création d'archive) + """ + if len(request.files) == 0: + return json_error(404, "Il n'y a pas de fichier joint") + + file = list(request.files.values())[0] + if file.filename == "": + return json_error(404, "Il n'y a pas de fichier joint") + + query = Justificatif.query.filter_by(id=justif_id) + if g.scodoc_dept: + query = query.join(Identite).filter_by(dept_id=g.scodoc_dept_id) + + justificatif_unique: Justificatif = query.first_or_404() + + archive_name: str = justificatif_unique.fichier + + archiver: JustificatifArchiver = JustificatifArchiver() + try: + archive_name: str = archiver.save_justificatif( + etudid=justificatif_unique.etudid, + filename=file.filename, + data=file.stream.read(), + archive_name=archive_name, + ) + + justificatif_unique.fichier = archive_name + + db.session.add(justificatif_unique) + db.session.commit() + + return jsonify({"response": "imported"}) + except ScoValueError as err: + return json_error(404, err.args[1]) + + # TODO: justificatif-export +@bp.route("/justificatif/export//", methods=["GET"]) +@api_web_bp.route("/justificatif/export//", methods=["GET"]) +@scodoc +@login_required +@permission_required(Permission.ScoView) +# @permission_required(Permission.ScoAssiduiteChange) +def justif_export(justif_id: int = None, filename: str = None): + """ + Retourne un fichier d'une archive d'un justificatif + """ + + query = Justificatif.query.filter_by(id=justif_id) + if g.scodoc_dept: + query = query.join(Identite).filter_by(dept_id=g.scodoc_dept_id) + + justificatif_unique: Justificatif = query.first_or_404() + + archive_name: str = justificatif_unique.fichier + if archive_name is None: + return json_error(404, "le justificatif ne possède pas de fichier") + + archiver: JustificatifArchiver = JustificatifArchiver() + + try: + return archiver.get_justificatif_file( + archive_name, justificatif_unique.etudid, filename + ) + except ScoValueError as err: + return json_error(404, err.args[1]) + + # TODO: justificatif-remove +@bp.route("/justificatif/remove/", methods=["POST"]) +@api_web_bp.route("/justificatif/remove/", methods=["POST"]) +@scodoc +@login_required +@permission_required(Permission.ScoView) +# @permission_required(Permission.ScoAssiduiteChange) +def justif_remove(justif_id: int = None): + """ + Supression d'un fichier ou d'une archive + + { + "remove": <"all"/"list"> + + "filenames"?: [ + , + ... + ] + } + """ + + data: dict = request.get_json(force=True) + + query = Justificatif.query.filter_by(id=justif_id) + if g.scodoc_dept: + query = query.join(Identite).filter_by(dept_id=g.scodoc_dept_id) + + justificatif_unique: Justificatif = query.first_or_404() + + archive_name: str = justificatif_unique.fichier + if archive_name is None: + return json_error(404, "le justificatif ne possède pas de fichier") + + remove: str = data.get("remove") + if remove is None or remove not in ("all", "list"): + return json_error(404, "param 'remove': Valeur invalide") + archiver: JustificatifArchiver = JustificatifArchiver() + etudid: int = justificatif_unique.etudid + try: + if remove == "all": + archiver.delete_justificatif(etudid=etudid, archive_name=archive_name) + justificatif_unique.fichier = None + db.session.add(justificatif_unique) + db.session.commit() + + else: + for fname in data.get("filenames", []): + archiver.delete_justificatif( + etudid=etudid, + archive_name=archive_name, + filename=fname, + ) + + if len(archiver.list_justificatifs(archive_name, etudid)) == 0: + archiver.delete_justificatif(etudid, archive_name) + justificatif_unique.fichier = None + db.session.add(justificatif_unique) + db.session.commit() + + except ScoValueError as err: + return json_error(404, err.args[1]) + + return jsonify({"response": "removed"}) + + # TODO: justificatif-list +@bp.route("/justificatif/list/", methods=["GET"]) +@api_web_bp.route("/justificatif/list/", methods=["GET"]) +@scodoc +@login_required +@permission_required(Permission.ScoView) +# @permission_required(Permission.ScoAssiduiteChange) +def justif_list(justif_id: int = None): + """ + Liste les fichiers du justificatif + """ + + query = Justificatif.query.filter_by(id=justif_id) + if g.scodoc_dept: + query = query.join(Identite).filter_by(dept_id=g.scodoc_dept_id) + + justificatif_unique: Justificatif = query.first_or_404() + + archive_name: str = justificatif_unique.fichier + + filenames: list[str] = [] + + archiver: JustificatifArchiver = JustificatifArchiver() + if archive_name is not None: + filenames = archiver.list_justificatifs( + archive_name, justificatif_unique.etudid + ) + + return jsonify(filenames) + # Partie justification # TODO: justificatif-justified diff --git a/app/scodoc/sco_archives_justificatifs.py b/app/scodoc/sco_archives_justificatifs.py index dbbefcba8..e64d788a1 100644 --- a/app/scodoc/sco_archives_justificatifs.py +++ b/app/scodoc/sco_archives_justificatifs.py @@ -1,4 +1,7 @@ from app.scodoc.sco_archives import BaseArchiver +from app.scodoc.sco_exceptions import ScoValueError +from app.models import Identite, Departement +from flask import g import os @@ -29,34 +32,39 @@ class JustificatifArchiver(BaseArchiver): etudid: int, filename: str, data: bytes or str, - archive_id: str = None, + archive_name: str = None, description: str = "", - ): + ) -> str: """ Ajoute un fichier dans une archive "justificatif" pour l'etudid donné + Retourne l'archive_name utilisé """ - if archive_id is None: + self._set_dept(etudid) + if archive_name is None: archive_id: str = self.create_obj_archive( oid=etudid, description=description ) else: - archive_id = self._true_archive_id(archive_id, etudid) + archive_id: str = self.get_id_from_name(etudid, archive_name) self.store(archive_id, filename, data) - def delete_justificatif(self, etudid: int, archive_id: str, filename: str = None): + return self.get_archive_name(archive_id) + + def delete_justificatif(self, etudid: int, archive_name: str, filename: str = None): """ Supprime une archive ou un fichier particulier de l'archivage de l'étudiant donné """ + self._set_dept(etudid) if str(etudid) not in self.list_oids(): raise ValueError(f"Aucune archive pour etudid[{etudid}]") - archive_id = self._true_archive_id(archive_id, etudid) + archive_id = self.get_id_from_name(etudid, archive_name) if filename is not None: if filename not in self.list_archive(archive_id): raise ValueError( - f"filename inconnu dans l'archive archive_id[{archive_id}] -> etudid[{etudid}]" + f"filename {filename} inconnu dans l'archive archive_id[{archive_id}] -> etudid[{etudid}]" ) path: str = os.path.join(self.get_obj_dir(etudid), archive_id, filename) @@ -72,19 +80,33 @@ class JustificatifArchiver(BaseArchiver): ) ) - def _true_archive_id(self, archive_id: str, etudid: int): + def list_justificatifs(self, archive_name: str, etudid: int) -> list[str]: """ - Test si l'archive_id est bien dans le dossier d'archive - Retourne le chemin complet de l'id + Retourne la liste des noms de fichiers dans l'archive donnée """ - archives: list[str] = [ - arc for arc in self.list_obj_archives(etudid) if archive_id in arc - ] + self._set_dept(etudid) + filenames: list[str] = [] + archive_id = self.get_id_from_name(etudid, archive_name) - if len(archives) == 0: - raise ValueError( - f"archive_id[{archive_id}] inconnu pour etudid[{etudid}]", - self.list_obj_archives(etudid), - ) + filenames = self.list_archive(archive_id) + return filenames - return archives[0] + def get_justificatif_file(self, archive_name: str, etudid: int, filename: str): + """ + Retourne une réponse de téléchargement de fichier si le fichier existe + """ + self._set_dept(etudid) + archive_id: str = self.get_id_from_name(etudid, archive_name) + if filename in self.list_archive(archive_id): + return self.get_archived_file(etudid, archive_name, filename) + raise ScoValueError( + f"Fichier {filename} introuvable dans l'archive {archive_name}" + ) + + def _set_dept(self, etudid: int): + if g.scodoc_dept is None or g.scodoc_dept_id is None: + etud: Identite = Identite.query.filter_by(id=etudid).first() + dept: Departement = Departement.query.filter_by(id=etud.dept_id).first() + + g.scodoc_dept = dept.acronym + g.scodoc_dept_id = dept.id diff --git a/tests/api/test_api_justificatif.txt b/tests/api/test_api_justificatif.txt new file mode 100644 index 000000000..370b0a4f6 --- /dev/null +++ b/tests/api/test_api_justificatif.txt @@ -0,0 +1 @@ +test de l'importation des fichiers / archive justificatif \ No newline at end of file diff --git a/tests/api/test_api_justificatif2.txt b/tests/api/test_api_justificatif2.txt new file mode 100644 index 000000000..370b0a4f6 --- /dev/null +++ b/tests/api/test_api_justificatif2.txt @@ -0,0 +1 @@ +test de l'importation des fichiers / archive justificatif \ No newline at end of file diff --git a/tests/api/test_api_justificatifs.py b/tests/api/test_api_justificatifs.py index 0c1ce6adb..875f767b4 100644 --- a/tests/api/test_api_justificatifs.py +++ b/tests/api/test_api_justificatifs.py @@ -7,7 +7,15 @@ Ecrit par HARTMANN Matthias from random import randint -from tests.api.setup_test_api import GET, POST_JSON, APIError, api_headers +from tests.api.setup_test_api import ( + GET, + POST_JSON, + APIError, + api_headers, + API_URL, + CHECK_CERTIFICATE, +) +import requests ETUDID = 1 FAUX = 42069 @@ -60,7 +68,7 @@ def check_failure_post(path, headers, data, err=None): if err is not None: assert api_err.payload["message"] == err else: - raise APIError("Le GET n'aurait pas du fonctionner") + raise APIError("Le POST n'aurait pas du fonctionner") def create_data(etat: str, day: str, raison: str = None): @@ -253,3 +261,114 @@ def test_route_delete(api_headers): assert len(res["errors"]) == 3 assert all([res["errors"][i] == "Justificatif non existant" for i in res["errors"]]) + + +# Gestion de l'archivage + + +def send_file(justif_id: int, filename: str, headers): + """ + Envoi un fichier vers la route d'importation + """ + with open(filename, "rb") as file: + url: str = API_URL + f"/justificatif/import/{justif_id}" + r = requests.post( + url, + files={filename: file}, + headers=headers, + verify=CHECK_CERTIFICATE, + ) + + if r.status_code != 200: + raise APIError(f"erreur status={r.status_code} !", r.json()) + else: + return r.json() + + +def check_failure_send( + justif_id: int, + headers, + filename: str = "/opt/scodoc/tests/api/test_api_justificatif.txt", + err: str = None, +): + try: + data = send_file(justif_id, filename, headers) + # ^ Renvoi un 404 + except APIError as api_err: + if err is not None: + assert api_err.payload["message"] == err + else: + raise APIError("Le POST n'aurait pas du fonctionner") + + +def test_import_justificatif(api_headers): + + # Bon fonctionnement + + filename: str = "/opt/scodoc/tests/api/test_api_justificatif.txt" + resp: dict = send_file(1, filename, api_headers) + assert "response" in resp + assert resp["response"] == "imported" + + filename: str = "/opt/scodoc/tests/api/test_api_justificatif2.txt" + resp: dict = send_file(1, filename, api_headers) + assert "response" in resp + assert resp["response"] == "imported" + + # Mauvais fonctionnement + + check_failure_send(FAUX, api_headers) + + +def test_list_justificatifs(api_headers): + + # Bon fonctionnement + + res: list = GET("/justificatif/list/1", api_headers) + + assert isinstance(res, list) + assert len(res) == 2 + + res: list = GET("/justificatif/list/2", api_headers) + + assert isinstance(res, list) + assert len(res) == 0 + + # Mauvais fonctionnement + + check_failure_get(f"/justificatif/list/{FAUX}", api_headers) + + +def test_remove_justificatif(api_headers): + + # Bon fonctionnement + + filename: str = "/opt/scodoc/tests/api/test_api_justificatif.txt" + send_file(2, filename, api_headers) + filename: str = "/opt/scodoc/tests/api/test_api_justificatif2.txt" + send_file(2, filename, api_headers) + + res: dict = POST_JSON("/justificatif/remove/1", {"remove": "all"}, api_headers) + assert res == {"response": "removed"} + assert len(GET("/justificatif/list/1", api_headers)) == 0 + + res: dict = POST_JSON( + "/justificatif/remove/2", + {"remove": "list", "filenames": ["test_api_justificatif2.txt"]}, + api_headers, + ) + assert res == {"response": "removed"} + assert len(GET("/justificatif/list/2", api_headers)) == 1 + + res: dict = POST_JSON( + "/justificatif/remove/2", + {"remove": "list", "filenames": ["test_api_justificatif.txt"]}, + api_headers, + ) + assert res == {"response": "removed"} + assert len(GET("/justificatif/list/2", api_headers)) == 0 + + # Mauvais fonctionnement + + check_failure_post("/justificatif/remove/2", api_headers, {}) + check_failure_post(f"/justificatif/remove/{FAUX}", api_headers, {"remove": "all"})