Add unit tests for JWT-based EKS detection in AwsEksResourceDetector

Author: alimx07

CHANGELOG.md

@@ -31,7 +31,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Fixed
 
 - `opentelemetry-sdk-extension-aws`: Fix `AwsEksResourceDetector` on clusters using the EKS Access Entries API mode where the `aws-auth` ConfigMap is absent; `_is_eks` now decodes the pod service-account JWT `iss` claim instead of querying the Kubernetes API.
-  ([#5080](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/5080))
+  ([#4414](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4414))
 - `opentelemetry-docker-tests`: Replace deprecated `SpanAttributes` from `opentelemetry.semconv.trace` with `opentelemetry.semconv._incubating.attributes`
  ([#4339](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/4339))
 - `opentelemetry-instrumentation-confluent-kafka`: Skip `recv` span creation when `poll()` returns no message or `consume()` returns an empty list, avoiding empty spans on idle polls

sdk-extension/opentelemetry-sdk-extension-aws/tests/resource/test_eks.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import base64
+import json
 import unittest
 from collections import OrderedDict
 from unittest.mock import mock_open, patch
@@ -25,6 +27,17 @@
     ResourceAttributes,
 )
 
+
+def _bearer_jwt(payload: dict) -> str:
+    header = base64.urlsafe_b64encode(b'{"alg":"RS256"}').rstrip(b"=").decode()
+    body = (
+        base64.urlsafe_b64encode(json.dumps(payload).encode())
+        .rstrip(b"=")
+        .decode()
+    )
+    return f"Bearer {header}.{body}.fakesig"
+
+
 MockEksResourceAttributes = {
     ResourceAttributes.CLOUD_PROVIDER: CloudProviderValues.AWS.value,
     ResourceAttributes.CLOUD_PLATFORM: CloudPlatformValues.AWS_EKS.value,
@@ -138,3 +151,52 @@ def test_if_no_eks_paths_should_not_raise(
             AwsEksResourceDetector(raise_on_error=True).detect()
         except RuntimeError:
             self.fail("Should not raise")
+
+    @patch(
+        "opentelemetry.sdk.extension.aws.resource.eks._get_k8s_cred_value",
+        return_value=_bearer_jwt(
+            {"iss": "https://oidc.eks.eu-west-2.amazonaws.com/id/EXAMPLE123"}
+        ),
+    )
+    @patch(
+        "opentelemetry.sdk.extension.aws.resource.eks._is_k8s",
+        return_value=True,
+    )
+    @patch(
+        "opentelemetry.sdk.extension.aws.resource.eks._get_cluster_info",
+        return_value=f"""{{
+  "data": {{
+    "cluster.name": "{MockEksResourceAttributes[ResourceAttributes.K8S_CLUSTER_NAME]}"
+  }}
+}}
+""",
+    )
+    @patch(
+        "builtins.open",
+        new_callable=mock_open,
+        read_data=f"14:name=systemd:/docker/{MockEksResourceAttributes[ResourceAttributes.CONTAINER_ID]}\n",
+    )
+    def test_eks_oidc_jwt_detected(
+        self,
+        mock_open_function,
+        mock_get_cluster_info,
+        mock_is_k8s,
+        mock_get_k8s_cred_value,
+    ):
+        actual = AwsEksResourceDetector().detect()
+        self.assertEqual(
+            actual.attributes.get(ResourceAttributes.CLOUD_PLATFORM),
+            CloudPlatformValues.AWS_EKS.value,
+        )
+
+    @patch(
+        "opentelemetry.sdk.extension.aws.resource.eks._get_k8s_cred_value",
+        return_value=_bearer_jwt({"iss": "https://accounts.google.com"}),
+    )
+    @patch(
+        "opentelemetry.sdk.extension.aws.resource.eks._is_k8s",
+        return_value=True,
+    )
+    def test_non_eks_jwt_returns_empty(self, mock_is_k8s, mock_get_k8s_cred_value):
+        actual = AwsEksResourceDetector().detect()
+        self.assertEqual(actual.attributes, {})