Fix pep8 gate

This fix propose two parts:
* introduce timeout (60s) to requests calls
* remove `file` scheme support for requests calls.

Change-Id: Ide2c2915ba5d6ff03933160b74f7206492276968
This commit is contained in:
ricolin 2023-03-14 09:17:54 +08:00
parent 6e8657912c
commit 6169eb26ed
5 changed files with 14 additions and 18 deletions

View File

@ -34,8 +34,6 @@ def get(url, allowed_schemes=('http', 'https')):
"""Get the data at the specified URL.
The URL must use the http: or https: schemes.
The file: scheme is also supported if you override
the allowed_schemes argument.
Raise an IOError if getting the data fails.
"""
LOG.info('Fetching data from %s', url)
@ -45,14 +43,8 @@ def get(url, allowed_schemes=('http', 'https')):
if components.scheme not in allowed_schemes:
raise URLFetchError(_('Invalid URL scheme %s') % components.scheme)
if components.scheme == 'file': # nosec
try:
return urllib.request.urlopen(url).read()
except urllib.error.URLError as uex:
raise URLFetchError(_('Failed to retrieve manifest: %s') % uex)
try:
resp = requests.get(url, stream=True)
resp = requests.get(url, stream=True, timeout=60)
resp.raise_for_status()
# We cannot use resp.text here because it would download the

View File

@ -68,7 +68,7 @@ def create_dirs():
def _get_public_ip():
return requests.get(PUBLIC_IP_URL).text
return requests.get(PUBLIC_IP_URL, timeout=60).text
def _build_subject_alt_names(config):
@ -95,7 +95,7 @@ def write_ca_cert(config, verify_ca):
'OpenStack-API-Version': 'container-infra latest'}
ca_cert_resp = requests.get(cluster_cert_url,
headers=headers,
verify=verify_ca)
verify=verify_ca, timeout=60)
with open(CA_CERT_PATH, 'w') as fp:
fp.write(ca_cert_resp.json()['pem'])
@ -141,7 +141,7 @@ def write_server_cert(config, csr_req, verify_ca):
csr_resp = requests.post(cert_url,
data=json.dumps(csr_req),
headers=headers,
verify=verify_ca)
verify=verify_ca, timeout=60)
with open(SERVER_CERT_PATH, 'w') as fp:
fp.write(csr_resp.json()['pem'])
@ -172,7 +172,8 @@ def get_user_token(config, verify_ca):
creds = creds_str % params
headers = {'Content-Type': 'application/json'}
url = config['AUTH_URL'] + '/auth/tokens'
r = requests.post(url, headers=headers, data=creds, verify=verify_ca)
r = requests.post(url, headers=headers, data=creds, verify=verify_ca,
timeout=60)
config['USER_TOKEN'] = r.headers['X-Subject-Token']
return config

View File

@ -481,7 +481,7 @@ class BaseTemplateDefinition(TemplateDefinition):
url += '/_config/size'
try:
result = requests.get(url).text
result = requests.get(url, timeout=60).text
except req_exceptions.RequestException as err:
LOG.error(err)
raise exception.GetClusterSizeFailed(
@ -521,7 +521,8 @@ class BaseTemplateDefinition(TemplateDefinition):
CONF.cluster.etcd_discovery_service_endpoint_format %
{'size': cluster.master_count})
try:
discovery_request = requests.get(discovery_endpoint)
discovery_request = requests.get(discovery_endpoint,
timeout=60)
if discovery_request.status_code != requests.codes.ok:
raise exception.GetDiscoveryUrlFailed(
discovery_endpoint=discovery_endpoint)

View File

@ -1182,7 +1182,7 @@ class TestClusterConductorWithK8s(base.TestCase):
'../../common/templates/environments/disable_lb_floating_ip.yaml',
],
env_files)
reqget.assert_called_once_with('http://etcd/test?size=1')
reqget.assert_called_once_with('http://etcd/test?size=1', timeout=60)
@patch('magnum.common.short_id.generate_id')
@patch('heatclient.common.template_utils.get_template_contents')

View File

@ -1376,7 +1376,8 @@ class AtomicK8sTemplateDefinitionTestCase(BaseK8sTemplateDefinitionTestCase):
k8s_def = k8sa_tdef.AtomicK8sTemplateDefinition()
discovery_url = k8s_def.get_discovery_url(mock_cluster)
mock_get.assert_called_once_with('http://etcd/test?size=10')
mock_get.assert_called_once_with('http://etcd/test?size=10',
timeout=60)
self.assertEqual(expected_discovery_url, mock_cluster.discovery_url)
self.assertEqual(expected_discovery_url, discovery_url)
@ -2043,7 +2044,8 @@ class AtomicSwarmTemplateDefinitionTestCase(base.TestCase):
swarm_def = swarm_tdef.AtomicSwarmTemplateDefinition()
discovery_url = swarm_def.get_discovery_url(mock_cluster)
mock_get.assert_called_once_with('http://etcd/test?size=1')
mock_get.assert_called_once_with('http://etcd/test?size=1',
timeout=60)
self.assertEqual(mock_cluster.discovery_url, expected_discovery_url)
self.assertEqual(discovery_url, expected_discovery_url)