Get a token for the current provider.
(self, username=None)
| 81 | self.request_called = 0 |
| 82 | |
| 83 | def get_token(self, username=None): |
| 84 | """Get a token for the current provider.""" |
| 85 | if ENVIRON == "test": |
| 86 | if username is None: |
| 87 | token_file = TOKEN_FILE |
| 88 | else: |
| 89 | token_file = os.path.join(TOKEN_DIR, username) |
| 90 | with open(token_file) as fid: # noqa: ASYNC101,RUF100 |
| 91 | return fid.read() |
| 92 | elif ENVIRON == "azure": |
| 93 | opts = parse_uri(self.uri_single)["options"] |
| 94 | token_aud = opts["authMechanismProperties"]["TOKEN_RESOURCE"] |
| 95 | return _get_azure_response(token_aud, username)["access_token"] |
| 96 | elif ENVIRON == "gcp": |
| 97 | opts = parse_uri(self.uri_single)["options"] |
| 98 | token_aud = opts["authMechanismProperties"]["TOKEN_RESOURCE"] |
| 99 | return _get_gcp_response(token_aud, username)["access_token"] |
| 100 | elif ENVIRON == "k8s": |
| 101 | return _get_k8s_token() |
| 102 | else: |
| 103 | raise ValueError(f"Unknown ENVIRON: {ENVIRON}") |
| 104 | |
| 105 | @asynccontextmanager |
| 106 | async def fail_point(self, command_args): |
no test coverage detected