Merge pull request #306 from j3-signalroom/305-add-create_kafka_api_key_pair-and-delete_kafka_api_key-and-unit-tests

Resolved #305.

Author: j3-signalroom

CHANGELOG.md

@@ -8,6 +8,7 @@ The format is base on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 ### Added
 - Issue [#297](https://github.com/j3-signalroom/cc-clients-lib/issues/297)
 - Issue [#299](https://github.com/j3-signalroom/cc-clients-lib/issues/299)
+- Issue [#305](https://github.com/j3-signalroom/cc-clients-lib/issues/305)
 
 ### Changed
 - Issue [#301](https://github.com/j3-signalroom/cc-clients-lib/issues/301)

CHANGELOG.pdf

@@ -105,6 +105,8 @@ Metric Type|Description
 The **Environment Client** provides the following methods:
 - `get_environment_list`
 - `get_kafka_cluster_list`
+- `create_kafka_api_key`
+- `delete_kafka_api_key`
 
 ## **2.0 Unit Tests**
 The library includes unit tests for each client. The tests are located in the `tests` directory.  To use them, you must clone the repo locally:
@@ -257,6 +259,8 @@ Unit Test|Command
 -|-
 Get list of all the Environments|`uv run pytest -s tests/test_environment_client.py::test_get_environment_list`
 Get list of the all the Kafka clusters|`uv run pytest -s tests/test_environment_client.py::test_get_kafka_cluster_list`
+Create a Kafka API Key|`uv run pytest -s tests/test_environment_client.py::test_create_kafka_api_key`
+Delete a Kafka API Key|`uv run pytest -s tests/test_environment_client.py::test_delete_kafka_api_key`
 
 Otherwise, to run all the tests, use the following command:
 ```shell

README.md

@@ -30,6 +30,67 @@ def __init__(self, environment_config: dict):
         self.environment_id = environment_config[ENVIRONMENT_CONFIG["environment_id"]]
         self.base_url = "https://api.confluent.cloud"
 
+    def create_kafka_api_key(self, kafka_cluster_id: str, principal_id: str) -> Tuple[int, str, Dict]:
+        """This function submits a RESTful API call to create a Kafka API key pair.
+        Reference: https://docs.confluent.io/cloud/current/api.html#tag/API-Keys-(iamv2)/operation/createIamV2ApiKey
+
+        Arg(s):
+            kafka_cluster_id (str):  The Kafka cluster ID.
+            principal_id (str): The principal ID for the Kafka API key.
+
+        Return(s):
+            Tuple[int, str, Dict]: A tuple of the HTTP status code, the error message (if any), and the Kafka API key pair.
+        """
+        payload = {
+            "spec": {
+                "display_name": "Kafka Cluster API Key",
+                "description": "API key for Kafka cluster operations",
+                "owner": {
+                    "id": principal_id
+                },
+                "resource": {
+                    "id": kafka_cluster_id
+                }
+            }
+        }
+
+        response = requests.post(url=f"{self.base_url}/iam/v2/api-keys",
+                                 auth=HTTPBasicAuth(self.confluent_cloud_api_key, self.confluent_cloud_api_secret),
+                                 json=payload)
+        
+        try:
+            # Raise HTTPError, if occurred.
+            response.raise_for_status()
+
+            api_key_pair = {}
+            api_key_pair["key"] = response.json().get("id")
+            api_key_pair["secret"] = response.json().get("spec").get("secret")
+
+            return response.status_code, "", api_key_pair
+        except requests.exceptions.RequestException as e:
+            return response.status_code, f"Fail to create the Kafka API key pair because {e}.  The error details are: {response.json() if response.content else {}}", response.json() if response.content else {}
+
+    def delete_kafka_api_key(self, api_key: str) -> Tuple[int, str]:
+        """This function submits a RESTful API call to delete a Kafka API key pair.
+        Reference: https://docs.confluent.io/cloud/current/api.html#tag/API-Keys-(iamv2)/operation/deleteIamV2ApiKey
+
+        Arg(s):
+            api_key (str):  The Kafka API key.
+
+        Return(s):
+            Tuple[int, str]: A tuple of the HTTP status code, and error message (if any).
+        """
+        response = requests.delete(url=f"{self.base_url}/iam/v2/api-keys/{api_key}",
+                                   auth=HTTPBasicAuth(self.confluent_cloud_api_key, self.confluent_cloud_api_secret))
+        
+        try:
+            # Raise HTTPError, if occurred.
+            response.raise_for_status()
+
+            return response.status_code, ""
+        except requests.exceptions.RequestException as e:
+            return response.status_code, f"Fail to delete the Kafka API key pair because {e}.  The error details are: {response.json() if response.content else {}}"
+        
     def get_environment_list(self, page_size: int = DEFAULT_PAGE_SIZE) -> Tuple[int, str, Dict]:
         """This function submits a RESTful API call to get a list of environments.
         Reference: https://docs.confluent.io/cloud/current/api.html#tag/Environments-(orgv2)/operation/listOrgV2Environments

README.pdf

@@ -1,5 +1,6 @@
 import json
 import logging
+import time
 from dotenv import load_dotenv
 import os
 import pytest
@@ -21,6 +22,8 @@
 
 # Initialize the global variables.
 environment_config = {}
+kafka_cluster_id = ""
+principal_id = ""
 
 
 @pytest.fixture(autouse=True)
@@ -34,6 +37,62 @@ def load_configurations():
     environment_config[ENVIRONMENT_CONFIG["confluent_cloud_api_secret"]] = os.getenv("CONFLUENT_CLOUD_API_SECRET")
     environment_config[ENVIRONMENT_CONFIG["environment_id"]] = os.getenv("ENVIRONMENT_ID")
 
+    global kafka_cluster_id
+    global principal_id
+
+    # Set the Kafka cluster ID and owner ID.
+    kafka_cluster_id = os.getenv("KAFKA_CLUSTER_ID")
+    principal_id = os.getenv("PRINCIPAL_ID")
+
+def test_create_kafka_api_key():
+    """Test the create_kafka_api_key() function."""
+
+    # Instantiate the EnvironmentClient class.
+    environment_client = EnvironmentClient(environment_config=environment_config)
+
+    http_status_code, error_message, api_key_pair = environment_client.create_kafka_api_key(kafka_cluster_id=kafka_cluster_id, principal_id=principal_id)
+ 
+    try:
+        assert http_status_code == HttpStatus.ACCEPTED, f"HTTP Status Code: {http_status_code}"
+
+        beautified = json.dumps(api_key_pair, indent=4, sort_keys=True)
+        logger.info(f"Kafka API Key Pair: {beautified}")
+    except AssertionError as e:
+        logger.error(e)
+        logger.error("HTTP Status Code: %d, Error Message: %s, Kafka API Key Pair: %s", http_status_code, error_message, api_key_pair)
+        return
+    
+def test_delete_kafka_api_key():
+    """Test the delete_kafka_api_key() function."""
+
+    # Instantiate the EnvironmentClient class.
+    environment_client = EnvironmentClient(environment_config=environment_config)
+
+    http_status_code, error_message, api_key_pair = environment_client.create_kafka_api_key(kafka_cluster_id=kafka_cluster_id, principal_id=principal_id)
+ 
+    try:
+        assert http_status_code == HttpStatus.ACCEPTED, f"HTTP Status Code: {http_status_code}"
+
+        beautified = json.dumps(api_key_pair, indent=4, sort_keys=True)
+        logger.info(f"Kafka API Key Pair: {beautified}")
+    except AssertionError as e:
+        logger.error(e)
+        logger.error("HTTP Status Code: %d, Error Message: %s, Kafka API Key Pair: %s", http_status_code, error_message, api_key_pair)
+        return
+
+    time.sleep(10)  # Wait for 10 seconds before deleting the API key.
+
+    http_status_code, error_message = environment_client.delete_kafka_api_key(api_key=api_key_pair["key"])
+ 
+    try:
+        assert http_status_code == HttpStatus.NO_CONTENT, f"HTTP Status Code: {http_status_code}"
+
+        logger.info(f"Successfully deleted Kafka API Key: {api_key_pair['key']}")
+    except AssertionError as e:
+        logger.error(e)
+        logger.error("HTTP Status Code: %d, Error Message: %s", http_status_code, error_message)
+        return  
+    
 def test_get_environment_list():
     """Test the get_environment_list() function."""