test/cql-pytest.py: add scylla_inject_error() utility

This patch adds a scylla_inject_error(), a context manager which tests
can use to temporarily enable some error injection while some test
code is running. It can be used to write tests that artificially
inject certain errors instead of trying to reach the elaborate (and
often requiring precise timing or high amounts of data) situation where
they occur naturally.

The error-injection API is Scylla-specific (it uses the Scylla REST API)
and does not work on "release"-mode builds (all other modes are supported),
so when Cassandra or release-mode build are being tested, the test which
uses scylla_inject_error() gets skipped.

Example usage:

```python
    from rest_api import scylla_inject_error
    with scylla_inject_error(cql, "injection_name", one_shot=True):
        # do something here
        ...
```

Signed-off-by: Nadav Har'El <nyh@scylladb.com>

Closes #12264

(cherry picked from commit 6d2e146aa6)

test/cql-pytest/rest_api.py

@@ -9,8 +9,9 @@
 import requests
 import nodetool
 import pytest
+from contextlib import contextmanager
 
-# Sends GET request to REST API. Response is restured as JSON.
+# Sends GET request to REST API. Response is returned as JSON.
 # If API isn't available, `pytest.skip()` is called.
 def get_request(cql, *path):
     if nodetool.has_rest_api(cql):
@@ -19,6 +20,31 @@ def get_request(cql, *path):
     else:
         pytest.skip("REST API not available")
 
+# Sends POST request to REST API. Response is returned as JSON or None
+# if the response body was empty (this is typical).
+# If API isn't available, `pytest.skip()` is called.
+def post_request(cql, *path):
+    if nodetool.has_rest_api(cql):
+        response = requests.post(f"{nodetool.rest_api_url(cql)}/{'/'.join(path)}")
+        if not response.text:
+            return None
+        return response.json()
+    else:
+        pytest.skip("REST API not available")
+
+# Sends DELETE request to REST API. Response is returned as JSON or None
+# if the response body was empty (this is typical).
+# If API isn't available, `pytest.skip()` is called.
+def delete_request(cql, *path):
+    if nodetool.has_rest_api(cql):
+        response = requests.delete(f"{nodetool.rest_api_url(cql)}/{'/'.join(path)}")
+        if not response.text:
+            return None
+        return response.json()
+    else:
+        pytest.skip("REST API not available")
+
+
 # Get column family's metric.
 # metric - name of matric
 # table - (optional) column family name to add to request's path
@@ -29,4 +55,24 @@ def get_column_family_metric(cql, metric, table=None):
         ks, cf = table.split('.')
         args.append(f"{ks}:{cf}")
         
-    return get_request(cql, *args)
+    return get_request(cql, *args)
+
+# scylla_inject_error() is a context manager, running a block of code with
+# the given error injection enabled - and automatically disabling the
+# injection when the block exits.
+# This error-injection feature uses Scylla's REST API, so it only works on
+# Scylla. Also, only works in specific build modes (dev, debug, sanitize).
+# When Cassandra or non-supporting build of Scylla is being tested, using
+# this function will cause the calling test to be skipped.
+@contextmanager
+def scylla_inject_error(cql, err, one_shot=False):
+    post_request(cql, f'v2/error_injection/injection/{err}?one_shot={one_shot}')
+    response = get_request(cql, f'v2/error_injection/injection')
+    print("Enabled error injections:", response)
+    if not err in response:
+        pytest.skip("Error injection not enabled in Scylla - try compiling in dev/debug/sanitize mode")
+    try:
+        yield
+    finally:
+        print("Disabling error injection", err)
+        delete_request(cql, f'v2/error_injection/injection/{err}')