Andal.LND/edx-platform
6
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge pull request #26706 from edx/BOM-2376-terrain

Browse Source 
Run Pyupgrade on terrain
This commit is contained in:
Awais Qureshi
2021-03-08 18:38:19 +05:00
committed by GitHub
parent a39befb548 51c9cc274a
commit 6832c95d8e
13 changed files with 72 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

9
common/djangoapps/terrain/stubs/edxnotes.py
View File

@@ -10,8 +10,7 @@ from datetime import datetime
from math import ceil
from uuid import uuid4
import six
from six.moves.urllib.parse import urlencode
from urllib.parse import urlencode
from .http import StubHttpRequestHandler, StubHttpService
@@ -62,7 +61,7 @@ class StubEdxNotesServiceHandler(StubHttpRequestHandler):
if method in self.URL_HANDLERS:
handlers_list = self.URL_HANDLERS[method]
else:
self.log_error("Unrecognized method '{method}'".format(method=method))
self.log_error(f"Unrecognized method '{method}'")
return
# Check the path (without querystring params) against our list of handlers
@@ -300,7 +299,7 @@ class StubEdxNotesService(StubHttpService):
HANDLER_CLASS = StubEdxNotesServiceHandler
def __init__(self, *args, **kwargs):
super(StubEdxNotesService, self).__init__(*args, **kwargs) # lint-amnesty, pylint: disable=super-with-arguments
super().__init__(*args, **kwargs)
self.notes = list()
def get_all_notes(self):
@@ -393,4 +392,4 @@ class StubEdxNotesService(StubHttpService):
"""
Search the `query(str)` text in the provided `data(list)`.
"""
return [note for note in data if six.text_type(query).strip() in note.get("text", "").split()]
return [note for note in data if str(query).strip() in note.get("text", "").split()]

20
common/djangoapps/terrain/stubs/http.py
View File

