Skip to content
Open
Changes from 2 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
45 changes: 37 additions & 8 deletions scripts/ci/test_source.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,12 +9,15 @@

import logging
import os
from pathlib import Path
import subprocess
import sys
import tempfile
import shutil
import shlex

from subprocess import check_output, CalledProcessError, run
from typing import Optional
from util import SRC_PATH

logger = logging.getLogger(__name__)
Expand Down Expand Up @@ -64,7 +67,7 @@ def run_command(cmd, check_return_code=False, cwd=None):
raise RuntimeError(f"{cmd} failed")


def test_extension():
def test_extension(whl_dir: Optional[Path] = None):
for pkg_name, ext_path in ALL_TESTS:
ext_name = ext_path.split('/')[-1]
logger.info(f'installing extension: {ext_name}')
Expand All @@ -86,22 +89,46 @@ def test_extension():
cmd = ['azdev', 'extension', 'remove', ext_name]
run_command(cmd, check_return_code=True)


def test_source_wheels():
if whl_dir is not None:
logger.info(f'installing extension wheel: {ext_name}')
wheel_path = next(whl_dir.glob(f"{ext_name}*.whl"))
Copy link

Copilot AI Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The next() call will raise StopIteration if no matching wheel file is found. Add a default value or handle the exception to provide a more informative error message. Consider: wheel_path = next(whl_dir.glob(f\"{ext_name}*.whl\"), None) followed by a check.

Suggested change
wheel_path = next(whl_dir.glob(f"{ext_name}*.whl"))
wheel_path = next(whl_dir.glob(f"{ext_name}*.whl"), None)
if wheel_path is None:
logger.error(f"No wheel file found for extension '{ext_name}' in directory '{whl_dir}'.")
raise FileNotFoundError(f"No wheel file found for extension '{ext_name}' in directory '{whl_dir}'.")

Copilot uses AI. Check for mistakes.
subprocess.run(
f"az extension add -y -s {wheel_path}".split(" "),
Copy link

Copilot AI Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Using .split(\" \") will incorrectly split paths that contain spaces. Use a list directly: [\"az\", \"extension\", \"add\", \"-y\", \"-s\", str(wheel_path)]

Suggested change
f"az extension add -y -s {wheel_path}".split(" "),
["az", "extension", "add", "-y", "-s", str(wheel_path)],

Copilot uses AI. Check for mistakes.
check=True,
)
subprocess.run(
f"az {ext_name} --help".split(" "),
Copy link

Copilot AI Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Using .split(\" \") will incorrectly split extension names that contain spaces (if any exist). Use a list directly: [\"az\", ext_name, \"--help\"]

Suggested change
f"az {ext_name} --help".split(" "),
["az", ext_name, "--help"],

Copilot uses AI. Check for mistakes.
check=True,
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
subprocess.run(
f"az extension remove -n {ext_name}".split(" "),
Copy link

Copilot AI Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Using .split(\" \") will incorrectly split extension names that contain spaces (if any exist). Use a list directly: [\"az\", \"extension\", \"remove\", \"-n\", ext_name]

Suggested change
f"az extension remove -n {ext_name}".split(" "),
["az", "extension", "remove", "-n", ext_name],

Copilot uses AI. Check for mistakes.
check=True,
)


def test_source_wheels(whl_dir: Optional[Path] = None):
# Test we can build all sources into wheels and that metadata from the wheel is valid
built_whl_dir = tempfile.mkdtemp()
source_extensions = [os.path.join(SRC_PATH, n) for n in os.listdir(SRC_PATH)
if os.path.isdir(os.path.join(SRC_PATH, n))]
for s in source_extensions:
ext_name = s.split('/')[-1]
if not os.path.isfile(os.path.join(s, 'setup.py')):
continue
try:
check_output(['python', 'setup.py', 'bdist_wheel', '-q', '-d', built_whl_dir], cwd=s)
check_output(['azdev', 'extension', 'build', ext_name, '--dist-dir', built_whl_dir])
except CalledProcessError as err:
raise("Unable to build extension {} : {}".format(s, err))
Copy link

Copilot AI Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The raise statement should instantiate an exception, not call it as a function. This should be raise RuntimeError(...) or similar exception class.

Suggested change
raise("Unable to build extension {} : {}".format(s, err))
raise RuntimeError("Unable to build extension {} : {}".format(s, err))

Copilot uses AI. Check for mistakes.
# Export built wheels so CI can publish them as artifacts
wheels_out_dir = os.environ.get('WHEELS_OUTPUT_DIR')
if wheels_out_dir:
wheels_out_dirs = [
os.environ.get('WHEELS_OUTPUT_DIR'),
str(whl_dir),
Copy link

Copilot AI Nov 12, 2025

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

When whl_dir is None, str(whl_dir) converts it to the string \"None\" instead of keeping it as None. This bypasses the null check on line 130. Use str(whl_dir) if whl_dir else None instead.

Suggested change
str(whl_dir),
str(whl_dir) if whl_dir else None,

Copilot uses AI. Check for mistakes.
]
for wheels_out_dir in wheels_out_dirs:
if not wheels_out_dir:
continue
try:
os.makedirs(wheels_out_dir, exist_ok=True)
for fname in os.listdir(built_whl_dir):
Expand All @@ -115,5 +142,7 @@ def test_source_wheels():


if __name__ == '__main__':
test_extension()
test_source_wheels()
with tempfile.TemporaryDirectory() as whl_dir:
whl_dir = Path(whl_dir)
test_source_wheels(whl_dir=whl_dir)
test_extension(whl_dir=whl_dir)
Loading