From 9cf893a872d92e790aad05517c99b320e90e74d9 Mon Sep 17 00:00:00 2001 From: Dalia Frank Date: Sun, 19 Apr 2026 17:52:11 +0300 Subject: [PATCH 1/2] Storage: Add DPA resource --- ocp_resources/data_protection_application.py | 124 ++++++++++++++++++ ocp_resources/resource.py | 1 + .../test_data_protection_application.py | 40 ++++++ 3 files changed, 165 insertions(+) create mode 100644 ocp_resources/data_protection_application.py create mode 100644 tests/test_resources/test_data_protection_application.py diff --git a/ocp_resources/data_protection_application.py b/ocp_resources/data_protection_application.py new file mode 100644 index 0000000000..07ca4575e2 --- /dev/null +++ b/ocp_resources/data_protection_application.py @@ -0,0 +1,124 @@ +# Generated using https://github.com/RedHatQE/openshift-python-wrapper/blob/main/scripts/resource/README.md + + +from typing import Any + +from ocp_resources.exceptions import MissingRequiredArgumentError +from ocp_resources.resource import NamespacedResource + + +class DataProtectionApplication(NamespacedResource): + """ + DataProtectionApplication is the Schema for the dpa API + """ + + api_group: str = NamespacedResource.ApiGroup.OADP_OPENSHIFT_IO + + def __init__( + self, + backup_images: bool | None = None, + backup_locations: list[Any] | None = None, + configuration: dict[str, Any] | None = None, + features: dict[str, Any] | None = None, + image_pull_policy: str | None = None, + pod_annotations: dict[str, Any] | None = None, + pod_dns_config: dict[str, Any] | None = None, + pod_dns_policy: str | None = None, + snapshot_locations: list[Any] | None = None, + unsupported_overrides: dict[str, Any] | None = None, + **kwargs: Any, + ) -> None: + r""" + Args: + backup_images (bool): backupImages is used to specify whether you want to deploy a registry + for enabling backup and restore of images + + backup_locations (list[Any]): backupLocations defines the list of desired configuration to use for + BackupStorageLocations + + configuration (dict[str, Any]): configuration is used to configure the data protection application's + server config + + features (dict[str, Any]): features defines the configuration for the DPA to enable the OADP tech + preview features + + image_pull_policy (str): which imagePullPolicy to use in all container images used by OADP. By + default, for images with sha256 or sha512 digest, OADP uses + IfNotPresent and uses Always for all other images. + + pod_annotations (dict[str, Any]): add annotations to pods deployed by operator + + pod_dns_config (dict[str, Any]): podDnsConfig defines the DNS parameters of a pod in addition to those + generated from DNSPolicy. + https://kubernetes.io/docs/concepts/services-networking/dns-pod- + service/#pod-dns-config + + pod_dns_policy (str): podDnsPolicy defines how a pod's DNS will be configured. + https://kubernetes.io/docs/concepts/services-networking/dns-pod- + service/#pod-s-dns-policy + + snapshot_locations (list[Any]): snapshotLocations defines the list of desired configuration to use for + VolumeSnapshotLocations + + unsupported_overrides (dict[str, Any]): unsupportedOverrides can be used to override images used in + deployments. Available keys are: - veleroImageFqin - + awsPluginImageFqin - legacyAWSPluginImageFqin - + openshiftPluginImageFqin - azurePluginImageFqin - + gcpPluginImageFqin - resticRestoreImageFqin - + kubevirtPluginImageFqin - operator-type + + """ + super().__init__(**kwargs) + + self.backup_images = backup_images + self.backup_locations = backup_locations + self.configuration = configuration + self.features = features + self.image_pull_policy = image_pull_policy + self.pod_annotations = pod_annotations + self.pod_dns_config = pod_dns_config + self.pod_dns_policy = pod_dns_policy + self.snapshot_locations = snapshot_locations + self.unsupported_overrides = unsupported_overrides + + def to_dict(self) -> None: + + super().to_dict() + + if not self.kind_dict and not self.yaml_file: + if self.configuration is None: + raise MissingRequiredArgumentError(argument="self.configuration") + + self.res["spec"] = {} + _spec = self.res["spec"] + + _spec["configuration"] = self.configuration + + if self.backup_images is not None: + _spec["backupImages"] = self.backup_images + + if self.backup_locations is not None: + _spec["backupLocations"] = self.backup_locations + + if self.features is not None: + _spec["features"] = self.features + + if self.image_pull_policy is not None: + _spec["imagePullPolicy"] = self.image_pull_policy + + if self.pod_annotations is not None: + _spec["podAnnotations"] = self.pod_annotations + + if self.pod_dns_config is not None: + _spec["podDnsConfig"] = self.pod_dns_config + + if self.pod_dns_policy is not None: + _spec["podDnsPolicy"] = self.pod_dns_policy + + if self.snapshot_locations is not None: + _spec["snapshotLocations"] = self.snapshot_locations + + if self.unsupported_overrides is not None: + _spec["unsupportedOverrides"] = self.unsupported_overrides + + # End of generated code diff --git a/ocp_resources/resource.py b/ocp_resources/resource.py index c87c7e64b9..76593f63b9 100644 --- a/ocp_resources/resource.py +++ b/ocp_resources/resource.py @@ -557,6 +557,7 @@ class ApiGroup: NMSTATE_IO: str = "nmstate.io" NODE_LABELLER_KUBEVIRT_IO: str = "node-labeller.kubevirt.io" NODEMAINTENANCE_KUBEVIRT_IO: str = "nodemaintenance.kubevirt.io" + OADP_OPENSHIFT_IO: str = "oadp.openshift.io" OBSERVABILITY_OPEN_CLUSTER_MANAGEMENT_IO: str = "observability.open-cluster-management.io" OCS_OPENSHIFT_IO: str = "ocs.openshift.io" OPENTELEMETRY_IO: str = "opentelemetry.io" diff --git a/tests/test_resources/test_data_protection_application.py b/tests/test_resources/test_data_protection_application.py new file mode 100644 index 0000000000..de24e2bbb1 --- /dev/null +++ b/tests/test_resources/test_data_protection_application.py @@ -0,0 +1,40 @@ +import pytest + +from ocp_resources.data_protection_application import DataProtectionApplication + + +@pytest.mark.incremental +class TestDataProtectionApplication: + @pytest.fixture(scope="class") + def dataprotectionapplication(self, fake_client): + return DataProtectionApplication( + client=fake_client, + name="test-dataprotectionapplication", + namespace="default", + configuration={"test-configuration": "test-value"}, + ) + + def test_01_create_dataprotectionapplication(self, dataprotectionapplication): + """Test creating DataProtectionApplication""" + deployed_resource = dataprotectionapplication.deploy() + assert deployed_resource + assert deployed_resource.name == "test-dataprotectionapplication" + assert dataprotectionapplication.exists + + def test_02_get_dataprotectionapplication(self, dataprotectionapplication): + """Test getting DataProtectionApplication""" + assert dataprotectionapplication.instance + assert dataprotectionapplication.kind == "DataProtectionApplication" + + def test_03_update_dataprotectionapplication(self, dataprotectionapplication): + """Test updating DataProtectionApplication""" + resource_dict = dataprotectionapplication.instance.to_dict() + resource_dict["metadata"]["labels"] = {"updated": "true"} + dataprotectionapplication.update(resource_dict=resource_dict) + assert dataprotectionapplication.labels["updated"] == "true" + + def test_04_delete_dataprotectionapplication(self, dataprotectionapplication): + """Test deleting DataProtectionApplication""" + dataprotectionapplication.clean_up(wait=False) + # Verify resource no longer exists after deletion + assert not dataprotectionapplication.exists From b71e488a62dbfa2acb32216f5ea505940d59ea57 Mon Sep 17 00:00:00 2001 From: Dalia Frank Date: Tue, 28 Apr 2026 11:00:30 +0300 Subject: [PATCH 2/2] remove tests file --- .../test_data_protection_application.py | 40 ------------------- 1 file changed, 40 deletions(-) delete mode 100644 tests/test_resources/test_data_protection_application.py diff --git a/tests/test_resources/test_data_protection_application.py b/tests/test_resources/test_data_protection_application.py deleted file mode 100644 index de24e2bbb1..0000000000 --- a/tests/test_resources/test_data_protection_application.py +++ /dev/null @@ -1,40 +0,0 @@ -import pytest - -from ocp_resources.data_protection_application import DataProtectionApplication - - -@pytest.mark.incremental -class TestDataProtectionApplication: - @pytest.fixture(scope="class") - def dataprotectionapplication(self, fake_client): - return DataProtectionApplication( - client=fake_client, - name="test-dataprotectionapplication", - namespace="default", - configuration={"test-configuration": "test-value"}, - ) - - def test_01_create_dataprotectionapplication(self, dataprotectionapplication): - """Test creating DataProtectionApplication""" - deployed_resource = dataprotectionapplication.deploy() - assert deployed_resource - assert deployed_resource.name == "test-dataprotectionapplication" - assert dataprotectionapplication.exists - - def test_02_get_dataprotectionapplication(self, dataprotectionapplication): - """Test getting DataProtectionApplication""" - assert dataprotectionapplication.instance - assert dataprotectionapplication.kind == "DataProtectionApplication" - - def test_03_update_dataprotectionapplication(self, dataprotectionapplication): - """Test updating DataProtectionApplication""" - resource_dict = dataprotectionapplication.instance.to_dict() - resource_dict["metadata"]["labels"] = {"updated": "true"} - dataprotectionapplication.update(resource_dict=resource_dict) - assert dataprotectionapplication.labels["updated"] == "true" - - def test_04_delete_dataprotectionapplication(self, dataprotectionapplication): - """Test deleting DataProtectionApplication""" - dataprotectionapplication.clean_up(wait=False) - # Verify resource no longer exists after deletion - assert not dataprotectionapplication.exists