reuse discover_targetd_packagfes

Co-authored-by: Copilot <copilot@github.com>

Author: l0lawrence

eng/tools/azure-sdk-tools/azpysdk/changelog.py

@@ -6,7 +6,9 @@
 from typing import Optional, List
 
 from .Check import Check, REPO_ROOT
+from ci_tools.functions import discover_targeted_packages
 from ci_tools.logging import logger
+from ci_tools.parsing import ParsedSetup
 
 # The expected Chronus package name and on-disk install location.
 # Chronus is pinned as a dev dependency in .github/chronus/package.json with
@@ -211,13 +213,17 @@ def _ensure_chronus_installed(self) -> None:
             )
             raise SystemExit(1)
 
-    def _detect_package_from_cwd(self) -> Optional[str]:
-        """If CWD is inside a package directory (``sdk/<service>/<package>``),
-        return the Chronus package **name** (the directory basename, e.g.
-        ``azure-core``).  Otherwise return ``None``.
+    def _find_package_root_from_cwd(self) -> Optional[str]:
+        """Find the package root directory when CWD is at or below ``sdk/<service>/<package>``.
 
-        The chronus config uses the pattern ``sdk/*/*`` for packages, so we
-        look for CWD being at or below ``<REPO_ROOT>/sdk/<service>/<pkg>``.
+        Walks up from the current directory to locate the package root,
+        unlike ``get_targeted_directories(target=".")`` which only works
+        when CWD is exactly the package root.  This lets developers run
+        changelog commands from subdirectories such as
+        ``sdk/core/azure-core/tests/``.
+
+        Returns the absolute path to the package root, or ``None`` if CWD
+        is not inside an ``sdk/<service>/<package>`` tree.
         """
         try:
             cwd = os.path.abspath(os.getcwd())
@@ -227,40 +233,38 @@ def _detect_package_from_cwd(self) -> Optional[str]:
             # On Windows, relpath raises ValueError when paths are on different drives
             return None
 
-        # rel should start with "sdk/<service>/<package>" (at least 3 components)
         parts = rel.replace("\\", "/").split("/")
         if len(parts) >= 3 and parts[0] == "sdk":
-            # Return the package name (third component, e.g. "azure-core")
-            return parts[2]
+            return os.path.join(repo, parts[0], parts[1], parts[2])
         return None
 
-    def _resolve_package_name(self, package: str) -> str:
-        """Resolve a user-supplied package argument to a Chronus package name.
-
-        Accepts either:
-        - A package name directly (e.g. ``azure-core``) — returned as-is.
-        - A relative or absolute path (e.g. ``sdk/core/azure-core``,
-          ``.\\sdk\\core\\azure-core\\``) — resolved to the package name.
+    def _resolve_package(self, package_arg: Optional[str]) -> Optional[str]:
+        """Resolve a package argument or CWD to a Chronus package name.
 
-        Chronus identifies packages by name (the directory basename under
-        ``sdk/<service>/``), not by path.
+        Uses ``discover_targeted_packages`` — the same discovery function
+        that powers ``get_targeted_directories`` — to locate packages by
+        path or bare name.  When *package_arg* is ``None``, the method
+        detects the package from CWD by walking up to the nearest
+        ``sdk/<service>/<package>`` directory.
         """
-        # If the path is absolute or starts with '.' resolve it relative to repo root
-        if os.path.isabs(package) or package.startswith("."):
-            try:
-                abs_path = os.path.abspath(os.path.join(os.getcwd(), package))
-                repo = os.path.abspath(REPO_ROOT)
-                package = os.path.relpath(abs_path, repo)
-            except ValueError:
-                pass
-        # Normalize separators and strip trailing slashes
-        package = package.replace("\\", "/").strip("/")
-        # If it looks like a path (sdk/<service>/<package>...), extract the name
-        parts = package.split("/")
-        if len(parts) >= 3 and parts[0] == "sdk":
-            return parts[2]
-        # Otherwise assume it's already a package name
-        return package
+        if package_arg:
+            found = discover_targeted_packages(package_arg, REPO_ROOT)
+            if found:
+                try:
+                    return ParsedSetup.from_path(found[0]).name
+                except Exception:
+                    return os.path.basename(found[0])
+            # Not found by discovery — pass through as-is
+            return package_arg
+
+        # No explicit package — detect from CWD
+        pkg_root = self._find_package_root_from_cwd()
+        if pkg_root is None:
+            return None
+        try:
+            return ParsedSetup.from_path(pkg_root).name
+        except Exception:
+            return os.path.basename(os.path.normpath(pkg_root))
 
     def _run_chronus(self, chronus_args: List[str]) -> int:
         """Run a chronus command from the repository root.
@@ -293,13 +297,10 @@ def _run_add(self, args: argparse.Namespace) -> int:
         ``azpysdk changelog add --kind breaking -m "Removed foo API"``).
         """
         chronus_args = ["add"]
-        package = args.package
-        if package:
-            package = self._resolve_package_name(package)
-        else:
-            package = self._detect_package_from_cwd()
-            if package:
-                logger.info(f"Detected package from current directory: {package}")
+        detected_from_cwd = not args.package
+        package = self._resolve_package(args.package)
+        if package and detected_from_cwd:
+            logger.info(f"Detected package from current directory: {package}")
         if package:
             chronus_args.append(package)
 
@@ -323,13 +324,10 @@ def _run_create(self, args: argparse.Namespace) -> int:
         via ``--package`` so only that package's changelog is generated.
         """
         chronus_args = ["changelog"]
-        package = args.package
-        if package:
-            package = self._resolve_package_name(package)
-        else:
-            package = self._detect_package_from_cwd()
-            if package:
-                logger.info(f"Detected package from current directory: {package}")
+        detected_from_cwd = not args.package
+        package = self._resolve_package(args.package)
+        if package and detected_from_cwd:
+            logger.info(f"Detected package from current directory: {package}")
         if not package:
             logger.error(
                 "No package specified and could not detect one from the current directory.\n"
@@ -356,13 +354,10 @@ def _run_status(self, args: argparse.Namespace) -> int:
         via ``--only`` so only that package's status is shown.
         """
         chronus_args = ["status"]
-        package = args.package
-        if package:
-            package = self._resolve_package_name(package)
-        else:
-            package = self._detect_package_from_cwd()
-            if package:
-                logger.info(f"Detected package from current directory: {package}")
+        detected_from_cwd = not args.package
+        package = self._resolve_package(args.package)
+        if package and detected_from_cwd:
+            logger.info(f"Detected package from current directory: {package}")
         if package:
             chronus_args.extend(["--only", package])
         return self._run_chronus(chronus_args)

eng/tools/azure-sdk-tools/tests/test_changelog_commands.py

@@ -3,6 +3,7 @@
 import argparse
 import os
 import sys
+from types import SimpleNamespace
 from unittest.mock import patch, MagicMock, call
 
 import pytest
@@ -303,7 +304,8 @@ def test_create_calls_chronus_changelog(self, mock_call, _mock_installed):
     def test_status_calls_chronus_status(self, mock_call, _mock_installed):
         parser = _build_parser()
         args = parser.parse_args(["changelog", "status"])
-        result = args.func(args)
+        with patch("os.getcwd", return_value=REPO_ROOT):
+            result = args.func(args)
         assert result == 0
         cmd = mock_call.call_args[0][0]
         assert cmd == [_CHRONUS_BIN, "status"]
@@ -328,42 +330,89 @@ def test_nonzero_exit_code_propagated(self, mock_call, _mock_installed):
 
 
 # ---------------------------------------------------------------------------
-# CWD detection unit tests
+# Package resolution unit tests
 # ---------------------------------------------------------------------------
 
 
-class TestDetectPackageFromCwd:
-    """Test the _detect_package_from_cwd helper directly."""
+class TestFindPackageRootFromCwd:
+    """Test the _find_package_root_from_cwd helper directly."""
 
     def test_at_package_root(self):
         c = changelog()
-        with patch("os.getcwd", return_value=os.path.join(REPO_ROOT, "sdk", "core", "azure-core")):
-            assert c._detect_package_from_cwd() == "azure-core"
+        pkg_root = os.path.join(REPO_ROOT, "sdk", "core", "azure-core")
+        with patch("os.getcwd", return_value=pkg_root):
+            assert c._find_package_root_from_cwd() == pkg_root
 
     def test_inside_package_subdir(self):
         c = changelog()
-        with patch("os.getcwd", return_value=os.path.join(REPO_ROOT, "sdk", "core", "azure-core", "azure", "core")):
-            assert c._detect_package_from_cwd() == "azure-core"
+        pkg_root = os.path.join(REPO_ROOT, "sdk", "core", "azure-core")
+        with patch("os.getcwd", return_value=os.path.join(pkg_root, "azure", "core")):
+            assert c._find_package_root_from_cwd() == pkg_root
 
     def test_at_repo_root(self):
         c = changelog()
         with patch("os.getcwd", return_value=REPO_ROOT):
-            assert c._detect_package_from_cwd() is None
+            assert c._find_package_root_from_cwd() is None
 
     def test_at_sdk_dir(self):
         c = changelog()
         with patch("os.getcwd", return_value=os.path.join(REPO_ROOT, "sdk")):
-            assert c._detect_package_from_cwd() is None
+            assert c._find_package_root_from_cwd() is None
 
     def test_at_service_dir(self):
         c = changelog()
         with patch("os.getcwd", return_value=os.path.join(REPO_ROOT, "sdk", "storage")):
-            assert c._detect_package_from_cwd() is None
+            assert c._find_package_root_from_cwd() is None
 
     def test_outside_repo(self):
         c = changelog()
         with patch("os.getcwd", return_value="/tmp"):
-            assert c._detect_package_from_cwd() is None
+            assert c._find_package_root_from_cwd() is None
+
+
+class TestResolvePackage:
+    """Test the _resolve_package helper that uses discover_targeted_packages."""
+
+    @patch("azpysdk.changelog.ParsedSetup")
+    @patch("azpysdk.changelog.discover_targeted_packages")
+    def test_explicit_path_uses_discover(self, mock_discover, mock_parsed_setup):
+        mock_discover.return_value = [os.path.join(REPO_ROOT, "sdk", "core", "azure-core")]
+        mock_parsed_setup.from_path.return_value = SimpleNamespace(name="azure-core")
+        c = changelog()
+        result = c._resolve_package("sdk/core/azure-core")
+        assert result == "azure-core"
+        mock_discover.assert_called_once_with("sdk/core/azure-core", REPO_ROOT)
+
+    @patch("azpysdk.changelog.ParsedSetup")
+    @patch("azpysdk.changelog.discover_targeted_packages")
+    def test_bare_name_uses_discover(self, mock_discover, mock_parsed_setup):
+        mock_discover.return_value = [os.path.join(REPO_ROOT, "sdk", "core", "azure-core")]
+        mock_parsed_setup.from_path.return_value = SimpleNamespace(name="azure-core")
+        c = changelog()
+        result = c._resolve_package("azure-core")
+        assert result == "azure-core"
+        mock_discover.assert_called_once_with("azure-core", REPO_ROOT)
+
+    @patch("azpysdk.changelog.discover_targeted_packages")
+    def test_bare_name_passthrough_when_not_discovered(self, mock_discover):
+        mock_discover.return_value = []
+        c = changelog()
+        result = c._resolve_package("some-nonexistent-package-name")
+        assert result == "some-nonexistent-package-name"
+
+    @patch("azpysdk.changelog.ParsedSetup")
+    def test_cwd_detection_uses_parsed_setup(self, mock_parsed_setup):
+        mock_parsed_setup.from_path.return_value = SimpleNamespace(name="azure-storage-blob")
+        c = changelog()
+        pkg_dir = os.path.join(REPO_ROOT, "sdk", "storage", "azure-storage-blob")
+        with patch("os.getcwd", return_value=pkg_dir):
+            result = c._resolve_package(None)
+        assert result == "azure-storage-blob"
+
+    def test_cwd_at_repo_root_returns_none(self):
+        c = changelog()
+        with patch("os.getcwd", return_value=REPO_ROOT):
+            assert c._resolve_package(None) is None
 
 
 # ---------------------------------------------------------------------------