Skip to content
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

Refactoring/Testing with Cody AI from Sourcegraph #55

Draft
wants to merge 16 commits into
base: main
Choose a base branch
from
266 changes: 34 additions & 232 deletions src/promote.py
Original file line number Diff line number Diff line change
Expand Up @@ -55,13 +55,43 @@
import yaml

from typing import Callable, Iterator, Union, Optional # noqa: F401
from validate import validate_promotion_lists, validate_charts, validate_images

Choose a reason for hiding this comment

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

✅ Getting better: Overall Code Complexity
The mean cyclomatic complexity decreases from 6.71 to 6.31, threshold = 4

Why does this problem occur?

This file has many conditional statements (e.g. if, for, while) across its implementation, leading to lower code health. Avoid adding more conditionals. Read more.


# Initialize logger
logger = logging.getLogger()
logger.addHandler(logging.StreamHandler())
logger = logging.getLogger(__name__)
logger.propagate = False
logger.setLevel(logging.DEBUG)



def validate_runtime_environment() -> None:
"""
Validate that the runtime environment has the tools we need and provided directories exist.

This function validates the runtime environment by checking if the `kustomize` command is available.

Example Usage:
```python
validate_runtime_environment()
```

Raises:
CalledProcessError: If the `kustomize` command is not available.

Returns:
None
"""

# Validate that the kustomize command is available
try:
logger.debug("Validating that kustomize is available...")
run(["kustomize", "version"])
except subprocess.CalledProcessError:
logger.fatal(
"kustomize is not available. Please install kustomize before running this script."
)
exit(1)

def run(args: list[str]) -> int:
"""
Run the given command and log the output.
Expand Down Expand Up @@ -118,142 +148,6 @@ def merge_manifests(a: dict, b: dict) -> dict:
return a


def find_duplicates(images: list, field: str) -> set:
"""
Find duplicate values for a specified field in a list of images.

Args:
images (list): The list of images to check for duplicates.
field (str): The field name to check for duplicates.

Returns:
set: A set containing the duplicate values for the specified field.
"""
duplicates = set()
image_names = [image[field] for image in images if field in image]
seen = set()
for image_name in image_names:
if image_name in seen:
duplicates.add(image_name)
else:
seen.add(image_name)

return duplicates


def validate_images(images):
"""
Validate a list of images to ensure they have the required fields and that the names and newNames are unique.

Args:
images (list): The list of images to validate.

Returns:
bool: True if all images are valid, False otherwise.
"""
originally_dict = False
# Convert to list if it is a dict (which is the case when we are validating
# images from a promotion file)
if isinstance(images, dict):
originally_dict = True
images = list(images.values())

# Ensure that all image names are unique
duplicates = find_duplicates(images, "name")
if len(duplicates) > 0:
logger.fatal(
f"Found duplicate image names: {' '.join(duplicates)}. Images must have unique names."
)
sys.exit(1)

# Ensure that all image newNames are unique
duplicates = find_duplicates(images, "newName")
if len(duplicates) > 0:
logger.fatal(
f"Found duplicate image newNames: {' '.join(duplicates)}. Image newNames must have unique names."
)
sys.exit(1)

for image in images:
# Validate that the image has the required fields
if "name" not in image:
logger.fatal(f"Image {image} is missing the required 'name' field.")
return False
if "fromOverlay" in image:
if "newName" in image:
logger.fatal(
f"Image {image} cannot set newName when fromOverlay is set."
)
return False
if "newTag" in image:
logger.fatal(
f"Image {image} cannot set newTag when fromOverlay is set."
)
return False
else:
if ("newTag" not in image) and ("newName" not in image):
logger.fatal(f"Image {image} must set newName, newTag or both.")
return False
# Validate that the image has the required fields if it was not a dict,
# which means that it is coming from a promotion file and not from a
# kustomization.yaml file.
if not originally_dict and "overlays" not in image:
logger.fatal(f"Image {image} is missing the required 'overlays' field.")
return False

return True
Comment on lines -144 to -204

Choose a reason for hiding this comment

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

✅ No longer an issue: Complex Method
validate_images is no longer above the threshold for cyclomatic complexity

Why does this problem occur?

This function has many conditional statements (e.g. if, for, while), leading to lower code health. Avoid adding more conditionals and code to it without refactoring. Read more.

Comment on lines -144 to -204

Choose a reason for hiding this comment

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

✅ No longer an issue: Bumpy Road Ahead
validate_images is no longer above the threshold for logical blocks with deeply nested code

Why does this problem occur?

The Bumpy Road code smell is a function that contains multiple chunks of nested conditional logic. The deeper the nesting and the more bumps, the lower the code health. Read more.



def validate_charts(charts):
"""
Validate that the charts to update have the required fields.

