autograder.fileop

  1"""
  2Fileops (File Operations) are simple operations on files
  3intended to be easy to implement on any platform.
  4
  5See:
  6 - https://github.com/edulinq/autograder-server/blob/main/docs/types.md#file-operation-fileop
  7 - https://github.com/edulinq/autograder-server/blob/main/internal/util/fileop.go
  8"""
  9
 10import fnmatch
 11import glob
 12import os
 13import re
 14import typing
 15
 16import edq.util.dirent
 17
 18import autograder.util.path
 19
 20FILE_OP_LONG_COPY: str = "copy"
 21FILE_OP_SHORT_COPY: str = "cp"
 22
 23FILE_OP_LONG_MOVE: str = "move"
 24FILE_OP_SHORT_MOVE: str = "mv"
 25
 26FILE_OP_LONG_MKDIR: str = "make-dir"
 27FILE_OP_SHORT_MKDIR: str = "mkdir"
 28
 29FILE_OP_LONG_REMOVE: str = "remove"
 30FILE_OP_SHORT_REMOVE: str = "rm"
 31
 32fileop_normal_name: typing.Dict[str, str] = {
 33    FILE_OP_SHORT_COPY: FILE_OP_LONG_COPY,
 34    FILE_OP_LONG_COPY: FILE_OP_LONG_COPY,
 35
 36    FILE_OP_SHORT_MOVE: FILE_OP_LONG_MOVE,
 37    FILE_OP_LONG_MOVE: FILE_OP_LONG_MOVE,
 38
 39    FILE_OP_SHORT_MKDIR: FILE_OP_LONG_MKDIR,
 40    FILE_OP_LONG_MKDIR: FILE_OP_LONG_MKDIR,
 41
 42    FILE_OP_SHORT_REMOVE: FILE_OP_LONG_REMOVE,
 43    FILE_OP_LONG_REMOVE: FILE_OP_LONG_REMOVE,
 44}
 45""" Map operations to their long/canonical name. """
 46
 47fileop_num_args: typing.Dict[str, int] = {
 48    FILE_OP_LONG_COPY: 2,
 49    FILE_OP_LONG_MOVE: 2,
 50    FILE_OP_LONG_MKDIR: 1,
 51    FILE_OP_LONG_REMOVE: 1,
 52}
 53""" The number of operations for each file operation. """
 54
 55FileOp = typing.List[str]
 56""" Alias file operations until they are formalized in a more robust class. """
 57
 58def validate(operation: typing.Union[None, typing.List[str], FileOp]) -> FileOp:
 59    """
 60    Validate a fileop.
 61    Raise on failure or invalid operation.
 62    Otherwise, return the same list that has been validated, cleaned, and typed.
 63    """
 64
 65    if (operation is None):
 66        raise ValueError("File operation is None.")
 67
 68    if (len(operation) == 0):
 69        raise ValueError("File operation is empty.")
 70
 71    command = fileop_normal_name.get(operation[0].lower(), None)
 72    if (command is None):
 73        raise ValueError(f"Unknown file operation: '{operation[0]}'.")
 74
 75    operation[0] = command
 76
 77    num_args = (len(operation) - 1)
 78    expected_num_args = fileop_num_args[command]
 79    if (expected_num_args != num_args):
 80        raise ValueError(f"Incorrect number of arguments for '{command}' file operation."
 81            + f" Expected {expected_num_args}, found {num_args}.")
 82
 83    # Check all path arguments.
 84    for i in range(1, len(operation)):
 85        path = operation[i]
 86
 87        if ('\\' in path):
 88            raise ValueError(f"Argument at index {i} ('{operation[i]}')"
 89                + " contains a backslash ('\\') or is not a POSIX path.")
 90
 91        path = os.path.normpath(path)
 92
 93        if (os.path.isabs(path)):
 94            raise ValueError(f"Argument at index {i} ('{operation[i]}') is an absolute path."
 95                + " Only relative paths are allowed.")
 96
 97        if (not autograder.util.path.is_local(path)):
 98            raise ValueError(f"Argument at index {i} ('{operation[i]}')"
 99                + " points outside of the its base directory."
100                + " File operation paths can not reference parent directories.")
101
102        if (path == "."):
103            raise ValueError(f"Argument at index {i} ('{operation[i]}')"
104                + " cannot point just to the current directory."
105                + " File operation paths must point to a"
106                + " dirent inside the current directory tree.")
107
108        try:
109            glob_pattern = fnmatch.translate(path)
110            re.compile(glob_pattern)
111        except Exception as error:
112            raise ValueError(f"Argument at index {i} ('{operation[i]}') is an invalid glob pattern: {error}.")  # pylint: disable=raise-missing-from
113
114        operation[i] = path
115
116    return operation
117
118def execute(operation: FileOp, base_dir: str) -> None:
119    """ Execute operation operation in the given directory. """
120
121    validate(operation)
122
123    command = operation[0]
124
125    if (command == FILE_OP_LONG_COPY):
126        source_path = _resolve_path(operation[1], base_dir)
127        dest_path = _resolve_path(operation[2], base_dir)
128
129        _handle_glob_file_operation(
130            source_path, dest_path, edq.util.dirent.copy,
131        )
132    elif (command == FILE_OP_LONG_MOVE):
133        source_path = _resolve_path(operation[1], base_dir)
134        dest_path = _resolve_path(operation[2], base_dir)
135
136        _handle_glob_file_operation(source_path, dest_path, edq.util.dirent.move)
137    elif (command == FILE_OP_LONG_MKDIR):
138        path = _resolve_path(operation[1], base_dir)
139
140        edq.util.dirent.mkdir(path)
141    elif (command == FILE_OP_LONG_REMOVE):
142        path_glob = _resolve_path(operation[1], base_dir)
143
144        _handle_glob_remove(path_glob)
145    else:
146        raise ValueError(f"Unknown file operation: '{command}'.")
147
148def validate_file_operations(operations: typing.List[typing.Union[None, typing.List[str]]]) -> typing.List[FileOp]:
149    """ Validate multiple file operations. """
150
151    return [validate(operation) for operation in operations]
152
153def exec_file_operations(operations: typing.List[FileOp], base_dir: str) -> None:
154    """ Execute multiple file operations in the given directory. """
155
156    for operation in operations:
157        execute(operation, base_dir)
158
159def _resolve_path(path: str, base_dir: str) -> str:
160    """ Resolve a path (which may be relative) in the given base directory. """
161
162    if (os.path.isabs(path)):
163        return os.path.normpath(path)
164
165    return os.path.normpath(os.path.join(base_dir, path))
166
167def _handle_glob_file_operation(source_path_glob: str, dest_path: str, operation: typing.Callable, **kwargs: typing.Any) -> None:
168    """ Resolve a path that may contain globs, and perform the given file system operation. """
169
170    source_paths = _prep_for_globs(source_path_glob, dest_path)
171
172    for source_path in source_paths:
173        if (source_path == dest_path):
174            continue
175
176        resolved_dest_path = dest_path
177
178        # Check for an operation into a dir.
179        if (os.path.isdir(dest_path)):
180            resolved_dest_path = os.path.join(dest_path, os.path.basename(source_path))
181
182        operation(source_path, resolved_dest_path, **kwargs)
183
184def _handle_glob_remove(path_glob: str) -> None:
185    """ Resolve a path that may contain a glob and remove the resolved paths. """
186
187    paths = glob.glob(path_glob)
188
189    for path in paths:
190        edq.util.dirent.remove(path)
191
192def _prep_for_globs(source_path_glob: str, dest_path: str) -> typing.List[str]:
193    """
194    Prepare for executing an operation in the presence of globs.
195
196    Do the following actions:
197    1) Resolve the source paths for globs.
198    2) Ensure that the source path matches at least one existing path.
199    3) If multiple source paths match, ensure that the dest exists and is a dir.
200    4) Return the resolved source paths.
201    """
202
203    source_paths = glob.glob(source_path_glob)
204
205    if (len(source_paths) == 0):
206        raise FileNotFoundError(f"No such file or directory: '{source_path_glob}'.")
207
208    if (len(source_paths) > 1):
209        edq.util.dirent.mkdir(dest_path)
210
211    return source_paths