autograder.util.net

  1"""
  2Utilities for network and HTTP.
  3"""
  4
  5import typing
  6
  7import edq.util.json
  8import requests
  9
 10import autograder.api.constants
 11import autograder.testing.asserts
 12import autograder.testing.constants
 13import autograder.testing.model
 14
 15CLEAN_REMOVE_HEADERS: typing.Set[str] = {
 16    'access-control-allow-origin',
 17    'transfer-encoding',
 18}
 19""" Headers to remove from API responses. """
 20
 21GRADING_ENDPOINTS: typing.Set[str] = {
 22    'courses/assignments/submissions/submit',
 23    'courses/assignments/submissions/proxy/regrade',
 24    'courses/assignments/submissions/proxy/resubmit',
 25    'courses/assignments/submissions/proxy/submit',
 26}
 27"""
 28These endpoints need to have additional timestamps normalized.
 29Note that the endpoint constants are copied to avoid cyclic dependencies.
 30"""
 31
 32FULL_NORMALIZE_TIMESTAMP_KEYS: typing.Set[str] = autograder.testing.asserts.NORMALIZE_TIMESTAMP_KEYS | {
 33    'grading_end_time',
 34    'grading_start_time',
 35    'proxy_end_time',
 36    'proxy_start_time',
 37    'regrade-cutoff',
 38}
 39""" Keys for timestamp values to normalize. """
 40
 41def clean_api_response(response: requests.Response, body: str) -> str:
 42    """
 43    Clean autograder API responses (so they can be stored consistently).
 44    """
 45
 46    # Clean response headers.
 47    for header in CLEAN_REMOVE_HEADERS:
 48        response.headers.pop(header, None)
 49
 50    # Most responses are JSON.
 51    try:
 52        data = edq.util.json.loads(body, strict = True)
 53    except Exception:
 54        # Response is not JSON.
 55        return body
 56
 57    # Standardize top-level (metadata) keys.
 58    data['start-timestamp'] = 0
 59    data['end-timestamp'] = 0
 60    data['id'] = '00000000-0000-0000-0000-000000000000'
 61
 62    # Standardize content keys.
 63
 64    # Clean specific timestamps.
 65
 66    endpoint = response.url.strip().split(f"api/{autograder.api.constants.API_VERSION}/")[-1]
 67
 68    timestamp_keys = autograder.testing.asserts.NORMALIZE_TIMESTAMP_KEYS
 69    if (endpoint in GRADING_ENDPOINTS):
 70        timestamp_keys = FULL_NORMALIZE_TIMESTAMP_KEYS
 71
 72    data = autograder.testing.asserts._normalize_timestamps(data, keys = timestamp_keys)
 73
 74    # Endpoint-Specific Tasks
 75
 76    if (endpoint in GRADING_ENDPOINTS):
 77        # Normalize submission IDs.
 78        content = data.get('content', None)
 79        if (content is not None):
 80            result = content.get('result', None)
 81            if (result is not None):
 82                parts = result['id'].split('::')
 83                parts[3] = str(autograder.testing.constants.TEST_TIMESTAMP)
 84                result['id'] = '::'.join(parts)
 85
 86                result['short-id'] = str(autograder.testing.constants.TEST_TIMESTAMP)
 87
 88                # A test case returns a stack trace, normalize it.
 89                if ("ModuleNotFoundError: No module named 'ZZZ'" in result.get('epilogue', '')):
 90                    result['epilogue'] = autograder.testing.constants.TEST_CRASH_EPILOGUE
 91    elif (endpoint == 'system/stacks'):
 92        # Replace payloads for stack traces completely to make it consistent.
 93        data['content'] = autograder.testing.model.STACK_TRACE_PAYLOAD
 94    elif (endpoint == 'courses/assignments/images/fetch'):
 95        # Write a dummy docker image (which is just a text file).
 96        content = data.get('content', None)
 97        if (content is not None):
 98            content['bytes'] = autograder.testing.constants.TEST_PAYLOAD_B64_GZIP_BYTES
 99            content['image-info']['created-timestamp'] = autograder.testing.constants.TEST_TIMESTAMP
100            content['image-info']['gzip-size-bytes'] = len(autograder.testing.constants.TEST_PAYLOAD_GZIP_BYTES)
101            content['image-info']['size-bytes'] = len(autograder.testing.constants.TEST_PAYLOAD_BYTES)
102    elif (endpoint == 'courses/assignments/images/info'):
103        content = data.get('content', None)
104        if (content is not None):
105            content['image-info']['created-timestamp'] = autograder.testing.constants.TEST_TIMESTAMP
106            content['image-info']['size-bytes'] = len(autograder.testing.constants.TEST_PAYLOAD_BYTES)
107    elif ('tokens' in endpoint):
108        # Normalize Tokens
109        content = data.get('content', None)
110        if (content is not None):
111            autograder.testing.asserts._noramlize_tokens(content)
112
113    # Convert body back to a string.
114    body = edq.util.json.dumps(data)
115
116    return body