Args:
charts (list): The list of charts to update.

Returns:
bool: True if all charts are valid, False otherwise.
"""
originally_dict = False
# Convert to list if it is a dict (which is the case when we are validating
# charts from a promotion file)
if isinstance(charts, dict):
originally_dict = True
charts = list(charts.values())

# Ensure that all chart names are unique
duplicates = find_duplicates(charts, "name")
if len(duplicates) > 0:
logger.fatal(
f"Found duplicate chart names: {' '.join(duplicates)}. Charts must have unique names."
)
sys.exit(1)

for chart in charts:
# Validate that the chart has the required fields
if "name" not in chart:
logger.fatal(f"Chart {chart} is missing the required 'name' field.")
return False
if "fromOverlay" in chart:
if "version" in chart:
logger.fatal(
f"Chart {chart} cannot set version when fromOverlay is set."
)
return False
else:
if "version" not in chart:
logger.fatal(f"Chart {chart} must set version.")
return False
# Validate that the chart has the required fields if it was not a dict,
# which means that it is coming from a promotion file and not from a
# kustomization.yaml file.
if not originally_dict and "overlays" not in chart:
logger.fatal(f"Chart {chart} is missing the required 'overlays' field.")
return False

return True
Comment on lines -207 to -254

Choose a reason for hiding this comment

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

✅ No longer an issue: Complex Method
validate_charts is no longer above the threshold for cyclomatic complexity

Why does this problem occur?

This function has many conditional statements (e.g. if, for, while), leading to lower code health. Avoid adding more conditionals and code to it without refactoring. Read more.

Comment on lines -207 to -254

Choose a reason for hiding this comment

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

✅ No longer an issue: Bumpy Road Ahead
validate_charts is no longer above the threshold for logical blocks with deeply nested code

Why does this problem occur?

The Bumpy Road code smell is a function that contains multiple chunks of nested conditional logic. The deeper the nesting and the more bumps, the lower the code health. Read more.



def read_images_from_overlay(overlay: str, deployment_dir: str) -> dict[str, dict]:
"""
Read the images from the specified overlay in a deployment directory and return a dictionary mapping image names to their corresponding image dictionaries.
Expand Down Expand Up @@ -599,35 +493,6 @@ def update_kustomize_charts(
return promotion_manifest


def validate_runtime_environment() -> None:
"""
Validate that the runtime environment has the tools we need and provided directories exist.

This function validates the runtime environment by checking if the `kustomize` command is available.

Example Usage:
```python
validate_runtime_environment()
```

Raises:
CalledProcessError: If the `kustomize` command is not available.

Returns:
None
"""

# Validate that the kustomize command is available
try:
logger.debug("Validating that kustomize is available...")
run(["kustomize", "version"])
except subprocess.CalledProcessError:
logger.fatal(
"kustomize is not available. Please install kustomize before running this script."
)
exit(1)


def get_deployment_dir() -> str:
"""
Get the deployment directory from the DEPLOYMENT_DIR env variable.
Expand Down Expand Up @@ -681,70 +546,6 @@ def load_promotion_json(type: str) -> dict:
return promotion_json


def validate_promotion_lists(
images_to_update: list[dict], charts_to_update: list[dict]
) -> None:
"""
Validate the provided promotion configuration.

Args:
images_to_update (list): The list of images to update.
charts_to_update (list): The list of charts to update.

Returns:
None

Raises:
SystemExit: If there are no images or charts to update.
"""
if len(images_to_update) == 0 and len(charts_to_update) == 0:
logger.fatal("No images or charts to update. Please provide either (or both):")
logger.fatal(
"- A JSON object of images to update via the IMAGES_TO_UPDATE env var or via stdin in the following format:"
)
logger.fatal(
"""
[
{
"name": "image-name",
# Either newTag or newName is required
"newName": "new-image-name",
"newTag": "new-image-tag",
# ... or fromOverlay is required
"fromOverlay": "overlay-name",
"overlays": ["target-env", "target-env2"]
}
]
"""
)
logger.fatal(
"- A JSON object of charts to update via the CHARTS_TO_UPDATE env var or via stdin in the following format:"
)
logger.fatal(
"""
[
{
"name": "chart-name",
# Either version is required
"version": "new-chart-version",
# ... or fromOverlay is required
"fromOverlay": "overlay-name",
# Optionally, update the release name
"releaseName": "new-release-name",
"overlays": ["target-env", "target-env2"]
}
]
"""
)
exit(1)

# Validate that the images to update have the required fields
validate_images(images_to_update)

# Validate that the charts to update have the required fields
validate_charts(charts_to_update)


def main():
validate_runtime_environment()

Expand All @@ -757,7 +558,8 @@ def main():
charts_to_update = load_promotion_json("charts")

# Exit with failure if there are no images or charts to update, printing usage information.
validate_promotion_lists(images_to_update, charts_to_update)
if not validate_promotion_lists(images_to_update, charts_to_update):
sys.exit(1)

# Get the list of images for each overlay
overlays_to_images = get_images_from_overlays(images_to_update, deployment_dir)
Expand Down
25 changes: 14 additions & 11 deletions src/tests/charts_test.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,8 @@
import unittest
import promote as promote
import os
import sys
sys.path.insert(0, os.path.abspath(os.path.join(os.path.dirname(__file__), '..')))
from promote import validate_charts, get_charts_from_overlays

overlay_no_name_or_version = [{"name": "lighthouse", "overlays": ["bar"]}]
overlay_new_name = [
Expand All @@ -18,40 +21,40 @@

class TestValidateChartsFromOverlays(unittest.TestCase):
def test_empty(self):
self.assertEqual(promote.validate_charts([]), True)
self.assertEqual(validate_charts([]), True)

def test_only_new_name(self):
self.assertEqual(
promote.validate_charts(overlay_new_name),
validate_charts(overlay_new_name),
False,
)

def test_only_new_version(self):
self.assertEqual(
promote.validate_charts(overlay_new_version),
validate_charts(overlay_new_version),
True,
)

def test_missing_new_name_and_version(self):
self.assertEqual(promote.validate_charts(overlay_no_name_or_version), False)
self.assertEqual(validate_charts(overlay_no_name_or_version), False)

def test_new_name_and_version(self):
self.assertEqual(promote.validate_charts(overlay_new_name_and_version), True)
self.assertEqual(validate_charts(overlay_new_name_and_version), True)


class TestGetChartsFromOverlays(unittest.TestCase):
def test_empty(self):
self.assertEqual(promote.get_charts_from_overlays([], "."), {})
self.assertEqual(get_charts_from_overlays([], "."), {})

def test_missing_new_name(self):
self.assertEqual(
promote.get_charts_from_overlays(overlay_no_name_or_version, "."),
get_charts_from_overlays(overlay_no_name_or_version, "."),
{"bar": [{"name": "lighthouse", "overlays": ["bar"]}]},
)

def test_new_name(self):
self.assertEqual(
promote.get_charts_from_overlays(overlay_new_name, "."),
get_charts_from_overlays(overlay_new_name, "."),
{
"bar": [
{
Expand All @@ -65,13 +68,13 @@ def test_new_name(self):

def test_new_tag(self):
self.assertEqual(
promote.get_charts_from_overlays(overlay_new_version, "."),
get_charts_from_overlays(overlay_new_version, "."),
{"bar": [{"name": "lighthouse", "version": "1.0.0", "overlays": ["bar"]}]},
)

def test_new_name_and_tag(self):
self.assertEqual(
promote.get_charts_from_overlays(overlay_new_name_and_version, "."),
get_charts_from_overlays(overlay_new_name_and_version, "."),
{
"bar": [
{
Expand Down
Loading
Loading