Skip to content

Fix #1828: Reduce cyclomatic complexity in argparser.py #1930

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
vinayK34 wants to merge 1 commit into
base: master
Choose a base branch
from
Open

Fix #1828: Reduce cyclomatic complexity in argparser.py #1930

Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
177 changes: 102 additions & 75 deletions httpie/cli/argparser.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,7 @@
from argparse import RawDescriptionHelpFormatter
from textwrap import dedent
from urllib.parse import urlsplit
from typing import Optional, Tuple, Union

from requests.utils import get_netrc_auth

Expand Down Expand Up @@ -280,80 +281,107 @@ def _process_ssl_cert(self):
self.args.cert_key_pass.prompt_password(self.args.cert_key)

def _process_auth(self):
"""Process authentication-related arguments."""
# TODO: refactor & simplify this method.
self.args.auth_plugin = None
default_auth_plugin = plugin_manager.get_auth_plugins()[0]
auth_type_set = self.args.auth_type is not None
url = urlsplit(self.args.url)

# Handle embedded URL credentials
if self.args.auth is None and not auth_type_set:
if url.username is not None:
# Handle http://username:password@hostname/
username = url.username
password = url.password or ''
self.args.auth = AuthCredentials(
key=username,
value=password,
sep=SEPARATOR_CREDENTIALS,
orig=SEPARATOR_CREDENTIALS.join([username, password])
)
self._handle_embedded_url_credentials(url)

# Process authentication if needed
if self.args.auth is not None or auth_type_set:
if not self.args.auth_type:
self.args.auth_type = default_auth_plugin.auth_type
plugin = plugin_manager.get_auth_plugin(self.args.auth_type)()

if (not self.args.ignore_netrc
and self.args.auth is None
and plugin.netrc_parse):
# Only host needed, so it’s OK URL not finalized.
netrc_credentials = get_netrc_auth(self.args.url)
if netrc_credentials:
self.args.auth = AuthCredentials(
key=netrc_credentials[0],
value=netrc_credentials[1],
sep=SEPARATOR_CREDENTIALS,
orig=SEPARATOR_CREDENTIALS.join(netrc_credentials)
)
self._process_auth_with_plugin(default_auth_plugin, auth_type_set, url)

if plugin.auth_require and self.args.auth is None:
self.error('--auth required')
# Handle .netrc if needed
if (not self.args.ignore_netrc
and self.args.auth is None
and self._get_auth_plugin(default_auth_plugin).netrc_parse):
self._process_netrc_credentials(url)

plugin.raw_auth = self.args.auth
self.args.auth_plugin = plugin
already_parsed = isinstance(self.args.auth, AuthCredentials)
# Handle auth requirement validation
if self._get_auth_plugin(default_auth_plugin).auth_require and self.args.auth is None:
self.error('--auth required')

if self.args.auth is None or not plugin.auth_parse:
self.args.auth = plugin.get_auth()
else:
if already_parsed:
# from the URL
credentials = self.args.auth
else:
credentials = parse_auth(self.args.auth)

if (not credentials.has_password()
and plugin.prompt_password):
if self.args.ignore_stdin:
# Non-tty stdin read by now
self.error(
'Unable to prompt for passwords because'
' --ignore-stdin is set.'
)
credentials.prompt_password(url.netloc)

if (credentials.key and credentials.value):
plugin.raw_auth = credentials.key + ":" + credentials.value

self.args.auth = plugin.get_auth(
username=credentials.key,
password=credentials.value,
)
# Handle authentication prompting and finalization
if self.args.auth is not None:
self._finalize_authentication(default_auth_plugin, url)

# Set no-op auth to force requests to ignore .netrc
if not self.args.auth and self.args.ignore_netrc:
# Set a no-op auth to force requests to ignore .netrc
# <https://github.com/psf/requests/issues/2773#issuecomment-174312831>
self.args.auth = ExplicitNullAuth()

def _handle_embedded_url_credentials(self, url):
"""Handle authentication credentials embedded in the URL."""
if url.username is not None:
# Handle http://username:***@hostname/
username = url.username
password = url.password or ''
self.args.auth = AuthCredentials(
key=username,
value=password,
sep=SEPARATOR_CREDENTIALS,
orig=SEPARATOR_CREDENTIALS.join([username, password])
)

