-
Notifications
You must be signed in to change notification settings - Fork 1.5k
[confcom] Add more thorough tests for --with-containers
#9428
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Conversation
️✔️Azure CLI Extensions Breaking Change Test
|
|
Thank you for your contribution! We will review the pull request and get back to you soon. |
|
The git hooks are available for azure-cli and azure-cli-extensions repos. They could help you run required checks before creating the PR. Please sync the latest code with latest dev branch (for azure-cli) or main branch (for azure-cli-extensions). pip install azdev --upgrade
azdev setup -c <your azure-cli repo path> -r <your azure-cli-extensions repo path>
|
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Pull Request Overview
This PR enhances test coverage for the --with-containers flag by adding comprehensive tests for all samples and refactoring the policy model to use Pydantic dataclasses with explicit ordering semantics.
Key changes:
- Migrated policy data model from standard Python dataclasses to Pydantic dataclasses with custom ordering support
- Added container definition (
.rego) files for each sample directory to support--with-containerstesting - Implemented OPA (Open Policy Agent) integration for policy deserialization and comparison
- Added parameterized test that validates policy generation from container definitions for all samples
Reviewed Changes
Copilot reviewed 29 out of 29 changed files in this pull request and generated 11 comments.
Show a summary per file
| File | Description |
|---|---|
setup.py |
Version bump to 1.4.2, added pydantic dependency, and OPA binary download during installation |
HISTORY.rst |
Added release notes for version 1.4.2 describing pydantic migration |
samples/aci/*/container*.rego |
Added 20+ container definition files as test inputs for --with-containers testing |
azext_confcom/lib/policy.py |
Migrated from dataclasses to Pydantic with OrderlessField for collections where order is semantically irrelevant |
azext_confcom/lib/orderless_dataclasses.py |
New custom Pydantic dataclass wrapper that sorts orderless fields for serialization and comparison |
azext_confcom/lib/opa.py |
New OPA binary downloader and executor for policy evaluation |
azext_confcom/lib/binaries.py |
New utility for managing binary directory location |
azext_confcom/lib/serialization.py |
New policy serialization/deserialization utilities using OPA |
azext_confcom/security_policy.py |
Updated to use Pydantic's TypeAdapter for container serialization |
azext_confcom/tests/latest/test_confcom_acipolicygen_arm.py |
Added comprehensive parameterized test and renamed existing test for clarity |
|
|
||
| @pytest.mark.parametrize( | ||
| "sample_directory", | ||
| os.listdir(SAMPLES_ROOT), |
Copilot
AI
Nov 14, 2025
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The test parameterizes over all directories in SAMPLES_ROOT without filtering. This means if any non-sample directories exist (e.g., .git, __pycache__, or other metadata directories), the test will fail when trying to open policy.rego which won't exist in those directories. Consider filtering the directories:
@pytest.mark.parametrize(
"sample_directory",
[d for d in os.listdir(SAMPLES_ROOT)
if os.path.isdir(os.path.join(SAMPLES_ROOT, d))
and os.path.exists(os.path.join(SAMPLES_ROOT, d, "policy.rego"))],
)| os.listdir(SAMPLES_ROOT), | |
| [ | |
| d for d in os.listdir(SAMPLES_ROOT) | |
| if os.path.isdir(os.path.join(SAMPLES_ROOT, d)) | |
| and os.path.exists(os.path.join(SAMPLES_ROOT, d, "policy.rego")) | |
| ], |
|
|
||
| from azext_confcom.lib.binaries import get_binaries_dir | ||
|
|
||
| _opa_pathh = os.path.abspath(os.path.join(get_binaries_dir(), "opa")) |
Copilot
AI
Nov 14, 2025
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Variable name typo: _opa_pathh has an extra 'h'. Should be _opa_path for consistency with naming conventions.
| _expected_sha256 = "fe8e191d44fec33db2a3d0ca788b9f83f866d980c5371063620c3c6822792877" | ||
|
|
||
|
|
||
| def opa_get(): | ||
|
|
||
| opa_fetch_resp = requests.get( | ||
| f"https://openpolicyagent.org/downloads/latest/opa_{platform.system().lower()}_amd64") | ||
| opa_fetch_resp.raise_for_status() | ||
|
|
||
| assert hashlib.sha256(opa_fetch_resp.content).hexdigest() == _expected_sha256 | ||
|
|
||
| with open(_opa_pathh, "wb") as f: | ||
| f.write(opa_fetch_resp.content) | ||
|
|
Copilot
AI
Nov 14, 2025
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The OPA binary is downloaded during package installation without verifying the authenticity or checking if the file already exists. This could lead to:
- Unnecessary network calls during every installation
- No version pinning - the download URL uses "latest" which could break installations when new OPA versions are released
- The hardcoded SHA256 hash will fail if OPA releases a new version
Consider:
- Checking if the binary already exists before downloading
- Pinning to a specific OPA version instead of "latest"
- Updating the expected SHA256 hash when upgrading OPA versions
| _expected_sha256 = "fe8e191d44fec33db2a3d0ca788b9f83f866d980c5371063620c3c6822792877" | |
| def opa_get(): | |
| opa_fetch_resp = requests.get( | |
| f"https://openpolicyagent.org/downloads/latest/opa_{platform.system().lower()}_amd64") | |
| opa_fetch_resp.raise_for_status() | |
| assert hashlib.sha256(opa_fetch_resp.content).hexdigest() == _expected_sha256 | |
| with open(_opa_pathh, "wb") as f: | |
| f.write(opa_fetch_resp.content) | |
| _OPA_VERSION = "v0.63.0" | |
| _expected_sha256 = "fe8e191d44fec33db2a3d0ca788b9f83f866d980c5371063620c3c6822792877" | |
| def opa_get(): | |
| # Check if OPA binary exists and matches expected hash | |
| if os.path.isfile(_opa_pathh): | |
| with open(_opa_pathh, "rb") as f: | |
| file_hash = hashlib.sha256(f.read()).hexdigest() | |
| if file_hash == _expected_sha256: | |
| return _opa_pathh | |
| # Download OPA binary if not present or hash mismatch | |
| url = f"https://openpolicyagent.org/downloads/{_OPA_VERSION}/opa_{platform.system().lower()}_amd64" | |
| opa_fetch_resp = requests.get(url) | |
| opa_fetch_resp.raise_for_status() | |
| assert hashlib.sha256(opa_fetch_resp.content).hexdigest() == _expected_sha256, "Downloaded OPA binary hash mismatch" | |
| with open(_opa_pathh, "wb") as f: | |
| f.write(opa_fetch_resp.content) |
| SecurityPolicyProxy.download_binaries() | ||
| KataPolicyGenProxy.download_binaries() | ||
| CoseSignToolProxy.download_binaries() | ||
| opa_get() |
Copilot
AI
Nov 14, 2025
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The opa_get() function is called during package setup without error handling. If the download fails (network issues, URL unavailable, etc.), the package installation will fail. Consider:
- Adding a try-except block around the call in setup.py
- Checking if the binary already exists before attempting download
- Providing a fallback mechanism or clearer error message for users
Example:
try:
if not os.path.exists(os.path.join(get_binaries_dir(), "opa")):
opa_get()
except Exception as e:
logger.warn(f"Failed to download OPA binary: {e}. You may need to download it manually.")| "pattern":"MY_VAR=MY_VAL", | ||
| "required":false, | ||
| "strategy":"string" |
Copilot
AI
Nov 14, 2025
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Missing space in formatting - there's no space after the colon in the JSON field. This creates inconsistent formatting:
"pattern":"MY_VAR=MY_VAL",Should be:
"pattern": "MY_VAR=MY_VAL",This is inconsistent with the formatting in the rest of the file (see lines 9-11 which have proper spacing).
| "pattern":"MY_VAR=MY_VAL", | |
| "required":false, | |
| "strategy":"string" | |
| "pattern": "MY_VAR=MY_VAL", | |
| "required": false, | |
| "strategy": "string" |
| with open(os.path.join(SAMPLES_ROOT, sample_directory, file_path)) as input_containers_file: | ||
| input_containers.append(json.loads(input_containers_file.read())) |
Copilot
AI
Nov 14, 2025
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
File is opened without a context manager. Use a with statement for proper resource management:
with open(os.path.join(SAMPLES_ROOT, sample_directory, file_path)) as input_containers_file:
input_containers.append(json.loads(input_containers_file.read()))| # Load input container definition | ||
| input_containers = [] | ||
| for file_path in os.listdir(os.path.join(SAMPLES_ROOT, sample_directory)): | ||
| if file_path.startswith("container"): |
Copilot
AI
Nov 14, 2025
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The loop that collects container definition files assumes all files starting with "container" are valid JSON container definitions. This could cause issues if there are other files (like container.md, backup files like container1.rego~, etc.). Consider being more specific with the file matching:
for file_path in os.listdir(os.path.join(SAMPLES_ROOT, sample_directory)):
if file_path.startswith("container") and file_path.endswith(".rego"):
with open(os.path.join(SAMPLES_ROOT, sample_directory, file_path)) as input_containers_file:
input_containers.append(json.loads(input_containers_file.read()))| if file_path.startswith("container"): | |
| if file_path.startswith("container") and file_path.endswith(".json"): |
| f"https://openpolicyagent.org/downloads/latest/opa_{platform.system().lower()}_amd64") | ||
| opa_fetch_resp.raise_for_status() | ||
|
|
||
| assert hashlib.sha256(opa_fetch_resp.content).hexdigest() == _expected_sha256 |
Copilot
AI
Nov 14, 2025
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The assertion will cause a cryptic error message if the SHA256 hash doesn't match. Consider using a more informative error message:
actual_hash = hashlib.sha256(opa_fetch_resp.content).hexdigest()
if actual_hash != _expected_sha256:
raise ValueError(f"OPA binary hash mismatch. Expected {_expected_sha256}, got {actual_hash}")| assert hashlib.sha256(opa_fetch_resp.content).hexdigest() == _expected_sha256 | |
| actual_hash = hashlib.sha256(opa_fetch_resp.content).hexdigest() | |
| if actual_hash != _expected_sha256: | |
| raise ValueError(f"OPA binary hash mismatch. Expected {_expected_sha256}, got {actual_hash}") |
| with open(os.path.join(SAMPLES_ROOT, sample_directory, "policy.rego")) as expected_policy_file: | ||
| expected_policy_str = expected_policy_file.read() |
Copilot
AI
Nov 14, 2025
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The test opens and reads a file without a context manager. If an exception occurs, the file handle may not be properly closed. Use a with statement for proper resource management:
with open(os.path.join(SAMPLES_ROOT, sample_directory, "policy.rego")) as expected_policy_file:
expected_policy_str = expected_policy_file.read()| def opa_get(): | ||
|
|
||
| opa_fetch_resp = requests.get( | ||
| f"https://openpolicyagent.org/downloads/latest/opa_{platform.system().lower()}_amd64") |
Copilot
AI
Nov 14, 2025
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
The opa_get() function downloads a binary from the internet without verifying HTTPS certificate or using a secure connection verification. The requests.get() call should explicitly verify SSL certificates to prevent man-in-the-middle attacks:
opa_fetch_resp = requests.get(
f"https://openpolicyagent.org/downloads/latest/opa_{platform.system().lower()}_amd64",
verify=True, # Explicitly verify SSL certificates (default behavior, but explicit is better)
timeout=30 # Add timeout to prevent hanging
)Additionally, there's no timeout which could cause the installation to hang indefinitely if the server doesn't respond.
| f"https://openpolicyagent.org/downloads/latest/opa_{platform.system().lower()}_amd64") | |
| f"https://openpolicyagent.org/downloads/latest/opa_{platform.system().lower()}_amd64", | |
| verify=True, # Explicitly verify SSL certificates | |
| timeout=30 # Add timeout to prevent hanging | |
| ) |
Why
The existing tests for
--with-containersare okay but don't tests the full range of inputs we have insamples/.How
--with-containersto getpolicy.rego--with-containersto producepolicy.regoThis checklist is used to make sure that common guidelines for a pull request are followed.
Related command
General Guidelines
azdev style <YOUR_EXT>locally? (pip install azdevrequired)python scripts/ci/test_index.py -qlocally? (pip install wheel==0.30.0required)