diff --git a/plugins/modules/df_customflow_version.py b/plugins/modules/df_customflow_version.py index 3d70c8b8..42c6f82b 100644 --- a/plugins/modules/df_customflow_version.py +++ b/plugins/modules/df_customflow_version.py @@ -1,7 +1,7 @@ #!/usr/bin/python # -*- coding: utf-8 -*- -# Copyright 2025 Cloudera, Inc. All Rights Reserved. +# Copyright 2026 Cloudera, Inc. All Rights Reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. @@ -17,50 +17,92 @@ DOCUMENTATION = r""" module: df_customflow_version -short_description: Import CustomFlow versions into the DataFlow Catalog +short_description: Import a new version into an existing CustomFlow in the DataFlow Catalog description: - - Import CustomFlow versions into the DataFlow Catalog + - Import a new version into an existing CustomFlow in the DataFlow Catalog author: - "Andre Araujo (@asdaraujo)" + - "Ronald Suplina (@rsuplina)" version_added: "2.0.0" -requirements: - - cdpy options: flow_crn: description: - - The name of the CustomFlow into which the version will be imported. + - The CRN of the existing CustomFlow into which the new version will be imported. type: str required: True file: description: - - The JSON file containing the CustomFlow definition to be imported as a new version. + - The path to the JSON file containing the CustomFlow definition to be imported as a new version. + - Mutually exclusive with O(content). + type: path + default: None + content: + description: + - The CustomFlow definition content as a string (JSON format) to be imported as a new version. + - Mutually exclusive with O(file). type: str - required: True + default: None comments: description: - - Comments associated to the version of the CustomFlow being imported. + - Comments associated to the new version of the CustomFlow being imported. type: str default: None required: False + tags: + description: + - The list of tags for the new flow definition version. + - Each tag should have a O(tags[].tag_name) (required) and optionally a O(tags[].tag_color). + type: list + elements: dict + required: False + suboptions: + tag_name: + description: + - The name of the version tag. + type: str + required: True + tag_color: + description: + - The color of the version tag. + type: str + required: False state: description: - - The declarative state of the CustomerFlow version + - The declarative state of the CustomFlow version. type: str required: False default: present choices: - present extends_documentation_fragment: - - cloudera.cloud.cdp_sdk_options - - cloudera.cloud.cdp_auth_options + - cloudera.cloud.cdp_client """ EXAMPLES = r""" -# Import a CustomFlow version into the DataFlow Catalog +# Note: These examples do not set authentication details. + +# Import a new CustomFlow version from a file - cloudera.cloud.df_customflow_version: - name: my-customflow-version-name + flow_crn: crn:cdp:df:us-west-1:tenant:flow:flow-123 file: /tmp/my-custom-flow-v2.json comments: Second version + +# Import a new CustomFlow version with content from a template/lookup +- cloudera.cloud.df_customflow_version: + flow_crn: crn:cdp:df:us-west-1:tenant:flow:flow-123 + content: "{{ lookup('file', 'my-flow-v2.json') }}" + comments: Second version from content + +# Import a new CustomFlow version with tags +- cloudera.cloud.df_customflow_version: + flow_crn: crn:cdp:df:us-west-1:tenant:flow:flow-123 + file: /tmp/my-custom-flow-v3.json + comments: Third version with tags + tags: + - tag_name: production + tag_color: blue + - tag_name: stable + tag_color: green """ RETURN = r""" @@ -93,62 +135,121 @@ description: The number of deployments of the artifact. returned: always type: int +sdk_out: + description: Returns the captured CDP SDK log. + returned: when supported + type: str +sdk_out_lines: + description: Returns a list of each line of the captured CDP SDK log. + returned: when supported + type: list + elements: str """ -from ansible.module_utils.basic import AnsibleModule -from ansible_collections.cloudera.cloud.plugins.module_utils.cdp_common import CdpModule +from typing import Optional + +from ansible_collections.cloudera.cloud.plugins.module_utils.common import ( + ServicesModule, +) +from ansible_collections.cloudera.cloud.plugins.module_utils.cdp_df import ( + CdpDfClient, + DataFlowModule, + format_tags_for_api, +) + +class DFCustomFlowVersion(DataFlowModule, ServicesModule): + def __init__(self): + super().__init__( + argument_spec=dict( + flow_crn=dict(required=True, type="str"), + file=dict(required=False, type="path"), + content=dict(required=False, type="str"), + comments=dict(required=False, type="str"), + tags=dict( + required=False, + type="list", + elements="dict", + options=dict( + tag_name=dict(required=True, type="str"), + tag_color=dict(required=False, type="str"), + ), + ), + state=dict( + type="str", + choices=["present"], + default="present", + ), + ), + mutually_exclusive=[ + ("file", "content"), + ], + required_if=[ + ("state", "present", ("file", "content"), True), + ], + supports_check_mode=True, + ) -class DFCustomFlowVersion(CdpModule): - def __init__(self, module): - super(DFCustomFlowVersion, self).__init__(module) + # Initialize parameters + self.flow_crn: str = self.get_param("flow_crn") + self.file: Optional[str] = self.get_param("file") + self.content: Optional[str] = self.get_param("content") + self.comments: Optional[str] = self.get_param("comments") + self.tags: Optional[list] = self.get_param("tags") + self.state: str = self.get_param("state") - # Set variables - self.flow_crn = self._get_param("flow_crn") - self.file = self._get_param("file") - self.comments = self._get_param("comments") - self.state = self._get_param("state") + # Initialize the DataFlow client + self.df_client = CdpDfClient(self.api_client) # Initialize return values - self.flow_version = None + self.flow_version = {} self.changed = False - # Execute logic process - self.process() - - @CdpModule._Decorators.process_debug def process(self): - flow = self.cdpy.df.describe_customflow(self.flow_crn) - if not flow: + existing_flow = self.df_client.get_flow_by_crn(self.flow_crn) + + if not existing_flow: self.module.fail_json( - msg="Flow definition with crn {} does not exist".format(self.flow_crn), + msg=f"Flow definition with CRN '{self.flow_crn}' does not exist", + ) + + # Only possible state is "present" - always creates a new version + self.changed = True + if not self.module.check_mode: + file_content = None + if self.file: + try: + with open(self.file, "r") as f: + file_content = f.read() + except Exception as e: + self.module.fail_json( + msg=f"Failed to read file '{self.file}': {str(e)}", + ) + elif self.content: + file_content = self.content + + api_tags = format_tags_for_api(self.tags) + + self.flow_version = self.df_client.import_flow_definition_version( + flow_crn=self.flow_crn, + file_content=file_content, + comments=self.comments, + tags=api_tags, ) - else: - # Only possible state is "present" - self.changed = True - if not self.module.check_mode: - self.flow_version = self.cdpy.df.import_customflow_version( - self.flow_crn, - self.file, - self.comments, - ) def main(): - module = AnsibleModule( - argument_spec=CdpModule.argument_spec( - flow_crn=dict(required=True, type="str"), - file=dict(required=True, type="str"), - comments=dict(required=False, type="str"), - state=dict(type="str", choices=["present"], default="present"), - ), - supports_check_mode=True, + result = DFCustomFlowVersion() + + output = dict( + changed=result.changed, + customflow_version=result.flow_version, ) - result = DFCustomFlowVersion(module) - output = dict(changed=result.changed, customflow_version=result.flow_version) + if result.debug_log: + output.update(sdk_out=result.log_out, sdk_out_lines=result.log_lines) - module.exit_json(**output) + result.module.exit_json(**output) if __name__ == "__main__": diff --git a/tests/unit/plugins/module_utils/cdp_df/test_df_customflow_version_api.py b/tests/unit/plugins/module_utils/cdp_df/test_df_customflow_version_api.py new file mode 100644 index 00000000..1c0f812c --- /dev/null +++ b/tests/unit/plugins/module_utils/cdp_df/test_df_customflow_version_api.py @@ -0,0 +1,537 @@ +# -*- coding: utf-8 -*- + +# Copyright 2026 Cloudera, Inc. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import pytest + +from ansible_collections.cloudera.cloud.plugins.module_utils.cdp_client import ( + CdpClient, +) +from ansible_collections.cloudera.cloud.plugins.module_utils.cdp_df import ( + CdpDfClient, + format_tags_for_api, +) + + +FLOW_CRN = "crn:cdp:df:us-west-1:tenant:flow:flow-123" +FLOW_NAME = "test-flow" +FLOW_VERSION_CRN = "crn:cdp:df:us-west-1:tenant:flow:flow-123/v1" +COLLECTION_CRN = "crn:cdp:df:us-west-1:tenant:collection:col-123" +SERVICE_CRN = "crn:cdp:df:us-west-1:tenant:service:svc-123" +SERVICE_NAME = "test-df-service" +ENV_CRN = "crn:cdp:environments:us-west-1:tenant:environment:env-123" + +FLOW_VERSION_SUMMARY = { + "crn": FLOW_VERSION_CRN, + "bucketIdentifier": "bucket-abc", + "author": "test-user", + "version": 1, + "timestamp": 1640000000000, + "deploymentCount": 0, + "comments": "Initial version", +} + +FLOW_DETAIL = { + "crn": FLOW_CRN, + "name": FLOW_NAME, + "versionCount": 1, + "createdTimestamp": 1640000000000, + "modifiedTimestamp": 1640000000000, + "description": "Test flow description", + "versions": [FLOW_VERSION_SUMMARY], +} + + +class TestFormatTagsForApi: + """Unit tests for the format_tags_for_api helper function.""" + + def test_none_input(self): + """Test that None input returns None.""" + assert format_tags_for_api(None) is None + + def test_empty_list(self): + """Test that empty list returns empty list.""" + assert format_tags_for_api([]) == [] + + def test_tags_with_color(self): + """Test conversion of tags with both name and color.""" + tags = [ + {"tag_name": "production", "tag_color": "blue"}, + {"tag_name": "stable", "tag_color": "green"}, + ] + result = format_tags_for_api(tags) + assert result == [ + {"tagName": "production", "tagColor": "blue"}, + {"tagName": "stable", "tagColor": "green"}, + ] + + def test_tags_without_color(self): + """Test that missing tag_color is omitted from the output.""" + tags = [{"tag_name": "production"}] + result = format_tags_for_api(tags) + assert result == [{"tagName": "production"}] + assert "tagColor" not in result[0] + + def test_mixed_tags(self): + """Test a mix of tags with and without color.""" + tags = [ + {"tag_name": "production", "tag_color": "blue"}, + {"tag_name": "stable"}, + ] + result = format_tags_for_api(tags) + assert result == [ + {"tagName": "production", "tagColor": "blue"}, + {"tagName": "stable"}, + ] + + +class TestCdpDfClientFlowDefinitions: + """Unit tests for CdpDfClient flow definition methods.""" + + def test_list_flow_definitions_default(self, mocker): + """Test listing all flow definitions with no filters.""" + mock_response = { + "flows": [ + {"crn": FLOW_CRN, "name": FLOW_NAME, "versionCount": 1}, + {"crn": "crn:other", "name": "other-flow", "versionCount": 2}, + ], + } + + api_client = mocker.create_autospec(CdpClient, instance=True) + api_client.post.return_value = mock_response + + client = CdpDfClient(api_client=api_client) + response = client.list_flow_definitions() + + assert "flows" in response + assert len(response["flows"]) == 2 + assert response["flows"][0]["name"] == FLOW_NAME + + api_client.post.assert_called_once_with( + "/api/v1/df/listFlowDefinitions", + data={"pageSize": 100}, + squelch={404: {"flows": []}}, + ) + + def test_list_flow_definitions_with_search(self, mocker): + """Test listing flow definitions filtered by search term.""" + mock_response = {"flows": [{"crn": FLOW_CRN, "name": FLOW_NAME}]} + + api_client = mocker.create_autospec(CdpClient, instance=True) + api_client.post.return_value = mock_response + + client = CdpDfClient(api_client=api_client) + client.list_flow_definitions(search_term=FLOW_NAME) + + call_data = api_client.post.call_args[1]["data"] + assert call_data["searchTerm"] == FLOW_NAME + + def test_list_flow_definitions_with_collection_crn(self, mocker): + """Test listing flow definitions filtered by collection CRN.""" + mock_response = {"flows": [{"crn": FLOW_CRN, "name": FLOW_NAME}]} + + api_client = mocker.create_autospec(CdpClient, instance=True) + api_client.post.return_value = mock_response + + client = CdpDfClient(api_client=api_client) + client.list_flow_definitions(collection_crn=COLLECTION_CRN) + + call_data = api_client.post.call_args[1]["data"] + assert call_data["collectionCrn"] == COLLECTION_CRN + + def test_list_flow_definitions_empty(self, mocker): + """Test listing flow definitions when none exist.""" + api_client = mocker.create_autospec(CdpClient, instance=True) + api_client.post.return_value = {"flows": []} + + client = CdpDfClient(api_client=api_client) + response = client.list_flow_definitions() + + assert "flows" in response + assert len(response["flows"]) == 0 + + def test_describe_flow(self, mocker): + """Test describing a flow definition by CRN.""" + mock_response = {"flow": {"flowDetail": FLOW_DETAIL}} + + api_client = mocker.create_autospec(CdpClient, instance=True) + api_client.post.return_value = mock_response + + client = CdpDfClient(api_client=api_client) + response = client.describe_flow(FLOW_CRN) + + assert response == mock_response + api_client.post.assert_called_once_with( + "/api/v1/df/describeFlow", + data={"flowCrn": FLOW_CRN}, + squelch={404: {}}, + ) + + def test_describe_flow_not_found(self, mocker): + """Test describing a flow that doesn't exist (404).""" + api_client = mocker.create_autospec(CdpClient, instance=True) + api_client.post.return_value = {} + + client = CdpDfClient(api_client=api_client) + response = client.describe_flow("nonexistent-crn") + + assert response == {} + + def test_get_flow_by_name(self, mocker): + """Test resolving a flow by name via list + describe.""" + list_mock = { + "flows": [ + {"crn": FLOW_CRN, "name": FLOW_NAME}, + {"crn": "crn:other", "name": "other-flow"}, + ], + } + describe_mock = {"flow": {"flowDetail": FLOW_DETAIL}} + + api_client = mocker.create_autospec(CdpClient, instance=True) + client = CdpDfClient(api_client=api_client) + + mocker.patch.object(client, "list_flow_definitions", return_value=list_mock) + mocker.patch.object(client, "describe_flow", return_value=describe_mock) + + result = client.get_flow_by_name(FLOW_NAME) + + assert result is not None + assert result["crn"] == FLOW_CRN + assert result["name"] == FLOW_NAME + + client.list_flow_definitions.assert_called_once_with(search_term=FLOW_NAME) + client.describe_flow.assert_called_once_with(FLOW_CRN) + + def test_get_flow_by_name_not_found(self, mocker): + """Test get_flow_by_name when no flow matches.""" + list_mock = {"flows": [{"crn": "crn:other", "name": "other-flow"}]} + + api_client = mocker.create_autospec(CdpClient, instance=True) + client = CdpDfClient(api_client=api_client) + mocker.patch.object(client, "list_flow_definitions", return_value=list_mock) + + result = client.get_flow_by_name("nonexistent-flow") + + assert result is None + client.list_flow_definitions.assert_called_once_with( + search_term="nonexistent-flow", + ) + + def test_get_flow_by_crn(self, mocker): + """Test resolving a flow by CRN via describe.""" + describe_mock = {"flow": {"flowDetail": FLOW_DETAIL}} + + api_client = mocker.create_autospec(CdpClient, instance=True) + client = CdpDfClient(api_client=api_client) + mocker.patch.object(client, "describe_flow", return_value=describe_mock) + + result = client.get_flow_by_crn(FLOW_CRN) + + assert result is not None + assert result["crn"] == FLOW_CRN + client.describe_flow.assert_called_once_with(FLOW_CRN) + + def test_get_flow_by_crn_not_found(self, mocker): + """Test get_flow_by_crn when the CRN doesn't exist.""" + api_client = mocker.create_autospec(CdpClient, instance=True) + client = CdpDfClient(api_client=api_client) + mocker.patch.object(client, "describe_flow", return_value={}) + + result = client.get_flow_by_crn("nonexistent-crn") + + assert result is None + + def test_import_flow_definition_minimal(self, mocker): + """Test importing a flow definition with only required parameters.""" + mock_response = FLOW_DETAIL + + api_client = mocker.create_autospec(CdpClient, instance=True) + api_client.post.return_value = mock_response + + client = CdpDfClient(api_client=api_client) + response = client.import_flow_definition( + name=FLOW_NAME, + file_content='{"flow": "content"}', + ) + + assert response == mock_response + + call_data = api_client.post.call_args[1]["data"] + assert call_data["name"] == FLOW_NAME + assert call_data["file"] == '{"flow": "content"}' + assert "description" not in call_data + assert "comments" not in call_data + assert "collectionCrn" not in call_data + assert "tags" not in call_data + + def test_import_flow_definition_all_params(self, mocker): + """Test importing a flow definition with all optional parameters.""" + mock_response = {**FLOW_DETAIL, "collectionCrn": COLLECTION_CRN} + + api_client = mocker.create_autospec(CdpClient, instance=True) + api_client.post.return_value = mock_response + + client = CdpDfClient(api_client=api_client) + response = client.import_flow_definition( + name=FLOW_NAME, + file_content='{"flow": "content"}', + description="Test description", + comments="Initial version", + collection_crn=COLLECTION_CRN, + tags=[{"tagName": "production", "tagColor": "blue"}], + ) + + assert response["collectionCrn"] == COLLECTION_CRN + + call_data = api_client.post.call_args[1]["data"] + assert call_data["description"] == "Test description" + assert call_data["comments"] == "Initial version" + assert call_data["collectionCrn"] == COLLECTION_CRN + assert call_data["tags"] == [{"tagName": "production", "tagColor": "blue"}] + + def test_import_flow_definition_version_minimal(self, mocker): + """Test importing a flow version with only required parameters.""" + mock_response = FLOW_VERSION_SUMMARY + + api_client = mocker.create_autospec(CdpClient, instance=True) + api_client.post.return_value = mock_response + + client = CdpDfClient(api_client=api_client) + response = client.import_flow_definition_version( + flow_crn=FLOW_CRN, + file_content='{"flow": "content"}', + ) + + assert response == mock_response + + call_data = api_client.post.call_args[1]["data"] + assert call_data["flowCrn"] == FLOW_CRN + assert call_data["file"] == '{"flow": "content"}' + assert "comments" not in call_data + assert "tags" not in call_data + + api_client.post.assert_called_once_with( + "/api/v1/df/importFlowDefinitionVersion", + data=call_data, + ) + + def test_import_flow_definition_version_with_comments_and_tags(self, mocker): + """Test importing a flow version with comments and tags.""" + mock_response = {**FLOW_VERSION_SUMMARY, "version": 2, "comments": "v2"} + + api_client = mocker.create_autospec(CdpClient, instance=True) + api_client.post.return_value = mock_response + + client = CdpDfClient(api_client=api_client) + response = client.import_flow_definition_version( + flow_crn=FLOW_CRN, + file_content='{"flow": "content"}', + comments="v2", + tags=[{"tagName": "stable"}], + ) + + assert response["version"] == 2 + assert response["comments"] == "v2" + + call_data = api_client.post.call_args[1]["data"] + assert call_data["comments"] == "v2" + assert call_data["tags"] == [{"tagName": "stable"}] + + def test_delete_flow(self, mocker): + """Test deleting a flow definition.""" + mock_response = {"flow": {"crn": FLOW_CRN, "name": FLOW_NAME}} + + api_client = mocker.create_autospec(CdpClient, instance=True) + api_client.post.return_value = mock_response + + client = CdpDfClient(api_client=api_client) + response = client.delete_flow(FLOW_CRN) + + assert response == mock_response + api_client.post.assert_called_once_with( + "/api/v1/df/deleteFlow", + data={"flowCrn": FLOW_CRN}, + ) + + +class TestCdpDfClientServices: + """Unit tests for CdpDfClient DataFlow service management methods.""" + + def test_list_services_default(self, mocker): + """Test listing all DataFlow services with no filters.""" + mock_response = { + "services": [ + { + "crn": SERVICE_CRN, + "name": SERVICE_NAME, + "status": {"state": "GOOD_HEALTH"}, + }, + ], + } + + api_client = mocker.create_autospec(CdpClient, instance=True) + api_client.post.return_value = mock_response + + client = CdpDfClient(api_client=api_client) + response = client.list_services() + + assert "services" in response + assert len(response["services"]) == 1 + assert response["services"][0]["crn"] == SERVICE_CRN + + api_client.post.assert_called_once_with( + "/api/v1/df/listServices", + data={"pageSize": 100}, + squelch={404: {"services": []}}, + ) + + def test_list_services_with_search(self, mocker): + """Test listing services filtered by search term.""" + mock_response = {"services": [{"crn": SERVICE_CRN, "name": SERVICE_NAME}]} + + api_client = mocker.create_autospec(CdpClient, instance=True) + api_client.post.return_value = mock_response + + client = CdpDfClient(api_client=api_client) + client.list_services(search_term=SERVICE_NAME) + + call_data = api_client.post.call_args[1]["data"] + assert call_data["searchTerm"] == SERVICE_NAME + + def test_list_services_empty(self, mocker): + """Test listing services when none exist.""" + api_client = mocker.create_autospec(CdpClient, instance=True) + api_client.post.return_value = {"services": []} + + client = CdpDfClient(api_client=api_client) + response = client.list_services() + + assert "services" in response + assert len(response["services"]) == 0 + + def test_describe_service(self, mocker): + """Test describing a DataFlow service by CRN.""" + mock_response = { + "service": { + "crn": SERVICE_CRN, + "name": SERVICE_NAME, + "environmentCrn": ENV_CRN, + "status": {"state": "GOOD_HEALTH"}, + }, + } + + api_client = mocker.create_autospec(CdpClient, instance=True) + api_client.post.return_value = mock_response + + client = CdpDfClient(api_client=api_client) + response = client.describe_service(SERVICE_CRN) + + assert response == mock_response + api_client.post.assert_called_once_with( + "/api/v1/df/describeService", + data={"serviceCrn": SERVICE_CRN}, + squelch={404: {}}, + ) + + def test_describe_service_not_found(self, mocker): + """Test describing a service that doesn't exist.""" + api_client = mocker.create_autospec(CdpClient, instance=True) + api_client.post.return_value = {} + + client = CdpDfClient(api_client=api_client) + response = client.describe_service("nonexistent-crn") + + assert response == {} + + def test_get_service_by_name(self, mocker): + """Test resolving a service by name via list + describe.""" + list_mock = { + "services": [ + { + "crn": SERVICE_CRN, + "name": SERVICE_NAME, + "status": {"state": "GOOD_HEALTH"}, + }, + { + "crn": "crn:other", + "name": "other-service", + "status": {"state": "GOOD_HEALTH"}, + }, + ], + } + describe_mock = { + "service": { + "crn": SERVICE_CRN, + "name": SERVICE_NAME, + "status": {"state": "GOOD_HEALTH"}, + }, + } + + api_client = mocker.create_autospec(CdpClient, instance=True) + client = CdpDfClient(api_client=api_client) + mocker.patch.object(client, "list_services", return_value=list_mock) + mocker.patch.object(client, "describe_service", return_value=describe_mock) + + result = client.get_service_by_name(SERVICE_NAME) + + assert result is not None + assert result["service"]["name"] == SERVICE_NAME + client.describe_service.assert_called_once_with(SERVICE_CRN) + + def test_get_service_by_name_not_found(self, mocker): + """Test get_service_by_name when no service matches.""" + list_mock = { + "services": [ + { + "crn": "crn:other", + "name": "other-service", + "status": {"state": "GOOD_HEALTH"}, + }, + ], + } + + api_client = mocker.create_autospec(CdpClient, instance=True) + client = CdpDfClient(api_client=api_client) + mocker.patch.object(client, "list_services", return_value=list_mock) + + result = client.get_service_by_name("nonexistent-service") + + assert result is None + + def test_get_service_by_name_skips_disabled(self, mocker): + """Test that get_service_by_name skips NOT_ENABLED services.""" + list_mock = { + "services": [ + { + "crn": SERVICE_CRN, + "name": SERVICE_NAME, + "status": {"state": "NOT_ENABLED"}, + }, + ], + } + + api_client = mocker.create_autospec(CdpClient, instance=True) + client = CdpDfClient(api_client=api_client) + mocker.patch.object(client, "list_services", return_value=list_mock) + describe_spy = mocker.patch.object(client, "describe_service") + + result = client.get_service_by_name(SERVICE_NAME) + + assert result is None + describe_spy.assert_not_called() diff --git a/tests/unit/plugins/module_utils/cdp_df/test_df_customflow_version_api_int.py b/tests/unit/plugins/module_utils/cdp_df/test_df_customflow_version_api_int.py new file mode 100644 index 00000000..0a0f741c --- /dev/null +++ b/tests/unit/plugins/module_utils/cdp_df/test_df_customflow_version_api_int.py @@ -0,0 +1,375 @@ +# -*- coding: utf-8 -*- + +# Copyright 2026 Cloudera, Inc. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import json +import pytest +import uuid +from typing import Generator + +from ansible_collections.cloudera.cloud.plugins.module_utils.cdp_df import CdpDfClient + +# Required environment variables for integration tests +REQUIRED_ENV_VARS = [ + "CDP_API_ENDPOINT", + "CDP_ACCESS_KEY_ID", + "CDP_PRIVATE_KEY", +] + +# Mark all tests in this module as integration tests requiring API credentials +pytestmark = pytest.mark.integration_api + + +@pytest.fixture +def df_client(test_cdp_client) -> CdpDfClient: + """Fixture to provide a DataFlow client for tests.""" + return CdpDfClient(api_client=test_cdp_client) + + +def _create_minimal_flow_definition(flow_name: str) -> str: + """Return a minimal NiFi flow definition as a JSON string.""" + return json.dumps( + { + "snapshotMetadata": { + "bucketIdentifier": None, + "flowIdentifier": str(uuid.uuid4()), + "version": 0, + "timestamp": 1771317050573, + "author": None, + "comments": None, + "link": None, + }, + "flowContents": { + "identifier": str(uuid.uuid4()), + "instanceIdentifier": None, + "name": flow_name, + "comments": None, + "position": None, + "processGroups": [], + "remoteProcessGroups": [], + "processors": [], + "inputPorts": [], + "outputPorts": [], + "connections": [], + "labels": [], + "funnels": [], + "controllerServices": [], + "versionedFlowCoordinates": None, + "parameterContextName": flow_name, + "defaultFlowFileExpiration": "0 sec", + "defaultBackPressureObjectThreshold": 10000, + "defaultBackPressureDataSizeThreshold": "1 GB", + "scheduledState": None, + "executionEngine": None, + "maxConcurrentTasks": None, + "statelessFlowTimeout": None, + "logFileSuffix": None, + "componentType": "PROCESS_GROUP", + "flowFileConcurrency": "UNBOUNDED", + "flowFileOutboundPolicy": "STREAM_WHEN_AVAILABLE", + "groupIdentifier": None, + }, + "externalControllerServices": None, + "parameterProviders": None, + "parameterContexts": { + flow_name: { + "identifier": str(uuid.uuid4()), + "instanceIdentifier": None, + "name": flow_name, + "comments": None, + "position": None, + "parameters": [], + "inheritedParameterContexts": [], + "description": None, + "parameterProvider": None, + "parameterGroupName": None, + "synchronized": None, + "componentType": "PARAMETER_CONTEXT", + "groupIdentifier": None, + }, + }, + "flowEncodingVersion": None, + "flow": None, + "bucket": None, + }, + ) + + +@pytest.fixture +def valid_df_flow(df_client) -> Generator[dict, None, None]: + """ + Fixture to create a temporary flow for testing and clean it up afterwards. + + Yields the created flow dict. The flow is deleted after the test regardless + of outcome. + """ + flow_name = f"test-df-api-{uuid.uuid4().hex[:8]}" + flow_content = _create_minimal_flow_definition(flow_name) + + flow = df_client.import_flow_definition( + name=flow_name, + file_content=flow_content, + description=f"Integration test flow - {flow_name}", + comments="Test version", + ) + + yield flow + + if flow and flow.get("crn"): + try: + df_client.delete_flow(flow_crn=flow["crn"]) + except Exception: + pass + + +@pytest.fixture +def valid_df_service(df_client): + """ + Fixture to find an active DataFlow service for testing. + + Returns a service summary dict. Skips the test if no active services exist. + """ + services = df_client.list_services().get("services", []) + + for svc in services: + state = svc.get("status", {}).get("state", "") + if ( + state not in CdpDfClient.DISABLED_STATES + and state not in CdpDfClient.FAILED_STATES + ): + details = df_client.describe_service(svc.get("crn")) + if details: + return svc + + pytest.skip("No active DataFlow services available for testing") + + +class TestCdpDfClientIntegration: + """Integration tests for CdpDfClient using the real CDP API.""" + + # ------------------------------------------------------------------------- + # Flow definition tests + # ------------------------------------------------------------------------- + + def test_list_flow_definitions_with_search(self, df_client, valid_df_flow): + """Test that search_term filters results by name.""" + flow_name = valid_df_flow["name"] + + response = df_client.list_flow_definitions(search_term=flow_name) + + assert "flows" in response + names = [f["name"] for f in response["flows"]] + assert flow_name in names + + def test_describe_flow(self, df_client, valid_df_flow): + """Test describing a flow definition by CRN returns expected fields.""" + flow_crn = valid_df_flow["crn"] + + response = df_client.describe_flow(flow_crn) + + assert response is not None + assert response != {} + + def test_describe_flow_not_found(self, df_client): + """Test that describing a nonexistent flow CRN returns an empty dict.""" + response = df_client.describe_flow( + "crn:cdp:df:us-west-1:00000000-0000-0000-0000-000000000000:flow:nonexistent", + ) + + assert response == {} + + def test_get_flow_by_name(self, df_client, valid_df_flow): + """Test resolving a flow by name.""" + flow_name = valid_df_flow["name"] + + result = df_client.get_flow_by_name(flow_name) + + assert result is not None + assert result["name"] == flow_name + assert "crn" in result + assert "versionCount" in result + + def test_get_flow_by_crn(self, df_client, valid_df_flow): + """Test resolving a flow by CRN.""" + flow_crn = valid_df_flow["crn"] + + result = df_client.get_flow_by_crn(flow_crn) + + assert result is not None + assert result["crn"] == flow_crn + assert "name" in result + + def test_get_flow_by_crn_not_found(self, df_client): + """Test that get_flow_by_crn returns None for an unknown CRN.""" + result = df_client.get_flow_by_crn( + "crn:cdp:df:us-west-1:00000000-0000-0000-0000-000000000000:flow:nonexistent", + ) + + assert result is None + + def test_import_flow_definition(self, df_client): + """Test importing a new flow definition and then deleting it.""" + flow_name = f"test-import-{uuid.uuid4().hex[:8]}" + flow_content = _create_minimal_flow_definition(flow_name) + + flow = None + try: + flow = df_client.import_flow_definition( + name=flow_name, + file_content=flow_content, + description="Import test flow", + comments="Initial version", + ) + + assert flow is not None + assert "crn" in flow + assert flow["name"] == flow_name + assert flow["versionCount"] == 1 + assert len(flow["versions"]) == 1 + assert flow["versions"][0]["version"] == 1 + assert flow["versions"][0]["comments"] == "Initial version" + finally: + if flow and flow.get("crn"): + df_client.delete_flow(flow_crn=flow["crn"]) + + def test_import_flow_definition_version(self, df_client, valid_df_flow): + """Test importing a new version into an existing flow.""" + flow_crn = valid_df_flow["crn"] + flow_name = valid_df_flow["name"] + flow_content = _create_minimal_flow_definition(flow_name) + + version = df_client.import_flow_definition_version( + flow_crn=flow_crn, + file_content=flow_content, + comments="Second version", + ) + + assert version is not None + assert "crn" in version + assert version["version"] == 2 + assert version["comments"] == "Second version" + + def test_import_flow_definition_version_with_tags(self, df_client, valid_df_flow): + """Test importing a flow version with tags.""" + flow_crn = valid_df_flow["crn"] + flow_name = valid_df_flow["name"] + flow_content = _create_minimal_flow_definition(flow_name) + + version = df_client.import_flow_definition_version( + flow_crn=flow_crn, + file_content=flow_content, + comments="Tagged version", + tags=[{"tagName": "production", "tagColor": "blue"}], + ) + + assert version is not None + assert version["version"] == 2 + + def test_delete_flow(self, df_client): + """Test deleting a flow definition.""" + flow_name = f"test-delete-{uuid.uuid4().hex[:8]}" + flow_content = _create_minimal_flow_definition(flow_name) + + flow = df_client.import_flow_definition( + name=flow_name, + file_content=flow_content, + comments="To be deleted", + ) + assert flow is not None + flow_crn = flow["crn"] + + df_client.delete_flow(flow_crn=flow_crn) + + # Verify the flow is no longer retrievable + result = df_client.get_flow_by_crn(flow_crn) + assert result is None + + def test_flow_definition_completeness(self, df_client, valid_df_flow): + """Test that a described flow contains all expected fields.""" + flow_crn = valid_df_flow["crn"] + result = df_client.get_flow_by_crn(flow_crn) + + assert result is not None + + expected_fields = [ + "crn", + "name", + "versionCount", + "createdTimestamp", + "modifiedTimestamp", + ] + for field in expected_fields: + assert field in result, f"Missing expected field: {field}" + + assert "versions" in result + assert isinstance(result["versions"], list) + + if result["versions"]: + version = result["versions"][0] + version_fields = ["crn", "version", "timestamp", "deploymentCount"] + for field in version_fields: + assert field in version, f"Missing expected version field: {field}" + + # ------------------------------------------------------------------------- + # Service tests + # ------------------------------------------------------------------------- + + def test_list_services(self, df_client): + """Test listing DataFlow services returns a valid structure.""" + response = df_client.list_services() + + assert "services" in response + assert isinstance(response["services"], list) + + if response["services"]: + service = response["services"][0] + assert "crn" in service + assert "name" in service + assert "status" in service + + def test_describe_service(self, df_client, valid_df_service): + """Test describing a DataFlow service returns expected fields.""" + service_crn = valid_df_service.get("crn") + + response = df_client.describe_service(service_crn) + + assert response is not None + assert response != {} + + def test_describe_service_not_found(self, df_client): + """Test that describing a nonexistent service CRN returns an empty dict.""" + response = df_client.describe_service( + "crn:cdp:df:us-west-1:00000000-0000-0000-0000-000000000000:service:nonexistent", + ) + + assert response == {} + + def test_get_service_by_name(self, df_client, valid_df_service): + """Test resolving a service by name.""" + service_name = valid_df_service.get("name") + + result = df_client.get_service_by_name(service_name) + + assert result is not None + + def test_get_service_by_name_not_found(self, df_client): + """Test that get_service_by_name returns None for an unknown name.""" + result = df_client.get_service_by_name("nonexistent-service-name-zzz-12345") + + assert result is None diff --git a/tests/unit/plugins/modules/df_customflow_version/test_df_customflow_version.py b/tests/unit/plugins/modules/df_customflow_version/test_df_customflow_version.py new file mode 100644 index 00000000..9c75e65b --- /dev/null +++ b/tests/unit/plugins/modules/df_customflow_version/test_df_customflow_version.py @@ -0,0 +1,322 @@ +# -*- coding: utf-8 -*- + +# Copyright 2026 Cloudera, Inc. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import pytest + +from ansible_collections.cloudera.cloud.tests.unit import ( + AnsibleFailJson, + AnsibleExitJson, +) + +from ansible_collections.cloudera.cloud.plugins.modules import df_customflow_version + + +BASE_URL = "https://cloudera.internal/api" +ACCESS_KEY = "test-access-key" +PRIVATE_KEY = "test-private-key" +FILE_ACCESS_KEY = "file-access-key" +FILE_PRIVATE_KEY = "file-private-key" +FILE_REGION = "default" + +FLOW_CRN = "crn:cdp:df:us-west-1:tenant:flow:flow-123" +VERSION_CRN = "crn:cdp:df:us-west-1:tenant:flow:flow-123/v2" +FLOW_FILE_CONTENT = '{"flow": "definition"}' + + +def test_df_customflow_version_import_success_from_file(module_args, mocker, tmp_path): + """Test importing a new CustomFlow version successfully from file.""" + + flow_file = tmp_path / "test-flow.json" + flow_file.write_text(FLOW_FILE_CONTENT) + + module_args( + { + "endpoint": BASE_URL, + "access_key": ACCESS_KEY, + "private_key": PRIVATE_KEY, + "flow_crn": FLOW_CRN, + "file": str(flow_file), + "comments": "Second version", + "state": "present", + }, + ) + + config = mocker.patch( + "ansible_collections.cloudera.cloud.plugins.module_utils.common.load_cdp_config", + ) + config.return_value = (FILE_ACCESS_KEY, FILE_PRIVATE_KEY, FILE_REGION) + + client = mocker.patch( + "ansible_collections.cloudera.cloud.plugins.modules.df_customflow_version.CdpDfClient", + autospec=True, + ).return_value + + client.get_flow_by_crn.return_value = { + "crn": FLOW_CRN, + "versionCount": 1, + } + + client.import_flow_definition_version.return_value = { + "crn": VERSION_CRN, + "version": 2, + "comments": "Second version", + "timestamp": 1640000000000, + "deploymentCount": 0, + } + + with pytest.raises(AnsibleExitJson) as result: + df_customflow_version.main() + + assert result.value.changed is True + assert result.value.customflow_version["crn"] == VERSION_CRN + assert result.value.customflow_version["version"] == 2 + + client.get_flow_by_crn.assert_called_once_with(FLOW_CRN) + client.import_flow_definition_version.assert_called_once() + call_args = client.import_flow_definition_version.call_args[1] + assert call_args["flow_crn"] == FLOW_CRN + assert call_args["file_content"] == FLOW_FILE_CONTENT + assert call_args["comments"] == "Second version" + + +def test_df_customflow_version_import_success_from_content(module_args, mocker): + """Test importing a new CustomFlow version successfully from content string.""" + + module_args( + { + "endpoint": BASE_URL, + "access_key": ACCESS_KEY, + "private_key": PRIVATE_KEY, + "flow_crn": FLOW_CRN, + "content": FLOW_FILE_CONTENT, + "comments": "Second version from content", + "state": "present", + }, + ) + + config = mocker.patch( + "ansible_collections.cloudera.cloud.plugins.module_utils.common.load_cdp_config", + ) + config.return_value = (FILE_ACCESS_KEY, FILE_PRIVATE_KEY, FILE_REGION) + + client = mocker.patch( + "ansible_collections.cloudera.cloud.plugins.modules.df_customflow_version.CdpDfClient", + autospec=True, + ).return_value + + client.get_flow_by_crn.return_value = { + "crn": FLOW_CRN, + "versionCount": 1, + } + + client.import_flow_definition_version.return_value = { + "crn": VERSION_CRN, + "version": 2, + "comments": "Second version from content", + "timestamp": 1640000000000, + "deploymentCount": 0, + } + + with pytest.raises(AnsibleExitJson) as result: + df_customflow_version.main() + + assert result.value.changed is True + assert result.value.customflow_version["crn"] == VERSION_CRN + assert result.value.customflow_version["version"] == 2 + + client.get_flow_by_crn.assert_called_once_with(FLOW_CRN) + client.import_flow_definition_version.assert_called_once() + call_args = client.import_flow_definition_version.call_args[1] + assert call_args["flow_crn"] == FLOW_CRN + assert call_args["file_content"] == FLOW_FILE_CONTENT + assert call_args["comments"] == "Second version from content" + + +def test_df_customflow_version_import_with_tags(module_args, mocker, tmp_path): + """Test importing a CustomFlow version with tags (verifies tag format conversion).""" + + flow_file = tmp_path / "test-flow.json" + flow_file.write_text(FLOW_FILE_CONTENT) + + module_args( + { + "endpoint": BASE_URL, + "access_key": ACCESS_KEY, + "private_key": PRIVATE_KEY, + "flow_crn": FLOW_CRN, + "file": str(flow_file), + "comments": "Tagged version", + "tags": [ + {"tag_name": "production", "tag_color": "blue"}, + {"tag_name": "stable"}, + ], + "state": "present", + }, + ) + + config = mocker.patch( + "ansible_collections.cloudera.cloud.plugins.module_utils.common.load_cdp_config", + ) + config.return_value = (FILE_ACCESS_KEY, FILE_PRIVATE_KEY, FILE_REGION) + + client = mocker.patch( + "ansible_collections.cloudera.cloud.plugins.modules.df_customflow_version.CdpDfClient", + autospec=True, + ).return_value + + client.get_flow_by_crn.return_value = { + "crn": FLOW_CRN, + "versionCount": 1, + } + + client.import_flow_definition_version.return_value = { + "crn": VERSION_CRN, + "version": 2, + "comments": "Tagged version", + "timestamp": 1640000000000, + "deploymentCount": 0, + } + + with pytest.raises(AnsibleExitJson) as result: + df_customflow_version.main() + + assert result.value.changed is True + + # Verify tags were converted from Ansible format (snake_case) to API format (camelCase) + call_args = client.import_flow_definition_version.call_args[1] + assert call_args["tags"] == [ + {"tagName": "production", "tagColor": "blue"}, + {"tagName": "stable"}, + ] + + +def test_df_customflow_version_nonexistent_flow(module_args, mocker): + """Test that the module fails when the referenced flow CRN does not exist.""" + + module_args( + { + "endpoint": BASE_URL, + "access_key": ACCESS_KEY, + "private_key": PRIVATE_KEY, + "flow_crn": FLOW_CRN, + "content": FLOW_FILE_CONTENT, + "state": "present", + }, + ) + + config = mocker.patch( + "ansible_collections.cloudera.cloud.plugins.module_utils.common.load_cdp_config", + ) + config.return_value = (FILE_ACCESS_KEY, FILE_PRIVATE_KEY, FILE_REGION) + + client = mocker.patch( + "ansible_collections.cloudera.cloud.plugins.modules.df_customflow_version.CdpDfClient", + autospec=True, + ).return_value + + # Mock: Flow does not exist + client.get_flow_by_crn.return_value = None + + with pytest.raises(AnsibleFailJson) as result: + df_customflow_version.main() + + assert result.value.failed is True + assert "does not exist" in result.value.msg + + # Verify import was NOT called + client.import_flow_definition_version.assert_not_called() + + +def test_df_customflow_version_check_mode(module_args, mocker, tmp_path): + """Test check mode: reports changed=True but does not call the API.""" + + flow_file = tmp_path / "test-flow.json" + flow_file.write_text(FLOW_FILE_CONTENT) + + module_args( + { + "endpoint": BASE_URL, + "access_key": ACCESS_KEY, + "private_key": PRIVATE_KEY, + "flow_crn": FLOW_CRN, + "file": str(flow_file), + "comments": "Check mode version", + "state": "present", + "_ansible_check_mode": True, + }, + ) + + config = mocker.patch( + "ansible_collections.cloudera.cloud.plugins.module_utils.common.load_cdp_config", + ) + config.return_value = (FILE_ACCESS_KEY, FILE_PRIVATE_KEY, FILE_REGION) + + client = mocker.patch( + "ansible_collections.cloudera.cloud.plugins.modules.df_customflow_version.CdpDfClient", + autospec=True, + ).return_value + + client.get_flow_by_crn.return_value = { + "crn": FLOW_CRN, + "versionCount": 1, + } + + with pytest.raises(AnsibleExitJson) as result: + df_customflow_version.main() + + assert result.value.changed is True + + # Verify import was NOT called (check mode) + client.import_flow_definition_version.assert_not_called() + + +def test_df_customflow_version_file_read_error(module_args, mocker): + """Test handling of file read errors.""" + + module_args( + { + "endpoint": BASE_URL, + "access_key": ACCESS_KEY, + "private_key": PRIVATE_KEY, + "flow_crn": FLOW_CRN, + "file": "/nonexistent/file.json", + "state": "present", + }, + ) + + config = mocker.patch( + "ansible_collections.cloudera.cloud.plugins.module_utils.common.load_cdp_config", + ) + config.return_value = (FILE_ACCESS_KEY, FILE_PRIVATE_KEY, FILE_REGION) + + client = mocker.patch( + "ansible_collections.cloudera.cloud.plugins.modules.df_customflow_version.CdpDfClient", + autospec=True, + ).return_value + + client.get_flow_by_crn.return_value = { + "crn": FLOW_CRN, + "versionCount": 1, + } + + with pytest.raises(AnsibleFailJson) as result: + df_customflow_version.main() + + assert "Failed to read file" in result.value.msg diff --git a/tests/unit/plugins/modules/df_customflow_version/test_df_customflow_version_int.py b/tests/unit/plugins/modules/df_customflow_version/test_df_customflow_version_int.py new file mode 100644 index 00000000..eba89a09 --- /dev/null +++ b/tests/unit/plugins/modules/df_customflow_version/test_df_customflow_version_int.py @@ -0,0 +1,482 @@ +# -*- coding: utf-8 -*- + +# Copyright 2026 Cloudera, Inc. All Rights Reserved. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from __future__ import absolute_import, division, print_function + +__metaclass__ = type + +import json +import os +import pytest +import random +import tempfile +import uuid +from contextlib import contextmanager +from typing import Callable, Generator + +from ansible_collections.cloudera.cloud.tests.unit import ( + AnsibleExitJson, + AnsibleFailJson, +) + +from ansible_collections.cloudera.cloud.plugins.module_utils.cdp_df import CdpDfClient +from ansible_collections.cloudera.cloud.plugins.modules import df_customflow_version + +# Required environment variables for integration tests +REQUIRED_ENV_VARS = [ + "CDP_API_ENDPOINT", + "CDP_ACCESS_KEY_ID", + "CDP_PRIVATE_KEY", +] + +# Mark all tests in this module as integration tests requiring API credentials +pytestmark = pytest.mark.integration_api + + +@pytest.fixture +def df_module_args(module_args, env_context) -> Callable[[dict], None]: + """Fixture to pre-populate common DataFlow module arguments.""" + + def wrapped_args(args=None): + if args is None: + args = {} + + args.update( + { + "endpoint": env_context["CDP_API_ENDPOINT"], + "access_key": env_context["CDP_ACCESS_KEY_ID"], + "private_key": env_context["CDP_PRIVATE_KEY"], + }, + ) + return module_args(args) + + return wrapped_args + + +@pytest.fixture +def df_client(test_cdp_client) -> CdpDfClient: + """Fixture to provide a DataFlow client for tests.""" + return CdpDfClient(api_client=test_cdp_client) + + +def create_minimal_flow_definition(flow_name: str) -> dict: + """ + Factory function to create a minimal NiFi flow definition. + + Args: + flow_name: The name of the flow + + Returns: + A minimal flow definition dictionary with random identifiers + """ + + return { + "snapshotMetadata": { + "bucketIdentifier": None, + "flowIdentifier": str(uuid.uuid4()), + "version": 0, + "timestamp": 1771317050573, + "author": None, + "comments": None, + "link": None, + }, + "flowContents": { + "identifier": str(uuid.uuid4()), + "instanceIdentifier": None, + "name": flow_name, + "comments": None, + "position": None, + "processGroups": [], + "remoteProcessGroups": [], + "processors": [], + "inputPorts": [], + "outputPorts": [], + "connections": [], + "labels": [], + "funnels": [], + "controllerServices": [], + "versionedFlowCoordinates": None, + "parameterContextName": flow_name, + "defaultFlowFileExpiration": "0 sec", + "defaultBackPressureObjectThreshold": 10000, + "defaultBackPressureDataSizeThreshold": "1 GB", + "scheduledState": None, + "executionEngine": None, + "maxConcurrentTasks": None, + "statelessFlowTimeout": None, + "logFileSuffix": None, + "componentType": "PROCESS_GROUP", + "flowFileConcurrency": "UNBOUNDED", + "flowFileOutboundPolicy": "STREAM_WHEN_AVAILABLE", + "groupIdentifier": None, + }, + "externalControllerServices": None, + "parameterProviders": None, + "parameterContexts": { + flow_name: { + "identifier": str(uuid.uuid4()), + "instanceIdentifier": None, + "name": flow_name, + "comments": None, + "position": None, + "parameters": [], + "inheritedParameterContexts": [], + "description": None, + "parameterProvider": None, + "parameterGroupName": None, + "synchronized": None, + "componentType": "PARAMETER_CONTEXT", + "groupIdentifier": None, + }, + }, + "flowEncodingVersion": None, + "flow": None, + "bucket": None, + } + + +@contextmanager +def temporary_flow_file(flow_name: str): + """ + Context manager to create a temporary flow definition file. + + Args: + flow_name: The name of the flow + + Yields: + The path to the temporary flow file + + Example: + with temporary_flow_file("my-flow") as flow_file: + # Use flow_file path + pass + """ + flow_definition = create_minimal_flow_definition(flow_name) + flow_content = json.dumps(flow_definition) + + with tempfile.TemporaryDirectory() as tmpdir: + temp_file_path = os.path.join(tmpdir, "flow.json") + with open(temp_file_path, "w") as f: + f.write(flow_content) + yield temp_file_path + + +@pytest.fixture +def df_flow_delete(df_client) -> Generator[Callable[[str], None], None, None]: + """Fixture to clean up DataFlow flows created during tests.""" + flow_crns = [] + + def _df_flow_delete(flow_crn: str): + flow_crns.append(flow_crn) + return + + yield _df_flow_delete + + # Cleanup: delete all tracked flows + for flow_crn in flow_crns: + try: + df_client.delete_flow(flow_crn=flow_crn) + except Exception: + pass + + +@pytest.fixture +def df_flow_create(df_client, df_flow_delete) -> Callable[[str, str, str], dict]: + """ + Fixture to create DataFlow flows and ensure cleanup. + + Returns a function that creates a flow and registers it for cleanup. + """ + + def _df_flow_create( + flow_name: str, + description: str = None, + comments: str = "Test Flow", + ) -> dict: + """ + Create a minimal DataFlow flow. + + Args: + flow_name: Name of the flow to create + description: Optional description for the flow + comments: Version comments (default: "Test Flow") + + Returns: + The created flow object from the API + """ + + flow_definition = create_minimal_flow_definition(flow_name) + flow_content = json.dumps(flow_definition) + + if description is None: + description = f"Test flow - {flow_name}" + + result = df_client.import_flow_definition( + name=flow_name, + file_content=flow_content, + description=description, + comments=comments, + ) + + if result and "crn" in result: + df_flow_delete(result["crn"]) + + return result + + return _df_flow_create + + +def test_df_flow_version_create_and_verify(df_flow_create, df_client): + """Test creating a flow version directly via the client.""" + + random_suffix = random.randint(100000, 999999) + flow_name = f"test-customflow-version-{random_suffix}" + + # Create the parent flow + flow = df_flow_create( + flow_name=flow_name, + description=f"Integration test flow for versioning - {flow_name}", + comments="Version 1", + ) + + assert flow is not None + assert "crn" in flow + assert flow["versionCount"] == 1 + + # Import a new version + flow_definition = create_minimal_flow_definition(flow_name) + flow_content = json.dumps(flow_definition) + + version = df_client.import_flow_definition_version( + flow_crn=flow["crn"], + file_content=flow_content, + comments="Version 2", + ) + + assert version is not None + assert "crn" in version + assert version["version"] == 2 + assert version["comments"] == "Version 2" + + # Verify the flow now has 2 versions + updated_flow = df_client.get_flow_by_crn(flow["crn"]) + assert updated_flow is not None + assert updated_flow["versionCount"] == 2 + + +def test_df_customflow_version_import_via_module( + df_module_args, + env_context, + df_flow_create, +): + """Test importing a CustomFlow version via the Ansible module using a file.""" + + random_suffix = random.randint(100000, 999999) + flow_name = f"test-customflow-version-{random_suffix}" + + # Create the parent flow first + flow = df_flow_create( + flow_name=flow_name, + description=f"Integration test flow for versioning - {flow_name}", + comments="Initial version", + ) + + assert flow is not None + assert "crn" in flow + + with temporary_flow_file(flow_name) as flow_file: + df_module_args( + { + "flow_crn": flow["crn"], + "file": flow_file, + "comments": "Second version", + "state": "present", + }, + ) + + with pytest.raises(AnsibleExitJson) as result: + df_customflow_version.main() + + assert result.value.changed is True + assert result.value.customflow_version is not None + assert result.value.customflow_version["version"] == 2 + assert result.value.customflow_version["comments"] == "Second version" + + +def test_df_customflow_version_import_with_content_via_module( + df_module_args, + env_context, + df_flow_create, +): + """Test importing a CustomFlow version using the content parameter.""" + + random_suffix = random.randint(100000, 999999) + flow_name = f"test-customflow-version-content-{random_suffix}" + + # Create the parent flow first + flow = df_flow_create( + flow_name=flow_name, + description=f"Integration test flow for versioning - {flow_name}", + comments="Initial version", + ) + + assert flow is not None + assert "crn" in flow + + # Create flow definition and convert to JSON string + flow_definition = create_minimal_flow_definition(flow_name) + flow_content = json.dumps(flow_definition) + + df_module_args( + { + "flow_crn": flow["crn"], + "content": flow_content, + "comments": "Second version from content", + "state": "present", + }, + ) + + with pytest.raises(AnsibleExitJson) as result: + df_customflow_version.main() + + assert result.value.changed is True + assert result.value.customflow_version is not None + assert result.value.customflow_version["version"] == 2 + assert result.value.customflow_version["comments"] == "Second version from content" + + +def test_df_customflow_version_import_with_tags_via_module( + df_module_args, + env_context, + df_flow_create, +): + """Test importing a CustomFlow version with tags via the Ansible module.""" + + random_suffix = random.randint(100000, 999999) + flow_name = f"test-customflow-version-tags-{random_suffix}" + + # Create the parent flow first + flow = df_flow_create( + flow_name=flow_name, + description=f"Integration test flow for versioning with tags - {flow_name}", + comments="Initial version", + ) + + assert flow is not None + assert "crn" in flow + + with temporary_flow_file(flow_name) as flow_file: + df_module_args( + { + "flow_crn": flow["crn"], + "file": flow_file, + "comments": "Second version with tags", + "tags": [ + {"tag_name": "production", "tag_color": "blue"}, + {"tag_name": "stable", "tag_color": "green"}, + {"tag_name": "tested"}, + ], + "state": "present", + }, + ) + + with pytest.raises(AnsibleExitJson) as result: + df_customflow_version.main() + + assert result.value.changed is True + assert result.value.customflow_version is not None + assert result.value.customflow_version["version"] == 2 + + +def test_df_customflow_version_multiple_versions_via_module( + df_module_args, + env_context, + df_flow_create, +): + """Test that each module invocation always creates a new version (non-idempotent by design).""" + + random_suffix = random.randint(100000, 999999) + flow_name = f"test-customflow-version-multi-{random_suffix}" + + # Create the parent flow first + flow = df_flow_create( + flow_name=flow_name, + description=f"Integration test flow for multiple versions - {flow_name}", + comments="Version 1", + ) + + assert flow is not None + assert "crn" in flow + + flow_definition = create_minimal_flow_definition(flow_name) + flow_content = json.dumps(flow_definition) + + # Import version 2 + df_module_args( + { + "flow_crn": flow["crn"], + "content": flow_content, + "comments": "Version 2", + "state": "present", + }, + ) + + with pytest.raises(AnsibleExitJson) as result: + df_customflow_version.main() + + assert result.value.changed is True + assert result.value.customflow_version["version"] == 2 + + # Import version 3 - same args, always changed + df_module_args( + { + "flow_crn": flow["crn"], + "content": flow_content, + "comments": "Version 3", + "state": "present", + }, + ) + + with pytest.raises(AnsibleExitJson) as result: + df_customflow_version.main() + + assert result.value.changed is True + assert result.value.customflow_version["version"] == 3 + + +def test_df_customflow_version_nonexistent_flow_via_module( + df_module_args, + env_context, +): + """Test that the module fails when the referenced flow CRN does not exist.""" + + flow_definition = create_minimal_flow_definition("nonexistent-flow") + flow_content = json.dumps(flow_definition) + + df_module_args( + { + "flow_crn": "crn:cdp:df:us-west-1:00000000-0000-0000-0000-000000000000:flow:nonexistent-flow-crn", + "content": flow_content, + "comments": "Should fail", + "state": "present", + }, + ) + + with pytest.raises(AnsibleFailJson) as result: + df_customflow_version.main() + + assert result.value.failed is True + assert "does not exist" in result.value.msg