tests/migrations: test migration validation

refuse to create migrations if cluster-wide or topic configs are unsupported

Author: bashtanov

tests/rptest/tests/data_migrations_api_test.py

@@ -33,8 +33,10 @@
 
 MIGRATION_LOG_ALLOW_LIST = [
     'Error during log recovery: cloud_storage::missing_partition_exception',
-    'Requested data migration does not exist',
-    'Invalid data migration state transition requested',
+    'Requested data migration does not exist',  # failed get migration by by id
+    'Invalid data migration state transition requested',  # failed migration deletes
+    'Data migration contains resources that does not exists or are already being migrated',  # failed create
+    r'/v1/migrations.*Requested feature is disabled'  # cloud storage disabled
 ]
 
 
@@ -193,8 +195,11 @@ def migration_is_absent(id: int):
     def wait_partitions_appear(self, topics: list[TopicSpec]):
         # we may be unlucky to query a slow node
         def topic_has_all_partitions(t: TopicSpec):
-            return t.partition_count == \
-                    len(self.client().describe_topic(t.name).partitions)
+            exp_part_cnt = len(self.client().describe_topic(t.name).partitions)
+            self.logger.debug(
+                f"topic {t.name} has {t.partition_count} partitions out of {exp_part_cnt} expected"
+            )
+            return t.partition_count == exp_part_cnt
 
         wait_until(lambda: all(topic_has_all_partitions(t) for t in topics),
                    timeout_sec=90,
@@ -234,6 +239,40 @@ def migration_id_if_exists():
         self.wait_migration_appear(migration_id)
         return migration_id
 
+    def assure_not_migratable(self, topic: TopicSpec):
+        out_migration = OutboundDataMigration(
+            [make_namespaced_topic(topic.name)], consumer_groups=[])
+        try:
+            self.create_and_wait(out_migration)
+            assert False
+        except requests.exceptions.HTTPError as e:
+            pass
+
+    @cluster(num_nodes=3, log_allow_list=MIGRATION_LOG_ALLOW_LIST)
+    def test_creating_with_topic_no_remote_writes(self):
+        self.redpanda.set_cluster_config(
+            {"cloud_storage_enable_remote_write": False}, expect_restart=True)
+        topic = TopicSpec(partition_count=3)
+        self.client().create_topic(topic)
+        self.wait_partitions_appear([topic])
+        self.redpanda.set_cluster_config(
+            {"cloud_storage_enable_remote_write": True}, expect_restart=True)
+        self.assure_not_migratable(topic)
+
+    @cluster(num_nodes=3, log_allow_list=MIGRATION_LOG_ALLOW_LIST)
+    @matrix(param_to_disable=[
+        "cloud_storage_enabled", "cloud_storage_disable_archiver_manager"
+    ])
+    def test_creating_when_cluster_misconfigured(self, param_to_disable):
+        self.redpanda.set_cluster_config({param_to_disable: False},
+                                         expect_restart=True)
+        topic = TopicSpec(partition_count=3)
+        self.client().create_topic(topic)
+        self.assure_not_migratable(topic)
+        # for scrubbing to complete
+        self.redpanda.set_cluster_config({param_to_disable: True},
+                                         expect_restart=True)
+
     @cluster(num_nodes=3, log_allow_list=MIGRATION_LOG_ALLOW_LIST)
     def test_creating_and_listing_migrations(self):
         topics = [TopicSpec(partition_count=3) for i in range(5)]