.pr_agent_auto_best_practices

Pattern 1: Refactor duplicated logic into a single helper to keep behavior consistent across call sites (e.g., shared prefix handling or image normalization), and replace repeated inline code with calls to that helper.

Example code before:

def find(name):
    prefix = f"acme-{tenant}-"
    if not name.startswith(prefix):
        name = prefix + name
    return api.find(name)

def delete(name):
    prefix = f"acme-{tenant}-"
    if not name.startswith(prefix):
        name = prefix + name
    return api.delete(name)

Example code after:

def _ensure_prefixed(name):
    prefix = f"acme-{tenant}-"
    return name if name.startswith(prefix) else prefix + name

def find(name):
    return api.find(_ensure_prefixed(name))

def delete(name):
    return api.delete(_ensure_prefixed(name))

Relevant past accepted suggestions:

Suggestion 1:

Extract image normalization helper method

Extract the image normalization logic into a shared helper method to avoid duplication and ensure consistent image processing across the codebase.

src/duplo_resource/service.py [710-716]

+def _normalize_image(self, image):
+  """Normalize image name by removing Docker registry prefixes."""
+  return image.removeprefix("docker.io/library/").removeprefix("docker.io/").removeprefix("library/")
+
 def get_pod_image(pod):
   """Get the image of a pod."""
-  raw_image=pod["Containers"][0]["Image"]
+  raw_image = pod["Containers"][0]["Image"]
   self.duplo.logger.debug(f"Retrieved raw image {raw_image} for {pod['InstanceId']}")
-  normalized_image = raw_image.removeprefix("docker.io/library/").removeprefix("docker.io/").removeprefix("library/")
+  normalized_image = self._normalize_image(raw_image)
   self.duplo.logger.debug(f"Returning normalized image {normalized_image} for {pod['InstanceId']}\n")
   return normalized_image

Suggestion 2:

Refactor duplicated prefix logic

The prefix logic is duplicated in both find() and delete() methods. This could lead to inconsistencies if the prefix format changes in one place but not the other. Extract this logic into a helper method to ensure consistent behavior.

src/duplo_resource/aws_secret.py [51-54]

-# if the name has the prefix we good, otherwise add it
-prefix = f"duploservices-{self.duplo.tenant}-"
-if not name.startswith(prefix):
-  name = prefix + name
+# Use a helper method to ensure consistent prefix handling
+name = self._ensure_name_has_prefix(name)

Suggestion 3:

Remove duplicated code block

Remove the duplicated code block that checks for and assigns IpcMode to the result dictionary.

src/duplo_resource/ecs_service.py [297-300]

-if "IpcMode" in task_def:
-  result["IpcMode"] = task_def["IpcMode"]
 if "IpcMode" in task_def:
   result["IpcMode"] = task_def["IpcMode"]

Suggestion 4:

Remove duplicate function definition

This function redefines the get_test_data function that's already imported at the top of the file. This duplication can lead to confusion about which function is being used. Remove this duplicate definition and use the imported function instead.

src/tests/test_cloudfront.py [23-27]

-def get_test_data(name) -> dict:
-  dir = pathlib.Path(__file__).parent.resolve()
-  f = f"{dir}/data/{name}.yaml"
-  with open(f, 'r') as stream:
-    return yaml.safe_load(stream)
+# Remove this duplicate function definition entirely, as it's already imported
+# from .conftest import get_test_data

Pattern 2: Remove dead or commented-out code to improve clarity and prevent confusion with the active implementation; keep files free of obsolete blocks and duplicate definitions.

Example code before:

# Old approach:
# items = []
# for x in data:
#     if x.enabled:
#         items.append(x.id)
items = [x.id for x in data if x.enabled]

Example code after:

items = [x.id for x in data if x.enabled]

Relevant past accepted suggestions:

Suggestion 1:

Remove commented dead code

Remove the commented-out runs-on configuration for the self-hosted runner to eliminate dead code and improve the workflow file's clarity.

.github/workflows/installer.yml [135-139]

-# runs-on:
-# - self-hosted
-# - darwin
-# - arm64
 runs-on: macos-latest-xlarge

Suggestion 2:

Remove commented-out code that has been replaced by active code

The commented-out code block should be removed since it's been replaced by a more efficient implementation using list comprehension.

src/duplo_resource/job.py [32-37]

 if fsct > 0:
   d = "\n".join([f["Description"] for f in fs])
   raise DuploFailedResource(f"Pod {pod['InstanceId']} raised {fsct} faults.\n{d}")