def _get_auth_plugin(self, default_auth_plugin):
"""Get the authentication plugin."""
if not self.args.auth_type:
self.args.auth_type = default_auth_plugin.auth_type
return plugin_manager.get_auth_plugin(self.args.auth_type)()

def _process_auth_with_plugin(self, default_auth_plugin, auth_type_set, url):
"""Process authentication with plugin."""
plugin = self._get_auth_plugin(default_auth_plugin)()
self.args.auth_plugin = plugin
plugin.raw_auth = self.args.auth

def _process_netrc_credentials(self, url):
"""Process .netrc credentials."""
netrc_credentials = get_netrc_auth(self.args.url)
if netrc_credentials:
self.args.auth = AuthCredentials(
key=netrc_credentials[0],
value=netrc_credentials[1],
sep=SEPARATOR_CREDENTIALS,
orig=SEPARATOR_CREDENTIALS.join(netrc_credentials)
)

def _finalize_authentication(self, default_auth_plugin, url):
"""Finalize authentication processing."""
plugin = self._get_auth_plugin(default_auth_plugin)()
already_parsed = isinstance(self.args.auth, AuthCredentials)

if self.args.auth is None or not plugin.auth_parse:
self.args.auth = plugin.get_auth()
else:
if already_parsed:
# from the URL
credentials = self.args.auth
else:
credentials = parse_auth(self.args.auth)

if (not credentials.has_password()
and plugin.prompt_password):
if self.args.ignore_stdin:
# Non-tty stdin read by now
self.error(
'Unable to prompt for passwords because'
' --ignore-stdin is set.'
)
credentials.prompt_password(url.netloc)

if (credentials.key and credentials.value):
plugin.raw_auth = credentials.key + ":" + credentials.value

self.args.auth = plugin.get_auth(
username=credentials.key,
password=creden…lue,
)

def _apply_no_options(self, no_options):
"""For every `--no-OPTION` in `no_options`, set `args.OPTION` to
its default value. This allows for un-setting of options, e.g.,
Expand All @@ -377,7 +405,7 @@ def _apply_no_options(self, no_options):
invalid.append(option)

if invalid:
self.error(f'unrecognized arguments: {" ".join(invalid)}')
self.error('unrecognized arguments: {}'.format(' '.join(invalid)))

def _body_from_file(self, fd):
"""Read the data from a file-like object.
Expand Down Expand Up @@ -474,7 +502,7 @@ def _parse_items(self):
if key != '':
self.error(
'Invalid file fields (perhaps you meant --form?):'
f' {",".join(self.args.files.keys())}')
'{}'.format(','.join(self.args.files.keys())))
if request_file is not None:
self.error("Can't read request from multiple files")
request_file = file
Expand All @@ -489,18 +517,18 @@ def _parse_items(self):
if content_type:
self.args.headers['Content-Type'] = content_type

def _check_options(self, value, option):
"""Check validity of output options."""
unknown = set(value) - OUTPUT_OPTIONS
if unknown:
self.error('Unknown output options: {}={}'.format(option, ','.join(unknown)))

def _process_output_options(self):
"""Apply defaults to output options, or validate the provided ones.

The default output options are stdout-type-sensitive.

"""

def check_options(value, option):
unknown = set(value) - OUTPUT_OPTIONS
if unknown:
self.error(f'Unknown output options: {option}={",".join(unknown)}')

if self.args.verbose:
self.args.all = True

Expand All @@ -519,8 +547,8 @@ def check_options(value, option):
if self.args.output_options_history is None:
self.args.output_options_history = self.args.output_options

check_options(self.args.output_options, '--print')
check_options(self.args.output_options_history, '--history-print')
self._check_options(self.args.output_options, '--print')
self._check_options(self.args.output_options_history, '--history-print')

if self.args.download and OUT_RESP_BODY in self.args.output_options:
# Response body is always downloaded with --download and it goes
Expand Down Expand Up @@ -601,13 +629,12 @@ def error(self, message):
self.print_usage(sys.stderr)
self.env.rich_error_console.print(
dedent(
f'''
'''
[bold]error[/bold]:
{message}

{}
[bold]for more information[/bold]:
run '{self.prog} --help' or visit https://httpie.io/docs/cli
'''.rstrip()
run '{}' --help' or visit https://httpie.io/docs/cli
'''.format(message, self.prog).rstrip()
)
)
self.exit(2)
self.exit(2)
Loading