Use _cache property as comments describe

setup.cfg

@@ -3,7 +3,7 @@ universal = 1
 
 [metadata]
 name = artifacts-keyring
-version = 0.3.4
+version = 1.0.0
 author = Microsoft Corporation
 url = https://github.com/Microsoft/artifacts-keyring
 license_file = LICENSE.txt
@@ -18,6 +18,10 @@ classifiers =
     Programming Language :: Python :: 3.5
     Programming Language :: Python :: 3.6
     Programming Language :: Python :: 3.7
+    Programming Language :: Python :: 3.8
+    Programming Language :: Python :: 3.9
+    Programming Language :: Python :: 3.10
+    Programming Language :: Python :: 3.11
 
 [options]
 package_dir=
@@ -38,6 +42,6 @@ keyring.backends =
     ArtifactsKeyringBackend = artifacts_keyring:ArtifactsKeyringBackend
 
 [tool:pytest]
-testpaths = src/tests
+testpaths = tests
 python_functions = test_*
 python_files = test_*.py

src/artifacts_keyring/__init__.py

@@ -6,11 +6,8 @@
 from __future__ import absolute_import
 
 __author__ = "Microsoft Corporation <[email protected]>"
-__version__ = "0.3.4"
+__version__ = "1.0.0"
 
-import json
-import subprocess
-import sys
 import warnings
 
 from .support import urlsplit
@@ -25,13 +22,12 @@ class ArtifactsKeyringBackend(keyring.backend.KeyringBackend):
         "pkgs.dev.azure.com",
         "pkgs.visualstudio.com",
         "pkgs.codedev.ms",
-        "pkgs.vsts.me"
+        "pkgs.vsts.me",
     )
     _PROVIDER = CredentialProvider
 
     priority = 9.9
 
-
     def __init__(self):
         # In-memory cache of user-pass combination, to allow
         # fast handling of applications that insist on querying
@@ -40,7 +36,6 @@ def __init__(self):
         # around for longer than necessary.
         self._cache = {}
 
-
     def get_credential(self, service, username):
         try:
             parsed = urlsplit(service)
@@ -53,32 +48,49 @@ def get_credential(self, service, username):
         if netloc is None or not netloc.endswith(self.SUPPORTED_NETLOC):
             return None
 
+        cached = self._cache.setdefault(service, {})
+
         provider = self._PROVIDER()
 
-        username, password = provider.get_credentials(service)
+        cred = provider.get_credentials(
+            service,
+            cached
+            if username is None
+            else {key: value for key, value in cached.items() if key == username},
+        )
 
-        if username and password:
-            self._cache[service, username] = password
-            return keyring.credentials.SimpleCredential(username, password)
+        if all(cred):
+            _, password = cred
+
+            if cred[0] == username:
+                self._cache[service][username] = password
+            else:
+                username, _ = cred
 
+            return keyring.credentials.SimpleCredential(username, password)
 
     def get_password(self, service, username):
-        password = self._cache.get((service, username), None)
+        password = self._cache.get(service, {}).pop(username, None)
         if password is not None:
             return password
 
-        creds = self.get_credential(service, None)
-        if creds and username == creds.username:
-            return creds.password
+        # Fills the cache
+        self.get_credential(service, username)
 
-        return None
+        # Empties the cache if username matches
+        password = self._cache.get(service, {}).pop(username, None)
+        if password is not None:
+            return password
 
+        return None
 
     def set_password(self, service, username, password):
         # Defer setting a password to the next backend
         raise NotImplementedError()
 
-
     def delete_password(self, service, username):
+        # Try and remove it from the cache out of an abundance of caution
+        self._cache.get(service, {}).pop(username, None)
+
         # Defer deleting a password to the next backend
         raise NotImplementedError()

src/artifacts_keyring/plugin.py

@@ -10,10 +10,7 @@
 import requests
 import subprocess
 import sys
-import warnings
-import shutil
 
-from . import __version__
 from .support import Popen
 
 
@@ -33,9 +30,18 @@ def __init__(self):
             self.exe = [tool_path]
         else:
             try:
