TESTS: test the curl wrapper with a command-line tool

In order to test the curl integration code, this patch adds a
command-line tool and tests that it's possible to drive a conversation
with the secrets responder using the tool.

Reviewed-by: Pavel Březina <pbrezina@redhat.com>
Reviewed-by: Lukáš Slebodník <lslebodn@redhat.com>

1 files changed, 142 insertions, 1 deletions

diff --git a/src/tests/intg/test_secrets.py b/src/tests/intg/test_secrets.py
index 36f43cd7a..cbc1a1f06 100644
--- a/src/tests/intg/test_secrets.py
+++ b/src/tests/intg/test_secrets.py
@@ -76,6 +76,7 @@ def create_sssd_secrets_fixture(request):
         for secdb_file in os.listdir(config.SECDB_PATH):
             os.unlink(config.SECDB_PATH + "/" + secdb_file)
     request.addfinalizer(sec_teardown)
+    return secpid
 
 
 @pytest.fixture
@@ -102,13 +103,27 @@ def setup_for_secrets(request):
     return None
 
 
+def get_secrets_socket():
+    return os.path.join(config.RUNSTATEDIR, "secrets.socket")
+
+
 @pytest.fixture
 def secrets_cli(request):
-    sock_path = os.path.join(config.RUNSTATEDIR, "secrets.socket")
+    sock_path = get_secrets_socket()
     cli = SecretsLocalClient(sock_path=sock_path)
     return cli
 
 
+@pytest.fixture
+def curlwrap_tool(request):
+    curlwrap_path = os.path.join(config.ABS_BUILDDIR,
+                                 "..", "..", "..", "tcurl-test-tool")
+    if os.access(curlwrap_path, os.X_OK):
+        return curlwrap_path
+
+    return None
+
+
 def test_crd_ops(setup_for_secrets, secrets_cli):
     """
     Test that the basic Create, Retrieve, Delete operations work
@@ -178,6 +193,132 @@ def test_crd_ops(setup_for_secrets, secrets_cli):
     assert str(err413.value).startswith("413")
 
 
+def run_curlwrap_tool(args, exp_http_code):
+    cmd = subprocess.Popen(args,
+                           stdin=subprocess.PIPE,
+                           stdout=subprocess.PIPE,
+                           stderr=subprocess.PIPE)
+    out, _ = cmd.communicate()
+
+    assert cmd.returncode == 0
+
+    out = out.decode('utf-8')
+    exp_http_code_str = "Request HTTP code: %d" % exp_http_code
+    assert exp_http_code_str in out
+
+    return out
+
+
+def test_curlwrap_crd_ops(setup_for_secrets,
+                          curlwrap_tool):
+    """
+    Test that the basic Create, Retrieve, Delete operations work using our
+    tevent libcurl code
+    """
+    if not curlwrap_tool:
+        pytest.skip("The tcurl tool is not available, skipping test")
+    sock_path = get_secrets_socket()
+
+    # listing an empty DB yields a 404
+    run_curlwrap_tool([curlwrap_tool,
+                       '-v', '-s', sock_path,
+                       'http://localhost/secrets/'],
+                      404)
+
+    # listing a non-existent secret yields a 404
+    run_curlwrap_tool([curlwrap_tool,
+                       '-v', '-s', sock_path,
+                       'http://localhost/secrets/foo'],
+                      404)
+
+    # set a secret foo:bar
+    run_curlwrap_tool([curlwrap_tool, '-p',
+                       '-v', '-s', sock_path,
+                       'http://localhost/secrets/foo',
+                       'bar'],
+                      200)
+
+    # list secrets
+    output = run_curlwrap_tool([curlwrap_tool,
+                                '-v', '-s', sock_path,
+                                'http://localhost/secrets/'],
+                               200)
+    assert "foo" in output
+
+    # get the foo secret
+    output = run_curlwrap_tool([curlwrap_tool,
+                                '-v', '-s', sock_path,
+                                'http://localhost/secrets/foo'],
+                               200)
+    assert "bar" in output
+
+    # Overwriting a secret is an error
+    run_curlwrap_tool([curlwrap_tool, '-p',
+                       '-v', '-s', sock_path,
+                       'http://localhost/secrets/foo',
+                       'baz'],
+                      409)
+
+    # Delete a secret
+    run_curlwrap_tool([curlwrap_tool, '-d',
+                       '-v', '-s', sock_path,
+                       'http://localhost/secrets/foo'],
+                      200)
+
+    # Delete a non-existent secret must yield a 404
+    run_curlwrap_tool([curlwrap_tool, '-d',
+                       '-v', '-s', sock_path,
+                       'http://localhost/secrets/foo'],
+                      404)
+
+
+def test_curlwrap_parallel(setup_for_secrets,
+                           curlwrap_tool):
+    """
+    The tevent libcurl wrapper is meant to be non-blocking. Test
+    its operation in parallel.
+    """
+    if not curlwrap_tool:
+        pytest.skip("The tcurl tool is not available, skipping test")
+    sock_path = get_secrets_socket()
+
+    secrets = dict()
+    nsecrets = 10
+
+    for i in range(0, nsecrets):
+        secrets["key" + str(i)] = "value" + str(i)
+
+    args = [curlwrap_tool, '-p', '-v', '-s', sock_path]
+    for skey, svalue in secrets.items():
+        args.extend(['http://localhost/secrets/%s' % skey, svalue])
+    run_curlwrap_tool(args, 200)
+
+    output = run_curlwrap_tool([curlwrap_tool,
+                                '-v', '-s', sock_path,
+                                'http://localhost/secrets/'],
+                               200)
+    for skey in secrets:
+        assert skey in output
+
+    args = [curlwrap_tool, '-g', '-v', '-s', sock_path]
+    for skey in secrets:
+        args.extend(['http://localhost/secrets/%s' % skey])
+    output = run_curlwrap_tool(args, 200)
+
+    for svalue in secrets.values():
+        assert svalue in output
+
+    args = [curlwrap_tool, '-d', '-v', '-s', sock_path]
+    for skey in secrets:
+        args.extend(['http://localhost/secrets/%s' % skey])
+    output = run_curlwrap_tool(args, 200)
+
+    run_curlwrap_tool([curlwrap_tool,
+                       '-v', '-s', sock_path,
+                       'http://localhost/secrets/'],
+                      404)
+
+
 def test_containers(setup_for_secrets, secrets_cli):
     """
     Test that storing secrets inside containers works