dt/data_transforms: Test for patch_transform_metadata endpoint

tests/rptest/tests/data_transforms_test.py

@@ -317,6 +317,57 @@ def all_offsets_removed():
             retry_on_exc=True,
         )
 
+    @cluster(num_nodes=3)
+    def test_patch(self):
+        input_topic = self.topics[0]
+        output_topic = self.topics[1]
+        self._deploy_wasm(name="identity-xform",
+                          input_topic=input_topic,
+                          output_topic=output_topic,
+                          wait_running=True)
+
+        def all_partitions_status(stat: str):
+            report = self._rpk.list_wasm()
+            return all(s.status == stat for s in report[0].status)
+
+        def env_is(env: dict[str, str]):
+            report = self._rpk.list_wasm()
+            return report[0].environment == env
+
+        rsp = self._admin.transforms_patch_meta("identity-xform", pause=True)
+        assert rsp.status_code == 200
+
+        wait_until(lambda: all_partitions_status("inactive"),
+                   timeout_sec=30,
+                   backoff_sec=1,
+                   err_msg=f"some partitions didn't become inactive",
+                   retry_on_exc=True)
+
+        env1 = {
+            "FOO": "bar",
+            "BAZ": "quux",
+        }
+        env2 = {"FOO": "bells"}
+
+        rsp = self._admin.transforms_patch_meta("identity-xform",
+                                                pause=False,
+                                                env=env1)
+        assert rsp.status_code == 200
+
+        wait_until(lambda: all_partitions_status("running") and env_is(env1),
+                   timeout_sec=30,
+                   backoff_sec=1,
+                   err_msg=f"some partitions didn't come back",
+                   retry_on_exc=True)
+
+        rsp = self._admin.transforms_patch_meta("identity-xform", env=env2)
+
+        wait_until(lambda: all_partitions_status("running") and env_is(env2),
+                   timeout_sec=30,
+                   backoff_sec=1,
+                   err_msg="some partitions didn't take the env update",
+                   retry_on_exc=True)
+
 
 class DataTransformsChainingTest(BaseDataTransformsTest):
     """