From af5ad2233f80956b8a1adb278930fd73ec95392b Mon Sep 17 00:00:00 2001
From: Ian Hoang <hoangia@amazon.com>
Date: Tue, 9 Jan 2024 13:46:22 -0600
Subject: [PATCH] Revert "renamed put pipeline to create ingest pipeline
 (#399)"

This reverts commit e73664af95218a1ff1afb8fa05bc9d0c7766ec33.
---
 osbenchmark/worker_coordinator/runner.py |  6 +++---
 osbenchmark/workload/workload.py         |  6 +++---
 tests/worker_coordinator/runner_test.py  | 17 ++++++-----------
 3 files changed, 12 insertions(+), 17 deletions(-)

diff --git a/osbenchmark/worker_coordinator/runner.py b/osbenchmark/worker_coordinator/runner.py
index 0e76427d7..51ad3f735 100644
--- a/osbenchmark/worker_coordinator/runner.py
+++ b/osbenchmark/worker_coordinator/runner.py
@@ -77,7 +77,7 @@ def register_default_runners():
     register_runner(workload.OperationType.RestoreSnapshot, RestoreSnapshot(), async_runner=True)
     # We treat the following as administrative commands and thus already start to wrap them in a retry.
     register_runner(workload.OperationType.ClusterHealth, Retry(ClusterHealth()), async_runner=True)
-    register_runner(workload.OperationType.CreateIngestPipeline, Retry(CreateIngestPipeline()), async_runner=True)
+    register_runner(workload.OperationType.PutPipeline, Retry(PutPipeline()), async_runner=True)
     register_runner(workload.OperationType.Refresh, Retry(Refresh()), async_runner=True)
     register_runner(workload.OperationType.CreateIndex, Retry(CreateIndex()), async_runner=True)
     register_runner(workload.OperationType.DeleteIndex, Retry(DeleteIndex()), async_runner=True)
@@ -1231,7 +1231,7 @@ def __repr__(self, *args, **kwargs):
         return "cluster-health"
 
 
-class CreateIngestPipeline(Runner):
+class PutPipeline(Runner):
     async def __call__(self, opensearch, params):
         await opensearch.ingest.put_pipeline(id=mandatory(params, "id", self),
                                      body=mandatory(params, "body", self),
@@ -1240,7 +1240,7 @@ async def __call__(self, opensearch, params):
                                      )
 
     def __repr__(self, *args, **kwargs):
-        return "create-ingest-pipeline"
+        return "put-pipeline"
 
 # TODO: refactor it after python client support search pipeline https://github.com/opensearch-project/opensearch-py/issues/474
 class CreateSearchPipeline(Runner):
diff --git a/osbenchmark/workload/workload.py b/osbenchmark/workload/workload.py
index d9b296846..1dc4ff530 100644
--- a/osbenchmark/workload/workload.py
+++ b/osbenchmark/workload/workload.py
@@ -590,7 +590,7 @@ class OperationType(Enum):
     # administrative actions
     ForceMerge = 1001
     ClusterHealth = 1002
-    CreateIngestPipeline = 1003
+    PutPipeline = 1003
     Refresh = 1004
     CreateIndex = 1005
     DeleteIndex = 1006
@@ -653,8 +653,8 @@ def from_hyphenated_string(cls, v):
             return OperationType.Bulk
         elif v == "raw-request":
             return OperationType.RawRequest
-        elif v == "create-ingest-pipeline":
-            return OperationType.CreateIngestPipeline
+        elif v == "put-pipeline":
+            return OperationType.PutPipeline
         elif v == "refresh":
             return OperationType.Refresh
         elif v == "create-index":
diff --git a/tests/worker_coordinator/runner_test.py b/tests/worker_coordinator/runner_test.py
index 728a663fe..01a2a3c3f 100644
--- a/tests/worker_coordinator/runner_test.py
+++ b/tests/worker_coordinator/runner_test.py
@@ -2737,13 +2737,13 @@ async def test_query_vector_search_with_custom_id_field_inside_source(self, open
         )
 
 
-class CreateIngestPipelineRunnerTests(TestCase):
+class PutPipelineRunnerTests(TestCase):
     @mock.patch("opensearchpy.OpenSearch")
     @run_async
     async def test_create_pipeline(self, opensearch):
         opensearch.ingest.put_pipeline.return_value = as_future()
 
-        r = runner.CreateIngestPipeline()
+        r = runner.PutPipeline()
 
         params = {
             "id": "rename",
@@ -2769,16 +2769,13 @@ async def test_create_pipeline(self, opensearch):
     async def test_param_body_mandatory(self, opensearch):
         opensearch.ingest.put_pipeline.return_value = as_future()
 
-        r = runner.CreateIngestPipeline()
+        r = runner.PutPipeline()
 
         params = {
             "id": "rename"
         }
         with self.assertRaisesRegex(exceptions.DataError,
-                                    "Parameter source "
-                                    "for operation 'create-ingest-pipeline' "
-                                    "did not provide the "
-                                    "mandatory parameter 'body'. "
+                                    "Parameter source for operation 'put-pipeline' did not provide the mandatory parameter 'body'. "
                                     "Add it to your parameter source and try again."):
             await r(opensearch, params)
 
@@ -2789,15 +2786,13 @@ async def test_param_body_mandatory(self, opensearch):
     async def test_param_id_mandatory(self, opensearch):
         opensearch.ingest.put_pipeline.return_value = as_future()
 
-        r = runner.CreateIngestPipeline()
+        r = runner.PutPipeline()
 
         params = {
             "body": {}
         }
         with self.assertRaisesRegex(exceptions.DataError,
-                                    "Parameter source for "
-                                    "operation 'create-ingest-pipeline' did"
-                                    " not provide the mandatory parameter 'id'. "
+                                    "Parameter source for operation 'put-pipeline' did not provide the mandatory parameter 'id'. "
                                     "Add it to your parameter source and try again."):
             await r(opensearch, params)