Analyze passes target python version

eng/pipelines/templates/jobs/ci.yml

@@ -64,9 +64,6 @@ parameters:
   - name: EnvVars
     type: object
     default: {}
-  - name: PythonVersionForAnalyze
-    type: string
-    default: ''
 
 jobs:
   - job: 'Build_Linux'
@@ -212,10 +209,10 @@ jobs:
 
     steps:
     - task: UsePythonVersion@0
-      displayName: "Use Python ${{ coalesce(parameters.PythonVersionForAnalyze, '$(PythonVersion)') }}"
+      displayName: "Use Python $(PythonVersion)"
       condition: succeededOrFailed()
       inputs:
-        versionSpec: ${{ coalesce(parameters.PythonVersionForAnalyze, '$(PythonVersion)') }}
+        versionSpec: $(PythonVersion)
     - template: /eng/pipelines/templates/steps/use-venv.yml
 
     - template: /eng/common/pipelines/templates/steps/check-spelling.yml

eng/pipelines/templates/stages/archetype-sdk-client.yml

@@ -84,9 +84,6 @@ parameters:
   - name: EnvVars
     type: object
     default: {}
-  - name: PythonVersionForAnalyze
-    type: string
-    default: ''
 
 extends:
   template: /eng/pipelines/templates/stages/1es-redirect.yml
@@ -123,7 +120,6 @@ extends:
               VerifyAutorest: ${{ parameters.VerifyAutorest }}
               TestProxy: ${{ parameters.TestProxy }}
               GenerateApiReviewForManualOnly: ${{ parameters.GenerateApiReviewForManualOnly }}
-              PythonVersionForAnalyze: ${{ parameters.PythonVersionForAnalyze }}
 
         variables:
           - template: /eng/pipelines/templates/variables/globals.yml

eng/scripts/dispatch_checks.py

@@ -15,6 +15,7 @@
 from ci_tools.scenario.generation import build_whl_for_req, replace_dev_reqs
 from ci_tools.logging import configure_logging, logger
 from ci_tools.environment_exclusions import is_check_enabled, CHECK_DEFAULTS
+from ci_tools.parsing import get_config_setting
 from devtools_testutils.proxy_startup import prepare_local_tool
 from packaging.requirements import Requirement
 
@@ -49,6 +50,7 @@ class ProxyProcess:
     "sdist",
     "devtest",
     "optional",
+    "import_all",
     "latestdependency",
     "mindependency",
 }
@@ -84,9 +86,7 @@ def _compare_req_to_injected_reqs(parsed_req, injected_packages: List[str]) -> b
     return any(parsed_req.name in req for req in injected_packages)
 
 
-def _inject_custom_reqs(
-    req_file: str, injected_packages: str, package_dir: str
-) -> None:
+def _inject_custom_reqs(req_file: str, injected_packages: str, package_dir: str) -> None:
     req_lines = []
     injected_list = [p for p in re.split(r"[\s,]", injected_packages) if p]
 
