autograder.filespec

  1"""
  2Filespecs (File Specifications) are objects in assignment configs that describe
  3files that need to exist.
  4These files can come from local files (path FileSpec), URLs (url FileSpec),
  5or git repos (git FileSpec).
  6"""
  7
  8import os
  9import urllib.parse
 10import typing
 11
 12import edq.net.request
 13import edq.util.dirent
 14import edq.util.git
 15
 16FILESPEC_TYPE_EMPTY: str = "empty"
 17FILESPEC_TYPE_NIL: str = "nil"
 18FILESPEC_TYPE_PATH: str = "path"
 19FILESPEC_TYPE_GIT: str = "git"
 20FILESPEC_TYPE_URL: str = "url"
 21
 22class FileSpec(typing.Dict[str, str]):
 23    """ Alias file specs until they are formalized in a more robust class. """
 24
 25def parse(data: typing.Union[None, str, typing.Dict[str, str], FileSpec]) -> FileSpec:
 26    """
 27    Parse and validate a FileSpec.
 28    Return a properly formatted FileSpec.
 29
 30    This parsing is more lenient than the parsing done by the autograder server:
 31    https://github.com/eriq-augustine/autograder-server/blob/main/common/filespec.go#L37
 32    """
 33
 34    if (data is None):
 35        return get_empty()
 36
 37    if (isinstance(data, str)):
 38        if (data == ""):
 39            return get_empty()
 40
 41        return get_path(data)
 42
 43    if (not isinstance(data, dict)):
 44        raise FileSpecError(f"FileSpec is not the correct type (str ot dict): '{data}' ({type(data)}).")
 45
 46    if ('type' not in data):
 47        raise FileSpecError(f"FileSpec is missing 'type' field: '{data}'.")
 48
 49    data['type'] = data['type'].strip().lower()
 50
 51    spec_type = data['type']
 52    if (spec_type == FILESPEC_TYPE_EMPTY):
 53        return get_empty()
 54    elif (spec_type == FILESPEC_TYPE_NIL):
 55        return get_nil()
 56    elif (spec_type == FILESPEC_TYPE_PATH):
 57        path = data.get('path', '').strip()
 58        if (path == ''):
 59            raise FileSpecError(f"Path FileSpec must have a non-empty path: '{data}'.")
 60
 61        return get_path(path, dest = data.get('dest', ''))
 62    elif (spec_type == FILESPEC_TYPE_GIT):
 63        path = data.get('path', '').strip()
 64        if (path == ''):
 65            raise FileSpecError(f"Git FileSpec must have a non-empty path: '{data}'.")
 66
 67        return get_git(path,
 68            dest = data.get('dest', ''),
 69            reference = data.get('reference', ''),
 70            username = data.get('username', ''),
 71            token = data.get('token', ''))
 72    elif (spec_type == FILESPEC_TYPE_URL):
 73        path = data.get('path', '').strip()
 74        if (path == ''):
 75            raise FileSpecError(f"URL FileSpec must have a non-empty path: '{data}'.")
 76
 77        return get_url(path, dest = data.get('dest', ''))
 78    else:
 79        raise FileSpecError(f"FileSpec has unkown type ('{spec_type}'): '{data}'.")
 80
 81def get_empty() -> FileSpec:
 82    """ Get an empty filespec. """
 83
 84    return FileSpec({
 85        "type": FILESPEC_TYPE_EMPTY,
 86    })
 87
 88def get_nil() -> FileSpec:
 89    """ Get a nil filespec. """
 90
 91    return FileSpec({
 92        "type": FILESPEC_TYPE_NIL,
 93    })
 94
 95def get_path(path: str, dest: str = '') -> FileSpec:
 96    """ Get a filespec that points to the given file system path. """
 97
 98    if (dest == ''):
 99        dest = os.path.basename(path)