-                sys_version = tuple(int(i) for i in
-                    subprocess.check_output(["dotnet", "--version"]).decode().strip().partition("-")[0].split("."))
-                get_runtime_path = lambda: "dotnet"
+                _sys_version = tuple(
+                    int(i)
+                    for i in subprocess.check_output(["dotnet", "--version"])
+                    .decode()
+                    .strip()
+                    .partition("-")[0]
+                    .split(".")
+                )
+
+                def get_runtime_path():
+                    return "dotnet"
+
             except Exception as e:
                 message = (
                     "Unable to find dependency dotnet, please manually install"
@@ -54,17 +60,24 @@ def __init__(self):
             self.exe = [get_runtime_path(), "exec", tool_path]
 
         if not os.path.exists(tool_path):
-            raise RuntimeError("Unable to find credential provider in the expected path: " + tool_path)
-
+            raise RuntimeError(
+                "Unable to find credential provider in the expected path: " + tool_path
+            )
 
-    def get_credentials(self, url):
+    def get_credentials(self, url, credentials):
         # Public feed short circuit: return nothing if not getting credentials for the upload endpoint
         # (which always requires auth) and the endpoint is public (can authenticate without credentials).
         if not self._is_upload_endpoint(url) and self._can_authenticate(url, None):
             return None, None
 
+        for cred in credentials.items():
+            if self._can_authenticate(url, cred):
+                return cred[0], cred[1]
+
         # Getting credentials with IsRetry=false; the credentials may come from the cache
-        username, password = self._get_credentials_from_credential_provider(url, is_retry=False)
+        username, password = self._get_credentials_from_credential_provider(
+            url, is_retry=False
+        )
 
         # Do not attempt to validate if the credentials could not be obtained
         if username is None or password is None:
@@ -77,24 +90,27 @@ def get_credentials(self, url):
         # The cached credentials are expired; get fresh ones with IsRetry=true
         return self._get_credentials_from_credential_provider(url, is_retry=True)
 
-
-    def _is_upload_endpoint(self, url):
-        url = url[: -1] if url[-1] == "/" else url
+    @staticmethod
+    def _is_upload_endpoint(url):
+        url = url[:-1] if url[-1] == "/" else url
         return url.endswith("pypi/upload")
 
-
-    def _can_authenticate(self, url, auth):
+    @staticmethod
+    def _can_authenticate(url, auth):
         response = requests.get(url, auth=auth)
 
-        return response.status_code < 500 and \
-            response.status_code != 401 and \
-            response.status_code != 403
-
+        return (
+            response.status_code < 500
+            and response.status_code != 401
+            and response.status_code != 403
+        )
 
     def _get_credentials_from_credential_provider(self, url, is_retry):
-        non_interactive = self._NON_INTERACTIVE_VAR_NAME in os.environ and \
-            os.environ[self._NON_INTERACTIVE_VAR_NAME] and \
-            str(os.environ[self._NON_INTERACTIVE_VAR_NAME]).lower() == "true"
+        non_interactive = (
+            self._NON_INTERACTIVE_VAR_NAME in os.environ
+            and os.environ[self._NON_INTERACTIVE_VAR_NAME]
+            and str(os.environ[self._NON_INTERACTIVE_VAR_NAME]).lower() == "true"
+        )
 
         proc = Popen(
             self.exe + [
@@ -106,13 +122,13 @@ def _get_credentials_from_credential_provider(self, url, is_retry):
             ],
             stdin=subprocess.PIPE,
             stdout=subprocess.PIPE,
-            stderr=subprocess.PIPE
+            stderr=subprocess.PIPE,
         )
 
         # Read all standard error first, which may either display
         # errors from the credential provider or instructions
         # from it for Device Flow authentication.
-        for stderr_line in iter(proc.stderr.readline, b''):
+        for stderr_line in iter(proc.stderr.readline, b""):
             line = stderr_line.decode("utf-8", "ignore")
             sys.stdout.write(line)
             sys.stdout.flush()
@@ -121,17 +137,24 @@ def _get_credentials_from_credential_provider(self, url, is_retry):
 
         if proc.returncode != 0:
             stderr = proc.stderr.read().decode("utf-8", "ignore")
-            raise RuntimeError("Failed to get credentials: process with PID {pid} exited with code {code}; additional error message: {error}"
-                .format(pid=proc.pid, code=proc.returncode, error=stderr))
+            raise RuntimeError(
+                "Failed to get credentials: process with PID {pid} exited with code {code}; additional error message: {error}".format(
+                    pid=proc.pid, code=proc.returncode, error=stderr
+                )
+            )
 
         try:
             # stdout is expected to be UTF-8 encoded JSON, so decoding errors are not ignored here.
             payload = proc.stdout.read().decode("utf-8")
         except ValueError:
-            raise RuntimeError("Failed to get credentials: the Credential Provider's output could not be decoded using UTF-8.")
+            raise RuntimeError(
+                "Failed to get credentials: the Credential Provider's output could not be decoded using UTF-8."
+            )
 
         try:
             parsed = json.loads(payload)
             return parsed["Username"], parsed["Password"]
         except ValueError:
-            raise RuntimeError("Failed to get credentials: the Credential Provider's output could not be parsed as JSON.")
+            raise RuntimeError(
+                "Failed to get credentials: the Credential Provider's output could not be parsed as JSON."
+            )

src/tests/test_backend.py → tests/test_backend.py

@@ -19,7 +19,8 @@
 
 
 class FakeProvider(object):
-    def get_credentials(self, service):
+    @staticmethod
+    def get_credentials(service, username):
         return "user" + service[-4:], "pass" + service[-4:]
 
 
@@ -87,14 +88,18 @@ def mock_requests_get(url, auth):
         response.status_code = int(url[:3])
         return response
 
-    monkeypatch.setattr(CredentialProvider, "_get_credentials_from_credential_provider", mock_get_credentials)
+    monkeypatch.setattr(
+        CredentialProvider,
+        "_get_credentials_from_credential_provider",
+        mock_get_credentials,
+    )
     monkeypatch.setattr(requests, "get", mock_requests_get)
 
     yield CredentialProvider()
 
 
 def test_get_credential_unsupported_host(only_backend):
-    assert keyring.get_credential("https://example.com", None) == None
+    assert keyring.get_credential("https://example.com", None) is None
 
 
 def test_get_credential(only_backend, fake_provider):
@@ -151,20 +156,30 @@ def test_cannot_delete_password(passwords, fake_provider):
 
 
 def test_retry_on_invalid_credentials(validating_provider):
-     # No credentials returned when it can already authenticate without them
-    username, password = validating_provider.get_credentials("200" + SUPPORTED_HOST)
-    assert username == None and password == None
+    # No credentials returned when it can already authenticate without them
+    username, password = validating_provider.get_credentials(
+        "200" + SUPPORTED_HOST, {}
+    )
+    assert username is None and password is None
 
     # Credentials returned from first call with IsRetry=false
-    username, password = validating_provider.get_credentials("200" + SUPPORTED_HOST + "pypi/upload")
-    assert password == False
+    username, password = validating_provider.get_credentials(
+        "200" + SUPPORTED_HOST + "pypi/upload", {}
+    )
+    assert password is False
 
     # Credentials returned from second call with IsRetry=true
-    username, password = validating_provider.get_credentials("401" + SUPPORTED_HOST)
-    assert password == True
-
-    username, password = validating_provider.get_credentials("403" + SUPPORTED_HOST)
-    assert password == True
-
-    username, password = validating_provider.get_credentials("500" + SUPPORTED_HOST)
-    assert password == True
+    username, password = validating_provider.get_credentials(
+        "401" + SUPPORTED_HOST, {}
+    )
+    assert password is True
+
+    username, password = validating_provider.get_credentials(
+        "403" + SUPPORTED_HOST, {}
+    )
+    assert password is True
+
+    username, password = validating_provider.get_credentials(
+        "500" + SUPPORTED_HOST, {}
+    )
+    assert password is True