diff --git a/.pre-commit-config.yaml b/.pre-commit-config.yaml index 60909695..f31ccb6e 100644 --- a/.pre-commit-config.yaml +++ b/.pre-commit-config.yaml @@ -37,16 +37,12 @@ repos: hooks: - id: mdformat name: Format markdown - - repo: https://github.com/pycqa/isort - rev: 8.0.1 + - repo: https://github.com/astral-sh/ruff-pre-commit + rev: v0.15.21 hooks: - - id: isort - name: isort (python) - args: [--profile, black, --filter-files] - - repo: https://github.com/psf/black - rev: 26.3.1 - hooks: - - id: black + - id: ruff-check + args: [--fix] + - id: ruff-format - repo: https://github.com/pre-commit/mirrors-mypy rev: v1.19.1 hooks: @@ -55,10 +51,3 @@ repos: additional_dependencies: - types-requests - boto3-stubs[sts,iam] - - repo: https://github.com/pylint-dev/pylint - rev: v4.0.5 - hooks: - - id: pylint - name: pylint (Python Linting) - files: ^(oktaawscli|tests)/ - args: [--rcfile=pylintrc, --output-format=colorized, --score=no] diff --git a/CONTRIBUTING.md b/CONTRIBUTING.md index ff90f51c..6b77485c 100644 --- a/CONTRIBUTING.md +++ b/CONTRIBUTING.md @@ -20,10 +20,7 @@ It's highly recommended to use virtualenv! pip install . ``` -- Ensure that you can run pylint against your code and no errors are returned. Pull Requests with pylint errors will be rejected. - - - Currently, automated builds are only checking for actual errors, as there are some refactoring and other such notices that need to be resolved. - - You can safely run `pylint --errors-only oktaawscli` to replicate what the build will be checking. +- Ensure `pre-commit run --all-files` passes (runs ruff for linting/formatting and mypy for type checking). Pull Requests with lint or type errors will be rejected. - Increment the version in `oktaawscli/version.py`, according to [SemVer](https://semver.org/). diff --git a/oktaawscli/__init__.py b/oktaawscli/__init__.py index 0a50a4c2..8cf708fa 100644 --- a/oktaawscli/__init__.py +++ b/oktaawscli/__init__.py @@ -1,3 +1,3 @@ """init""" -from .version import __version__ +from .version import __version__ as __version__ diff --git a/oktaawscli/aws_auth.py b/oktaawscli/aws_auth.py index 6bf320be..ccf42d4e 100644 --- a/oktaawscli/aws_auth.py +++ b/oktaawscli/aws_auth.py @@ -68,10 +68,7 @@ def choose_aws_role(self, assertion): self.logger.info("Using predefined role: %s" % self.role) return predefined_role else: - self.logger.info( - "Predefined role, %s, not found in the list of roles assigned to you." - % self.role - ) + self.logger.info("Predefined role, %s, not found in the list of roles assigned to you." % self.role) self.logger.info("Please choose a role.") if len(roles) == 1: @@ -106,9 +103,7 @@ def get_sts_token(self, role_arn, principal_arn, assertion, duration): if profile is not None: os.environ["AWS_PROFILE"] = profile except ProfileNotFound: - self.logger.exception( - "Unable to handle AWS_PROFILE=%s" % os.environ["AWS_PROFILE"] - ) + self.logger.exception("Unable to handle AWS_PROFILE=%s" % os.environ["AWS_PROFILE"]) response = sts.assume_role_with_saml( RoleArn=role_arn, @@ -137,9 +132,7 @@ def check_sts_token(self, profile): return False elif not parser.has_section(profile): - self.logger.info( - "No existing credentials found. Requesting new credentials." - ) + self.logger.info("No existing credentials found. Requesting new credentials.") return False session = boto3.Session(profile_name=profile) @@ -149,23 +142,17 @@ def check_sts_token(self, profile): except (ClientError, NoCredentialsError) as ex: if str(ex) == "Unable to locate credentials": - self.logger.info( - "No credentials have been located. Requesting new credentials." - ) + self.logger.info("No credentials have been located. Requesting new credentials.") return False elif ex.response["Error"]["Code"] == "ExpiredToken": - self.logger.info( - "Temporary credentials have expired. Requesting new credentials." - ) + self.logger.info("Temporary credentials have expired. Requesting new credentials.") return False print("AWS credentials are valid. Nothing to do.") self.logger.info("STS credentials are valid. Nothing to do.") return True - def write_sts_token( - self, profile, access_key_id, secret_access_key, session_token, region=None - ): + def write_sts_token(self, profile, access_key_id, secret_access_key, session_token, region=None): """Writes STS auth information to credentials file""" region = region or self.region output = "json" @@ -225,9 +212,7 @@ def __extract_available_roles_from(assertion): roles = [] role_tuple = namedtuple("RoleTuple", ["principal_arn", "role_arn"]) root = ET.fromstring(base64.b64decode(assertion)) - for saml2attribute in root.iter( - "{urn:oasis:names:tc:SAML:2.0:assertion}Attribute" - ): + for saml2attribute in root.iter("{urn:oasis:names:tc:SAML:2.0:assertion}Attribute"): if saml2attribute.get("Name") == aws_attribute_role: for saml2attributevalue in saml2attribute.iter(attribute_value_urn): roles.append(role_tuple(*saml2attributevalue.text.split(","))) @@ -257,12 +242,8 @@ def __get_role_info(self, roles, assertion): current_date = date.today() alias_age = current_date - last_updated if alias_age.days >= 7 or alias is None: - self.logger.info( - "Refreshing cached alias for role %s" % role.role_arn - ) - alias = self.__get_account_alias( - role.role_arn, role.principal_arn, assertion - ) + self.logger.info("Refreshing cached alias for role %s" % role.role_arn) + alias = self.__get_account_alias(role.role_arn, role.principal_arn, assertion) last_updated = current_date if alias is None: continue @@ -301,13 +282,9 @@ def __get_account_alias(self, role_arn, principal_arn, assertion): try: sts = boto3.client("sts") except ProfileNotFound: - self.logger.exception( - "Unable to handle AWS_PROFILE=%s" % os.environ["AWS_PROFILE"] - ) + self.logger.exception("Unable to handle AWS_PROFILE=%s" % os.environ["AWS_PROFILE"]) try: - saml_resp = sts.assume_role_with_saml( - RoleArn=role_arn, PrincipalArn=principal_arn, SAMLAssertion=assertion - ) + saml_resp = sts.assume_role_with_saml(RoleArn=role_arn, PrincipalArn=principal_arn, SAMLAssertion=assertion) except ClientError: self.logger.warning( "Unable to assume role '%s', cannot get account alias", @@ -337,9 +314,7 @@ def __get_account_alias(self, role_arn, principal_arn, assertion): exc_info=self.debug, ) else: - self.logger.exception( - "Unknown Error. Unable to get account alias for role %s", role_arn - ) + self.logger.exception("Unknown Error. Unable to get account alias for role %s", role_arn) return "unknown" @staticmethod @@ -347,9 +322,7 @@ def __create_options_from(roles): options = [] for index, role in enumerate(roles): # role[0] is the role arn, role[2] is the account alias - options.append( - "[%s]: %s : %s" % (str(index + 1).ljust(2), role[2].ljust(27), role[0]) - ) + options.append("[%s]: %s : %s" % (str(index + 1).ljust(2), role[2].ljust(27), role[0])) return options def __find_predefined_role_from(self, roles): diff --git a/oktaawscli/okta_auth.py b/oktaawscli/okta_auth.py index 07a318dc..c33399fd 100644 --- a/oktaawscli/okta_auth.py +++ b/oktaawscli/okta_auth.py @@ -25,9 +25,7 @@ class OktaAuth: """Handles auth to Okta and returns SAML assertion""" - def __init__( - self, okta_profile, verbose, logger, totp_token, okta_auth_config, debug=False - ): + def __init__(self, okta_profile, verbose, logger, totp_token, okta_auth_config, debug=False): self.okta_profile = okta_profile self.totp_token = totp_token self.logger = logger @@ -53,37 +51,28 @@ def primary_auth(self): with locked(self.token_path, timeout=INTERACTIVE_LOCK_TIMEOUT_SECONDS): refreshed = self.get_cached_session_id() if refreshed is not None and refreshed != session_id: - self.logger.info( - "Cached Okta session was refreshed by another process; using it." - ) + self.logger.info("Cached Okta session was refreshed by another process; using it.") return refreshed return self.get_session(self._run_authn_flow()) def _run_authn_flow(self): """Runs the Okta authn POST and returns a sessionToken. Caller holds the lock.""" - self.logger.warning( - "Cached Okta session is missing or invalid. Authenticating now..." - ) + self.logger.warning("Cached Okta session is missing or invalid. Authenticating now...") auth_data = { "username": self.okta_auth_config.username_for(self.okta_profile), "password": self.okta_auth_config.password_for(self.okta_profile), } # https://developer.okta.com/docs/reference/api/authn/ - resp_json = self._okta_json_request( - "POST", "/api/v1/authn", "_run_authn_flow", json=auth_data - ) + resp_json = self._okta_json_request("POST", "/api/v1/authn", "_run_authn_flow", json=auth_data) if "status" in resp_json: status = resp_json["status"] if status == "MFA_REQUIRED": - return self.verify_mfa( - resp_json["_embedded"]["factors"], resp_json["stateToken"] - ) + return self.verify_mfa(resp_json["_embedded"]["factors"], resp_json["stateToken"]) if status == "SUCCESS": return resp_json["sessionToken"] if status == "MFA_ENROLL": self.logger.warning( - "MFA not enrolled. Cannot continue. " - "Please enroll an MFA factor in the Okta Web UI first!" + "MFA not enrolled. Cannot continue. Please enroll an MFA factor in the Okta Web UI first!" ) sys.exit(2) if status == "LOCKED_OUT": @@ -134,9 +123,7 @@ def verify_mfa(self, factors_list, state_token): if self.factor: if self.factor == factor_provider: factor_choice = index - self.logger.info( - "Using pre-selected factor choice from ~/.okta-aws" - ) + self.logger.info("Using pre-selected factor choice from ~/.okta-aws") break else: print("%d: %s" % (index + 1, factor_name)) @@ -146,12 +133,9 @@ def verify_mfa(self, factors_list, state_token): self.okta_profile, supported_factors[factor_choice]["provider"] ) self.logger.info( - "Performing secondary authentication using: %s" - % supported_factors[factor_choice]["provider"] - ) - session_token = self.verify_single_factor( - supported_factors[factor_choice], state_token + "Performing secondary authentication using: %s" % supported_factors[factor_choice]["provider"] ) + session_token = self.verify_single_factor(supported_factors[factor_choice], state_token) else: print("MFA required, but no supported factors enrolled! Exiting.") exit(1) @@ -178,9 +162,7 @@ def verify_single_factor(self, factor, state_token): elif resp_json["status"] == "MFA_CHALLENGE": print("Waiting for push verification...") while True: - resp = requests.post( - resp_json["_links"]["next"]["href"], json=req_data - ) + resp = requests.post(resp_json["_links"]["next"]["href"], json=req_data) resp_json = resp.json() if resp_json["status"] == "SUCCESS": return resp_json["sessionToken"] @@ -204,9 +186,7 @@ def get_session(self, session_token): """Gets a session cookie from a session token""" data = {"sessionToken": session_token} # https://developer.okta.com/docs/guides/ie-limitations/main/#sessions-apis - resp = self._okta_json_request( - "POST", "/api/v1/sessions", "get_session", json=data - ) + resp = self._okta_json_request("POST", "/api/v1/sessions", "get_session", json=data) self.cache_session_id(resp["id"], resp["expiresAt"]) return resp["id"] @@ -236,9 +216,7 @@ def get_cached_session_id(self): expiration_date = datetime.min if session_info.get("expiration_date"): - expiration_date = datetime.strptime( - session_info.get("expiration_date"), "%Y-%m-%dT%H:%M:%S.%fZ" - ) + expiration_date = datetime.strptime(session_info.get("expiration_date"), "%Y-%m-%dT%H:%M:%S.%fZ") current_time = datetime.utcnow() if max([current_time, expiration_date]) == expiration_date: @@ -252,17 +230,11 @@ def check_for_desync(self, session_id): sid = "sid=%s" % session_id headers = {"Cookie": sid} # https://developer.okta.com/docs/api/openapi/okta-management/management/tag/User/#tag/User/operation/getUser - raw_resp = requests.get( - self.https_base_url + "/api/v1/users/me", headers=headers - ) + raw_resp = requests.get(self.https_base_url + "/api/v1/users/me", headers=headers) raw_resp.raise_for_status() return False except requests.HTTPError as e: - if ( - e.response is None - or e.response.status_code != 403 - or "Invalid session" not in e.response.text - ): + if e.response is None or e.response.status_code != 403 or "Invalid session" not in e.response.text: raise e message = "Okta session invalidated. Refreshing token now..." self.logger.error(message) @@ -321,9 +293,7 @@ def get_apps(self, session_id): sid = "sid=%s" % session_id headers = {"Cookie": sid} # https://developer.okta.com/docs/api/openapi/okta-management/management/tag/UserResources/#tag/UserResources/operation/listAppLinks - resp = self._okta_json_request( - "GET", "/api/v1/users/me/appLinks", "get_apps", headers=headers - ) + resp = self._okta_json_request("GET", "/api/v1/users/me/appLinks", "get_apps", headers=headers) aws_apps = [] for app in resp: @@ -342,9 +312,7 @@ def get_apps(self, session_id): print("%d: %s" % (index + 1, app["label"])) if app_choice is None: app_choice = int(input("Please select AWS app: ")) - 1 - self.okta_auth_config.save_chosen_app_for_profile( - self.okta_profile, aws_apps[app_choice]["label"] - ) + self.okta_auth_config.save_chosen_app_for_profile(self.okta_profile, aws_apps[app_choice]["label"]) return aws_apps[app_choice]["label"], aws_apps[app_choice]["linkUrl"] diff --git a/oktaawscli/okta_auth_config.py b/oktaawscli/okta_auth_config.py index 5efc2755..97cf0d45 100644 --- a/oktaawscli/okta_auth_config.py +++ b/oktaawscli/okta_auth_config.py @@ -88,9 +88,7 @@ def region_for(self, okta_profile, default="us-east-1"): def get_check_valid_creds(self, okta_profile): """Gets if should check if AWS creds are valid from config""" - check_valid_creds = self._value.get( - okta_profile, "check-valid-creds", fallback="True" - ) + check_valid_creds = self._value.get(okta_profile, "check-valid-creds", fallback="True") self.logger.info("Check if credentials are valid: %s" % check_valid_creds) return check_valid_creds @@ -102,25 +100,17 @@ def get_store_role(self, okta_profile): def get_auto_write_profile(self, okta_profile): """Gets if should auto write aws creds to ~/.aws/credentials from config""" - auto_write_profile = self._value.get( - okta_profile, "auto-write-profile", fallback=True - ) - self.logger.info( - "Should write profile to ~/.aws/credentials: %s" % auto_write_profile - ) + auto_write_profile = self._value.get(okta_profile, "auto-write-profile", fallback=True) + self.logger.info("Should write profile to ~/.aws/credentials: %s" % auto_write_profile) return auto_write_profile def get_session_duration(self, okta_profile): """Gets STS session duration from config as an int""" # AWS docs say default duration is 1 hour (3600 seconds) - session_duration = int( - self._value.get(okta_profile, "session-duration", fallback="3600") - ) + session_duration = int(self._value.get(okta_profile, "session-duration", fallback="3600")) if session_duration > 43200 or session_duration < 3600: - self.logger.info( - "Invalid session duration specified, defaulting to 1 hour." - ) + self.logger.info("Invalid session duration specified, defaulting to 1 hour.") session_duration = 3600 self.logger.info("Configured session duration: %s seconds" % session_duration) diff --git a/oktaawscli/okta_awscli.py b/oktaawscli/okta_awscli.py index 9a205aad..89da1f5b 100644 --- a/oktaawscli/okta_awscli.py +++ b/oktaawscli/okta_awscli.py @@ -49,9 +49,7 @@ def get_credentials( print("Copying AWS profile creds to default") exit(0) - okta = OktaAuth( - okta_profile, verbose, logger, totp_token, okta_auth_config, debug=debug - ) + okta = OktaAuth(okta_profile, verbose, logger, totp_token, okta_auth_config, debug=debug) _, assertion = okta.get_assertion() role = aws_auth.choose_aws_role(assertion) role_arn, principal_arn, alias = role @@ -73,12 +71,8 @@ def get_credentials( session_token = sts_token["SessionToken"] print("Credentials valid for %s hours" % round(duration / 3600, 1)) if (profile_name is None or export) and not write_default: - logger.info( - "Either profile name not given or export flag set, will output to console." - ) - exports = console_output( - access_key_id, secret_access_key, session_token, verbose - ) + logger.info("Either profile name not given or export flag set, will output to console.") + exports = console_output(access_key_id, secret_access_key, session_token, verbose) if cache: cache = open("%s/.okta-credentials.cache" % (os.path.expanduser("~"),), "w") cache.write(exports) @@ -97,16 +91,12 @@ def get_credentials( logger.debug("Setting region=%s via okta-profile=%s", region, okta_profile) elif account_region is not None and account_region != default_region: region = account_region - logger.debug( - "Setting region=%s via account profile=%s", region, profile_name - ) + logger.debug("Setting region=%s via account profile=%s", region, profile_name) else: region = default_region logger.debug("Setting region=%s via defaults", region) - logger.info( - "Export flag not set, will write credentials to ~/.aws/credentials." - ) + logger.info("Export flag not set, will write credentials to ~/.aws/credentials.") aws_auth.write_sts_token( profile=profile_name, access_key_id=access_key_id, @@ -154,9 +144,7 @@ def console_output(access_key_id, secret_access_key, session_token, verbose): @click.command() @click.option("-v", "--verbose", is_flag=True, help="Enables verbose mode") -@click.option( - "-w", "--write-default", is_flag=True, help="Writes to both default and account" -) +@click.option("-w", "--write-default", is_flag=True, help="Writes to both default and account") @click.option("-V", "--version", is_flag=True, help="Outputs version number and exits") @click.option("-d", "--debug", is_flag=True, help="Enables debug mode") @click.option( @@ -260,10 +248,7 @@ def main( except Timeout as exc: # Use print() so click's CliRunner captures the message in result.output; # the logger writes to stderr which CliRunner doesn't capture by default. - print( - "Could not acquire lock on %s — another okta-awscli process is " - "holding it. Try again." % exc.lock_file - ) + print("Could not acquire lock on %s — another okta-awscli process is holding it. Try again." % exc.lock_file) exit(1) if awscli_args: diff --git a/pylintrc b/pylintrc deleted file mode 100644 index 321d2499..00000000 --- a/pylintrc +++ /dev/null @@ -1,55 +0,0 @@ -[MASTER] - -[MESSAGES CONTROL] -# F0401: Unable to import — handled by mypy/imports -# E0611: No name in module — false positives across boto3 etc. -# E1101: %s has no %r member — false positives on dynamic AWS clients -# W0212: Access to protected member — tests reach into mangled names intentionally -# W0703/W0718: Catching too general Exception -# R0901: Too many ancestors -# W0511: TODO/FIXME warnings -# W0231: __init__ from base class not called -# W0127: Self-assigning variable — `input = input` no-op at okta_auth.py:20 (legacy py2/3 carry-over; the actual `raw_input` shim is in okta_auth_config.py) -# C0209: consider-using-f-string — legacy % formatting is intentional -# R0911/R0912/R0913/R0914/R0915/R0917: too-many-* — CLI entry point + auth flows -# W0603: global statement — harmless if reintroduced -# W0622: Redefining builtins — `input/raw_input` compat shim -# W0613: Unused argument — CLI click signatures -# W1201/W1203: lazy logging format — legacy % style is fine -# R1705: no-else-return — readability preference -# C0301: line-too-long — handled by black -# R1716: chained-comparison — disabled defensively; no current violations after this PR's cleanup -# R1710: inconsistent-return-statements — explicit-None additions risk behavior drift -# R1722: consider-using-sys-exit — legacy `exit()` calls are intentional in this CLI; both reach the same termination path here -# W1514: unspecified-encoding — paths are dotfiles assumed to be UTF-8 -# R1732: consider-using-with — bare `open(path, "a").close()` touch idiom in __init__ paths -# C0415: import-outside-toplevel — tests use lazy imports for isolation -# C0116: missing-function-docstring — many test methods rely on class docstrings -# W0621: redefined-outer-name — tests reimport `mock` inside methods intentionally -# W0404: reimported — same as above -disable=F0401,E0611,E1101,W0212,W0703,W0718,R0901,W0511,W0231,W0127, - C0209,R0911,R0912,R0913,R0914,R0915,R0917,W0603,W0622,W0613,W1201,W1203, - R1705,C0301,R1716,R1710,R1722,W1514,R1732,C0415,C0116,W0621,W0404 - -[BASIC] -good-names=i,j,k,v,e,f,ex,bs,fd,p,cm,_ -function-rgx=[a-z_][a-z0-9_]{2,60}$ -method-rgx=[a-z_][a-z0-9_]{2,80}$ -variable-rgx=[a-z_][a-z0-9_]{2,40}$ - -[FORMAT] -max-line-length=120 -max-module-lines=1000 - -[DESIGN] -max-args=15 -max-locals=20 -max-statements=60 -max-attributes=15 -max-branches=15 - -[SIMILARITIES] -min-similarity-lines=6 -ignore-comments=yes -ignore-docstrings=yes -ignore-imports=yes diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 00000000..f65ebd5f --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,7 @@ +[tool.ruff] +line-length = 120 +target-version = "py311" + +[tool.ruff.lint] +# E/W: pycodestyle, F: pyflakes, I: import sorting (replaces isort). +select = ["E", "W", "F", "I"] diff --git a/tests/test_aws_creds_path.py b/tests/test_aws_creds_path.py index 44c01d41..e26d0e53 100644 --- a/tests/test_aws_creds_path.py +++ b/tests/test_aws_creds_path.py @@ -25,9 +25,7 @@ def setUp(self): def test_creds_file_defaults_to_home_dot_aws_credentials(self): auth = _make_aws_auth() - self.assertEqual( - auth.creds_file, os.path.join(self.tempdir, ".aws", "credentials") - ) + self.assertEqual(auth.creds_file, os.path.join(self.tempdir, ".aws", "credentials")) self.assertEqual(auth.creds_dir, os.path.join(self.tempdir, ".aws")) @@ -64,9 +62,7 @@ def test_write_sts_token_writes_to_overridden_path(self): self.assertTrue(os.path.isfile(self.override_path)) # Default ~/.aws/credentials should NOT have been created. - self.assertFalse( - os.path.exists(os.path.join(self.tempdir, ".aws", "credentials")) - ) + self.assertFalse(os.path.exists(os.path.join(self.tempdir, ".aws", "credentials"))) config = ConfigParser() config.read(self.override_path) diff --git a/tests/test_locking.py b/tests/test_locking.py index 7e690c65..74774cef 100644 --- a/tests/test_locking.py +++ b/tests/test_locking.py @@ -157,9 +157,7 @@ def test_acquires_lock_on_credentials_file(self): from oktaawscli import _locking as locking_module auth = self._make_aws_auth("test_profile") - with mock.patch( - "oktaawscli.aws_auth.locked", wraps=locking_module.locked - ) as mock_locked: + with mock.patch("oktaawscli.aws_auth.locked", wraps=locking_module.locked) as mock_locked: auth.write_sts_token( "test_profile", "AKIA_TEST", @@ -172,10 +170,7 @@ def test_two_parallel_writes_preserve_both_profiles(self): from configparser import ConfigParser ctx = multiprocessing.get_context("fork") - procs = [ - ctx.Process(target=_child_write_sts, args=(self.tempdir, f"profile_{i}")) - for i in range(2) - ] + procs = [ctx.Process(target=_child_write_sts, args=(self.tempdir, f"profile_{i}")) for i in range(2)] for p in procs: p.start() for p in procs: @@ -215,9 +210,7 @@ def test_acquires_lock_on_credentials_file(self): from oktaawscli import _locking as locking_module auth = self._make_aws_auth("source") - with mock.patch( - "oktaawscli.aws_auth.locked", wraps=locking_module.locked - ) as mock_locked: + with mock.patch("oktaawscli.aws_auth.locked", wraps=locking_module.locked) as mock_locked: auth.copy_to_default("source") mock_locked.assert_called_once_with(auth.creds_file) @@ -284,9 +277,7 @@ def test_acquires_lock_on_alias_info_file(self): ] auth = self._make_aws_auth("test") - with mock.patch( - "oktaawscli.aws_auth.locked", wraps=locking_module.locked - ) as mock_locked: + with mock.patch("oktaawscli.aws_auth.locked", wraps=locking_module.locked) as mock_locked: auth._AwsAuth__get_role_info(roles, b"unused-because-cache-is-fresh") mock_locked.assert_called_once_with(self.info_path) @@ -330,9 +321,7 @@ def fake_alias(*args, **kwargs): timed_out.append(True) return "fresh-alias" - with mock.patch.object( - auth, "_AwsAuth__get_account_alias", side_effect=fake_alias - ): + with mock.patch.object(auth, "_AwsAuth__get_account_alias", side_effect=fake_alias): result = auth._AwsAuth__get_role_info(roles, b"unused") self.assertEqual(timed_out, [True]) @@ -399,14 +388,10 @@ def test_preserves_existing_token_when_writer_raises(self): token_path = os.path.join(self.tempdir, ".okta-token") with open(token_path, "w") as f: - f.write( - '{"session_id": "original", "expiration_date": "2099-01-01T00:00:00.000Z"}' - ) + f.write('{"session_id": "original", "expiration_date": "2099-01-01T00:00:00.000Z"}') auth = self._make_okta_auth() - with mock.patch( - "oktaawscli.okta_auth.json.dumps", side_effect=RuntimeError("boom") - ): + with mock.patch("oktaawscli.okta_auth.json.dumps", side_effect=RuntimeError("boom")): with self.assertRaises(RuntimeError): auth.cache_session_id("new_sess", "2099-01-01T00:00:00.000Z") @@ -454,9 +439,7 @@ def test_get_apps_exits_cleanly_on_error_response(self): mock_resp = mock.MagicMock() mock_resp.json.return_value = error_response mock_resp.status_code = 401 - with mock.patch( - "oktaawscli.okta_auth.requests.request", return_value=mock_resp - ): + with mock.patch("oktaawscli.okta_auth.requests.request", return_value=mock_resp): with self.assertRaises(SystemExit) as cm: auth.get_apps("stale_sid") self.assertEqual(cm.exception.code, 1) @@ -475,9 +458,7 @@ def test_get_session_exits_cleanly_on_error_response(self): mock_resp = mock.MagicMock() mock_resp.json.return_value = error_response mock_resp.status_code = 401 - with mock.patch( - "oktaawscli.okta_auth.requests.request", return_value=mock_resp - ): + with mock.patch("oktaawscli.okta_auth.requests.request", return_value=mock_resp): with self.assertRaises(SystemExit) as cm: auth.get_session("bad_session_token") self.assertEqual(cm.exception.code, 1) @@ -490,11 +471,11 @@ def test_fast_path_skips_lock_when_cached_session_valid(self): from unittest import mock auth = self._make_okta_auth() - with mock.patch.object( - auth, "get_cached_session_id", return_value="cached_sid" - ), mock.patch.object(auth, "check_for_desync", return_value=False), mock.patch( - "oktaawscli.okta_auth.locked" - ) as mock_locked: + with ( + mock.patch.object(auth, "get_cached_session_id", return_value="cached_sid"), + mock.patch.object(auth, "check_for_desync", return_value=False), + mock.patch("oktaawscli.okta_auth.locked") as mock_locked, + ): result = auth.primary_auth() self.assertEqual(result, "cached_sid") mock_locked.assert_not_called() @@ -514,22 +495,19 @@ def test_slow_path_acquires_lock_when_cache_is_empty(self): fake_resp.json.return_value = {"status": "SUCCESS", "sessionToken": "stoken"} fake_resp.status_code = 200 - with mock.patch.object( - auth, "get_cached_session_id", return_value=None - ), mock.patch.object( - auth, "get_session", return_value="fresh_sid" - ) as mock_get_session, mock.patch( - "oktaawscli.okta_auth.locked", - wraps=locking_module.locked, - ) as mock_locked, mock.patch( - "oktaawscli.okta_auth.requests.request", return_value=fake_resp + with ( + mock.patch.object(auth, "get_cached_session_id", return_value=None), + mock.patch.object(auth, "get_session", return_value="fresh_sid") as mock_get_session, + mock.patch( + "oktaawscli.okta_auth.locked", + wraps=locking_module.locked, + ) as mock_locked, + mock.patch("oktaawscli.okta_auth.requests.request", return_value=fake_resp), ): result = auth.primary_auth() self.assertEqual(result, "fresh_sid") - mock_locked.assert_called_once_with( - auth.token_path, timeout=INTERACTIVE_LOCK_TIMEOUT_SECONDS - ) + mock_locked.assert_called_once_with(auth.token_path, timeout=INTERACTIVE_LOCK_TIMEOUT_SECONDS) mock_get_session.assert_called_once_with("stoken") def test_slow_path_uses_session_refreshed_by_peer_while_waiting(self): @@ -540,24 +518,23 @@ def test_slow_path_uses_session_refreshed_by_peer_while_waiting(self): from oktaawscli._locking import INTERACTIVE_LOCK_TIMEOUT_SECONDS auth = self._make_okta_auth() - with mock.patch.object( - auth, - "get_cached_session_id", - side_effect=[None, "peer_refreshed_sid"], - ) as mock_get_cached, mock.patch.object( - auth, "check_for_desync" - ) as mock_desync, mock.patch( - "oktaawscli.okta_auth.locked", - wraps=locking_module.locked, - ) as mock_locked, mock.patch( - "oktaawscli.okta_auth.requests.request" - ) as mock_post: + with ( + mock.patch.object( + auth, + "get_cached_session_id", + side_effect=[None, "peer_refreshed_sid"], + ) as mock_get_cached, + mock.patch.object(auth, "check_for_desync") as mock_desync, + mock.patch( + "oktaawscli.okta_auth.locked", + wraps=locking_module.locked, + ) as mock_locked, + mock.patch("oktaawscli.okta_auth.requests.request") as mock_post, + ): result = auth.primary_auth() self.assertEqual(result, "peer_refreshed_sid") - mock_locked.assert_called_once_with( - auth.token_path, timeout=INTERACTIVE_LOCK_TIMEOUT_SECONDS - ) + mock_locked.assert_called_once_with(auth.token_path, timeout=INTERACTIVE_LOCK_TIMEOUT_SECONDS) self.assertEqual(mock_get_cached.call_count, 2) mock_post.assert_not_called() mock_desync.assert_not_called() @@ -606,9 +583,10 @@ def test_get_apps_retries_on_rate_limit_then_succeeds(self): self._success_apps_response(), ] - with mock.patch( - "oktaawscli.okta_auth.requests.request", side_effect=responses - ) as mock_get, mock.patch("oktaawscli.okta_auth.time.sleep") as mock_sleep: + with ( + mock.patch("oktaawscli.okta_auth.requests.request", side_effect=responses) as mock_get, + mock.patch("oktaawscli.okta_auth.time.sleep") as mock_sleep, + ): label, link = auth.get_apps("sid") self.assertEqual(label, "AWS Prod") @@ -622,10 +600,13 @@ def test_get_apps_exits_after_exhausting_retries(self): auth = self._make_okta_auth() auth.app = "AWS Prod" - with mock.patch( - "oktaawscli.okta_auth.requests.request", - return_value=self._rate_limit_response(), - ) as mock_get, mock.patch("oktaawscli.okta_auth.time.sleep"): + with ( + mock.patch( + "oktaawscli.okta_auth.requests.request", + return_value=self._rate_limit_response(), + ) as mock_get, + mock.patch("oktaawscli.okta_auth.time.sleep"), + ): with self.assertRaises(SystemExit) as cm: auth.get_apps("sid") @@ -647,10 +628,10 @@ def test_get_session_retries_on_rate_limit_then_succeeds(self): responses = [self._rate_limit_response(), success] - with mock.patch( - "oktaawscli.okta_auth.requests.request", side_effect=responses - ) as mock_post, mock.patch.object(auth, "cache_session_id"), mock.patch( - "oktaawscli.okta_auth.time.sleep" + with ( + mock.patch("oktaawscli.okta_auth.requests.request", side_effect=responses) as mock_post, + mock.patch.object(auth, "cache_session_id"), + mock.patch("oktaawscli.okta_auth.time.sleep"), ): sid = auth.get_session("stoken") @@ -670,10 +651,13 @@ def test_non_rate_limit_error_exits_without_retry(self): } non_rate_limit_resp.status_code = 401 - with mock.patch( - "oktaawscli.okta_auth.requests.request", - return_value=non_rate_limit_resp, - ) as mock_get, mock.patch("oktaawscli.okta_auth.time.sleep") as mock_sleep: + with ( + mock.patch( + "oktaawscli.okta_auth.requests.request", + return_value=non_rate_limit_resp, + ) as mock_get, + mock.patch("oktaawscli.okta_auth.time.sleep") as mock_sleep, + ): with self.assertRaises(SystemExit): auth.get_apps("sid")