feat(credentials): add referenced credential element management (#7850)

pyoaev/contracts/contract_config.py

@@ -7,6 +7,7 @@
 from pyoaev import utils
 from pyoaev.contracts.contract_utils import ContractCardinality, ContractVariable
 from pyoaev.contracts.variable_helper import VariableHelper
+from pyoaev.credential.types import CredentialType
 
 
 class SupportedLanguage(str, Enum):
@@ -31,11 +32,13 @@ class ContractFieldType(str, Enum):
     AssetGroup: str = "asset-group"
     AiTarget: str = "ai-target"
     Payload: str = "payload"
+    CredentialReference: str = "credential-reference"
 
 
 class ContractFieldKey(str, Enum):
     Asset: str = "assets"
     AssetGroup: str = "asset_groups"
+    CredentialReference: str = "credential_reference"
 
 
 class ContractOutputType(str, Enum):
@@ -365,6 +368,19 @@ def get_type(self) -> str:
         return ContractFieldType.Attachment.value
 
 
+@dataclass
+class ContractReferencedCredential(ContractElement):
+    key: str = field(default=ContractFieldKey.CredentialReference.value, init=False)
+    label: str = "Select a credential reference"
+    mandatory: bool = True
+    credential_reference_type: Optional[CredentialType] = None
+    multiple: bool = True
+
+    @property
+    def get_type(self) -> str:
+        return ContractFieldType.CredentialReference.value
+
+
 @dataclass
 class ContractExpectations(ContractCardinalityElement):
     cardinality: str = ContractCardinality.Multiple

pyoaev/credential/__init__.py

@@ -0,0 +1,15 @@
+from .types import CredentialType
+
+
+def build_single_referenced_credential_element(provider_name: str):
+    from .utils import (
+        build_single_referenced_credential_element as _build_single_referenced_credential_element,
+    )
+
+    return _build_single_referenced_credential_element(provider_name)
+
+
+__all__ = [
+    "CredentialType",
+    "build_single_referenced_credential_element",
+]

pyoaev/credential/types.py

@@ -0,0 +1,16 @@
+"""Credential reference types shared with the OpenAEV platform.
+
+This enum is the Python mirror of the OpenAEV backend credential-type enum
+used by credential secret references. Values must stay label-for-label in sync
+with the platform so contracts can declare credential references without any
+translation layer.
+"""
+
+from enum import Enum
+
+
+class CredentialType(str, Enum):
+    IDENTITY = "IDENTITY"
+    CLOUD_AWS = "CLOUD_AWS"
+    CLOUD_AZURE = "CLOUD_AZURE"
+    CLOUD_GCP = "CLOUD_GCP"

pyoaev/credential/utils.py

@@ -0,0 +1,41 @@
+"""Helpers centralizing how injectors declare credential-reference fields.
+
+Injectors can call ``build_single_referenced_credential_element`` instead of
+re-implementing provider-to-credential-type mapping in each project. This keeps
+contract generation consistent across the Python ecosystem and aligns the
+serialized contract payload with the OpenAEV platform's expected values.
+"""
+
+from typing import Optional
+
+from pyoaev.contracts.contract_config import ContractReferencedCredential
+from pyoaev.credential.types import CredentialType
+
+_PROVIDER_TO_CREDENTIAL_TYPE = {
+    "aws": CredentialType.CLOUD_AWS,
+    "eks": CredentialType.CLOUD_AWS,
+    "azure": CredentialType.CLOUD_AZURE,
+    "gcp": CredentialType.CLOUD_GCP,
+}
+
+
+def _resolve_credential_type(provider_name: str) -> Optional[CredentialType]:
+    normalized_provider = provider_name.casefold()
+    return _PROVIDER_TO_CREDENTIAL_TYPE.get(normalized_provider)
+
+
+def build_single_referenced_credential_element(
+    provider_name: str,
+) -> ContractReferencedCredential:
+    """Build a credential-reference field for a provider-specific contract
+    with ``multiple`` value at ``False``.
+
+    This is the centralized entry point injectors should use when they need the
+    OpenAEV inject form to ask for one referenced credential. The helper keeps
+    provider-to-``CredentialType`` mapping consistent across injector projects.
+    """
+
+    return ContractReferencedCredential(
+        credential_reference_type=_resolve_credential_type(provider_name),
+        multiple=False,
+    )

test/contracts/test_contract_referenced_credential.py

@@ -0,0 +1,53 @@
+import json
+import unittest
+
+from pyoaev import utils
+from pyoaev.contracts.contract_config import (
+    ContractFieldKey,
+    ContractFieldType,
+    ContractReferencedCredential,
+)
+from pyoaev.credential.types import CredentialType
+
+
+def _serialize(field):
+    return json.loads(json.dumps(field, cls=utils.EnhancedJSONEncoder))
+
+
+class ContractReferencedCredentialTest(unittest.TestCase):
+    def test_defaults_match_platform_contract_expectations(self):
+        field = ContractReferencedCredential()
+        serialized = _serialize(field)
+
+        self.assertEqual(field.key, ContractFieldKey.CredentialReference.value)
+        self.assertEqual(field.label, "Select a credential reference")
+        self.assertTrue(field.mandatory)
+        self.assertTrue(field.multiple)
+        self.assertIsNone(field.credential_reference_type)
+        self.assertEqual(field.type, ContractFieldType.CredentialReference.value)
+        self.assertEqual(serialized["key"], "credential_reference")
+        self.assertEqual(serialized["type"], "credential-reference")
+        self.assertEqual(serialized["label"], "Select a credential reference")
+        self.assertTrue(serialized["mandatory"])
+        self.assertTrue(serialized["multiple"])
+        self.assertIsNone(serialized["credential_reference_type"])
+
+    def test_explicit_credential_type_serializes_to_platform_label(self):
+        field = ContractReferencedCredential(
+            credential_reference_type=CredentialType.CLOUD_AZURE
+        )
+
+        self.assertEqual(field.credential_reference_type, CredentialType.CLOUD_AZURE)
+        self.assertEqual(_serialize(field)["credential_reference_type"], "CLOUD_AZURE")
+
+    def test_identity_credential_type_serializes_to_platform_label(self):
+        field = ContractReferencedCredential(
+            credential_reference_type=CredentialType.IDENTITY
+        )
+
+        self.assertEqual(field.credential_reference_type, CredentialType.IDENTITY)
+        self.assertEqual(_serialize(field)["credential_reference_type"], "IDENTITY")
+
+
+if __name__ == "__main__":
+    unittest.main()

test/credential/test_credential_types.py

@@ -0,0 +1,31 @@
+import unittest
+
+from pyoaev.credential import CredentialType as PublicCredentialType
+from pyoaev.credential.types import CredentialType
+
+
+class CredentialTypeTest(unittest.TestCase):
+    def test_every_value_matches_the_supported_platform_labels(self):
+        expected_labels = {
+            "IDENTITY",
+            "CLOUD_AWS",
+            "CLOUD_AZURE",
+            "CLOUD_GCP",
+        }
+        actual_labels = {member.value for member in CredentialType}
+        self.assertEqual(actual_labels, expected_labels)
+
+    def test_identity_wire_label(self):
+        self.assertEqual(CredentialType.IDENTITY.value, "IDENTITY")
+        self.assertEqual(CredentialType.IDENTITY, "IDENTITY")
+
+    def test_package_re_exports_credential_type(self):
+        self.assertIs(PublicCredentialType, CredentialType)
+
+    def test_aws_wire_label(self):
+        self.assertEqual(CredentialType.CLOUD_AWS.value, "CLOUD_AWS")
+        self.assertEqual(CredentialType.CLOUD_AWS, "CLOUD_AWS")
+
+
+if __name__ == "__main__":
+    unittest.main()

test/credential/test_credential_utils.py

@@ -0,0 +1,60 @@
+import json
+import unittest
+
+from pyoaev import utils
+from pyoaev.contracts.contract_config import ContractReferencedCredential
+from pyoaev.credential import (
+    build_single_referenced_credential_element as public_build_single_referenced_credential_element,
+)
+from pyoaev.credential.types import CredentialType
+from pyoaev.credential.utils import build_single_referenced_credential_element
+
+
+def _serialize(field):
+    return json.loads(json.dumps(field, cls=utils.EnhancedJSONEncoder))
+
+
+class CredentialUtilsTest(unittest.TestCase):
+    def test_aws_provider_maps_to_aws_credential_reference(self):
+        field = build_single_referenced_credential_element("aws")
+
+        self.assertIsInstance(field, ContractReferencedCredential)
+        self.assertEqual(field.credential_reference_type, CredentialType.CLOUD_AWS)
+
+    def test_eks_provider_maps_to_aws_credential_reference(self):
+        field = build_single_referenced_credential_element("eks")
+
+        self.assertEqual(field.credential_reference_type, CredentialType.CLOUD_AWS)
+
+    def test_azure_provider_maps_to_azure_credential_reference(self):
+        field = build_single_referenced_credential_element("azure")
+
+        self.assertEqual(field.credential_reference_type, CredentialType.CLOUD_AZURE)
+
+    def test_gcp_provider_maps_to_gcp_credential_reference(self):
+        field = build_single_referenced_credential_element("gcp")
+
+        self.assertEqual(field.credential_reference_type, CredentialType.CLOUD_GCP)
+
+    def test_provider_mapping_is_case_insensitive(self):
+        field = build_single_referenced_credential_element("AWS")
+
+        self.assertEqual(field.credential_reference_type, CredentialType.CLOUD_AWS)
+
+    def test_package_public_helper_builds_the_same_field(self):
+        field = public_build_single_referenced_credential_element("azure")
+
+        self.assertIsInstance(field, ContractReferencedCredential)
+        self.assertEqual(field.credential_reference_type, CredentialType.CLOUD_AZURE)
+        self.assertEqual(_serialize(field)["credential_reference_type"], "CLOUD_AZURE")
+        self.assertFalse(field.multiple)
+
+    def test_unknown_provider_leaves_credential_type_empty(self):
+        field = build_single_referenced_credential_element("openstack")
+
+        self.assertIsNone(field.credential_reference_type)
+        self.assertIsNone(_serialize(field)["credential_reference_type"])
+
+
+if __name__ == "__main__":
+    unittest.main()