-# for f in faults:
-#   if f["Resource"].get("Name", None) == pod["InstanceId"]:
-#     raise DuploFailedResource(f"Pod {pod['InstanceId']} raised a fault.\n{f['Description']}")

Apply this suggestion

Suggestion 3:

Remove duplicate function definition

This function redefines the get_test_data function that's already imported at the top of the file. This duplication can lead to confusion about which function is being used. Remove this duplicate definition and use the imported function instead.

src/tests/test_cloudfront.py [23-27]

-def get_test_data(name) -> dict:
-  dir = pathlib.Path(__file__).parent.resolve()
-  f = f"{dir}/data/{name}.yaml"
-  with open(f, 'r') as stream:
-    return yaml.safe_load(stream)
+# Remove this duplicate function definition entirely, as it's already imported
+# from .conftest import get_test_data

Suggestion 4:

Remove duplicated code block

Remove the duplicated code block that checks for and assigns IpcMode to the result dictionary.

src/duplo_resource/ecs_service.py [297-300]

-if "IpcMode" in task_def:
-  result["IpcMode"] = task_def["IpcMode"]
 if "IpcMode" in task_def:
   result["IpcMode"] = task_def["IpcMode"]

Pattern 3: Ensure tests and API calls use consistent resource naming and parameter validation; validate mutually exclusive inputs explicitly and use the same normalized identifier across create/find/update/delete operations.

Example code before:

# Inconsistent name usage across tests
r.create({"FriendlyName": f"acme-{tenant}-svc"})
r.delete("svc")  # missing prefix

# Ambiguous parameter validation
if [image, container_image, init_image].count(None) != 2:
    raise Error("invalid")

Example code after:

full_name = ensure_prefix("svc")
r.create({"FriendlyName": full_name})
r.delete(full_name)

if not any([image, container_image, init_image]):
    raise Error("Provide one of image, container_image, or init_image.")
if sum(x is not None for x in [image, container_image, init_image]) > 1:
    raise Error("Provide only one of image, container_image, or init_image.")

Relevant past accepted suggestions:

Suggestion 1:

Fix inconsistent resource naming

**kwargs):

"""Helper function to execute a test and handle errors."""
try:
```
   return func(*args, **kwargs)
```
except DuploError as e:
```
   pytest.fail(f"Test failed: {e}")
```

class TestAsg:

 @pytest.mark.integration
 @pytest.mark.dependency(name="create_asg", scope="session")
 @pytest.mark.order(1)

def test_create_asg(self, duplo):
```
   r = duplo.load("asg")
```

def test_create_asg(self, asg_resource):

   r, asg_name = asg_resource
   body = {

```
       "FriendlyName": "duploctl",
```

       "FriendlyName": asg_name,
       "Zone": 1,
       "IsEbsOptimized": False,
       "DesiredCapacity": 1,

@@ -31,72 +46,46 @@ "IsClusterAutoscaled": True, "IsMinion": True }

```
   try:
```
```
       r.create(body, wait=True)
```
```
   except DuploError as e:
```

       pytest.fail(f"Failed to create ASG: {e}")

```
   execute_test(r.create, body, wait=True)
   time.sleep(60)
```
@pytest.mark.integration @pytest.mark.dependency(depends=["create_asg"], scope="session") @pytest.mark.order(2)

def test_find_asg(self, duplo):
```
   r = duplo.load("asg")
```
```
   tenant = r.tenant["AccountName"]
```
```
   try:
```

       asg = r.find(f"duploservices-{tenant}-duploctl")

```
   except DuploError as e:
```

       pytest.fail(f"Failed to find ASG: {e}")

   assert asg["FriendlyName"] == f"duploservices-{tenant}-duploctl"

def test_find_asg(self, asg_resource):
```
   r, asg_name = asg_resource
```

   asg = execute_test(r.find, asg_name)

```
   assert asg["FriendlyName"] == asg_name
```
@pytest.mark.integration @pytest.mark.dependency(depends=["create_asg"], scope="session") @pytest.mark.order(3)

def test_update_asg(self, duplo):
```
   r = duplo.load("asg")
```
```
   tenant = r.tenant["AccountName"]
```
```
   body = {
```

       "FriendlyName": f"duploservices-{tenant}-duploctl",

```
       "MinSize": 2,
```
```
       "MaxSize": 3
```
```
   }
```
```
   try:
```
```
       response = r.update(body)
```
```
   except DuploError as e:
```

       pytest.fail(f"Failed to update ASG: {e}")

