Add authentication support for Azure Artifacts in version retrieval

This commit enhances the version retrieval process by adding support for authenticated queries to Azure Artifacts. The main function now retrieves credentials from command-line arguments or environment variables, ensuring seamless integration in non-interactive environments. The query_registry_version and _query_azure_artifacts_version functions are updated to accept username and password parameters, allowing for secure access to package versions. Additionally, tests are introduced to verify the functionality of authenticated queries, improving overall reliability and user experience.

Author: alelom

src/python_package_folder/python_package_folder.py

@@ -186,6 +186,21 @@ def build_cmd() -> None:
                 repository = args.publish if args.publish else None
                 repository_url = args.repository_url if args.publish else None
 
+                # Get credentials for authenticated registry queries (especially Azure Artifacts)
+                # Try to get them from args or environment variables
+                query_username = args.username
+                query_password = args.password
+                
+                # If not provided via args, check environment variables (for Azure Artifacts)
+                if repository == "azure" and not query_username:
+                    query_username = os.getenv("TWINE_USERNAME") or os.getenv("PYPI_USERNAME")
+                if repository == "azure" and not query_password:
+                    query_password = (
+                        os.getenv("TWINE_PASSWORD")
+                        or os.getenv("PYPI_PASSWORD")
+                        or os.getenv("AZURE_ARTIFACTS_TOKEN")
+                    )
+                
                 if is_subfolder:
                     # Workflow 1: subfolder build
                     # src_dir is guaranteed to be relative to project_root due to is_subfolder check
@@ -199,6 +214,8 @@ def build_cmd() -> None:
                         subfolder_path=subfolder_rel_path,
                         repository=repository,
                         repository_url=repository_url,
+                        username=query_username,
+                        password=query_password,
                     )
                 else:
                     # Workflow 2: main package
@@ -221,6 +238,8 @@ def build_cmd() -> None:
                         subfolder_path=None,
                         repository=repository,
                         repository_url=repository_url,
+                        username=query_username,
+                        password=query_password,
                     )
 
                 if resolved_version:

src/python_package_folder/version_calculator.py

@@ -25,6 +25,8 @@ def query_registry_version(
     package_name: str,
     repository: str,
     repository_url: str | None = None,
+    username: str | None = None,
+    password: str | None = None,
 ) -> str | None:
     """
     Query package registry for the latest published version.
@@ -33,6 +35,8 @@ def query_registry_version(
         package_name: Package name to query
         repository: Repository type ('pypi', 'testpypi', or 'azure')
         repository_url: Repository URL (required for Azure Artifacts)
+        username: Optional username for authenticated queries (Azure Artifacts)
+        password: Optional password/token for authenticated queries (Azure Artifacts)
 
     Returns:
         Latest version string or None if not found/unsupported
@@ -54,7 +58,7 @@ def query_registry_version(
                 logger.warning("Azure Artifacts repository URL not provided")
                 return None
             logger.info(f"Querying Azure Artifacts for package '{package_name}' at {repository_url}")
-            version = _query_azure_artifacts_version(package_name, repository_url)
+            version = _query_azure_artifacts_version(package_name, repository_url, username, password)
             if version:
                 logger.info(f"Found version {version} on Azure Artifacts")
             else:
@@ -184,6 +188,8 @@ def _extract_version_from_filename(self, filename: str) -> str | None:
 def _query_azure_artifacts_version(
     package_name: str,
     repository_url: str,
+    username: str | None = None,
+    password: str | None = None,
 ) -> str | None:
     """
     Query Azure Artifacts for the latest version.
@@ -194,6 +200,8 @@ def _query_azure_artifacts_version(
     Args:
         package_name: Package name to query
         repository_url: Azure Artifacts repository URL
+        username: Optional username for authentication
+        password: Optional password/token for authentication
 
     Returns:
         Latest version string or None if not found/unsupported
@@ -212,8 +220,15 @@ def _query_azure_artifacts_version(
         return None
 
     try:
-        logger.info(f"Fetching Azure Artifacts simple index for '{package_name}'...")
-        response = requests.get(simple_index_url, timeout=10)
+        # Prepare authentication if credentials are provided
+        auth = None
+        if username and password:
+            auth = (username, password)
+            logger.info(f"Fetching Azure Artifacts simple index for '{package_name}' with authentication...")
+        else:
+            logger.info(f"Fetching Azure Artifacts simple index for '{package_name}' (no authentication)...")
+        
+        response = requests.get(simple_index_url, auth=auth, timeout=10)
         logger.info(f"Azure Artifacts response: status={response.status_code}, content_length={len(response.text)} bytes")
 
         if response.status_code == 401:
@@ -628,6 +643,8 @@ def resolve_version(
     subfolder_path: Path | None = None,
     repository: str | None = None,
     repository_url: str | None = None,
+    username: str | None = None,
+    password: str | None = None,
 ) -> tuple[str | None, str | None]:
     """
     Resolve the next version using conventional commits.
@@ -640,6 +657,8 @@ def resolve_version(
         subfolder_path: Optional path to subfolder (relative to project_root)
         repository: Optional target repository ('pypi', 'testpypi', or 'azure')
         repository_url: Optional repository URL (required for Azure Artifacts)
+        username: Optional username for authenticated registry queries (Azure Artifacts)
+        password: Optional password/token for authenticated registry queries (Azure Artifacts)
 
     Returns:
         Tuple of (version string if a release is determined, error message if any)
@@ -651,7 +670,7 @@ def resolve_version(
     baseline_version = None
     if repository and package_name:
         logger.info(f"Attempting to query {repository} for baseline version of '{package_name}'")
-        baseline_version = query_registry_version(package_name, repository, repository_url)
+        baseline_version = query_registry_version(package_name, repository, repository_url, username, password)
 
     # Step 2: Fallback to git tags if registry query failed
     if not baseline_version:

tests/test_version_calculator.py

@@ -151,6 +151,39 @@ def test_query_azure_artifacts_version_empty_html(self, mock_get: MagicMock) ->
         # Should return None when no versions found in HTML
         assert version is None
 
+    @patch("python_package_folder.version_calculator.requests.get")
+    def test_query_azure_artifacts_version_with_auth(self, mock_get: MagicMock) -> None:
+        """Test querying Azure Artifacts with authentication."""
+        mock_response = Mock()
+        mock_response.status_code = 200
+        mock_response.text = """<!DOCTYPE html>
+<html>
+  <head>
+    <title>Links for test-package</title>
+  </head>
+  <body>
+    <h1>Links for test-package</h1>
+    <a href="test-package-1.0.0-py3-none-any.whl">test-package-1.0.0-py3-none-any.whl</a>
+    <a href="test-package-1.1.0-py3-none-any.whl">test-package-1.1.0-py3-none-any.whl</a>
+  </body>
+</html>"""
+        mock_get.return_value = mock_response
+
+        version = query_registry_version(
+            "test-package",
+            "azure",
+            repository_url="https://pkgs.dev.azure.com/ORG/PROJECT/_packaging/FEED/pypi/upload",
+            username="testuser",
+            password="testtoken",
+        )
+        # Should parse HTML and return the latest version
+        assert version == "1.1.0"
+        
+        # Verify authentication was used
+        mock_get.assert_called_once()
+        call_args = mock_get.call_args
+        assert call_args[1]["auth"] == ("testuser", "testtoken")
+
     @patch("python_package_folder.version_calculator.requests.get")
     def test_query_registry_version_error_handling(self, mock_get: MagicMock) -> None:
         """Test that registry query errors are handled gracefully."""