@@ -115,8 +115,7 @@ def _inject_custom_reqs(
         all_adjustments = installable + [
             line_tuple[0].strip()
             for line_tuple in req_lines
-            if line_tuple[0].strip()
-            and not _compare_req_to_injected_reqs(line_tuple[1], all_filter_names)
+            if line_tuple[0].strip() and not _compare_req_to_injected_reqs(line_tuple[1], all_filter_names)
         ]
     else:
         all_adjustments = installable
@@ -138,6 +137,7 @@ async def run_check(
     mark_arg: Optional[str],
     dest_dir: Optional[str] = None,
     service: Optional[str] = None,
+    python_version: Optional[str] = None,
 ) -> CheckResult:
     """Run a single check (subprocess) within a concurrency semaphore, capturing output and timing.
 
@@ -161,6 +161,8 @@ async def run_check(
     async with semaphore:
         start = time.time()
         cmd = base_args + [check, "--isolate", package]
+        if python_version and check not in INSTALL_AND_TEST_CHECKS:
+            cmd += ["--python", python_version]
         if service:
             cmd += ["--service", service]
         if mark_arg:
@@ -172,9 +174,7 @@ async def run_check(
         env["PROXY_URL"] = f"http://localhost:{proxy_port}"
 
         if in_ci():
-            env["PROXY_ASSETS_FOLDER"] = os.path.join(
-                root_dir, ".assets_distributed", str(proxy_port)
-            )
+            env["PROXY_ASSETS_FOLDER"] = os.path.join(root_dir, ".assets_distributed", str(proxy_port))
         try:
             logger.info(" ".join(cmd))
             proc = await asyncio.create_subprocess_exec(
@@ -194,9 +194,7 @@ async def run_check(
         stderr = stderr_b.decode(errors="replace")
         exit_code = proc.returncode or 0
         status = "OK" if exit_code == 0 else f"FAIL({exit_code})"
-        logger.info(
-            f"[END   {idx}/{total}] {check} :: {package} -> {status} in {duration:.2f}s"
-        )
+        logger.info(f"[END   {idx}/{total}] {check} :: {package} -> {status} in {duration:.2f}s")
         # Print captured output after completion to avoid interleaving
         header = f"===== OUTPUT: {check} :: {package} (exit {exit_code}) ====="
         trailer = "=" * len(header)
@@ -220,9 +218,7 @@ async def run_check(
         # finally, we need to clean up any temp dirs created by --isolate
         if in_ci():
             package_name = os.path.basename(os.path.normpath(package))
-            isolate_dir = os.path.join(
-                root_dir, ".venv", package_name, f".venv_{check}"
-            )
+            isolate_dir = os.path.join(root_dir, ".venv", package_name, f".venv_{check}")
             ISOLATE_DIRS_TO_CLEAN.append(isolate_dir)
         return CheckResult(package, check, exit_code, duration, stdout, stderr)
 
@@ -247,14 +243,10 @@ def summarize(results: List[CheckResult]) -> int:
     print("-" * len(header))
     for r in sorted(results, key=lambda x: (x.exit_code != 0, x.package, x.check)):
         status = "OK" if r.exit_code == 0 else f"FAIL({r.exit_code})"
-        print(
-            f"{r.package.ljust(pkg_w)}  {r.check.ljust(chk_w)}  {status.ljust(8)}  {r.duration:>10.2f}"
-        )
+        print(f"{r.package.ljust(pkg_w)}  {r.check.ljust(chk_w)}  {status.ljust(8)}  {r.duration:>10.2f}")
     worst = max((r.exit_code for r in results), default=0)
     failed = [r for r in results if r.exit_code != 0]
-    print(
-        f"\nTotal checks: {len(results)} | Failed: {len(failed)} | Worst exit code: {worst}"
-    )
+    print(f"\nTotal checks: {len(results)} | Failed: {len(failed)} | Worst exit code: {worst}")
     return worst
 
 
@@ -292,14 +284,10 @@ async def run_all_checks(
     dependency_tools_path = os.path.join(root_dir, "eng", "dependency_tools.txt")
 
     if in_ci():
-        logger.info(
-            "Replacing relative requirements in eng/test_tools.txt with prebuilt wheels."
-        )
+        logger.info("Replacing relative requirements in eng/test_tools.txt with prebuilt wheels.")
         replace_dev_reqs(test_tools_path, root_dir, wheel_dir)
 
-        logger.info(
-            "Replacing relative requirements in eng/dependency_tools.txt with prebuilt wheels."
-        )
+        logger.info("Replacing relative requirements in eng/dependency_tools.txt with prebuilt wheels.")
         replace_dev_reqs(dependency_tools_path, root_dir, wheel_dir)
 
     for pkg in packages:
@@ -321,15 +309,19 @@ async def run_all_checks(
         if not is_check_enabled(package, check, CHECK_DEFAULTS.get(check, True)):
             logger.warning(f"Skipping disabled check {check} for package {package}")
             continue
-        logger.info(
-            f"Assigning proxy port {next_proxy_port} to check {check} for package {package}"
-        )
-        scheduled.append((package, check, next_proxy_port))
+        logger.info(f"Assigning proxy port {next_proxy_port} to check {check} for package {package}")
+
+        # Check if this package overrides the Python version for analysis
+        pkg_python_version = get_config_setting(package, "analyze_python_version", None)
+        if pkg_python_version:
+            logger.info(f"Package {package} overrides analyze Python version to {pkg_python_version}")
+
+        scheduled.append((package, check, next_proxy_port, pkg_python_version))
         next_proxy_port += 1
 
     total = len(scheduled)
 
-    for idx, (package, check, proxy_port) in enumerate(scheduled, start=1):
+    for idx, (package, check, proxy_port, pkg_python_version) in enumerate(scheduled, start=1):
         tasks.append(
             asyncio.create_task(
                 run_check(
@@ -343,6 +335,7 @@ async def run_all_checks(
                     mark_arg,
                     dest_dir,
                     service,
+                    pkg_python_version,
                 )
             )
         )
@@ -358,15 +351,13 @@ async def run_all_checks(
         raise
     # Normalize exceptions
     norm_results: List[CheckResult] = []
-    for res, (package, check, _) in zip(results, scheduled):
+    for res, (package, check, _, _) in zip(results, scheduled):
         if isinstance(res, CheckResult):
             norm_results.append(res)
         elif isinstance(res, Exception):
             norm_results.append(CheckResult(package, check, 99, 0.0, "", str(res)))
         else:
-            norm_results.append(
-                CheckResult(package, check, 98, 0.0, "", f"Unknown result type: {res}")
-            )
+            norm_results.append(CheckResult(package, check, 98, 0.0, "", f"Unknown result type: {res}"))
     return summarize(norm_results)
 
 
@@ -442,15 +433,11 @@ def handler(signum, frame):
         ),
     )
 
-    parser.add_argument(
-        "--disablecov", help=("Flag. Disables code coverage."), action="store_true"
-    )
+    parser.add_argument("--disablecov", help=("Flag. Disables code coverage."), action="store_true")
 
     parser.add_argument(
         "--service",
-        help=(
-            "Name of service directory (under sdk/) to test. Example: --service applicationinsights"
-        ),
+        help=("Name of service directory (under sdk/) to test. Example: --service applicationinsights"),
     )
 
     parser.add_argument(
@@ -517,9 +504,7 @@ def handler(signum, frame):
     else:
         target_dir = root_dir
 
-    logger.info(
-        f"Beginning discovery for {args.service} and root dir {root_dir}. Resolving to {target_dir}."
-    )
+    logger.info(f"Beginning discovery for {args.service} and root dir {root_dir}. Resolving to {target_dir}.")
 
     # ensure that recursive virtual envs aren't messed with by this call
     os.environ.pop("VIRTUAL_ENV", None)
@@ -536,9 +521,7 @@ def handler(signum, frame):
     )
 
     if len(targeted_packages) == 0:
-        logger.info(
-            f"No packages collected for targeting string {args.glob_string} and root dir {root_dir}. Exit 0."
-        )
+        logger.info(f"No packages collected for targeting string {args.glob_string} and root dir {root_dir}. Exit 0.")
         exit(0)
 
     logger.info(f"Executing checks with the executable {sys.executable}.")
@@ -570,9 +553,7 @@ def handler(signum, frame):
         try:
             proxy_executable = prepare_local_tool(root_dir)
         except Exception as exc:
-            logger.error(
-                f"Unable to prepare test proxy executable for recording restore: {exc}"
-            )
+            logger.error(f"Unable to prepare test proxy executable for recording restore: {exc}")
             sys.exit(1)
 
     logger.info(
@@ -583,9 +564,7 @@ def handler(signum, frame):
     proxy_processes: List[ProxyProcess] = []
     try:
         if in_ci():
-            logger.info(
-                f"Ensuring {len(checks)} test proxies are running for requested checks..."
-            )
+            logger.info(f"Ensuring {len(checks)} test proxies are running for requested checks...")
         # Pass through service if set and not "auto"
         effective_service = args.service if (args.service and args.service != "auto") else None
         exit_code = asyncio.run(

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

@@ -14,11 +14,13 @@
 from ci_tools.parsing import ParsedSetup
 from ci_tools.functions import (
     discover_targeted_packages,
+    find_whl,
+)
+from ci_tools.venv import (
     get_venv_call,
     install_into_venv,
     get_venv_python,
     get_pip_command,
-    find_whl,
 )
 from ci_tools.variables import discover_repo_root, in_ci
 from ci_tools.logging import logger
@@ -71,10 +73,14 @@ def run(self, args: argparse.Namespace) -> int:
         """
         return 0
 
-    def create_venv(self, isolate: bool, venv_location: str) -> str:
-        """Abstraction for creating a virtual environment."""
+    def create_venv(self, isolate: bool, venv_location: str, python_version: Optional[str] = None) -> str:
+        """Abstraction for creating a virtual environment.
+
+        :param python_version: If provided, passed to ``uv venv --python`` during initial creation only.
+            Has no effect on venv reuse or lookup.
+        """
         if isolate:
-            venv_cmd = get_venv_call(sys.executable)
+            venv_cmd = get_venv_call(sys.executable, python_version=python_version)
             venv_python = get_venv_python(venv_location)
             if os.path.exists(venv_python):
                 logger.info(f"Reusing existing venv at {venv_python}")
@@ -109,16 +115,30 @@ def create_venv(self, isolate: bool, venv_location: str) -> str:
         # if we don't need to isolate, just return the python executable that we're invoking
         return sys.executable
 
-    def get_executable(self, isolate: bool, check_name: str, executable: str, package_folder: str) -> Tuple[str, str]:
+    def get_executable(
+        self,
+        isolate: bool,
+        check_name: str,
+        executable: str,
+        package_folder: str,
+        python_version: Optional[str] = None,
+    ) -> Tuple[str, str]:
         """Get the Python executable that should be used for this check."""
         # Keep venvs under a shared repo-level folder to prevent nested import errors during pytest collection
         package_name = os.path.basename(os.path.normpath(package_folder))
         shared_venv_root = os.path.join(REPO_ROOT, ".venv", package_name)
         os.makedirs(shared_venv_root, exist_ok=True)
-        venv_location = os.path.join(shared_venv_root, f".venv_{check_name}")
+
+        # version-qualify the venv dir when a specific Python is requested to avoid cache collisions
+        venv_suffix = f".venv_{check_name}"
+        if python_version:
+            sanitized = python_version.replace(".", "").replace("-", "")
+            venv_suffix = f".venv_{check_name}_py{sanitized}"
+        venv_location = os.path.join(shared_venv_root, venv_suffix)
+
         # if isolation is required, the executable we get back will align with the venv
         # otherwise we'll just get sys.executable and install in current
-        executable = self.create_venv(isolate, venv_location)
+        executable = self.create_venv(isolate, venv_location, python_version=python_version)
         staging_directory = os.path.join(venv_location, ".staging")
         os.makedirs(staging_directory, exist_ok=True)
         return executable, staging_directory

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

@@ -92,7 +92,13 @@ def run(self, args: argparse.Namespace) -> int:
                 os.chdir(parsed.folder)
             package_dir = parsed.folder
             package_name = parsed.name
-            executable, staging_directory = self.get_executable(args.isolate, args.command, sys.executable, package_dir)
+            executable, staging_directory = self.get_executable(
+                args.isolate,
+                args.command,
+                sys.executable,
+                package_dir,
+                python_version=getattr(args, "python_version", None),
+            )
             logger.info(f"Processing {package_name} for apistub check")
 
             # install dependencies

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

@@ -42,7 +42,13 @@ def run(self, args: argparse.Namespace) -> int:
                 os.chdir(parsed.folder)
             package_dir = parsed.folder
             package_name = parsed.name
-            executable, staging_directory = self.get_executable(args.isolate, args.command, sys.executable, package_dir)
+            executable, staging_directory = self.get_executable(
+                args.isolate,
+                args.command,
+                sys.executable,
+                package_dir,
+                python_version=getattr(args, "python_version", None),
+            )
             logger.info(f"Processing {package_name} for bandit check")
 
             self.install_dev_reqs(executable, args, package_dir)

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

@@ -45,7 +45,13 @@ def run(self, args: argparse.Namespace) -> int:
             package_dir = parsed.folder
             package_name = parsed.name
 
-            executable, staging_directory = self.get_executable(args.isolate, args.command, sys.executable, package_dir)
+            executable, staging_directory = self.get_executable(
+                args.isolate,
+                args.command,
+                sys.executable,
+                package_dir,
+                python_version=getattr(args, "python_version", None),
+            )
             logger.info(f"Processing {package_name} for black check")
 
             self.install_dev_reqs(executable, args, package_dir)