diff --git a/nbgrader/apps/api.py b/nbgrader/apps/api.py index f78d1344f..7e8f1e359 100644 --- a/nbgrader/apps/api.py +++ b/nbgrader/apps/api.py @@ -1,9 +1,8 @@ -import glob -import re import sys import os import logging import warnings +import typing from traitlets.config import LoggingConfigurable, Config, get_config from traitlets import Instance, Enum, Unicode, observe @@ -123,7 +122,7 @@ def gradebook(self): """ return Gradebook(self.coursedir.db_url, self.course_id) - def get_source_assignments(self): + def get_source_assignments(self) -> typing.Set[str]: """Get the names of all assignments in the `source` directory. Returns @@ -132,29 +131,14 @@ def get_source_assignments(self): A set of assignment names """ - filenames = glob.glob(self.coursedir.format_path( - self.coursedir.source_directory, - student_id='.', - assignment_id='*')) - - assignments = set([]) - for filename in filenames: - # skip files that aren't directories - if not os.path.isdir(filename): - continue - - # parse out the assignment name - regex = self.coursedir.format_path( - self.coursedir.source_directory, - student_id='.', - assignment_id='(?P.*)', - escape=True) - matches = re.match(regex, filename) - if matches: - assignments.add(matches.groupdict()['assignment_id']) - - return assignments + return { + entry['assignment_id'] for entry in + self.coursedir.find_assignments( + nbgrader_step=self.coursedir.source_directory, + student_id=".", + ) + } def get_released_assignments(self): """Get the names of all assignments that have been released to the @@ -194,32 +178,14 @@ def get_submitted_students(self, assignment_id): A set of student ids """ - # get the names of all student submissions in the `submitted` directory - filenames = glob.glob(self.coursedir.format_path( - self.coursedir.submitted_directory, - student_id='*', - assignment_id=assignment_id)) - - students = set([]) - for filename in filenames: - # skip files that aren't directories - if not os.path.isdir(filename): - continue - - # parse out the student id - if assignment_id == "*": - assignment_id = ".*" - regex = self.coursedir.format_path( - self.coursedir.submitted_directory, - student_id='(?P.*)', - assignment_id=assignment_id, - escape=True) - matches = re.match(regex, filename) - if matches: - students.add(matches.groupdict()['student_id']) - - return students + return { + entry['student_id'] for entry in + self.coursedir.find_assignments( + nbgrader_step=self.coursedir.submitted_directory, + assignment_id=assignment_id + ) + } def get_submitted_timestamp(self, assignment_id, student_id): """Gets the timestamp of a submitted assignment. @@ -243,10 +209,7 @@ def get_submitted_timestamp(self, assignment_id, student_id): student_id, assignment_id)) - timestamp_pth = os.path.join(assignment_dir, 'timestamp.txt') - if os.path.exists(timestamp_pth): - with open(timestamp_pth, 'r') as fh: - return parse_utc(fh.read().strip()) + return self.coursedir.get_existing_timestamp(assignment_dir) def get_autograded_students(self, assignment_id): """Get the ids of students whose submission for a given assignment @@ -420,6 +383,7 @@ def get_notebooks(self, assignment_id): A list of dictionaries containing information about each notebook """ + notebooks = [] with self.gradebook as gb: try: assignment = gb.find_assignment(assignment_id) @@ -428,7 +392,6 @@ def get_notebooks(self, assignment_id): # if the assignment exists in the database if assignment and assignment.notebooks: - notebooks = [] for notebook in assignment.notebooks: x = notebook.to_dict() x["average_score"] = gb.average_notebook_score(notebook.name, assignment.name) @@ -439,23 +402,13 @@ def get_notebooks(self, assignment_id): # if it doesn't exist in the database else: - sourcedir = self.coursedir.format_path( - self.coursedir.source_directory, - student_id='.', - assignment_id=assignment_id) - escaped_sourcedir = self.coursedir.format_path( - self.coursedir.source_directory, + for notebook in self.coursedir.find_notebooks( + nbgrader_step=self.coursedir.source_directory, student_id='.', - assignment_id=assignment_id, - escape=True) - - notebooks = [] - for filename in glob.glob(os.path.join(sourcedir, "*.ipynb")): - regex = re.escape(os.path.sep).join([escaped_sourcedir, "(?P.*).ipynb"]) - matches = re.match(regex, filename) - notebook_id = matches.groupdict()['notebook_id'] + assignment_id=assignment_id + ): notebooks.append({ - "name": notebook_id, + "name": notebook['notebook_id'], "id": None, "average_score": 0, "average_code_score": 0, diff --git a/nbgrader/converters/autograde.py b/nbgrader/converters/autograde.py index d3b4e8bac..afb13c3f1 100644 --- a/nbgrader/converters/autograde.py +++ b/nbgrader/converters/autograde.py @@ -181,11 +181,11 @@ def _init_preprocessors(self) -> None: for pp in preprocessors: self.exporter.register_preprocessor(pp) - def convert_single_notebook(self, notebook_filename: str) -> None: + def convert_single_notebook(self, notebook_filename: str, assignment_id: str, student_id: str) -> None: self.log.info("Sanitizing %s", notebook_filename) self._sanitizing = True self._init_preprocessors() - super(Autograde, self).convert_single_notebook(notebook_filename) + super(Autograde, self).convert_single_notebook(notebook_filename, assignment_id, student_id) notebook_filename = os.path.join(self.writer.build_directory, os.path.basename(notebook_filename)) self.log.info("Autograding %s", notebook_filename) @@ -193,6 +193,6 @@ def convert_single_notebook(self, notebook_filename: str) -> None: self._init_preprocessors() try: with utils.setenv(NBGRADER_EXECUTION='autograde'): - super(Autograde, self).convert_single_notebook(notebook_filename) + super(Autograde, self).convert_single_notebook(notebook_filename, assignment_id, student_id) finally: self._sanitizing = True diff --git a/nbgrader/converters/base.py b/nbgrader/converters/base.py index 166386cc0..d88f09ab1 100644 --- a/nbgrader/converters/base.py +++ b/nbgrader/converters/base.py @@ -1,10 +1,9 @@ import os -import glob -import re import shutil import sqlalchemy import traceback import importlib +from pathlib import Path from rapidfuzz import fuzz from traitlets.config import LoggingConfigurable, Config @@ -160,53 +159,75 @@ def _format_dest(self, assignment_id: str, student_id: str, escape: bool = False def init_notebooks(self) -> None: self.assignments = {} self.notebooks = [] - assignment_glob = self._format_source(self.coursedir.assignment_id, self.coursedir.student_id) - for assignment in glob.glob(assignment_glob): - notebook_glob = os.path.join(assignment, self.coursedir.notebook_id + ".ipynb") - found = glob.glob(notebook_glob) - if len(found) == 0: - self.log.warning("No notebooks were matched by '%s'", notebook_glob) + + assignments = self.coursedir.find_assignments( + nbgrader_step=self._input_directory, + student_id=self.coursedir.student_id, + assignment_id=self.coursedir.assignment_id + ) + + for assignment in assignments: + assignment_pair = (assignment["assignment_id"], assignment["student_id"]) + assignment_dir = self._format_source(*assignment_pair) + notebooks = [ + str(notebook) for notebook in + Path(assignment_dir).glob(self.coursedir.notebook_id + ".ipynb") + ] + + if len(notebooks) == 0: + self.log.warning( + "No notebooks matching '%s' were found within '%s'", + self.coursedir.notebook_id + ".ipynb", + assignment_dir + ) continue - self.assignments[assignment] = found + + self.assignments[assignment_pair] = notebooks if len(self.assignments) == 0: - msg = "No notebooks were matched by '%s'" % assignment_glob + msg = "No notebooks were matched by assignment_id='%s', student_id='%s'" % ( + self.coursedir.assignment_id, + self.coursedir.student_id + ) self.log.error(msg) - assignment_glob2 = self._format_source("*", self.coursedir.student_id) - found = glob.glob(assignment_glob2) - if found: - scores = sorted([(fuzz.ratio(assignment_glob, x), x) for x in found]) - self.log.error("Did you mean: %s", scores[-1][1]) - - raise NbGraderException(msg) + options = [ + entry["assignment_id"] for entry in + self.coursedir.find_assignments( + nbgrader_step=self._input_directory, + student_id=self.coursedir.student_id, + ) + ] - def init_single_notebook_resources(self, notebook_filename: str) -> typing.Dict[str, typing.Any]: - regexp = re.escape(os.path.sep).join([ - self._format_source("(?P.*)", "(?P.*)", escape=True), - "(?P.*).ipynb" - ]) + if options: + most_similar = max( + options, + key=lambda x: fuzz.ratio(self.coursedir.assignment_id, x) + ) + self.log.error("Did you mean: %s", most_similar) - m = re.match(regexp, notebook_filename) - if m is None: - msg = "Could not match '%s' with regexp '%s'" % (notebook_filename, regexp) - self.log.error(msg) raise NbGraderException(msg) - gd = m.groupdict() + def init_single_notebook_resources( + self, + notebook_filename: str, + assignment_id: str, + student_id: str + ) -> typing.Dict[str, typing.Any]: + notebook_id = Path(notebook_filename).stem - self.log.debug("Student: %s", gd['student_id']) - self.log.debug("Assignment: %s", gd['assignment_id']) - self.log.debug("Notebook: %s", gd['notebook_id']) + self.log.debug("Student: %s", student_id) + self.log.debug("Assignment: %s", assignment_id) + self.log.debug("Notebook: %s", notebook_id) resources = {} - resources['unique_key'] = gd['notebook_id'] - resources['output_files_dir'] = '%s_files' % gd['notebook_id'] + resources['unique_key'] = notebook_id + resources['output_files_dir'] = '%s_files' % notebook_id resources['nbgrader'] = {} - resources['nbgrader']['student'] = gd['student_id'] - resources['nbgrader']['assignment'] = gd['assignment_id'] - resources['nbgrader']['notebook'] = gd['notebook_id'] + resources['nbgrader']['student'] = student_id + resources['nbgrader']['assignment'] = assignment_id + resources['nbgrader']['notebook'] = notebook_id resources['nbgrader']['db_url'] = self.coursedir.db_url return resources @@ -330,7 +351,7 @@ def set_permissions(self, assignment_id: str, student_id: str) -> None: except PermissionError: self.log.warning("Could not update permissions of %s to make it groupshared", rootdir) - def convert_single_notebook(self, notebook_filename: str) -> None: + def convert_single_notebook(self, notebook_filename: str, assignment_id: str, student_id: str) -> None: """ Convert a single notebook. @@ -340,15 +361,15 @@ def convert_single_notebook(self, notebook_filename: str) -> None: 3. Write the exported notebook to file """ self.log.info("Converting notebook %s", notebook_filename) - resources = self.init_single_notebook_resources(notebook_filename) + resources = self.init_single_notebook_resources(notebook_filename, assignment_id, student_id) output, resources = self.exporter.from_filename(notebook_filename, resources=resources) self.write_single_notebook(output, resources) def convert_notebooks(self) -> None: errors = [] - def _handle_failure(gd: typing.Dict[str, str]) -> None: - dest = os.path.normpath(self._format_dest(gd['assignment_id'], gd['student_id'])) + def _handle_failure(assignment_id: str, student_id: str) -> None: + dest = os.path.normpath(self._format_dest(assignment_id, student_id)) if self.coursedir.notebook_id == "*": if os.path.exists(dest): self.log.warning("Removing failed assignment: {}".format(dest)) @@ -361,36 +382,32 @@ def _handle_failure(gd: typing.Dict[str, str]) -> None: self.log.warning("Removing failed notebook: {}".format(path)) remove(path) - for assignment in sorted(self.assignments.keys()): + for assignment_key in sorted(self.assignments.keys()): # initialize the list of notebooks and the exporter - self.notebooks = sorted(self.assignments[assignment]) - - # parse out the assignment and student ids - regexp = self._format_source("(?P.*)", "(?P.*)", escape=True) - m = re.match(regexp, assignment) - if m is None: - msg = "Could not match '%s' with regexp '%s'" % (assignment, regexp) - self.log.error(msg) - raise NbGraderException(msg) - gd = m.groupdict() + self.notebooks = sorted(self.assignments[assignment_key]) + assignment_id, student_id = assignment_key try: # determine whether we actually even want to process this submission - should_process = self.init_destination(gd['assignment_id'], gd['student_id']) + should_process = self.init_destination(assignment_id, student_id) if not should_process: continue self.run_pre_convert_hook() # initialize the destination - self.init_assignment(gd['assignment_id'], gd['student_id']) + self.init_assignment(assignment_id, student_id) # convert all the notebooks for notebook_filename in self.notebooks: - self.convert_single_notebook(notebook_filename) + self.convert_single_notebook( + notebook_filename, + assignment_id, + student_id + ) # set assignment permissions - self.set_permissions(gd['assignment_id'], gd['student_id']) + self.set_permissions(assignment_id, student_id) self.run_post_convert_hook() except UnresponsiveKernelError: @@ -403,11 +420,11 @@ def _handle_failure(gd: typing.Dict[str, str]) -> None: "have to manually edit the students' code (for example, to " "just throw an error rather than enter an infinite loop). ", assignment) - errors.append((gd['assignment_id'], gd['student_id'])) - _handle_failure(gd) + errors.append((assignment_id, student_id)) + _handle_failure(assignment_id, student_id) except sqlalchemy.exc.OperationalError: - _handle_failure(gd) + _handle_failure(assignment_id, student_id) self.log.error(traceback.format_exc()) msg = ( "There was an error accessing the nbgrader database. This " @@ -419,7 +436,7 @@ def _handle_failure(gd: typing.Dict[str, str]) -> None: raise NbGraderException(msg) except SchemaTooOldError: - _handle_failure(gd) + _handle_failure(assignment_id, student_id) msg = ( "One or more notebooks in the assignment use an old version \n" "of the nbgrader metadata format. Please **back up your class files \n" @@ -429,7 +446,7 @@ def _handle_failure(gd: typing.Dict[str, str]) -> None: raise NbGraderException(msg) except SchemaTooNewError: - _handle_failure(gd) + _handle_failure(assignment_id, student_id) msg = ( "One or more notebooks in the assignment use an newer version \n" "of the nbgrader metadata format. Please update your version of \n" @@ -439,15 +456,15 @@ def _handle_failure(gd: typing.Dict[str, str]) -> None: raise NbGraderException(msg) except KeyboardInterrupt: - _handle_failure(gd) + _handle_failure(assignment_id, student_id) self.log.error("Canceled") raise except Exception: - self.log.error("There was an error processing assignment: %s", assignment) + self.log.error("There was an error processing assignment: %s", assignment_id) self.log.error(traceback.format_exc()) - errors.append((gd['assignment_id'], gd['student_id'])) - _handle_failure(gd) + errors.append((assignment_id, student_id)) + _handle_failure(assignment_id, student_id) if len(errors) > 0: for assignment_id, student_id in errors: @@ -487,4 +504,4 @@ def run_post_convert_hook(self): student=self.coursedir.student_id, notebooks=self.notebooks) except Exception: - self.log.info('Post-convert hook failed', exc_info=True) \ No newline at end of file + self.log.info('Post-convert hook failed', exc_info=True) diff --git a/nbgrader/converters/generate_assignment.py b/nbgrader/converters/generate_assignment.py index 231e59bd4..278beb762 100644 --- a/nbgrader/converters/generate_assignment.py +++ b/nbgrader/converters/generate_assignment.py @@ -1,6 +1,5 @@ -import os -import re from textwrap import dedent +from pathlib import Path from traitlets import List, Bool, default @@ -93,20 +92,17 @@ def __init__(self, coursedir: CourseDirectory = None, **kwargs: Any) -> None: def _clean_old_notebooks(self, assignment_id: str, student_id: str) -> None: with Gradebook(self.coursedir.db_url, self.coursedir.course_id) as gb: assignment = gb.find_assignment(assignment_id) - regexp = re.escape(os.path.sep).join([ - self._format_source("(?P.*)", "(?P.*)", escape=True), - "(?P.*).ipynb" - ]) + assignment_dir = Path(self._format_source(assignment_id, student_id)) # find a set of notebook ids for new notebooks new_notebook_ids = set([]) for notebook in self.notebooks: - m = re.match(regexp, notebook) - if m is None: - raise NbGraderException("Could not match '%s' with regexp '%s'", notebook, regexp) - gd = m.groupdict() - if gd['assignment_id'] == assignment_id and gd['student_id'] == student_id: - new_notebook_ids.add(gd['notebook_id']) + notebook = Path(notebook) + + if assignment_dir not in notebook.parents: + continue + + new_notebook_ids.add(notebook.stem) # pull out the existing notebook ids old_notebook_ids = set(x.name for x in assignment.notebooks) diff --git a/nbgrader/converters/generate_solution.py b/nbgrader/converters/generate_solution.py index 63a44d849..5c1f6767b 100644 --- a/nbgrader/converters/generate_solution.py +++ b/nbgrader/converters/generate_solution.py @@ -1,5 +1,3 @@ -import os -import re from textwrap import dedent from traitlets import List, Bool, default diff --git a/nbgrader/converters/generate_source_with_tests.py b/nbgrader/converters/generate_source_with_tests.py index 2cebdb87b..296b59360 100644 --- a/nbgrader/converters/generate_source_with_tests.py +++ b/nbgrader/converters/generate_source_with_tests.py @@ -1,6 +1,3 @@ -import os -import re - from traitlets import List, default from .base import BaseConverter diff --git a/nbgrader/coursedir.py b/nbgrader/coursedir.py index 9242e56a7..d50a85894 100644 --- a/nbgrader/coursedir.py +++ b/nbgrader/coursedir.py @@ -1,15 +1,17 @@ import os import re - +import sys +import datetime +import typing +from pathlib import Path from textwrap import dedent +import logging from traitlets.config import LoggingConfigurable from traitlets import Integer, Bool, Unicode, List, default, validate, TraitError -from .utils import full_split, parse_utc +from .utils import parse_utc, is_ignored from traitlets.utils.bunch import Bunch -import datetime -from typing import Optional class CourseDirectory(LoggingConfigurable): @@ -302,26 +304,165 @@ def _validate_root(self, proposal: Bunch) -> str: ) ).tag(config=True) - def format_path(self, nbgrader_step: str, student_id: str, assignment_id: str, escape: bool = False) -> str: + def format_path( + self, + nbgrader_step: str, + student_id: str = '.', + assignment_id: str = '.', + escape: bool = False + ) -> str: + """ + Produce an absolute path to an assignment directory. + + When escape=True, the non-passed elements of the path are regex-escaped, so the + resulting string can be used as a pattern to match path components. + """ kwargs = dict( nbgrader_step=nbgrader_step, student_id=student_id, assignment_id=assignment_id ) - + base = Path(self.root) + structure = Path(self.directory_structure.format(**kwargs)) if escape: - structure = [x.format(**kwargs) for x in full_split(self.directory_structure)] - if len(structure) == 0 or not structure[0].startswith(os.sep): - base = [re.escape(self.root)] + if structure.is_absolute(): + anchor = structure.anchor + parts = list(structure.parts) else: - base = [] - path = re.escape(os.path.sep).join(base + structure) + anchor = base.anchor + parts = [re.escape(part) for part in base.parts if part != anchor] + parts += list(structure.parts) + return re.escape(anchor) + re.escape(os.path.sep).join(parts) else: - path = os.path.join(self.root, self.directory_structure.format(**kwargs)) + return str(base / structure) - return path + def find_assignments(self, + nbgrader_step: str = "*", + student_id: str = "*", + assignment_id: str = "*", + ) -> typing.List[typing.Dict]: + """ + Find all directories that match the given criteria. + + The default value for each acts as a wildcard. To exclude a directory, use + a value of "." (e.g. nbgrader_step="source", student_id="."). + + Returns: + A list of dicts containing input values, one per matching directory. + """ + + results = [] + + kwargs = dict( + nbgrader_step=nbgrader_step, + student_id=student_id, + assignment_id=assignment_id + ) + + # Locate all matching directories using a glob + dirs = list( + filter( + lambda p: ( + p.is_dir() + and not is_ignored(str(p.relative_to(self.root)), self.ignore) + ), + Path(self.root).glob(self.directory_structure.format(**kwargs)) + ) + ) + + if not dirs: + return results + + # Create a regex to capture the wildcard values + pattern_args = { + key: value.replace("*", f"(?P<{key}>.*)") + for key, value in kwargs.items() + } + + # Convert to a Path and back to a string to remove any instances of `/.` + pattern = str(self.directory_structure) + + # Escape backslashes on Windows before doing any other escaping + if sys.platform == 'win32': + pattern = pattern.replace('\\', r'\\') + + pattern = str(Path(self.directory_structure.format(**pattern_args))) + + for dir in dirs: + match = re.match(pattern, str(dir.relative_to(self.root))) + if match: + results.append({ **kwargs, **match.groupdict(), 'path': dir }) + + return results + + def find_notebooks( + self, + nbgrader_step: str = "*", + student_id: str = "*", + assignment_id: str = "*", + notebook_id: str = "*", + ext: str = "ipynb", + ) -> typing.List[typing.Dict]: + """ + Find all notebooks that match the given criteria. + + The default value for each acts as a wildcard. To exclude a directory, use + a value of "." (e.g. nbgrader_step="source", student_id="."). + + Returns: + A list of dicts containing input values, one per matching directory. + """ + + results = [] + + kwargs = dict( + nbgrader_step=nbgrader_step, + student_id=student_id, + assignment_id=assignment_id, + notebook_id=notebook_id, + ext=ext, + ) + + pattern = os.path.join(self.directory_structure, "{notebook_id}.{ext}") + + # Locate all matching files using a glob + files = list( + filter( + lambda p: p.is_file() and not is_ignored(p.name, self.ignore), + Path(self.root).glob(pattern.format(**kwargs)) + ) + ) + + if not files: + return results + + pattern_args = { + key: value.replace("*", f"(?P<{key}>.*)") + for key, value in kwargs.items() + } + + logging.error("unescaped pattern: %s", pattern) + + # Escape backslashes on Windows before doing any other escaping + if sys.platform == 'win32': + pattern = pattern.replace('\\', r'\\') + + logging.error("winescaped pattern: %s", pattern) + + # Convert to a Path and back to a string to remove any instances of `/.` + pattern = str(Path(pattern.replace(".", r"\.").format(**pattern_args))) + + logging.error("final pattern: %s", pattern) + + for file in files: + logging.error("file: %s", file.relative_to(self.root)) + match = re.match(pattern, str(file.relative_to(self.root))) + if match: + results.append({ **kwargs, **match.groupdict(), "path": file }) + + return results - def get_existing_timestamp(self, dest_path: str) -> Optional[datetime.datetime]: + def get_existing_timestamp(self, dest_path: str) -> typing.Optional[datetime.datetime]: """Get the timestamp, as a datetime object, of an existing submission.""" timestamp_path = os.path.join(dest_path, 'timestamp.txt') if os.path.exists(timestamp_path): diff --git a/nbgrader/exchange/default/release_feedback.py b/nbgrader/exchange/default/release_feedback.py index b9370f550..c825bf757 100644 --- a/nbgrader/exchange/default/release_feedback.py +++ b/nbgrader/exchange/default/release_feedback.py @@ -1,12 +1,10 @@ import os import shutil -import glob -import re from stat import S_IRUSR, S_IWUSR, S_IXUSR, S_IRGRP, S_IWGRP, S_IXGRP, S_IXOTH, S_ISGID from nbgrader.exchange.abc import ExchangeReleaseFeedback as ABCExchangeReleaseFeedback from .exchange import Exchange -from nbgrader.utils import notebook_hash, make_unique_key +from nbgrader.utils import notebook_hash class ExchangeReleaseFeedback(Exchange, ABCExchangeReleaseFeedback): @@ -37,40 +35,37 @@ def copy_files(self): else: exclude_students = set() - html_files = glob.glob(os.path.join(self.src_path, "*.html")) - for html_file in html_files: - regexp = re.escape(os.path.sep).join([ - self.coursedir.format_path( - self.coursedir.feedback_directory, - "(?P.*)", - self.coursedir.assignment_id, escape=True), - "(?P.*).html" - ]) - - m = re.match(regexp, html_file) - if m is None: - msg = "Could not match '%s' with regexp '%s'" % (html_file, regexp) - self.log.error(msg) - continue - - gd = m.groupdict() - student_id = gd['student_id'] - notebook_id = gd['notebook_id'] + for feedback in self.coursedir.find_notebooks( + nbgrader_step=self.coursedir.feedback_directory, + student_id=self.coursedir.student_id or "*", + assignment_id=self.coursedir.assignment_id, + notebook_id="*", + ext="html" + ): + html_file = feedback['path'] + student_id = feedback['student_id'] if student_id in exclude_students: self.log.debug("Skipping student '{}'".format(student_id)) continue - feedback_dir = os.path.split(html_file)[0] + notebook_id = feedback['notebook_id'] + feedback_dir = html_file.parent - timestamp = open(os.path.join(feedback_dir, 'timestamp.txt')).read() - with open(os.path.join(feedback_dir, "submission_secret.txt")) as fh: - submission_secret = fh.read() + timestamp = feedback_dir.joinpath('timestamp.txt').read_text() + submission_secret = feedback_dir.joinpath("submission_secret.txt").read_text() checksum = notebook_hash(secret=submission_secret, notebook_id=notebook_id) dest = os.path.join(self.dest_path, "{}-tmp.html".format(checksum)) - self.log.info("Releasing feedback for student '{}' on assignment '{}/{}/{}' ({})".format( - student_id, self.coursedir.course_id, self.coursedir.assignment_id, notebook_id, timestamp)) + self.log.info( + "Releasing feedback for student '%s' on assignment '%s/%s/%s' (%s)", + student_id, + self.coursedir.course_id, + feedback["assignment_id"], + notebook_id, + timestamp + ) + shutil.copy(html_file, dest) # Copy to temporary location and mv to update atomically. updated_feedback = os.path.join(self.dest_path, "{}.html". format(checksum)) diff --git a/nbgrader/tests/apps/conftest.py b/nbgrader/tests/apps/conftest.py index e1173224b..9729b548c 100644 --- a/nbgrader/tests/apps/conftest.py +++ b/nbgrader/tests/apps/conftest.py @@ -1,134 +1 @@ -import os -import tempfile -import shutil -import pytest -import sys - -from textwrap import dedent - -from _pytest.fixtures import SubRequest - -from ...api import Gradebook -from ...utils import rmtree - - -@pytest.fixture -def db(request: SubRequest) -> str: - path = tempfile.mkdtemp(prefix='tmp-dbdir-') - dbpath = os.path.join(path, "nbgrader_test.db") - - def fin() -> None: - rmtree(path) - request.addfinalizer(fin) - - return "sqlite:///" + dbpath - - -@pytest.fixture -def course_dir(request: SubRequest) -> str: - path = tempfile.mkdtemp(prefix='tmp-coursedir-') - - def fin() -> None: - rmtree(path) - request.addfinalizer(fin) - - return path - - -@pytest.fixture -def temp_cwd(request: SubRequest, course_dir: str) -> str: - orig_dir = os.getcwd() - path = tempfile.mkdtemp(prefix='tmp-cwd-') - os.chdir(path) - - with open("nbgrader_config.py", "w") as fh: - fh.write(dedent( - """ - c = get_config() - c.CourseDirectory.root = r"{}" - """.format(course_dir) - )) - - def fin() -> None: - os.chdir(orig_dir) - rmtree(path) - request.addfinalizer(fin) - - return path - - -@pytest.fixture -def jupyter_config_dir(request): - path = tempfile.mkdtemp(prefix='tmp-configdir-') - - def fin(): - rmtree(path) - request.addfinalizer(fin) - - return path - - -@pytest.fixture -def jupyter_data_dir(request): - path = tempfile.mkdtemp(prefix='tmp-datadir-') - - def fin(): - rmtree(path) - request.addfinalizer(fin) - - return path - - -@pytest.fixture -def fake_home_dir(request, monkeypatch): - ''' - this fixture creates a temporary home directory. This prevents existing - nbgrader_config.py files in the user directory to interfer with the tests. - ''' - path = tempfile.mkdtemp(prefix='tmp-homedir-') - - def fin(): - rmtree(path) - request.addfinalizer(fin) - - monkeypatch.setenv('HOME', str(path)) - - return path - - -@pytest.fixture -def env(request, jupyter_config_dir, jupyter_data_dir): - env = os.environ.copy() - env['JUPYTER_DATA_DIR'] = jupyter_data_dir - env['JUPYTER_CONFIG_DIR'] = jupyter_config_dir - return env - - -@pytest.fixture -def exchange(request): - path = tempfile.mkdtemp(prefix='tmp-exchange-') - - def fin(): - rmtree(path) - request.addfinalizer(fin) - - return path - - -@pytest.fixture -def cache(request): - path = tempfile.mkdtemp(prefix='tmp-cache-') - - def fin(): - rmtree(path) - request.addfinalizer(fin) - - return path - -notwindows = pytest.mark.skipif( - sys.platform == 'win32', - reason='This functionality of nbgrader is unsupported on Windows') - -windows = pytest.mark.skipif( - sys.platform != 'win32', - reason='This test is only to be run on Windows') +from ..conftest import notwindows, windows diff --git a/nbgrader/tests/apps/test_api.py b/nbgrader/tests/apps/test_api.py index 8b989ffe1..40b242130 100644 --- a/nbgrader/tests/apps/test_api.py +++ b/nbgrader/tests/apps/test_api.py @@ -1,8 +1,6 @@ import pytest import sys import os -import shutil -import filecmp from os.path import join from traitlets.config import Config @@ -12,8 +10,8 @@ from ...coursedir import CourseDirectory from ...utils import rmtree, get_username, parse_utc from .. import run_nbgrader -from .base import BaseTestApp from .conftest import notwindows, windows +from .base import BaseTestApp @pytest.fixture diff --git a/nbgrader/tests/conftest.py b/nbgrader/tests/conftest.py new file mode 100644 index 000000000..02b5efdb7 --- /dev/null +++ b/nbgrader/tests/conftest.py @@ -0,0 +1,134 @@ +import os +import tempfile +import pytest +import sys + +from textwrap import dedent + +from _pytest.fixtures import SubRequest + +from ..api import Gradebook +from ..utils import rmtree + + +notwindows = pytest.mark.skipif( + sys.platform == 'win32', + reason='This functionality of nbgrader is unsupported on Windows') + +windows = pytest.mark.skipif( + sys.platform != 'win32', + reason='This test is only to be run on Windows') + + +@pytest.fixture +def db(request: SubRequest) -> str: + path = tempfile.mkdtemp(prefix='tmp-dbdir-') + dbpath = os.path.join(path, "nbgrader_test.db") + + def fin() -> None: + rmtree(path) + request.addfinalizer(fin) + + return "sqlite:///" + dbpath + + +@pytest.fixture +def course_dir(request: SubRequest) -> str: + path = tempfile.mkdtemp(prefix='tmp-coursedir-') + + def fin() -> None: + rmtree(path) + request.addfinalizer(fin) + + return path + + +@pytest.fixture +def temp_cwd(request: SubRequest, course_dir: str) -> str: + orig_dir = os.getcwd() + path = tempfile.mkdtemp(prefix='tmp-cwd-') + os.chdir(path) + + with open("nbgrader_config.py", "w") as fh: + fh.write(dedent( + """ + c = get_config() + c.CourseDirectory.root = r"{}" + """.format(course_dir) + )) + + def fin() -> None: + os.chdir(orig_dir) + rmtree(path) + request.addfinalizer(fin) + + return path + + +@pytest.fixture +def jupyter_config_dir(request): + path = tempfile.mkdtemp(prefix='tmp-configdir-') + + def fin(): + rmtree(path) + request.addfinalizer(fin) + + return path + + +@pytest.fixture +def jupyter_data_dir(request): + path = tempfile.mkdtemp(prefix='tmp-datadir-') + + def fin(): + rmtree(path) + request.addfinalizer(fin) + + return path + + +@pytest.fixture +def fake_home_dir(request, monkeypatch): + ''' + this fixture creates a temporary home directory. This prevents existing + nbgrader_config.py files in the user directory to interfer with the tests. + ''' + path = tempfile.mkdtemp(prefix='tmp-homedir-') + + def fin(): + rmtree(path) + request.addfinalizer(fin) + + monkeypatch.setenv('HOME', str(path)) + + return path + + +@pytest.fixture +def env(request, jupyter_config_dir, jupyter_data_dir): + env = os.environ.copy() + env['JUPYTER_DATA_DIR'] = jupyter_data_dir + env['JUPYTER_CONFIG_DIR'] = jupyter_config_dir + return env + + +@pytest.fixture +def exchange(request): + path = tempfile.mkdtemp(prefix='tmp-exchange-') + + def fin(): + rmtree(path) + request.addfinalizer(fin) + + return path + + +@pytest.fixture +def cache(request): + path = tempfile.mkdtemp(prefix='tmp-cache-') + + def fin(): + rmtree(path) + request.addfinalizer(fin) + + return path diff --git a/nbgrader/tests/test_coursedir.py b/nbgrader/tests/test_coursedir.py new file mode 100644 index 000000000..2dec8ea78 --- /dev/null +++ b/nbgrader/tests/test_coursedir.py @@ -0,0 +1,119 @@ +import pytest +import os +import re +from pathlib import Path + +from traitlets.config import Config + +from nbgrader.coursedir import CourseDirectory + +def pluck(key, collection): + return sorted([x[key] for x in collection], key=lambda x: x) + +def touch(path): + path.parent.mkdir(parents=True, exist_ok=True) + path.touch() + +@pytest.mark.parametrize("root", [ + None, # Keep the course_dir fixture + os.path.sep + "[special]~root", + "C:\\Users\\Student", +]) +def test_coursedir_format_path(course_dir, root): + config = Config() + if root is None: + root = course_dir + config.CourseDirectory.root = root + coursedir = CourseDirectory(config=config) + + # The default includes the un-escaped root + path = os.path.join(coursedir.root, "step", "student_id", "assignment1") + assert coursedir.format_path("step", "student_id", "assignment1") == path + + # The escape=True option escapes the root and path separators + root = Path(coursedir.root) + expected = re.escape(root.anchor) + re.escape(os.path.sep).join([ + *[re.escape(part) for part in root.parts[1:]], + "step", "student_id", "(?P.*)" + ]) + + actual = coursedir.format_path("step", "student_id", "(?P.*)", escape=True) + assert actual == expected + + # The escape=True option produces a regex pattern which can match paths + match = re.match(actual, path) + assert match is not None + assert match.group("assignment_id") == "assignment1" + +def test_find_assignments(course_dir): + config = Config({"CourseDirectory": {"root": course_dir}}) + coursedir = CourseDirectory(config=config) + + root = Path(coursedir.root) + + root.joinpath("source", ".", "assn").mkdir(parents=True) + root.joinpath("release", ".", "assn").mkdir(parents=True) + root.joinpath("submitted", "student1", "assn").mkdir(parents=True) + root.joinpath("submitted", "student2", "assn").mkdir(parents=True) + root.joinpath("autograded", "student1", "assn").mkdir(parents=True) + root.joinpath("autograded", "student2", "assn").mkdir(parents=True) + root.joinpath("feedback", "student1", "assn").mkdir(parents=True) + root.joinpath("feedback", "student2", "assn").mkdir(parents=True) + + + assert pluck("assignment_id", coursedir.find_assignments("source", ".")) == [ + "assn", + ] + + assert pluck("student_id", coursedir.find_assignments("submitted")) == [ + "student1", "student2", + ] + + assert pluck("nbgrader_step", coursedir.find_assignments(student_id="student1")) == [ + "autograded", "feedback", "submitted", + ] + + # Finding by assignment has the shortcoming of not being able to find across + # student_id="." and student_id="*" at the same time. + assn = coursedir.find_assignments(assignment_id="assn", student_id=".") + assert pluck("nbgrader_step", assn) == ["release", "source"] + + assn = coursedir.find_assignments(assignment_id="assn") + assert pluck("nbgrader_step", assn) == [ + "autograded", "autograded", "feedback", "feedback", "submitted", "submitted", + ] + +def test_find_notebooks(course_dir): + config = Config({"CourseDirectory": {"root": course_dir}}) + coursedir = CourseDirectory(config=config) + + root = Path(coursedir.root) + + touch(root.joinpath("source", ".", "assn", "assn.ipynb")) + touch(root.joinpath("release", ".", "assn", "assn.ipynb")) + touch(root.joinpath("submitted", "student1", "assn", "assn.ipynb")) + touch(root.joinpath("submitted", "student2", "assn", "assn.ipynb")) + touch(root.joinpath("autograded", "student1", "assn", "assn.ipynb")) + touch(root.joinpath("autograded", "student2", "assn", "assn.ipynb")) + touch(root.joinpath("feedback", "student1", "assn", "assn.html")) + touch(root.joinpath("feedback", "student2", "assn", "assn.html")) + + assert pluck("ext", coursedir.find_notebooks(assignment_id="assn", ext="*")) == [ + "html", + "html", + "ipynb", + "ipynb", + "ipynb", + "ipynb", + ] + + + # By default, we only look for .ipynb files + assert pluck("nbgrader_step", coursedir.find_notebooks(student_id="student1")) == [ + "autograded", "submitted", + ] + + assert pluck("nbgrader_step", coursedir.find_notebooks(student_id="student1", ext="*")) == [ + "autograded", "feedback", "submitted", + ] + diff --git a/nbgrader/tests/ui-tests/formgrader.spec.ts b/nbgrader/tests/ui-tests/formgrader.spec.ts index d99b7dda5..9a05845d9 100644 --- a/nbgrader/tests/ui-tests/formgrader.spec.ts +++ b/nbgrader/tests/ui-tests/formgrader.spec.ts @@ -956,6 +956,7 @@ test.describe('#localFormgrader', () => { '#jp-mainmenu-nbgrader li[data-command="nbgrader:open-formgrader-local"]' ); + // Ensure that the local formgrader setting is checked. const [settings, settingsTab] = await openSettings(page); const formgraderSettings = settings.locator( '.jp-PluginList-entry[data-id="@jupyter/nbgrader:formgrader"]' @@ -970,6 +971,9 @@ test.describe('#localFormgrader', () => { await expect(settingsTab).toHaveAttribute('class', /jp-mod-dirty/); await expect(settingsTab).not.toHaveAttribute('class', /jp-mod-dirty/); + // Make sure the global nbgrader exchange and cache directories are junk + createEnv(testDir, "tmpPath", "/dev/null", "/dev/null", isWindows); + // Add a local formgrader in another directory const newDirectory = path.resolve(testDir, 'localFormgrader'); @@ -979,7 +983,7 @@ test.describe('#localFormgrader', () => { fs.mkdirSync(newDirectory); fs.copyFileSync( path.resolve(testDir, "nbgrader_config.py"), - path.resolve(testDir, tmpPath, "nbgrader_config.py") + path.resolve(newDirectory, "nbgrader_config.py") ); var text_to_append = ` diff --git a/nbgrader/tests/utils/conftest.py b/nbgrader/tests/utils/conftest.py deleted file mode 100644 index 2fbea8ef7..000000000 --- a/nbgrader/tests/utils/conftest.py +++ /dev/null @@ -1,6 +0,0 @@ -import pytest -import sys - -notwindows = pytest.mark.skipif( - sys.platform == 'win32', - reason='This functionality of nbgrader is unsupported on Windows') diff --git a/nbgrader/tests/utils/test_utils.py b/nbgrader/tests/utils/test_utils.py index 8f4c2ea07..d10d1ad06 100644 --- a/nbgrader/tests/utils/test_utils.py +++ b/nbgrader/tests/utils/test_utils.py @@ -17,7 +17,7 @@ create_grade_cell, create_solution_cell, create_grade_and_solution_cell) -from .conftest import notwindows +from ..conftest import notwindows @pytest.fixture def temp_cwd(request): diff --git a/nbgrader/utils.py b/nbgrader/utils.py index f9fbdb421..a733682ea 100644 --- a/nbgrader/utils.py +++ b/nbgrader/utils.py @@ -276,12 +276,13 @@ def is_ignored(filename: str, ignore_globs: List[str] = None) -> bool: """Determines whether a filename should be ignored, based on whether it matches any file glob in the given list. Note that this only matches on the base filename itself, not the full path.""" + if ignore_globs is None: return False - dirname = os.path.dirname(filename) + + basename = os.path.basename(filename) for expr in ignore_globs: - globs = glob.glob(os.path.join(dirname, expr)) - if filename in globs: + if fnmatch.fnmatch(basename, expr): return True return False @@ -368,16 +369,6 @@ def find_all_notebooks(path): return notebooks -def full_split(path: str) -> Tuple[str, ...]: - rest, last = os.path.split(path) - if last == path: - return (path,) - elif rest == path: - return (rest,) - else: - return full_split(rest) + (last,) - - @contextlib.contextmanager def chdir(dirname: str) -> Iterator: currdir = os.getcwd()