refactor: centralize credential and param extraction for eks workload list#1089
Conversation
Greptile SummaryThis PR centralizes repeated credential and parameter extraction across
Confidence Score: 4/5Not safe to merge as-is — cluster name filtering is silently broken for One P1 logic bug:
Important Files Changed
Flowchart%%{init: {'theme': 'neutral'}}%%
flowchart TD
A[sources dict] --> B[extract_workload_params]
B --> C{eks key present?}
C -- No --> D[raise ValueError]
C -- Yes --> E[Extract cluster_name, namespace, eks_backend]
E --> F[Extract creds via _eks_creds\nrole_arn, external_id, region, credentials]
F --> G[Return merged params dict]
G --> H[list_eks_pods]
G --> I[list_eks_deployments]
G --> J[list_eks_clusters ⚠️]
J --> K[cluster_names always None\nfiltering never applied]
style J fill:#ff9999
style K fill:#ff9999
|
![greptile-apps[bot]](https://avatars.githubusercontent.com/in/867647?s=60&v=4){kind=small}
| def extract_workload_params(sources: dict[str, dict]) -> dict[str, Any]: | ||
| """Extract common parameters for workload list operations (pods/deployments)""" | ||
|
|
||
| eks = sources.get("eks") | ||
| if eks is None: | ||
| raise ValueError("Sources dictionary must contain an 'eks' key with cluster configuration") | ||
|
|
||
| return { | ||
| "cluster_name": eks.get("cluster_name", ""), | ||
| "namespace": eks.get("namespace") or "all", | ||
| "eks_backend": eks.get("_backend"), | ||
| **_eks_creds(eks), | ||
| } |
There was a problem hiding this comment.
cluster_names filtering silently dropped for EKSListClustersTool
extract_workload_params is now used as extract_params for all three tools including EKSListClustersTool, but it never extracts cluster_names (plural). The old _list_clusters_extract_params returned cluster_names: eks.get("cluster_names", []), which list_eks_clusters uses to filter results (if cluster_names: clusters = [c for c in clusters if c in cluster_names]). With this helper, that parameter is never populated, so the filter is always skipped and all clusters are returned regardless of any configured cluster_names in the EKS source.
extract_workload_params is correctly shared for pods and deployments, but EKSListClustersTool needs its own extractor (or the helper needs a cluster_names variant).
| def test_missing_eks_raises_error(): | ||
| """Test ValueError when 'eks' key is missing""" | ||
| sources = {"other": {}} | ||
|
|
||
| try: | ||
| extract_workload_params(sources) | ||
| raise AssertionError("Should have raised ValueError") | ||
| except ValueError as e: | ||
| assert "must contain an 'eks' key" in str(e) |
There was a problem hiding this comment.
pytest.raises instead of manual try/except
The manual try/except/AssertionError pattern works but is non-idiomatic pytest. Prefer pytest.raises for exception assertions.
| def test_missing_eks_raises_error(): | |
| """Test ValueError when 'eks' key is missing""" | |
| sources = {"other": {}} | |
| try: | |
| extract_workload_params(sources) | |
| raise AssertionError("Should have raised ValueError") | |
| except ValueError as e: | |
| assert "must contain an 'eks' key" in str(e) | |
| def test_missing_eks_raises_error(): | |
| """Test ValueError when 'eks' key is missing""" | |
| sources = {"other": {}} | |
| with pytest.raises(ValueError, match="must contain an 'eks' key"): | |
| extract_workload_params(sources) |
Don't forget to add import pytest at the top of the file.
Context Used: Testing conventions and patterns (source)
Note: If this suggestion doesn't match your team's coding style, reply to this and let me know. I'll remember it for next time!
VaibhavUpreti
left a comment
VaibhavUpreti
left a comment
There was a problem hiding this comment.
Great job @cokerrd !! Welcome to the OpenSRE community
|
🔥 Another one. @cokerrd said "here's a PR" and maintainers said "ship it". That's how it's done.
👋 Join us on Discord - OpenSRE : hang out, contribute, or hunt for features and issues. Everyone's welcome. |
2 participants
Fixes #
#895
Describe the changes you have made in this PR -
app/tools/utils/eks_workload_helper.pythat centralizes the repeated credential and parameter extraction.Code Understanding and AI Usage
Did you use AI assistance (ChatGPT, Claude, Copilot, etc.) to write any part of this code?
If you used AI assistance:
Explain your implementation approach:
Checklist before requesting a review
Note: Please check Allow edits from maintainers if you would like us to assist in the PR.