100
101    return FileSpec({
102        "type": FILESPEC_TYPE_PATH,
103        "path": path,
104        "dest": dest,
105    })
106
107def get_git(path: str, dest: str = '', reference: str = '', username: str = '', token: str = '') -> FileSpec:
108    """ Get a filespec that points to the given Git reference. """
109
110    if (dest == ''):
111        dest = os.path.splitext(os.path.basename(path))[0]
112
113    if ((username != '') and (token == '')):
114        raise FileSpecError(("If username is specified on a Git FileSpec,"
115            + " then token must also be specified."))
116
117    return FileSpec({
118        "type": FILESPEC_TYPE_GIT,
119        "path": path,
120        "dest": dest,
121        "reference": reference,
122        "username": username,
123        "token": token,
124    })
125
126def get_url(path: str, dest: str = '') -> FileSpec:
127    """ Get a filespec that points to the given URL. """
128
129    if (dest == ''):
130        url = urllib.parse.urlparse(path)
131        dest = os.path.basename(url.path)
132
133    return FileSpec({
134        "type": FILESPEC_TYPE_URL,
135        "path": path,
136        "dest": dest,
137    })
138
139def copy(filespec: FileSpec, base_dir: str, dest_dir: str, only_contents: bool) -> None:
140    """ Copy the filespec from the source to the given destination dir. """
141
142    spec_type = filespec['type']
143    if (spec_type in [FILESPEC_TYPE_EMPTY, FILESPEC_TYPE_NIL]):
144        # noop
145        pass
146    elif (spec_type == FILESPEC_TYPE_PATH):
147        _copy_path(filespec['path'], filespec['dest'], base_dir, dest_dir, only_contents)
148    elif (spec_type == FILESPEC_TYPE_GIT):
149        _copy_git(filespec['path'], filespec['dest'], dest_dir,
150                reference = filespec['reference'],
151                username = filespec['username'],
152                token = filespec['token'])
153    elif (spec_type == FILESPEC_TYPE_URL):
154        _copy_url(filespec['path'], filespec['dest'], dest_dir)
155    else:
156        raise FileSpecError(f"FileSpec has unkown type ('{spec_type}'): '{filespec}'.")
157
158def _copy_path(path: str, dest: str, base_dir: str, dest_dir: str, only_contents: bool) -> None:
159    """ Copy a path filespec. """
160
161    source_path = path
162    if ((not os.path.isabs(source_path)) and (base_dir != '')):
163        source_path = os.path.join(base_dir, path)
164
165    if (only_contents):
166        dest_path = dest_dir
167
168        if (dest != ''):
169            dest_path = os.path.join(dest_dir, dest)
170    else:
171        filename = dest
172        if (filename == ''):
173            filename = os.path.basename(path)
174
175        dest_path = os.path.join(dest_dir, filename)
176
177    if (only_contents):
178        edq.util.dirent.copy_contents(source_path, dest_path)
179    else:
180        edq.util.dirent.copy(source_path, dest_path)
181
182def _copy_git(path: str, dest: str, dest_dir: str,
183        reference: typing.Union[str, None] = None,
184        username: typing.Union[str, None] = None,
185        token: typing.Union[str, None] = None,
186        ) -> None:
187    """ Copy a Git filespec. """
188
189    if (reference == ''):
190        reference = None
191
192    if (username == ''):
193        username = None
194
195    if (token == ''):
196        token = None
197
198    dest_path = os.path.join(dest_dir, dest)
199
200    edq.util.git.ensure_repo(path, dest_path, update = True,
201        ref = reference, username = username, token = token)
202
203def _copy_url(path: str, dest: str, dest_dir: str) -> None:
204    """ Copy a URL filespec. """
205
206    dest_path = os.path.join(dest_dir, dest)
207    response, body = edq.net.request.make_get(path)
208    edq.util.dirent.write_file(dest_path, body, encoding = response.encoding, strip = False, newline = False)
209
210class FileSpecError(ValueError):
211    """ Error for the validation and execution of filespecs. """