diff --git a/canvasapi/assignment.py b/canvasapi/assignment.py index 4b015668..8cfc14ad 100644 --- a/canvasapi/assignment.py +++ b/canvasapi/assignment.py @@ -1,3 +1,5 @@ +from typing import List + from canvasapi.canvas_object import CanvasObject from canvasapi.exceptions import CanvasException, RequiredFieldMissing from canvasapi.paginated_list import PaginatedList @@ -22,6 +24,71 @@ def __init__(self, requester, attributes): def __str__(self): return "{} ({})".format(self.name, self.id) + def bulk_submit(self, submission, files=None, **kwargs): + """ + Makes a submission of multiple files for an assigment. + + :calls: `POST /api/v1/courses/:course_id/assignments/:assignment_id/submissions \ + `_ + + :param submission: The attributes of the submission. + :type submission: dict + :param files: A list of files to upload with the submission. (Optional, + defaults to `None`. Submission type must be `online_upload`) + :type files: a list of files or str + + :rtype: :class:`canvasapi.submission.Submission` + """ + if isinstance(submission, dict) and "submission_type" in submission: + kwargs["submission"] = submission + else: + raise RequiredFieldMissing( + "Dictionary with key 'submission_type' is required." + ) + + if files: + if submission.get("submission_type") != "online_upload": + raise ValueError( + "To upload files, 'submission['submission_type']' must be 'online_upload'." + ) + file_ids = self.bulk_upload(files, **kwargs) + kwargs["submission"]["file_ids"] = file_ids + + response = self._requester.request( + "POST", + "courses/{}/assignments/{}/submissions".format(self.course_id, self.id), + _kwargs=combine_kwargs(**kwargs), + ) + response_json = response.json() + response_json.update(course_id=self.course_id) + + return Submission(self._requester, response_json) + + def bulk_upload(self, files: List[FileOrPathLike], user="self", **kwargs): + """ + Upload multiples files to a submission. + + :param files: A list of files or path of files to upload + :type files: list of FileLike + param user: The object or ID of the related user, or 'self' for the + current user. Defaults to 'self'. + :type user: :class:`canvasapi.user.User`, int, or str + + :returns: A list of file ids corresponding to files uploaded \ + to the assigment. + :rtype: list + """ + file_ids = [] + for file in files: + upload_response = self.upload_to_submission(file, user, **kwargs) + + if upload_response[0]: + file_ids.append(upload_response[1]["id"]) + else: + raise CanvasException(f"File {file} upload failed.") + + return file_ids + def create_override(self, **kwargs): """ Create an override for this assignment. @@ -443,7 +510,7 @@ def upload_to_submission(self, file: FileOrPathLike, user="self", **kwargs): self.course_id, self.id, user_id ), file, - **kwargs + **kwargs, ).start() diff --git a/tests/test_assignment.py b/tests/test_assignment.py index b86314b6..8583b829 100644 --- a/tests/test_assignment.py +++ b/tests/test_assignment.py @@ -41,6 +41,141 @@ def test__init__overrides(self, m): self.assertEqual(len(assignment.overrides), 1) self.assertIsInstance(assignment.overrides[0], AssignmentOverride) + # bulk_submit() + def test_bulk_submit(self, m): + register_uris({"assignment": ["submit"]}, m) + + sub_type = "online_upload" + sub_dict = {"submission_type": sub_type} + submission = self.assignment.bulk_submit(sub_dict) + + self.assertIsInstance(submission, Submission) + self.assertTrue(hasattr(submission, "submission_type")) + self.assertEqual(submission.submission_type, sub_type) + + def test_bulk_submit_fail(self, m): + with self.assertRaises(RequiredFieldMissing): + self.assignment.bulk_submit({}) + + def test_bulk_submit_file(self, m): + register_uris({"assignment": ["submit", "upload", "upload_final"]}, m) + + filename = "testfile_assignment_{}".format(uuid.uuid4().hex) + + try: + with open(filename, "w+") as file: + sub_type = "online_upload" + sub_dict = {"submission_type": sub_type} + submission = self.assignment.bulk_submit(sub_dict, file) + + self.assertIsInstance(submission, Submission) + self.assertTrue(hasattr(submission, "submission_type")) + self.assertEqual(submission.submission_type, sub_type) + finally: + cleanup_file(filename) + + def test_bulk_submit_multiple_files(self, m): + register_uris({"assignment": ["submit", "upload", "upload_final"]}, m) + + filename_one = "testfile_assignment_{}".format(uuid.uuid4().hex) + filename_two = "testfile_assignment_{}".format(uuid.uuid4().hex) + + try: + with open(filename_one, "w+") as f1, open(filename_two, "w+") as f2: + files = [f1, f2] + + sub_type = "online_upload" + sub_dict = {"submission_type": sub_type} + submission = self.assignment.bulk_submit(sub_dict, files) + + self.assertIsInstance(submission, Submission) + self.assertTrue(hasattr(submission, "submission_type")) + self.assertEqual(submission.submission_type, sub_type) + finally: + cleanup_file(filename_one) + cleanup_file(filename_two) + + def test_bulk_submit_wrong_type(self, m): + filename = "testfile_assignment_{}".format(uuid.uuid4().hex) + sub_type = "online_text_entry" + sub_dict = {"submission_type": sub_type} + + with self.assertRaises(ValueError): + self.assignment.bulk_submit(sub_dict, filename) + + def test_bulk_submit_upload_failure(self, m): + register_uris({"assignment": ["submit", "upload", "upload_fail"]}, m) + + filename_one = "testfile_assignment_{}".format(uuid.uuid4().hex) + filename_two = "testfile_assignment_{}".format(uuid.uuid4().hex) + + try: + with open(filename_one, "w+") as f1, open(filename_two, "w+") as f2: + files = [f1, f2] + + sub_type = "online_upload" + sub_dict = {"submission_type": sub_type} + with self.assertRaises(CanvasException): + self.assignment.bulk_submit(sub_dict, files) + finally: + cleanup_file(filename_one) + cleanup_file(filename_two) + + # bulk_upload() + def test_bulk_upload_self(self, m): + register_uris({"assignment": ["upload", "upload_final"]}, m) + + filename_one = "testfile_assignment_{}".format(uuid.uuid4().hex) + filename_two = "testfile_assignment_{}".format(uuid.uuid4().hex) + + try: + with open(filename_one, "w+") as f1, open(filename_two, "w+") as f2: + files = [f1, f2] + + file_ids = self.assignment.bulk_upload(files) + + self.assertIsNotNone(file_ids) + self.assertIsInstance(file_ids, list) + finally: + cleanup_file(filename_one) + cleanup_file(filename_two) + + def test_bulk_upload_failure(self, m): + register_uris({"assignment": ["upload", "upload_fail"]}, m) + + filename_one = "testfile_assignment_{}".format(uuid.uuid4().hex) + filename_two = "testfile_assignment_{}".format(uuid.uuid4().hex) + + try: + with open(filename_one, "w+") as f1, open(filename_two, "w+") as f2: + files = [f1, f2] + + with self.assertRaises(CanvasException): + self.assignment.bulk_upload(files) + finally: + cleanup_file(filename_one) + cleanup_file(filename_two) + + def test_bulk_upload_user(self, m): + register_uris({"assignment": ["upload_by_id", "upload_final"]}, m) + + filename_one = "testfile_assignment_{}".format(uuid.uuid4().hex) + filename_two = "testfile_assignment_{}".format(uuid.uuid4().hex) + + user_id = 1 + + try: + with open(filename_one, "w+") as f1, open(filename_two, "w+") as f2: + files = [f1, f2] + + file_ids = self.assignment.bulk_upload(files, user_id) + + self.assertIsNotNone(file_ids) + self.assertIsInstance(file_ids, list) + finally: + cleanup_file(filename_one) + cleanup_file(filename_two) + # create_override() def test_create_override(self, m): register_uris({"assignment": ["create_override"]}, m)