def test_update_asg(self, asg_resource):
```
   r, asg_name = asg_resource
```

   body = {"FriendlyName": asg_name, "MinSize": 2, "MaxSize": 3}

```
   response = execute_test(r.update, body)
   assert "Successfully updated asg" in response["message"]
```
@pytest.mark.integration @pytest.mark.dependency(depends=["create_asg"], scope="session") @pytest.mark.order(4)

def test_list_asgs(self, duplo):
```
   r = duplo.load("asg")
```
```
   try:
```
```
       asgs = r.list()
```
```
   except DuploError as e:
```

       pytest.fail(f"Failed to list ASGs: {e}")

```
   assert isinstance(asgs, list)
```
```
   assert len(asgs) > 0
```

def test_list_asgs(self, asg_resource):
```
   r, _ = asg_resource
```
```
   asgs = execute_test(r.list)
```
```
   assert isinstance(asgs, list) and len(asgs) > 0
```
@pytest.mark.integration @pytest.mark.dependency(depends=["create_asg"], scope="session") @pytest.mark.order(5)

def test_scale_asg(self, duplo):
```
   r = duplo.load("asg")
```
```
   tenant = r.tenant["AccountName"]
```
```
   try:
```

       response = r.scale(f"duploservices-{tenant}-duploctl", min=1, max=2)

```
   except DuploError as e:
```

       pytest.fail(f"Failed to scale ASG: {e}")

def test_scale_asg(self, asg_resource):
```
   r, asg_name = asg_resource
```

   response = execute_test(r.scale, asg_name, min=1, max=2)
   assert "Successfully updated asg" in response["message"]

@pytest.mark.integration @pytest.mark.dependency(depends=["create_asg"], scope="session") @pytest.mark.order(6)

def test_delete_asg(self, duplo):
```
   r = duplo.load("asg")
```
```
   try:
```
```
       response = r.delete("duploctl")
```
```
   except DuploError as e:
```

       pytest.fail(f"Failed to delete ASG: {e}")

def test_delete_asg(self, asg_resource):
```
   r, _ = asg_resource
```

   response = execute_test(r.delete, "duploctl")
   assert "Successfully deleted asg" in response["message"]








**The delete method is using just "duploctl" as the ASG name, but other test methods use the full name with tenant prefix (e.g., "duploservices-{tenant}-duploctl"). This inconsistency could cause the test to fail if the API expects the full name.**

[src/tests/test_asg.py [93-102]](https://github.com/duplocloud/duploctl/pull/135/files#diff-a8011d44133b22fa8b9ed402ff058dc4469923ba774e2939b019502d43ae4a69R93-R102)

```diff
 @pytest.mark.integration
 @pytest.mark.dependency(depends=["create_asg"], scope="session")
 @pytest.mark.order(6)
 def test_delete_asg(self, duplo):
     r = duplo.load("asg")
+    tenant = r.tenant["AccountName"]
     try:
-        response = r.delete("duploctl")
+        response = r.delete(f"duploservices-{tenant}-duploctl")
     except DuploError as e:
         pytest.fail(f"Failed to delete ASG: {e}")
     assert "Successfully deleted asg" in response["message"]

Suggestion 2:

Fix parameter validation logic

The current condition is incorrect. It requires exactly one of the three parameters to be non-None, but doesn't properly handle the case when all three are None. Change the condition to explicitly check that at least one parameter is provided.

src/duplo_resource/service.py [249-250]

-if [image, container_image, init_container_image].count(None) != 2:
+if not any([image, container_image, init_container_image]):
   raise DuploError("Provide a service image, container images, or init container images.")
+if sum(x is not None for x in [image, container_image, init_container_image]) > 1:
+  raise DuploError("Provide only one of: service image, container images, or init container images.")

Suggestion 3:

Refactor duplicated prefix logic

The prefix logic is duplicated in both find() and delete() methods. This could lead to inconsistencies if the prefix format changes in one place but not the other. Extract this logic into a helper method to ensure consistent behavior.

src/duplo_resource/aws_secret.py [51-54]

-# if the name has the prefix we good, otherwise add it
-prefix = f"duploservices-{self.duplo.tenant}-"
-if not name.startswith(prefix):
-  name = prefix + name
+# Use a helper method to ensure consistent prefix handling
+name = self._ensure_name_has_prefix(name)

[Auto-generated best practices - 2025-11-28]

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Uh oh!

.pr_agent_auto_best_practices

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Clone this wiki locally