@@ -44,7 +44,7 @@ def require_params(method, *required_keys):
elif method == "POST":
params = self.post_dict
else:
raise ValueError("Unsupported method '{method}'".format(method=method))
raise ValueError(f"Unsupported method '{method}'")
# Check for required values
missing = []
@@ -63,7 +63,7 @@ def require_params(method, *required_keys):
return decorator
class StubHttpRequestHandler(BaseHTTPRequestHandler, object):
class StubHttpRequestHandler(BaseHTTPRequestHandler):
"""
Handler for the stub HTTP service.
"""
@@ -157,15 +157,15 @@ class StubHttpRequestHandler(BaseHTTPRequestHandler, object):
if self.path == "/set_config" or self.path == "/set_config/":
if len(self.post_dict) > 0:
for key, value in six.iteritems(self.post_dict):
for key, value in self.post_dict.items():
self.log_message(u"Set config '{0}' to '{1}'".format(key, value))
self.log_message(f"Set config '{key}' to '{value}'")
try:
value = json.loads(value)
except ValueError:
self.log_message(u"Could not parse JSON: {0}".format(value))
self.log_message(f"Could not parse JSON: {value}")
self.send_response(400)
else:
@@ -185,7 +185,7 @@ class StubHttpRequestHandler(BaseHTTPRequestHandler, object):
`content` (str) and `headers` (dict).
"""
self.log_message(
"Sent HTTP response: {0} with content '{1}' and headers {2}".format(status_code, content, headers)
f"Sent HTTP response: {status_code} with content '{content}' and headers {headers}"
)
if headers is None:
@@ -202,7 +202,7 @@ class StubHttpRequestHandler(BaseHTTPRequestHandler, object):
self.end_headers()
if content is not None:
if not six.PY2 and isinstance(content, six.text_type):
if not six.PY2 and isinstance(content, str):
content = content.encode('utf-8')
self.wfile.write(content)
@@ -221,7 +221,7 @@ class StubHttpRequestHandler(BaseHTTPRequestHandler, object):
"""
if not args:
format_str = six.moves.urllib.parse.unquote(format_str)
return u"{0} - - [{1}] {2}\n".format(
return "{} - - [{}] {}\n".format(
self.client_address[0],
self.log_date_time_string(),
format_str % args
@@ -234,7 +234,7 @@ class StubHttpRequestHandler(BaseHTTPRequestHandler, object):
self.send_response(200)
class StubHttpService(ThreadingMixIn, HTTPServer, object):
class StubHttpService(ThreadingMixIn, HTTPServer):
"""
Stub HTTP service implementation.
"""
@@ -260,7 +260,7 @@ class StubHttpService(ThreadingMixIn, HTTPServer, object):
server_thread.start()
# Log the port we're using to help identify port conflict errors
LOGGER.debug('Starting service on port {0}'.format(self.port))
LOGGER.debug(f'Starting service on port {self.port}')
def shutdown(self):
"""

30
common/djangoapps/terrain/stubs/lti.py
View File

@@ -15,9 +15,9 @@ import hashlib
import logging
import os
import textwrap
from unittest import mock
from uuid import uuid4
import mock
import oauthlib.oauth1
import requests
import six
@@ -245,15 +245,15 @@ class StubLtiHandler(StubHttpRequestHandler):
lti_endpoint = self.server.config.get('lti_endpoint', self.DEFAULT_LTI_ENDPOINT)
return lti_endpoint in self.path
def _oauth_sign(self, url, body, content_type=u'application/x-www-form-urlencoded', method=u'POST'):
def _oauth_sign(self, url, body, content_type='application/x-www-form-urlencoded', method='POST'):
"""
Signs request and returns signed Authorization header.
"""
client_key = self.server.config.get('client_key', self.DEFAULT_CLIENT_KEY)
client_secret = self.server.config.get('client_secret', self.DEFAULT_CLIENT_SECRET)
client = oauthlib.oauth1.Client(
client_key=six.text_type(client_key),
client_secret=six.text_type(client_secret)
client_key=str(client_key),
client_secret=str(client_secret)
)
headers = {
# This is needed for body encoding:
@@ -265,17 +265,17 @@ class StubLtiHandler(StubHttpRequestHandler):
sha1.update(body.encode('utf-8'))
oauth_body_hash = base64.b64encode(sha1.digest()).decode('utf-8')
mock_request = mock.Mock(
uri=six.text_type(six.moves.urllib.parse.unquote(url)),
uri=str(six.moves.urllib.parse.unquote(url)),
headers=headers,
body=u"",
decoded_body=u"",
http_method=six.text_type(method),
body="",
decoded_body="",
http_method=str(method),
)
params = client.get_oauth_params(mock_request)
mock_request.oauth_params = params
mock_request.oauth_params.append((u'oauth_body_hash', oauth_body_hash))
mock_request.oauth_params.append(('oauth_body_hash', oauth_body_hash))
sig = client.get_oauth_signature(mock_request)
mock_request.oauth_params.append((u'oauth_signature', sig))
mock_request.oauth_params.append(('oauth_signature', sig))
new_headers = parameters.prepare_headers(mock_request.oauth_params, headers, realm=None)
return new_headers['Authorization']
@@ -295,17 +295,17 @@ class StubLtiHandler(StubHttpRequestHandler):
Returns `True` if signatures are correct, otherwise `False`.
"""
client_secret = six.text_type(self.server.config.get('client_secret', self.DEFAULT_CLIENT_SECRET))
client_secret = str(self.server.config.get('client_secret', self.DEFAULT_CLIENT_SECRET))
host = os.environ.get('BOK_CHOY_HOSTNAME', '127.0.0.1')
port = self.server.server_address[1]
lti_base = self.DEFAULT_LTI_ADDRESS.format(host=host, port=port)
lti_endpoint = self.server.config.get('lti_endpoint', self.DEFAULT_LTI_ENDPOINT)
url = lti_base + lti_endpoint
request = mock.Mock()
request.params = [(six.text_type(k), six.text_type(v)) for k, v in params.items()]
request.uri = six.text_type(url)
request.http_method = u'POST'
request.signature = six.text_type(client_signature)
request.params = [(str(k), str(v)) for k, v in params.items()]
request.uri = str(url)
request.http_method = 'POST'
request.signature = str(client_signature)
return signature.verify_hmac_sha1(request, client_secret)

8
common/djangoapps/terrain/stubs/start.py
View File

@@ -47,7 +47,7 @@ def get_args():
config_dict = _parse_config_args(sys.argv[3:])
if service_name not in SERVICES:
print("Unrecognized service '{0}'. Valid choices are: {1}".format(
print("Unrecognized service '{}'. Valid choices are: {}".format(
service_name, ", ".join(list(SERVICES.keys()))))
sys.exit(1)
@@ -57,7 +57,7 @@ def get_args():
raise ValueError
except ValueError:
print("Port '{0}' must be a positive integer".format(port_num))
print(f"Port '{port_num}' must be a positive integer")
sys.exit(1)
return service_name, port_num, config_dict
@@ -79,7 +79,7 @@ def _parse_config_args(args):
config_dict[components[0]] = "=".join(components[1:])
except: # lint-amnesty, pylint: disable=bare-except
print("Warning: could not interpret config value '{0}'".format(config_str))
print(f"Warning: could not interpret config value '{config_str}'")
return config_dict
@@ -89,7 +89,7 @@ def main():
Start a server; shut down on keyboard interrupt signal.
"""
service_name, port_num, config_dict = get_args()
print("Starting stub service '{0}' on port {1}...".format(service_name, port_num))
print(f"Starting stub service '{service_name}' on port {port_num}...")
server = SERVICES[service_name](port_num=port_num)
server.config.update(config_dict)

5
common/djangoapps/terrain/stubs/tests/test_edxnotes.py
View File

@@ -10,7 +10,6 @@ from uuid import uuid4
import ddt
import requests
import six
from six.moves import range
from ..edxnotes import StubEdxNotesService
@@ -24,7 +23,7 @@ class StubEdxNotesServiceTest(unittest.TestCase):
"""
Start the stub server.
"""
super(StubEdxNotesServiceTest, self).setUp() # lint-amnesty, pylint: disable=super-with-arguments
super().setUp()
self.server = StubEdxNotesService()
dummy_notes = self._get_dummy_notes(count=5)
self.server.add_notes(dummy_notes)
@@ -192,7 +191,7 @@ class StubEdxNotesServiceTest(unittest.TestCase):
updated_note = self._get_notes()[0]
assert 'new test text' == updated_note['text']
assert note['id'] == updated_note['id']
six.assertCountEqual(self, note, updated_note)
self.assertCountEqual(note, updated_note)
response = requests.get(self._get_url("api/v1/annotations/does_not_exist"))
assert response.status_code == 404

19
common/djangoapps/terrain/stubs/tests/test_http.py
View File

@@ -7,7 +7,6 @@ import json
import unittest
import requests
import six
from common.djangoapps.terrain.stubs.http import StubHttpRequestHandler, StubHttpService, require_params
@@ -15,10 +14,10 @@ from common.djangoapps.terrain.stubs.http import StubHttpRequestHandler, StubHtt
class StubHttpServiceTest(unittest.TestCase): # lint-amnesty, pylint: disable=missing-class-docstring
def setUp(self):
super(StubHttpServiceTest, self).setUp() # lint-amnesty, pylint: disable=super-with-arguments
super().setUp()
self.server = StubHttpService()
self.addCleanup(self.server.shutdown)
self.url = "http://127.0.0.1:{0}/set_config".format(self.server.port)
self.url = f"http://127.0.0.1:{self.server.port}/set_config"
def test_configure(self):
"""
@@ -34,12 +33,12 @@ class StubHttpServiceTest(unittest.TestCase): # lint-amnesty, pylint: disable=m
'test_key': 'test_val',
},
'test_empty_dict': {},
'test_unicode': u'\u2603 the snowman',
'test_unicode': '\u2603 the snowman',
'test_none': None,
'test_boolean': False
}
for key, val in six.iteritems(params):
for key, val in params.items():
# JSON-encode each parameter
post_params = {key: json.dumps(val)}
@@ -47,7 +46,7 @@ class StubHttpServiceTest(unittest.TestCase): # lint-amnesty, pylint: disable=m
assert response.status_code == 200
# Check that the expected values were set in the configuration
for key, val in six.iteritems(params):
for key, val in params.items():
assert self.server.config.get(key) == val
def test_bad_json(self):
@@ -60,12 +59,12 @@ class StubHttpServiceTest(unittest.TestCase): # lint-amnesty, pylint: disable=m
def test_unicode_non_json(self):
# Send unicode without json-encoding it
response = requests.put(self.url, data={'test_unicode': u'\u2603 the snowman'})
response = requests.put(self.url, data={'test_unicode': '\u2603 the snowman'})
assert response.status_code == 400
def test_unknown_path(self):
response = requests.put(
"http://127.0.0.1:{0}/invalid_url".format(self.server.port),
f"http://127.0.0.1:{self.server.port}/invalid_url",
data="{}"
)
assert response.status_code == 404
@@ -91,10 +90,10 @@ class RequireParamTest(unittest.TestCase):
"""
def setUp(self):
super(RequireParamTest, self).setUp() # lint-amnesty, pylint: disable=super-with-arguments
super().setUp()
self.server = RequireHttpService()
self.addCleanup(self.server.shutdown)
self.url = "http://127.0.0.1:{port}".format(port=self.server.port)
self.url = f"http://127.0.0.1:{self.server.port}"
def test_require_get_param(self):

14
common/djangoapps/terrain/stubs/tests/test_lti_stub.py
View File

@@ -4,10 +4,10 @@ Unit tests for stub LTI implementation.
import unittest
from unittest.mock import Mock, patch
import requests
import six
from mock import Mock, patch
from urllib.request import urlopen # pylint: disable=wrong-import-order
from common.djangoapps.terrain.stubs.lti import StubLtiService
@@ -20,9 +20,9 @@ class StubLtiServiceTest(unittest.TestCase):
Used for lettuce BDD tests in lms/courseware/features/lti.feature
"""
def setUp(self):
super(StubLtiServiceTest, self).setUp() # lint-amnesty, pylint: disable=super-with-arguments
super().setUp()
self.server = StubLtiService()
self.uri = 'http://127.0.0.1:{}/'.format(self.server.port)
self.uri = f'http://127.0.0.1:{self.server.port}/'
self.launch_uri = self.uri + 'correct_lti_endpoint'
self.addCleanup(self.server.shutdown)
self.payload = {
@@ -74,7 +74,7 @@ class StubLtiServiceTest(unittest.TestCase):
grade_uri = self.uri + 'grade'
with patch('common.djangoapps.terrain.stubs.lti.requests.post') as mocked_post:
mocked_post.return_value = Mock(content='Test response', status_code=200)
response = six.moves.urllib.request.urlopen(grade_uri, data=b'')
response = urlopen(grade_uri, data=b'')
assert b'Test response' in response.read()
@patch('common.djangoapps.terrain.stubs.lti.signature.verify_hmac_sha1', return_value=True)
@@ -84,7 +84,7 @@ class StubLtiServiceTest(unittest.TestCase):
grade_uri = self.uri + 'lti2_outcome'
with patch('common.djangoapps.terrain.stubs.lti.requests.put') as mocked_put:
mocked_put.return_value = Mock(status_code=200)
response = six.moves.urllib.request.urlopen(grade_uri, data=b'')
response = urlopen(grade_uri, data=b'')
assert b'LTI consumer (edX) responded with HTTP 200' in response.read()
@patch('common.djangoapps.terrain.stubs.lti.signature.verify_hmac_sha1', return_value=True)
@@ -94,5 +94,5 @@ class StubLtiServiceTest(unittest.TestCase):
grade_uri = self.uri + 'lti2_delete'
with patch('common.djangoapps.terrain.stubs.lti.requests.put') as mocked_put:
mocked_put.return_value = Mock(status_code=200)
response = six.moves.urllib.request.urlopen(grade_uri, data=b'')
response = urlopen(grade_uri, data=b'')
assert b'LTI consumer (edX) responded with HTTP 200' in response.read()

6
common/djangoapps/terrain/stubs/tests/test_video.py
View File

@@ -33,16 +33,16 @@ class StubVideoServiceTest(unittest.TestCase):
"""
Start the stub server.
"""
super(StubVideoServiceTest, self).setUp() # lint-amnesty, pylint: disable=super-with-arguments
super().setUp()
self.server = VideoSourceHttpService()
self.server.config['root_dir'] = '{}/data/video'.format(settings.TEST_ROOT)
self.server.config['root_dir'] = f'{settings.TEST_ROOT}/data/video'
self.addCleanup(self.server.shutdown)
def test_get_hls_manifest(self):
"""
Verify that correct hls manifest is received.
"""
response = requests.get("http://127.0.0.1:{port}/hls/history.m3u8".format(port=self.server.port))
response = requests.get(f"http://127.0.0.1:{self.server.port}/hls/history.m3u8")
assert response.ok
assert response.text == HLS_MANIFEST_TEXT.lstrip()
assert response.headers['Access-Control-Allow-Origin'] == '*'

8
common/djangoapps/terrain/stubs/tests/test_xqueue_stub.py
View File

@@ -6,14 +6,14 @@ Unit tests for stub XQueue implementation.
import ast
import json
import unittest
from unittest import mock
import mock
import requests
from ..xqueue import StubXQueueService
class FakeTimer(object):
class FakeTimer:
"""
Fake timer implementation that executes immediately.
"""
@@ -27,9 +27,9 @@ class FakeTimer(object):
class StubXQueueServiceTest(unittest.TestCase): # lint-amnesty, pylint: disable=missing-class-docstring
def setUp(self):
super(StubXQueueServiceTest, self).setUp() # lint-amnesty, pylint: disable=super-with-arguments
super().setUp()
self.server = StubXQueueService()
self.url = "http://127.0.0.1:{0}/xqueue/submit".format(self.server.port)
self.url = f"http://127.0.0.1:{self.server.port}/xqueue/submit"
self.addCleanup(self.server.shutdown)
# Patch the timer async calls

4
common/djangoapps/terrain/stubs/tests/test_youtube_stub.py
View File

@@ -13,9 +13,9 @@ from ..youtube import StubYouTubeService
class StubYouTubeServiceTest(unittest.TestCase): # lint-amnesty, pylint: disable=missing-class-docstring
def setUp(self):
super(StubYouTubeServiceTest, self).setUp() # lint-amnesty, pylint: disable=super-with-arguments
super().setUp()
self.server = StubYouTubeService()
self.url = "http://127.0.0.1:{0}/".format(self.server.port)
self.url = f"http://127.0.0.1:{self.server.port}/"
self.server.config['time_to_response'] = 0.0
self.addCleanup(self.server.shutdown)

2
common/djangoapps/terrain/stubs/video_source.py
View File

@@ -25,7 +25,7 @@ class VideoSourceRequestHandler(SimpleHTTPRequestHandler):
becomes /gizmo.mp4
"""
root_dir = self.server.config.get('root_dir')
path = '{}{}'.format(root_dir, path)
path = f'{root_dir}{path}'
return path.split('?')[0]
def end_headers(self):

17
common/djangoapps/terrain/stubs/xqueue.py
View File

@@ -14,7 +14,6 @@ import copy
import json
from threading import Timer
import six
from requests import post
from openedx.core.djangolib.markup import HTML
@@ -38,7 +37,7 @@ class StubXQueueHandler(StubHttpRequestHandler):
Sends back an immediate success/failure response.
It then POSTS back to the client with grading results.
"""
msg = "XQueue received POST request {0} to path {1}".format(self.post_dict, self.path)
msg = f"XQueue received POST request {self.post_dict} to path {self.path}"
self.log_message(msg)
# Respond only to grading requests
@@ -100,7 +99,7 @@ class StubXQueueHandler(StubHttpRequestHandler):
self.send_response(
200, content=response_str, headers={'Content-type': 'text/plain'}
)
self.log_message("XQueue: sent response {0}".format(response_str))
self.log_message(f"XQueue: sent response {response_str}")
else:
self.send_response(500)
@@ -138,7 +137,7 @@ class StubXQueueHandler(StubHttpRequestHandler):
# Multiple matches, so abort and log an error
else:
self.log_error(
"Multiple response patterns matched '{0}'".format(xqueue_body_json),
f"Multiple response patterns matched '{xqueue_body_json}'",
)
return
@@ -159,7 +158,7 @@ class StubXQueueHandler(StubHttpRequestHandler):
}
post(postback_url, data=data)
self.log_message("XQueue: sent grading response {0} to {1}".format(data, postback_url))
self.log_message(f"XQueue: sent grading response {data} to {postback_url}")
def _register_submission(self, xqueue_body_json):
"""
@@ -174,7 +173,7 @@ class StubXQueueHandler(StubHttpRequestHandler):
xqueue_body = json.loads(xqueue_body_json)
except ValueError:
self.log_error(
"Could not decode XQueue body as JSON: '{0}'".format(xqueue_body_json))
f"Could not decode XQueue body as JSON: '{xqueue_body_json}'")
else:
@@ -187,12 +186,12 @@ class StubXQueueHandler(StubHttpRequestHandler):
response = post(url, data={'grader_payload': grader_payload})
if not response.ok:
self.log_error(
"Could register submission at URL '{0}'. Status was {1}".format(
"Could register submission at URL '{}'. Status was {}".format(
url, response.status_code))
else:
self.log_message(
"XQueue body is missing 'grader_payload' key: '{0}'".format(xqueue_body)
f"XQueue body is missing 'grader_payload' key: '{xqueue_body}'"
)
def _is_grade_request(self):
@@ -222,6 +221,6 @@ class StubXQueueService(StubHttpService):
"""
return list({
key: value
for key, value in six.iteritems(self.config)
for key, value in self.config.items()
if key not in self.NON_QUEUE_CONFIG_KEYS
}.items())

6
common/djangoapps/terrain/stubs/youtube.py
View File

@@ -52,7 +52,7 @@ class StubYouTubeHandler(StubHttpRequestHandler):
Handle a GET request from the client and sends response back.
"""
self.log_message(
"Youtube provider received GET request to path {}".format(self.path)
f"Youtube provider received GET request to path {self.path}"
)
if 'get_config' in self.path:
@@ -137,7 +137,7 @@ class StubYouTubeHandler(StubHttpRequestHandler):
response = "{cb}({data})".format(cb=callback, data=json.dumps(data)).encode('utf-8')
self.send_response(200, content=response, headers={'Content-type': 'text/html'})
self.log_message("Youtube: sent response {}".format(message))
self.log_message(f"Youtube: sent response {message}")
def _send_private_video_response(self, message):
"""
@@ -161,7 +161,7 @@ class StubYouTubeHandler(StubHttpRequestHandler):
response = "{cb}({data})".format(cb=callback, data=json.dumps(data)).encode('utf-8')
self.send_response(200, content=response, headers={'Content-type': 'text/html'})
self.log_message("Youtube: sent response {}".format(message))
self.log_message(f"Youtube: sent response {message}")
class StubYouTubeService(StubHttpService):