diff --git a/.travis.yml b/.travis.yml index b98ad20..50b0400 100644 --- a/.travis.yml +++ b/.travis.yml @@ -3,10 +3,6 @@ python: - "3.6" - "3.7" - "3.8" -# command to install dependencies -install: - - pip install pytest - - pip install hypothesis # command to run tests script: - - pytest \ No newline at end of file + - python setup.py test diff --git a/Changelog.md b/Changelog.md index d1fe712..78fdbc5 100644 --- a/Changelog.md +++ b/Changelog.md @@ -1,13 +1,17 @@ # Changelog -##[v0.1.1] +## [v0.2.0] +- Added new methods for adding and removing extra marks (#15) +- Rewrite everything to use the requests library (#26) + +## [v0.1.1] - Fixed bug which caused file uploads to be PUT requests by default instead of POST (#22) -##[v0.1.0] +## [v0.1.0] - this release includes functions to call all API routes for MarkUs version 1.9.0 (#10, #12, #13, #14, #18) - updated documentation to give correct instructions for pip installation (#11) - made some functions private that should have been private initially (#17) -##[v0.0.1] +## [v0.0.1] - initial release! - this release includes functions to call all API routes for MarkUs version 1.8.0 diff --git a/markusapi/markusapi.py b/markusapi/markusapi.py index cad8988..6ca169d 100644 --- a/markusapi/markusapi.py +++ b/markusapi/markusapi.py @@ -14,21 +14,17 @@ # (c) by the authors, 2008 - 2020. # -import http.client import json import mimetypes -from typing import Optional, List, Union, Dict, Tuple +import requests +from typing import Optional, List, Union, Dict from datetime import datetime -from urllib.parse import urlparse, urlencode - -ResponseType = Tuple[int, str, bytes] +from .response_parser import parse_response class Markus: """A class for interfacing with the MarkUs API.""" - API_PATH = "/api" # The root api path. - def __init__(self, api_key: str, url: str) -> None: """ Initialize an instance of the Markus class. @@ -41,20 +37,26 @@ def __init__(self, api_key: str, url: str) -> None: url -- the root domain of the MarkUs instance. """ self.api_key = api_key - self.parsed_url = urlparse(url.strip()) - assert self.parsed_url.scheme in ["http", "https"] + self.url = url + + @property + def _auth_header(self): + return {"Authorization": f"MarkUsAuth {self.api_key}"} + + def _url(self, tail=""): + return f"{self.url}/api/{tail}.json" - def get_all_users(self) -> List[dict]: + @parse_response("json") + def get_all_users(self) -> requests.Response: """ Return a list of every user in the MarkUs instance. Each user is a dictionary object, with the following keys: 'id', 'user_name', 'first_name', 'last_name', 'type', 'grace_credits', 'notes_count', 'hidden'. """ - params = None - response = self._submit_request(params, "/api/users.json", "GET") - return Markus._decode_json_response(response) + return requests.get(self._url("users"), headers=self._auth_header) + @parse_response("json") def new_user( self, user_name: str, @@ -63,7 +65,7 @@ def new_user( last_name: str, section_name: Optional[str] = None, grace_credits: Optional[str] = None, - ) -> ResponseType: + ) -> requests.Response: """ Add a new user to the MarkUs database. Returns a list containing the response's status, @@ -74,71 +76,66 @@ def new_user( params["section_name"] = section_name if grace_credits is not None: params["grace_credits"] = grace_credits - return self._submit_request(params, "/api/users", "POST") + return requests.post(self._url("users"), params=params, headers=self._auth_header) - def get_assignments(self) -> List[dict]: + @parse_response("json") + def get_assignments(self) -> requests.Response: """ Return a list of all assignments. """ - params = None - response = self._submit_request(params, "/api/assignments.json", "GET") - return Markus._decode_json_response(response) + return requests.get(self._url("assignments"), headers=self._auth_header) - def get_groups(self, assignment_id: int) -> List[dict]: + @parse_response("json") + def get_groups(self, assignment_id: int) -> requests.Response: """ Return a list of all groups associated with the given assignment. """ - params = None - path = Markus._get_path(assignments=assignment_id, groups=None) + ".json" - response = self._submit_request(params, path, "GET") - return Markus._decode_json_response(response) + return requests.get(self._url(f"assignments/{assignment_id}/groups"), headers=self._auth_header) - def get_groups_by_name(self, assignment_id: int) -> dict: + @parse_response("json") + def get_groups_by_name(self, assignment_id: int) -> requests.Response: """ Return a dictionary mapping group names to group ids. """ - params = None - path = Markus._get_path(assignments=assignment_id, groups=None, group_ids_by_name=None) + ".json" - response = self._submit_request(params, path, "GET") - return Markus._decode_json_response(response) + return requests.get( + self._url(f"assignments/{assignment_id}/groups/group_ids_by_name"), headers=self._auth_header + ) - def get_group(self, assignment_id: int, group_id: int) -> dict: + @parse_response("json") + def get_group(self, assignment_id: int, group_id: int) -> requests.Response: """ Return the group info associated with the given id and assignment. """ - params = None - path = Markus._get_path(assignments=assignment_id, groups=group_id) + ".json" - response = self._submit_request(params, path, "GET") - return Markus._decode_json_response(response) + return requests.get(self._url(f"assignments/{assignment_id}/groups/{group_id}"), headers=self._auth_header) - def get_feedback_files(self, assignment_id: int, group_id: int) -> List[dict]: + @parse_response("json") + def get_feedback_files(self, assignment_id: int, group_id: int) -> requests.Response: """ Get the feedback files info associated with the assignment and group. """ - params = {} - path = Markus._get_path(assignments=assignment_id, groups=group_id, feedback_files=None) + ".json" - response = self._submit_request(params, path, "GET") - return Markus._decode_json_response(response) + return requests.get( + self._url(f"assignments/{assignment_id}/groups/{group_id}/feedback_files"), headers=self._auth_header + ) - def get_feedback_file(self, assignment_id: int, group_id: int, feedback_file_id: int) -> str: + @parse_response("content") + def get_feedback_file(self, assignment_id: int, group_id: int, feedback_file_id: int) -> requests.Response: """ Get the feedback file associated with the given id, assignment and group. WARNING: This will fail for non-text feedback files """ - params = {} - path = Markus._get_path(assignments=assignment_id, groups=group_id, feedback_files=feedback_file_id) + ".json" - response = self._submit_request(params, path, "GET") - return Markus._decode_text_response(response) + return requests.get( + self._url(f"assignments/{assignment_id}/groups/{group_id}/feedback_files/{feedback_file_id}"), + headers=self._auth_header, + ) - def get_grades_summary(self, assignment_id: int) -> str: + @parse_response("text") + def get_grades_summary(self, assignment_id: int) -> requests.Response: """ Get grades summary csv file as a string. """ - params = {} - path = Markus._get_path(assignments=assignment_id, grades_summary=None) + ".json" - response = self._submit_request(params, path, "GET") - return Markus._decode_text_response(response) + return requests.get(self._url(f"assignments/{assignment_id}/grades_summary"), headers=self._auth_header) + @parse_response("json") def new_marks_spreadsheet( self, short_identifier: str, @@ -146,8 +143,8 @@ def new_marks_spreadsheet( date: Optional[datetime] = None, is_hidden: bool = True, show_total: bool = True, - grade_entry_items: Optional[Dict[str, Union[str, bool, float]]] = None, - ) -> ResponseType: + grade_entry_items: Optional[List[Dict[str, Union[str, bool, float]]]] = None, + ) -> requests.Response: """ Create a new marks spreadsheet """ @@ -159,9 +156,9 @@ def new_marks_spreadsheet( "show_total": show_total, "grade_entry_items": grade_entry_items, } - path = Markus._get_path(grade_entry_forms=None) + ".json" - return self._submit_request(params, path, "POST", content_type="application/json") + return requests.post(self._url("grade_entry_forms"), params=params, headers=self._auth_header) + @parse_response("json") def update_marks_spreadsheet( self, spreadsheet_id: int, @@ -170,8 +167,8 @@ def update_marks_spreadsheet( date: Optional[datetime] = None, is_hidden: Optional[bool] = None, show_total: Optional[bool] = None, - grade_entry_items: Optional[Dict[str, Union[str, bool, float]]] = None, - ) -> ResponseType: + grade_entry_items: Optional[List[Dict[str, Union[str, bool, float]]]] = None, + ) -> requests.Response: """ Update an existing marks spreadsheet """ @@ -186,34 +183,32 @@ def update_marks_spreadsheet( for name in list(params): if params[name] is None: params.pop(name) - path = Markus._get_path(grade_entry_forms=spreadsheet_id) + ".json" - return self._submit_request(params, path, "PUT", content_type="application/json") + return requests.put(self._url(f"grade_entry_forms/{spreadsheet_id}"), params=params, headers=self._auth_header) + @parse_response("json") def update_marks_spreadsheets_grades( self, spreadsheet_id: int, user_name: str, grades_per_column: Dict[str, float] - ) -> ResponseType: + ) -> requests.Response: params = {"user_name": user_name, "grade_entry_items": grades_per_column} - path = Markus._get_path(grade_entry_forms=spreadsheet_id, update_grades=None) + ".json" - return self._submit_request(params, path, "PUT", content_type="application/json") + return requests.put( + self._url(f"grade_entry_forms/{spreadsheet_id}/update_grades"), json=params, headers=self._auth_header + ) - def get_marks_spreadsheets(self) -> List[dict]: + @parse_response("json") + def get_marks_spreadsheets(self) -> requests.Response: """ Get all marks spreadsheets. """ - params = {} - path = Markus._get_path(grade_entry_forms=None) + ".json" - response = self._submit_request(params, path, "GET") - return Markus._decode_json_response(response) + return requests.get(self._url(f"grade_entry_forms"), headers=self._auth_header) - def get_marks_spreadsheet(self, spreadsheet_id: int) -> str: + @parse_response("text") + def get_marks_spreadsheet(self, spreadsheet_id: int) -> requests.Response: """ Get the marks spreadsheet associated with the given id. """ - params = {} - path = Markus._get_path(grade_entry_forms=spreadsheet_id) + ".json" - response = self._submit_request(params, path, "GET") - return Markus._decode_text_response(response) + return requests.get(self._url(f"grade_entry_forms/{spreadsheet_id}"), headers=self._auth_header) + @parse_response("json") def upload_feedback_file( self, assignment_id: int, @@ -222,7 +217,7 @@ def upload_feedback_file( contents: Union[str, bytes], mime_type: Optional[str] = None, overwrite: bool = True, - ) -> ResponseType: + ) -> requests.Response: """ Upload a feedback file to Markus. @@ -234,29 +229,39 @@ def upload_feedback_file( mime_type -- mime type of title file, if None then the mime type will be guessed based on the file extension overwrite -- whether to overwrite a feedback file with the same name that already exists in Markus """ - feedback_file_id = None + url_content = f"assignments/{assignment_id}/groups/{group_id}/feedback_files" if overwrite: feedback_files = self.get_feedback_files(assignment_id, group_id) feedback_file_id = next((ff.get("id") for ff in feedback_files if ff.get("filename") == title), None) - path = Markus._get_path(assignments=assignment_id, groups=group_id, feedback_files=None) - request_type = "POST" - if feedback_file_id: - path = "{}/{}".format(path, feedback_file_id) - request_type = "PUT" - - return self._do_file_upload(path, title, contents, mime_type, request_type) + if feedback_file_id is not None: + url_content += f"/{feedback_file_id}" + else: + overwrite = False + files = {"file_content": (title, contents)} + params = {"filename": title, "mime_type": mime_type or mimetypes.guess_type(title)[0]} + if overwrite: + return requests.put(self._url(url_content), files=files, params=params, headers=self._auth_header) + else: + return requests.post(self._url(url_content), files=files, params=params, headers=self._auth_header) + @parse_response("json") def upload_test_group_results( - self, assignment_id: int, group_id: int, test_run_id: int, test_output: str - ) -> ResponseType: + self, assignment_id: int, group_id: int, test_run_id: int, test_output: Union[str, Dict] + ) -> requests.Response: """ Upload test results to Markus """ + if not isinstance(test_output, str): + test_output = json.dumps(test_output) params = {"test_run_id": test_run_id, "test_output": test_output} - path = Markus._get_path(assignments=assignment_id, groups=group_id, test_group_results=None) - return self._submit_request(params, path, "POST") + return requests.post( + self._url(f"assignments/{assignment_id}/groups/{group_id}/test_group_results"), + json=params, + headers=self._auth_header, + ) + @parse_response("json") def upload_annotations( - self, assignment_id: int, group_id: int, annotations: dict, force_complete: bool = False - ) -> ResponseType: + self, assignment_id: int, group_id: int, annotations: List, force_complete: bool = False + ) -> requests.Response: """ Each element of annotations must be a dictionary with the following keys: - filename @@ -270,20 +275,26 @@ def upload_annotations( This currently only works for plain-text file submissions. """ params = {"annotations": annotations, "force_complete": force_complete} - path = Markus._get_path(assignments=assignment_id, groups=group_id, add_annotations=None) - return self._submit_request(params, path, "POST", "application/json") + return requests.post( + self._url(f"assignments/{assignment_id}/groups/{group_id}/add_annotations"), + json=params, + headers=self._auth_header, + ) - def get_annotations(self, assignment_id: int, group_id: Optional[int] = None) -> List[Dict]: + @parse_response("json") + def get_annotations(self, assignment_id: int, group_id: Optional[int] = None) -> requests.Response: """ Return a list of dictionaries containing information for each annotation in the assignment with id = assignment_id. If group_id is not None, return only annotations for the given group. """ - params = None - path = Markus._get_path(assignments=assignment_id, groups=group_id, annotations=None) + ".json" - response = self._submit_request(params, path, "GET") - return Markus._decode_json_response(response) + return requests.get( + self._url(f"assignments/{assignment_id}/groups/{group_id}/annotations"), headers=self._auth_header + ) - def update_marks_single_group(self, criteria_mark_map: dict, assignment_id: int, group_id: int) -> ResponseType: + @parse_response("json") + def update_marks_single_group( + self, criteria_mark_map: dict, assignment_id: int, group_id: int + ) -> requests.Response: """ Update the marks of a single group. Only the marks specified in criteria_mark_map will be changed. @@ -299,19 +310,56 @@ def update_marks_single_group(self, criteria_mark_map: dict, assignment_id: int, assignment_id -- the assignment's id group_id -- the id of the group whose marks we are updating """ - params = criteria_mark_map - path = Markus._get_path(assignments=assignment_id, groups=group_id, update_marks=None) - return self._submit_request(params, path, "PUT") + return requests.put( + self._url(f"assignments/{assignment_id}/groups/{group_id}/update_marks"), + json=criteria_mark_map, + headers=self._auth_header, + ) - def update_marking_state(self, assignment_id: int, group_id: int, new_marking_state: str) -> ResponseType: + @parse_response("json") + def update_marking_state(self, assignment_id: int, group_id: int, new_marking_state: str) -> requests.Response: """ Update marking state for a single group to either 'complete' or 'incomplete' """ params = {"marking_state": new_marking_state} - path = Markus._get_path(assignments=assignment_id, groups=group_id, update_marking_state=None) - return self._submit_request(params, path, "PUT") - + return requests.put( + self._url(f"assignments/{assignment_id}/groups/{group_id}/update_marking_state"), + params=params, + headers=self._auth_header, + ) + + @parse_response("json") + def create_extra_marks( + self, assignment_id: int, group_id: int, extra_marks: float, description: str + ) -> requests.Response: + """ + Create new extra mark for the particular group. + Mark specified in extra_marks will be created + """ + params = {"extra_marks": extra_marks, "description": description} + return requests.post( + self._url(f"assignments/{assignment_id}/groups/{group_id}/create_extra_marks"), + params=params, + headers=self._auth_header, + ) + + @parse_response("json") + def remove_extra_marks( + self, assignment_id: int, group_id: int, extra_marks: float, description: str + ) -> requests.Response: + """ + Remove the extra mark for the particular group. + Mark specified in extra_marks will be removed + """ + params = {"extra_marks": extra_marks, "description": description} + return requests.delete( + self._url(f"assignments/{assignment_id}/groups/{group_id}/remove_extra_marks"), + params=params, + headers=self._auth_header, + ) + + @parse_response("content") def get_files_from_repo( self, assignment_id: int, group_id: int, filename: Optional[str] = None, collected: bool = True - ) -> Optional[bytes]: + ) -> requests.Response: """ Return file content from the submission of a single group. If is specified, return the content of a single file, otherwise return the content of a zipfile containing @@ -322,21 +370,27 @@ def get_files_from_repo( The method returns None if there are no files to collect. """ - path = Markus._get_path(assignments=assignment_id, groups=group_id, submission_files=None) + ".json" params = {} if collected: params["collected"] = collected if filename: params["filename"] = filename - response = self._submit_request(params, path, "GET") - if response[0] == 200: - return response[2] - - def upload_folder_to_repo(self, assignment_id: int, group_id: int, folder_path: str) -> ResponseType: - path = Markus._get_path(assignments=assignment_id, groups=group_id, submission_files=None, create_folders=None) + return requests.get( + self._url(f"assignments/{assignment_id}/groups/{group_id}/submission_files"), + params=params, + headers=self._auth_header, + ) + + @parse_response("json") + def upload_folder_to_repo(self, assignment_id: int, group_id: int, folder_path: str) -> requests.Response: params = {"folder_path": folder_path} - return self._submit_request(params, path, "POST") + return requests.post( + self._url(f"assignments/{assignment_id}/groups/{group_id}/submission_files/create_folders"), + params=params, + headers=self._auth_header, + ) + @parse_response("json") def upload_file_to_repo( self, assignment_id: int, @@ -344,7 +398,7 @@ def upload_file_to_repo( file_path: str, contents: Union[str, bytes], mime_type: Optional[str] = None, - ) -> ResponseType: + ) -> requests.Response: """ Upload a file at file_path with content contents to the assignment directory in the repo for group with id group_id. @@ -354,149 +408,70 @@ def upload_file_to_repo( of the assignment with id assignment_id should be A1 and the file_path argument should be: 'somesubdir/myfile.txt' """ - path = Markus._get_path(assignments=assignment_id, groups=group_id, submission_files=None) - - return self._do_file_upload(path, file_path, contents, mime_type) + files = {"file_content": (file_path, contents)} + params = {"filename": file_path, "mime_type": mime_type or mimetypes.guess_type(file_path)[0]} + return requests.post( + self._url(f"assignments/{assignment_id}/groups/{group_id}/submission_files"), + files=files, + params=params, + headers=self._auth_header, + ) - def remove_file_from_repo(self, assignment_id: int, group_id: int, file_path: str) -> ResponseType: + @parse_response("json") + def remove_file_from_repo(self, assignment_id: int, group_id: int, file_path: str) -> requests.Response: """ Remove a file at file_path from the assignment directory in the repo for group with id group_id. The file_path should be a relative path from the assignment directory of a repository. - For example, if you want to upload a file to A1/somesubdir/myfile.txt then the short identifier + For example, if you want to remove a file A1/somesubdir/myfile.txt then the short identifier of the assignment with id assignment_id should be A1 and the file_path argument should be: 'somesubdir/myfile.txt' """ - path = Markus._get_path(assignments=assignment_id, groups=group_id, submission_files=None, remove_file=None) params = {"filename": file_path} - return self._submit_request(params, path, "DELETE") + return requests.delete( + self._url(f"assignments/{assignment_id}/groups/{group_id}/submission_files/remove_file"), + params=params, + headers=self._auth_header, + ) + + @parse_response("json") + def remove_folder_from_repo(self, assignment_id: int, group_id: int, folder_path: str) -> requests.Response: + """ + Remove a folder at folder_path and all its contents for group with id grou;_id. - def remove_folder_from_repo(self, assignment_id: int, group_id: int, folder_path: str) -> ResponseType: - path = Markus._get_path(assignments=assignment_id, groups=group_id, submission_files=None, remove_folder=None) + The file_path should be a relative path from the assignment directory of a repository. + For example, if you want to remove a folder A1/somesubdir/ then the short identifier + of the assignment with id assignment_id should be A1 and the folder_path argument should be: + 'somesubdir/' + """ params = {"folder_path": folder_path} - return self._submit_request(params, path, "DELETE") + return requests.delete( + self._url(f"assignments/{assignment_id}/groups/{group_id}/submission_files/remove_folder"), + params=params, + headers=self._auth_header, + ) - def get_test_specs(self, assignment_id: int) -> Dict: + @parse_response("json") + def get_test_specs(self, assignment_id: int) -> requests.Response: """ Get the test spec settings for an assignment with id . """ - path = Markus._get_path(assignments=assignment_id, test_specs=None) + ".json" - params = None - response = self._submit_request(params, path, "GET") - return Markus._decode_json_response(response) + return requests.get(self._url(f"assignments/{assignment_id}/test_specs"), headers=self._auth_header) - def update_test_specs(self, assignment_id: int, specs: Dict) -> ResponseType: + @parse_response("json") + def update_test_specs(self, assignment_id: int, specs: Dict) -> requests.Response: """ Update the test spec settings for an assignment with id to be . """ - path = Markus._get_path(assignments=assignment_id, update_test_specs=None) + ".json" params = {"specs": specs} - return self._submit_request(params, path, "POST", content_type="application/json") + return requests.post( + self._url(f"assignments/{assignment_id}/update_test_specs"), json=params, headers=self._auth_header + ) - def get_test_files(self, assignment_id: int) -> Optional[bytes]: + @parse_response("content") + def get_test_files(self, assignment_id: int) -> requests.Response: """ Return the content of a zipfile containing the content of all files uploaded for automated testing of the assignment with id . """ - path = Markus._get_path(assignments=assignment_id, test_files=None) + ".json" - params = None - response = self._submit_request(params, path, "GET") - if response[0] == 200: - return response[2] - - # Helpers - - def _submit_request( - self, - params: Optional[dict], - path: str, - request_type: str, - content_type: str = "application/x-www-form-urlencoded", - ) -> ResponseType: - """ Return result from _do_submit_request after formatting the params and setting headers """ - headers = {"Content-type": content_type} - if params is not None: - if content_type == "application/x-www-form-urlencoded": - # simple params, sent as form query string (needs url encoding of reserved and non-alphanumeric chars) - params = urlencode(params) - elif content_type == "multipart/form-data": - # complex params like binary files, sent as-is (assumes already-encoded data) - pass - elif content_type == "application/json": - # json-encoded params - params = json.dumps(params) - if not isinstance(params, str): - raise ValueError("If the params are not a string type object please provide a valid content_type") - if request_type == "GET": # we only want this for GET requests - headers["Accept"] = "text/plain" - return self._do_submit_request(params, path, request_type, headers) - - def _do_submit_request(self, params: Optional[str], path: str, request_type: str, headers: dict) -> ResponseType: - """ - Perform the HTTP/HTTPS request. Return a list - containing the response's status, reason, and content. - - Keyword arguments: - params -- contains the parameters of the request - path -- route to the resource we are targetting - request_type -- the desired HTTP method (usually 'GET' or 'POST') - """ - auth_header = "MarkUsAuth {}".format(self.api_key) - headers["Authorization"] = auth_header - if self.parsed_url.scheme == "http": - conn = http.client.HTTPConnection(self.parsed_url.netloc) - elif self.parsed_url.scheme == "https": - conn = http.client.HTTPSConnection(self.parsed_url.netloc) - else: - raise ValueError("Panic! Neither http nor https URL.") - conn.request(request_type, self.parsed_url.path + path, params, headers) - resp = conn.getresponse() - lst = (resp.status, resp.reason, resp.read()) - conn.close() - return lst - - def _do_file_upload( - self, - path: str, - file_path: str, - contents: Union[str, bytes], - mime_type: Optional[str] = None, - request_type: str = "POST", - ) -> ResponseType: - """ - Helper that performs requests of to that involves uploading a file named - containing . If is None, the mime type of the file will be guessed. - """ - if mime_type is None: - mime_type = mimetypes.guess_type(file_path)[0] - if mime_type is None: - raise ValueError( - "if the mime_type argument is not given you must provide a title file with a valid extension" - ) - - if isinstance(contents, str): - params = {"filename": file_path, "file_content": contents, "mime_type": mime_type} - content_type = "application/x-www-form-urlencoded" - else: # binary data - params = { - "filename": file_path.encode("utf-8"), - "file_content": contents, - "mime_type": mime_type.encode("utf-8"), - } - content_type = "multipart/form-data" - return self._submit_request(params, path, request_type, content_type) - - @staticmethod - def _get_path(**kwargs: Optional[int]) -> str: - path = "/".join([str(v) for vals in kwargs.items() for v in vals if v is not None]) - return f"{Markus.API_PATH}/{path}" - - @staticmethod - def _decode_text_response(resp: ResponseType) -> str: - """Converts response from _submit_request into a utf-8 string.""" - return resp[2].decode("utf-8") - - @staticmethod - def _decode_json_response(resp: ResponseType) -> Union[List[Dict], Dict]: - """Converts response from _submit_request into python dict.""" - return json.loads(Markus._decode_text_response(resp)) + return requests.get(self._url(f"assignments/{assignment_id}/test_files"), headers=self._auth_header) diff --git a/markusapi/response_parser.py b/markusapi/response_parser.py new file mode 100644 index 0000000..b8add4a --- /dev/null +++ b/markusapi/response_parser.py @@ -0,0 +1,45 @@ +from functools import wraps +from typing import List, Union, Dict, Callable + + +def parse_response(expected: str) -> Callable: + """ + Decorator for a function that returns a requests.Response object. + This decorator parses that response depending on the value of . + + If the response indicates the request failed (status >= 400) a dictionary + containing the response status and message will be returned. Otherwise, + the content will be parsed and a dictionary or list will be returned if + expected == 'json', a string will be returned if expected == 'text' and + a binary string will be returned if expected == 'content'. + + This also updates the return annotation for the wrapped function according + to the expected return value type. + """ + + def _parser(f): + @wraps(f) + def _f(*args, **kwargs): + response = f(*args, **kwargs) + if not response.ok or expected == "json": + return response.json() + if expected == "content": + return response.content + if expected == "text": + return response.text + return response.json() + + f.__annotations__["return"] = _get_expected_return(expected) + return _f + + return _parser + + +def _get_expected_return(expected: str) -> type: + if expected == "json": + return Union[Dict[str, str], List[Dict[str, str]]] + elif expected == "content": + return Union[Dict[str, str], bytes] + elif expected == "text": + return Union[Dict[str, str], bytes] + return Dict[str, str] diff --git a/markusapi/tests/test_markusapi.py b/markusapi/tests/test_markusapi.py index 3baab5d..978d306 100644 --- a/markusapi/tests/test_markusapi.py +++ b/markusapi/tests/test_markusapi.py @@ -1,445 +1,681 @@ +import abc import pytest -import typing -import mimetypes -import json -import http.client -from hypothesis import given -from hypothesis import strategies as st -from unittest.mock import patch -from markusapi import Markus - - -def strategies_from_signature(method): - mapping = {k: st.from_type(v) for k, v in typing.get_type_hints(method).items() if k != "return"} - return st.fixed_dictionaries(mapping) - - -def dummy_markus(scheme="http"): - return Markus("", f"{scheme}://localhost:8080") - - -DUMMY_RETURNS = { - "_submit_request": b'{"f": "foo"}', - "_decode_json_response": {"f": "foo"}, - "_decode_text_response": '{"f": "foo"}', - "path": "/fake/path", -} - - -def file_name_strategy(): - exts = "|".join([f"\\{ext}" for ext in mimetypes.types_map.keys()]) - return st.from_regex(fr"\w+({exts})", fullmatch=True) - - -class TestMarkusAPICalls: - def test_init_set_attributes(self): - obj = dummy_markus() - assert isinstance(obj, Markus) - - def test_init_bad_scheme(self): - try: - dummy_markus("ftp") - except AssertionError: - return - pytest.fail() - - def test_init_parse_url(self): - api_key = "" - url = "https://markus.com/api/users?id=1" - obj = Markus(api_key, url) - assert obj.parsed_url.scheme == "https" - assert obj.parsed_url.netloc == "markus.com" - assert obj.parsed_url.path == "/api/users" - assert obj.parsed_url.query == "id=1" - - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_json_response") - def test_get_all_users(self, _decode_json_response, _submit_request): - dummy_markus().get_all_users() - _submit_request.assert_called_with(None, "/api/users.json", "GET") - _decode_json_response.assert_called_with(_submit_request.return_value) - - @given(kwargs=strategies_from_signature(Markus.new_user)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - def test_new_user(self, _submit_request, kwargs): - dummy_markus().new_user(**kwargs) - _submit_request.assert_called_once() - call_args = _submit_request.call_args[0][0].values() - assert all(v in call_args for k, v in kwargs.items() if v is not None) - - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_json_response") - def test_get_assignments(self, _decode_json_response, _submit_request): - dummy_markus().get_assignments() - _submit_request.assert_called_with(None, "/api/assignments.json", "GET") - _decode_json_response.assert_called_with(_submit_request.return_value) - - @given(kwargs=strategies_from_signature(Markus.get_groups)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_json_response") - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_get_groups(self, _get_path, _decode_json_response, _submit_request, kwargs): - dummy_markus().get_groups(**kwargs) - _get_path.assert_called_with(assignments=kwargs["assignment_id"], groups=None) - _submit_request.assert_called_with(None, f"{_get_path.return_value}.json", "GET") - _decode_json_response.assert_called_with(_submit_request.return_value) - - @given(kwargs=strategies_from_signature(Markus.get_groups_by_name)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_json_response") - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_get_groups_by_name(self, _get_path, _decode_json_response, _submit_request, kwargs): - dummy_markus().get_groups_by_name(**kwargs) - _get_path.assert_called_with(assignments=kwargs["assignment_id"], groups=None, group_ids_by_name=None) - _submit_request.assert_called_with(None, f"{_get_path.return_value}.json", "GET") - _decode_json_response.assert_called_with(_submit_request.return_value) - - @given(kwargs=strategies_from_signature(Markus.get_group)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_json_response") - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_get_group(self, _get_path, _decode_json_response, _submit_request, kwargs): - dummy_markus().get_group(**kwargs) - _get_path.assert_called_with(assignments=kwargs["assignment_id"], groups=kwargs["group_id"]) - _submit_request.assert_called_with(None, f"{_get_path.return_value}.json", "GET") - _decode_json_response.assert_called_with(_submit_request.return_value) - - @given(kwargs=strategies_from_signature(Markus.get_feedback_files)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_json_response") - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_get_feedback_files(self, _get_path, _decode_json_response, _submit_request, kwargs): - dummy_markus().get_feedback_files(**kwargs) - _get_path.assert_called_with( - assignments=kwargs["assignment_id"], groups=kwargs["group_id"], feedback_files=None - ) - _submit_request.assert_called_with({}, f"{_get_path.return_value}.json", "GET") - _decode_json_response.assert_called_with(_submit_request.return_value) - - @given(kwargs=strategies_from_signature(Markus.get_feedback_file)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_text_response") - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_get_feedback_file(self, _get_path, _decode_text_response, _submit_request, kwargs): - dummy_markus().get_feedback_file(**kwargs) - _get_path.assert_called_with( - assignments=kwargs["assignment_id"], groups=kwargs["group_id"], feedback_files=kwargs["feedback_file_id"] - ) - _submit_request.assert_called_with({}, f"{_get_path.return_value}.json", "GET") - _decode_text_response.assert_called_with(_submit_request.return_value) - - @given(kwargs=strategies_from_signature(Markus.get_grades_summary)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_text_response") - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_get_grades_summary(self, _get_path, _decode_text_response, _submit_request, kwargs): - dummy_markus().get_grades_summary(**kwargs) - _get_path.get_grades_summary(assignments=kwargs["assignment_id"], grades_summary=None) - _submit_request.assert_called_with({}, f"{_get_path.return_value}.json", "GET") - _decode_text_response.assert_called_with(_submit_request.return_value) - - @given(kwargs=strategies_from_signature(Markus.new_marks_spreadsheet)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_new_marks_spreadsheet(self, _get_path, _submit_request, kwargs): - dummy_markus().new_marks_spreadsheet(**kwargs) - _get_path.assert_called_with(grade_entry_forms=None) +import markusapi +import datetime +from unittest.mock import patch, PropertyMock, Mock + +FAKE_API_KEY = "fake_api_key" +FAKE_URL = "http://example.com" + + +@pytest.fixture +def api(): + return markusapi.Markus(FAKE_API_KEY, FAKE_URL) + + +class AbstractTestClass(abc.ABC): + @classmethod + @pytest.fixture + def response_mock(cls): + with patch(f"requests.{cls.request_verb}") as mock: + type(mock.return_value).ok = PropertyMock(return_value=True) + mock.return_value.content = "content" + mock.return_value.text = "text" + mock.return_value.json.return_value = "json" + yield mock + + @classmethod + @pytest.fixture + def bad_response_mock(cls, response_mock): + type(response_mock.return_value).ok = PropertyMock(return_value=False) + yield response_mock + + @staticmethod + @pytest.fixture + def basic_call(api): + pass + + @property + @abc.abstractmethod + def request_verb(self): + pass + + @property + @abc.abstractmethod + def response_format(self): + pass + + @property + @abc.abstractmethod + def url(self): + pass + + def test_json_response_data_on_failure(self, bad_response_mock, basic_call): + assert basic_call == "json" + + def test_correct_response_data_on_success(self, response_mock, basic_call): + assert basic_call == self.response_format + + def test_called_with_correct_athorization(self, response_mock, basic_call): + _, kwargs = response_mock.call_args + assert kwargs["headers"]["Authorization"] == f"MarkUsAuth {FAKE_API_KEY}" + + def test_called_with_correct_url(self, response_mock, basic_call): + args, _ = response_mock.call_args + assert args[0] == f"{FAKE_URL}/api/{self.url}.json" + + +class TestGetAllUsers(AbstractTestClass): + request_verb = "get" + response_format = "json" + url = "users" + + @staticmethod + @pytest.fixture + def basic_call(api): + return api.get_all_users() + + +class TestNewUser(AbstractTestClass): + request_verb = "post" + response_format = "json" + url = "users" + + @staticmethod + @pytest.fixture + def basic_call(api): + return api.new_user("test", "Student", "first", "last", "section", "3") + + def test_called_with_basic_params(self, api, response_mock): + api.new_user("test", "Student", "first", "last") + params = {"user_name": "test", "type": "Student", "first_name": "first", "last_name": "last"} + _, kwargs = response_mock.call_args + assert kwargs["params"] == params + + def test_called_with_section(self, api, response_mock): + api.new_user("test", "Student", "first", "last", section_name="section") params = { - "short_identifier": kwargs["short_identifier"], - "description": kwargs["description"], - "date": kwargs["date"], - "is_hidden": kwargs["is_hidden"], - "show_total": kwargs["show_total"], - "grade_entry_items": kwargs["grade_entry_items"], + "user_name": "test", + "type": "Student", + "first_name": "first", + "last_name": "last", + "section_name": "section", } - _submit_request.assert_called_with( - params, _get_path.return_value + ".json", "POST", content_type="application/json" - ) - - @given(kwargs=strategies_from_signature(Markus.update_marks_spreadsheet)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_update_marks_spreadsheet(self, _get_path, _submit_request, kwargs): - dummy_markus().update_marks_spreadsheet(**kwargs) - _get_path.assert_called_with(grade_entry_forms=kwargs["spreadsheet_id"]) + _, kwargs = response_mock.call_args + assert kwargs["params"] == params + + def test_called_with_grace_credits(self, api, response_mock): + api.new_user("test", "Student", "first", "last", grace_credits="3") + params = { + "user_name": "test", + "type": "Student", + "first_name": "first", + "last_name": "last", + "grace_credits": "3", + } + _, kwargs = response_mock.call_args + assert kwargs["params"] == params + + +class TestGetAssignments(AbstractTestClass): + request_verb = "get" + response_format = "json" + url = "assignments" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.get_assignments() + + +class TestGetGroups(AbstractTestClass): + request_verb = "get" + response_format = "json" + url = "assignments/1/groups" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.get_groups(1) + + +class TestGetGroupsByName(AbstractTestClass): + request_verb = "get" + response_format = "json" + url = "assignments/1/groups/group_ids_by_name" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.get_groups_by_name(1) + + +class TestGetGroup(AbstractTestClass): + request_verb = "get" + response_format = "json" + url = "assignments/1/groups/1" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.get_group(1, 1) + + +class TestGetFeedbackFiles(AbstractTestClass): + request_verb = "get" + response_format = "json" + url = "assignments/1/groups/1/feedback_files" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.get_feedback_files(1, 1) + + +class TestGetFeedbackFile(AbstractTestClass): + request_verb = "get" + response_format = "content" + url = "assignments/1/groups/1/feedback_files/1" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.get_feedback_file(1, 1, 1) + + +class TestGetGradesSummary(AbstractTestClass): + request_verb = "get" + response_format = "text" + url = "assignments/1/grades_summary" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.get_grades_summary(1) + + +class TestNewMarksSpreadsheet(AbstractTestClass): + request_verb = "post" + response_format = "json" + url = "grade_entry_forms" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.new_marks_spreadsheet("test", "description", datetime.datetime.now()) + + def test_called_with_basic_params(self, api, response_mock): + now = datetime.datetime.now() + api.new_marks_spreadsheet("test", "description", now) + params = { + "short_identifier": "test", + "description": "description", + "date": now, + "is_hidden": True, + "show_total": True, + "grade_entry_items": None, + } + _, kwargs = response_mock.call_args + assert kwargs["params"] == params + + def test_called_with_is_hidden(self, api, response_mock): + now = datetime.datetime.now() + api.new_marks_spreadsheet("test", "description", now, is_hidden=False) + params = { + "short_identifier": "test", + "description": "description", + "date": now, + "is_hidden": False, + "show_total": True, + "grade_entry_items": None, + } + _, kwargs = response_mock.call_args + assert kwargs["params"] == params + + def test_called_with_is_show_total(self, api, response_mock): + now = datetime.datetime.now() + api.new_marks_spreadsheet("test", "description", now, show_total=False) + params = { + "short_identifier": "test", + "description": "description", + "date": now, + "is_hidden": True, + "show_total": False, + "grade_entry_items": None, + } + _, kwargs = response_mock.call_args + assert kwargs["params"] == params + + def test_called_with_is_show_grade_entry_items(self, api, response_mock): + now = datetime.datetime.now() + ge_items = [{"name": "a", "out_of": 4}] + api.new_marks_spreadsheet("test", "description", now, grade_entry_items=ge_items) + params = { + "short_identifier": "test", + "description": "description", + "date": now, + "is_hidden": True, + "show_total": True, + "grade_entry_items": ge_items, + } + _, kwargs = response_mock.call_args + assert kwargs["params"] == params + + +class TestUpdateMarksSpreadsheet(AbstractTestClass): + request_verb = "put" + response_format = "json" + url = "grade_entry_forms/1" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.update_marks_spreadsheet(1, "test", "description", datetime.datetime.now()) + + def test_called_with_basic_params(self, api, response_mock): + now = datetime.datetime.now() + api.update_marks_spreadsheet(1, "test", "description", now) + params = {"short_identifier": "test", "description": "description", "date": now} + _, kwargs = response_mock.call_args + assert kwargs["params"] == params + + def test_called_with_is_hidden(self, api, response_mock): + now = datetime.datetime.now() + api.update_marks_spreadsheet(1, "test", "description", now, is_hidden=False) + params = {"short_identifier": "test", "description": "description", "date": now, "is_hidden": False} + _, kwargs = response_mock.call_args + assert kwargs["params"] == params + + def test_called_with_is_show_total(self, api, response_mock): + now = datetime.datetime.now() + api.update_marks_spreadsheet(1, "test", "description", now, show_total=False) + params = {"short_identifier": "test", "description": "description", "date": now, "show_total": False} + _, kwargs = response_mock.call_args + assert kwargs["params"] == params + + def test_called_with_is_show_grade_entry_items(self, api, response_mock): + now = datetime.datetime.now() + ge_items = [{"name": "a", "out_of": 4}] + api.update_marks_spreadsheet(1, "test", "description", now, grade_entry_items=ge_items) params = { - "short_identifier": kwargs["short_identifier"], - "description": kwargs["description"], - "date": kwargs["date"], - "is_hidden": kwargs["is_hidden"], - "show_total": kwargs["show_total"], - "grade_entry_items": kwargs["grade_entry_items"], + "short_identifier": "test", + "description": "description", + "date": now, + "grade_entry_items": ge_items, } - for name in list(params): - if params[name] is None: - params.pop(name) - _submit_request.assert_called_with( - params, _get_path.return_value + ".json", "PUT", content_type="application/json" - ) - - @given(kwargs=strategies_from_signature(Markus.update_marks_spreadsheets_grades)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_update_marks_spreadsheets_grades(self, _get_path, _submit_request, kwargs): - dummy_markus().update_marks_spreadsheets_grades(**kwargs) - _get_path.assert_called_with(grade_entry_forms=kwargs["spreadsheet_id"], update_grades=None) - params = {"user_name": kwargs["user_name"], "grade_entry_items": kwargs["grades_per_column"]} - _submit_request.assert_called_with( - params, _get_path.return_value + ".json", "PUT", content_type="application/json" - ) - - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_json_response", return_value=[DUMMY_RETURNS["_decode_json_response"]]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_get_marks_spreadsheets(self, _get_path, _decode_json_response, _submit_request): - dummy_markus().get_marks_spreadsheets() - _get_path.assert_called_with(grade_entry_forms=None) - _submit_request.assert_called_with({}, f"{_get_path.return_value}.json", "GET") - _decode_json_response.assert_called_with(_submit_request.return_value) - - @given(kwargs=strategies_from_signature(Markus.get_marks_spreadsheet)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_text_response") - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_get_marks_spreadsheet(self, _get_path, _decode_text_response, _submit_request, kwargs): - dummy_markus().get_marks_spreadsheet(**kwargs) - _get_path.assert_called_with(grade_entry_forms=kwargs["spreadsheet_id"]) - _submit_request.assert_called_with({}, f"{_get_path.return_value}.json", "GET") - _decode_text_response.assert_called_with(_submit_request.return_value) - - @given(kwargs=strategies_from_signature(Markus.upload_feedback_file), filename=file_name_strategy()) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_json_response", return_value=[DUMMY_RETURNS["_decode_json_response"]]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_upload_feedback_file_good_title(self, _get_path, _decode_json_response, _submit_request, kwargs, filename): - dummy_markus().upload_feedback_file(**{**kwargs, "title": filename}) - _get_path.assert_called_with( - assignments=kwargs["assignment_id"], groups=kwargs["group_id"], feedback_files=None - ) - params, path, request_type, _content_type = _submit_request.call_args[0] - assert path == _get_path.return_value - assert params.keys() == {"filename", "file_content", "mime_type"} - - @given(kwargs=strategies_from_signature(Markus.upload_feedback_file), filename=file_name_strategy()) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_upload_feedback_file_overwrite(self, _get_path, _submit_request, kwargs, filename): - with patch.object(Markus, "_decode_json_response", return_value=[{"id": 1, "filename": filename}]): - dummy_markus().upload_feedback_file(**{**kwargs, "title": filename}) - _get_path.assert_called_with( - assignments=kwargs["assignment_id"], groups=kwargs["group_id"], feedback_files=None - ) - _params, _path, request_type, _content_type = _submit_request.call_args[0] - assert request_type == ("PUT" if kwargs["overwrite"] else "POST") - - @given( - kwargs=strategies_from_signature(Markus.upload_feedback_file), filename=st.from_regex(r"\w+", fullmatch=True) - ) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_json_response", return_value=[DUMMY_RETURNS["_decode_json_response"]]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_upload_feedback_file_bad_title(self, _get_path, _decode_json_response, _submit_request, kwargs, filename): - with pytest.raises(ValueError): - dummy_markus().upload_feedback_file(**{**kwargs, "title": filename, "mime_type": None}) - - @given(kwargs=strategies_from_signature(Markus.upload_test_group_results)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_upload_test_group_results(self, _get_path, _submit_request, kwargs): - dummy_markus().upload_test_group_results(**kwargs) - params = {"test_run_id": kwargs["test_run_id"], "test_output": kwargs["test_output"]} - _get_path.assert_called_with( - assignments=kwargs["assignment_id"], groups=kwargs["group_id"], test_group_results=None - ) - _submit_request.assert_called_with(params, _get_path.return_value, "POST") - - @given(kwargs=strategies_from_signature(Markus.upload_annotations)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_upload_annotations(self, _get_path, _submit_request, kwargs): - dummy_markus().upload_annotations(**kwargs) - params = {"annotations": kwargs["annotations"], "force_complete": kwargs["force_complete"]} - _get_path.assert_called_with( - assignments=kwargs["assignment_id"], groups=kwargs["group_id"], add_annotations=None - ) - _submit_request.assert_called_with(params, _get_path.return_value, "POST", "application/json") - - @given(kwargs=strategies_from_signature(Markus.get_annotations)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_json_response") - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_get_annotations(self, _get_path, _decode_json_response, _submit_request, kwargs): - dummy_markus().get_annotations(**kwargs) - _get_path.assert_called_with(assignments=kwargs["assignment_id"], groups=kwargs["group_id"], annotations=None) - _submit_request.assert_called_with(None, f"{_get_path.return_value}.json", "GET") - _decode_json_response.assert_called_with(_submit_request.return_value) - - @given(kwargs=strategies_from_signature(Markus.update_marks_single_group)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_update_marks_single_group(self, _get_path, _submit_request, kwargs): - dummy_markus().update_marks_single_group(**kwargs) - _get_path.assert_called_with(assignments=kwargs["assignment_id"], groups=kwargs["group_id"], update_marks=None) - _submit_request.assert_called_with(kwargs["criteria_mark_map"], _get_path.return_value, "PUT") - - @given( - kwargs=strategies_from_signature(Markus.upload_folder_to_repo), - foldername=st.from_regex(fr"([a-z]+/?)+", fullmatch=True), - ) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_json_response", return_value=[DUMMY_RETURNS["_decode_json_response"]]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_upload_folder_to_repo(self, _get_path, _decode_json_response, _submit_request, kwargs, foldername): - dummy_markus().upload_folder_to_repo(**{**kwargs, "folder_path": foldername}) - _get_path.assert_called_with( - assignments=kwargs["assignment_id"], groups=kwargs["group_id"], submission_files=None, create_folders=None - ) - params, path, request_type = _submit_request.call_args[0] - assert path == _get_path.return_value - assert params.keys() == {"folder_path"} - - @given(kwargs=strategies_from_signature(Markus.upload_file_to_repo), filename=file_name_strategy()) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_json_response", return_value=[DUMMY_RETURNS["_decode_json_response"]]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_upload_file_to_repo(self, _get_path, _decode_json_response, _submit_request, kwargs, filename): - dummy_markus().upload_file_to_repo(**{**kwargs, "file_path": filename}) - _get_path.assert_called_with( - assignments=kwargs["assignment_id"], groups=kwargs["group_id"], submission_files=None - ) - params, path, request_type, _content_type = _submit_request.call_args[0] - assert path == _get_path.return_value - assert params.keys() == {"filename", "file_content", "mime_type"} - - @given(kwargs=strategies_from_signature(Markus.remove_file_from_repo), filename=file_name_strategy()) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_json_response", return_value=[DUMMY_RETURNS["_decode_json_response"]]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_remove_file_from_repo(self, _get_path, _decode_json_response, _submit_request, kwargs, filename): - dummy_markus().remove_file_from_repo(**{**kwargs, "file_path": filename}) - _get_path.assert_called_with( - assignments=kwargs["assignment_id"], groups=kwargs["group_id"], submission_files=None, remove_file=None - ) - params, path, request_type = _submit_request.call_args[0] - assert path == _get_path.return_value - assert params.keys() == {"filename"} - - @given(kwargs=strategies_from_signature(Markus.remove_folder_from_repo), foldername=st.from_regex(fr"([a-z]+/?)+")) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_json_response", return_value=[DUMMY_RETURNS["_decode_json_response"]]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_remove_folder_from_repo(self, _get_path, _decode_json_response, _submit_request, kwargs, foldername): - dummy_markus().remove_folder_from_repo(**{**kwargs, "folder_path": foldername}) - _get_path.assert_called_with( - assignments=kwargs["assignment_id"], groups=kwargs["group_id"], submission_files=None, remove_folder=None - ) - params, path, request_type = _submit_request.call_args[0] - assert path == _get_path.return_value - assert params.keys() == {"folder_path"} - - @given(kwargs=strategies_from_signature(Markus.get_files_from_repo)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_get_files_from_repo(self, _get_path, _submit_request, kwargs): - dummy_markus().get_files_from_repo(**{**kwargs}) - _get_path.assert_called_with( - assignments=kwargs["assignment_id"], groups=kwargs["group_id"], submission_files=None - ) - params, path, request_type = _submit_request.call_args[0] - assert path == _get_path.return_value + ".json" - if kwargs.get("filename"): - assert "filename" in params.keys() - if kwargs.get("collected"): - assert "collected" in params.keys() - - @given(kwargs=strategies_from_signature(Markus.get_test_specs)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_decode_json_response") - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_get_test_specs(self, _get_path, _decode_json_response, _submit_request, kwargs): - dummy_markus().get_test_specs(**kwargs) - _get_path.assert_called_with(assignments=kwargs["assignment_id"], test_specs=None) - _submit_request.assert_called_with(None, f"{_get_path.return_value}.json", "GET") - _decode_json_response.assert_called_with(_submit_request.return_value) - - @given(kwargs=strategies_from_signature(Markus.update_test_specs)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_update_test_specs(self, _get_path, _submit_request, kwargs): - dummy_markus().update_test_specs(**{**kwargs}) - _get_path.assert_called_with(assignments=kwargs["assignment_id"], update_test_specs=None) - specs = {"specs": kwargs["specs"]} - _submit_request.assert_called_with( - specs, f"{_get_path.return_value}.json", "POST", content_type="application/json" - ) - - @given(kwargs=strategies_from_signature(Markus.get_test_files)) - @patch.object(Markus, "_submit_request", return_value=DUMMY_RETURNS["_submit_request"]) - @patch.object(Markus, "_get_path", return_value=DUMMY_RETURNS["path"]) - def test_get_test_files(self, _get_path, _submit_request, kwargs): - dummy_markus().get_test_files(**kwargs) - _get_path.assert_called_with(assignments=kwargs["assignment_id"], test_files=None) - _submit_request.assert_called_with(None, f"{_get_path.return_value}.json", "GET") - - -class TestMarkusAPIHelpers: - @given( - kwargs=strategies_from_signature(Markus._submit_request), - content_type=st.sampled_from(["application/x-www-form-urlencoded", "application/json"]), - ) - @patch.object(Markus, "_do_submit_request") - def test_submit_request_check_types(self, do_submit_request, kwargs, content_type): - dummy_markus()._submit_request(**{**kwargs, "content_type": content_type}) - params, _path, _request_type, headers = do_submit_request.call_args[0] - assert isinstance(params, (str, type(None))) - assert isinstance(headers, dict) - - @given( - kwargs=strategies_from_signature(Markus._submit_request), - content_type=st.sampled_from(["multipart/form-data", "bad/content/type"]), - ) - @patch.object(Markus, "_do_submit_request") - def test_submit_request_check_catches_invalid(self, do_submit_request, kwargs, content_type): - try: - dummy_markus()._submit_request(**{**kwargs, "content_type": content_type}) - except ValueError: - return - params, _path, _request_type, headers = do_submit_request.call_args[0] - assert isinstance(params, (str, type(None))) - - @given(kwargs=strategies_from_signature(Markus._do_submit_request)) - @patch.object(http.client.HTTPConnection, "request") - @patch.object(http.client.HTTPConnection, "getresponse") - @patch.object(http.client.HTTPConnection, "close") - def test__do_submit_request_http(self, request, getresponse, close, kwargs): - dummy_markus("http")._do_submit_request(**kwargs) - request.assert_called_once() - getresponse.assert_called_once() - close.assert_called_once() - - @given(kwargs=strategies_from_signature(Markus._do_submit_request)) - @patch.object(http.client.HTTPSConnection, "request") - @patch.object(http.client.HTTPSConnection, "getresponse") - @patch.object(http.client.HTTPSConnection, "close") - def test__do_submit_request_https(self, request, getresponse, close, kwargs): - dummy_markus("https")._do_submit_request(**kwargs) - request.assert_called_once() - getresponse.assert_called_once() - close.assert_called_once() - - @given(kwargs=st.dictionaries(st.text(), st.text())) - def test_get_path(self, kwargs): - path = Markus._get_path(**kwargs) - for k, v in kwargs.items(): - assert k + (f"/{v}" if v is not None else "") in path - - @given(strategies_from_signature(Markus._decode_text_response)) - def test_decode_text_response(self, **kwargs): - result = Markus._decode_text_response(**kwargs) - assert isinstance(result, str) - - @given(in_dict=st.dictionaries(st.text(), st.text())) - def test_decode_text_response(self, in_dict): - res = json.dumps(in_dict).encode() - result = Markus._decode_text_response((200, "", res)) - assert isinstance(result, str) + _, kwargs = response_mock.call_args + assert kwargs["params"] == params + + +class TestUpdateMarksSpreadsheetGrades(AbstractTestClass): + request_verb = "put" + response_format = "json" + url = "grade_entry_forms/1/update_grades" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.update_marks_spreadsheets_grades(1, "some_user", {"some_column": 2}) + + def test_called_with_basic_params(self, api, response_mock): + api.update_marks_spreadsheets_grades(1, "some_user", {"some_column": 2}) + params = {"user_name": "some_user", "grade_entry_items": {"some_column": 2}} + _, kwargs = response_mock.call_args + assert kwargs["json"] == params + + +class TestGetMarksSpreadsheets(AbstractTestClass): + request_verb = "get" + response_format = "json" + url = "grade_entry_forms" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.get_marks_spreadsheets() + + +class TestGetMarksSpreadsheet(AbstractTestClass): + request_verb = "get" + response_format = "text" + url = "grade_entry_forms/1" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.get_marks_spreadsheet(1) + + +class TestUploadFeedbackFileReplace(AbstractTestClass): + request_verb = "put" + response_format = "json" + url = "assignments/1/groups/1/feedback_files/1" + + @staticmethod + @pytest.fixture + def basic_call(api): + api.get_feedback_files = Mock(return_value=[{"id": 1, "filename": "test.txt"}]) + yield api.upload_feedback_file(1, 1, "test.txt", "feedback info") + + def test_discovers_mime_type(self, api, response_mock): + api.get_feedback_files = Mock(return_value=[{"id": 1, "filename": "test.txt"}]) + api.upload_feedback_file(1, 1, "test.txt", "feedback info") + _, kwargs = response_mock.call_args + assert kwargs["params"]["mime_type"] == "text/plain" + + def test_called_with_mime_type(self, api, response_mock): + api.get_feedback_files = Mock(return_value=[{"id": 1, "filename": "test.txt"}]) + api.upload_feedback_file(1, 1, "test.txt", "feedback info", mime_type="application/octet-stream") + params = {"filename": "test.txt", "mime_type": "application/octet-stream"} + _, kwargs = response_mock.call_args + assert kwargs["params"] == params + + def test_sends_file_data(self, api, response_mock): + api.get_feedback_files = Mock(return_value=[{"id": 1, "filename": "test.txt"}]) + api.upload_feedback_file(1, 1, "test.txt", "feedback info") + files = {"file_content": ("test.txt", "feedback info")} + _, kwargs = response_mock.call_args + assert kwargs["files"] == files + + +class TestUploadFeedbackFileNew(AbstractTestClass): + request_verb = "post" + response_format = "json" + url = "assignments/1/groups/1/feedback_files" + + @staticmethod + @pytest.fixture + def basic_call(api): + api.get_feedback_files = Mock(return_value=[{"id": 1, "filename": "other.txt"}]) + yield api.upload_feedback_file(1, 1, "test.txt", "feedback info") + + +class TestUploadFeedbackFileNoOverwrite(TestUploadFeedbackFileNew): + @staticmethod + @pytest.fixture + def basic_call(api): + api.get_feedback_files = Mock(return_value=[{"id": 1, "filename": "test.txt"}]) + yield api.upload_feedback_file(1, 1, "test.txt", "feedback info", overwrite=False) + + +class TestUploadTestGroupResultsJsonString(AbstractTestClass): + request_verb = "post" + response_format = "json" + url = "assignments/1/groups/1/test_group_results" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.upload_test_group_results(1, 1, 1, '{"data": []}') + + def test_called_wth_basic_args(self, api, response_mock): + api.upload_test_group_results(1, 1, 1, '{"data": []}') + params = {"test_run_id": 1, "test_output": '{"data": []}'} + _, kwargs = response_mock.call_args + assert kwargs["json"] == params + + +class TestUploadTestGroupResultsDict(AbstractTestClass): + request_verb = "post" + response_format = "json" + url = "assignments/1/groups/1/test_group_results" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.upload_test_group_results(1, 1, 1, {"data": []}) + + def test_dict_changed_to_json_string(self, api, response_mock): + api.upload_test_group_results(1, 1, 1, {"data": []}) + _, kwargs = response_mock.call_args + assert kwargs["json"]["test_output"] == '{"data": []}' + + +class TestUploadAnnotations(AbstractTestClass): + request_verb = "post" + response_format = "json" + url = "assignments/1/groups/1/add_annotations" + annotations = [ + { + "filename": "test.txt", + "annotation_category_name": "category", + "content": "something", + "line_start": 1, + "line_end": 2, + "column_start": 3, + "column_end": 10, + } + ] + + @classmethod + @pytest.fixture + def basic_call(cls, api): + yield api.upload_annotations(1, 1, cls.annotations) + + def test_called_with_basic_params(self, api, response_mock): + api.upload_annotations(1, 1, self.annotations) + params = {"annotations": self.annotations, "force_complete": False} + _, kwargs = response_mock.call_args + assert kwargs["json"] == params + + def test_called_with_force_complete(self, api, response_mock): + api.upload_annotations(1, 1, self.annotations, True) + params = {"annotations": self.annotations, "force_complete": True} + _, kwargs = response_mock.call_args + assert kwargs["json"] == params + + +class TestGetAnnotations(AbstractTestClass): + request_verb = "get" + response_format = "json" + url = "assignments/1/groups/1/annotations" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.get_annotations(1, 1) + + +class TestUpdateMarksSingleGroup(AbstractTestClass): + request_verb = "put" + response_format = "json" + url = "assignments/1/groups/1/update_marks" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.update_marks_single_group({"criteria_a": 10}, 1, 1) + + def test_called_with_basic_params(self, api, response_mock): + api.update_marks_single_group({"criteria_a": 10}, 1, 1) + _, kwargs = response_mock.call_args + assert kwargs["json"] == {"criteria_a": 10} + + +class TestUpdateMarkingState(AbstractTestClass): + request_verb = "put" + response_format = "json" + url = "assignments/1/groups/1/update_marking_state" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.update_marking_state(1, 1, "collected") + + def test_called_with_basic_params(self, api, response_mock): + api.update_marking_state(1, 1, "collected") + _, kwargs = response_mock.call_args + assert kwargs["params"] == {"marking_state": "collected"} + + +class TestCreateExtraMarks(AbstractTestClass): + request_verb = "post" + response_format = "json" + url = "assignments/1/groups/1/create_extra_marks" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.create_extra_marks(1, 1, 10, "a bonus!") + + def test_called_with_basic_params(self, api, response_mock): + api.create_extra_marks(1, 1, 10, "a bonus!") + _, kwargs = response_mock.call_args + assert kwargs["params"] == {"extra_marks": 10, "description": "a bonus!"} + + +class TestRemoveExtraMarks(AbstractTestClass): + request_verb = "delete" + response_format = "json" + url = "assignments/1/groups/1/remove_extra_marks" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.remove_extra_marks(1, 1, 10, "a bonus!") + + def test_called_with_basic_params(self, api, response_mock): + api.remove_extra_marks(1, 1, 10, "a bonus!") + _, kwargs = response_mock.call_args + assert kwargs["params"] == {"extra_marks": 10, "description": "a bonus!"} + + +class TestGetFilesFromRepo(AbstractTestClass): + request_verb = "get" + response_format = "content" + url = "assignments/1/groups/1/submission_files" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.get_files_from_repo(1, 1) + + def test_called_with_basic_params(self, api, response_mock): + api.get_files_from_repo(1, 1) + _, kwargs = response_mock.call_args + assert kwargs["params"] == {"collected": True} + + def test_called_with_collected(self, api, response_mock): + api.get_files_from_repo(1, 1, collected=False) + _, kwargs = response_mock.call_args + assert kwargs["params"] == {} + + def test_called_with_filename(self, api, response_mock): + api.get_files_from_repo(1, 1, filename="test.txt") + _, kwargs = response_mock.call_args + assert kwargs["params"] == {"collected": True, "filename": "test.txt"} + + +class TestUploadFolderToRepo(AbstractTestClass): + request_verb = "post" + response_format = "json" + url = "assignments/1/groups/1/submission_files/create_folders" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.upload_folder_to_repo(1, 1, "subdir") + + def test_called_with_basic_params(self, api, response_mock): + api.upload_folder_to_repo(1, 1, "subdir") + _, kwargs = response_mock.call_args + assert kwargs["params"] == {"folder_path": "subdir"} + + +class TestUploadFileToRepo(AbstractTestClass): + request_verb = "post" + response_format = "json" + url = "assignments/1/groups/1/submission_files" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.upload_file_to_repo(1, 1, "test.txt", "some content") + + def test_discovers_mime_type(self, api, response_mock): + api.upload_file_to_repo(1, 1, "test.txt", "some content") + _, kwargs = response_mock.call_args + assert kwargs["params"]["mime_type"] == "text/plain" + + def test_called_with_mime_type(self, api, response_mock): + api.upload_file_to_repo(1, 1, "test.txt", "feedback info", mime_type="application/octet-stream") + params = {"filename": "test.txt", "mime_type": "application/octet-stream"} + _, kwargs = response_mock.call_args + assert kwargs["params"] == params + + def test_sends_file_data(self, api, response_mock): + api.upload_file_to_repo(1, 1, "test.txt", "some content") + files = {"file_content": ("test.txt", "some content")} + _, kwargs = response_mock.call_args + assert kwargs["files"] == files + + +class TestRemoveFileFromRepo(AbstractTestClass): + request_verb = "delete" + response_format = "json" + url = "assignments/1/groups/1/submission_files/remove_file" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.remove_file_from_repo(1, 1, "test.txt") + + def test_called_with_basic_params(self, api, response_mock): + api.remove_file_from_repo(1, 1, "test.txt") + _, kwargs = response_mock.call_args + assert kwargs["params"] == {"filename": "test.txt"} + + +class TestRemoveFolderFromRepo(AbstractTestClass): + request_verb = "delete" + response_format = "json" + url = "assignments/1/groups/1/submission_files/remove_folder" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.remove_folder_from_repo(1, 1, "subdir") + + def test_called_with_basic_params(self, api, response_mock): + api.remove_folder_from_repo(1, 1, "subdir") + _, kwargs = response_mock.call_args + assert kwargs["params"] == {"folder_path": "subdir"} + + +class TestGetTestSpecs(AbstractTestClass): + request_verb = "get" + response_format = "json" + url = "assignments/1/test_specs" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.get_test_specs(1) + + +class TestUpdateTestSpecs(AbstractTestClass): + request_verb = "post" + response_format = "json" + url = "assignments/1/update_test_specs" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.update_test_specs(1, {}) + + def test_called_with_basic_params(self, api, response_mock): + specs = {"some": ["fake", "data"]} + api.update_test_specs(1, specs) + _, kwargs = response_mock.call_args + assert kwargs["json"] == {"specs": specs} + + +class TestGetTestFiles(AbstractTestClass): + request_verb = "get" + response_format = "content" + url = "assignments/1/test_files" + + @staticmethod + @pytest.fixture + def basic_call(api): + yield api.get_test_files(1) diff --git a/setup.cfg b/setup.cfg index 224a779..6afcf47 100644 --- a/setup.cfg +++ b/setup.cfg @@ -1,2 +1,5 @@ [metadata] -description-file = README.md \ No newline at end of file +description-file = README.md + +[aliases] +test=pytest diff --git a/setup.py b/setup.py index 17d6d56..7218c6d 100644 --- a/setup.py +++ b/setup.py @@ -5,7 +5,7 @@ setuptools.setup( name="markusapi", - version="0.1.1", + version="0.2.0", author="Alessio Di Sandro, Misha Schwartz", author_email="mschwa@cs.toronto.edu", description="Interface to interact with MarkUs API", @@ -13,6 +13,9 @@ long_description_content_type="text/markdown", url="https://github.com/MarkUsProject/markus-api", packages=setuptools.find_packages(), + install_requires=["requests==2.24.0"], + tests_require=["pytest==5.3.1"], + setup_requires=["pytest-runner"], classifiers=[ "Programming Language :: Python :: 3", "License :: OSI Approved :: MIT License",