[rdma-core,2/6] pyverbs: Add unittests for completion-related classes

Commit Message

Noa Osherovich March 24, 2019, 2:10 p.m. UTC

Add unittests for creation and destruction of CQ an CompChannel:
- Creation and deletion of a CompChannel (good flow).
- Creation and deletion of a CQ (good flow), with and without a
  completion channel.
- Bad flow checks: Verify failure for illegal comp_vector / cqe number.

Signed-off-by: Noa Osherovich <[e-mail]>
Reviewed-by: Maor Gottlieb <[e-mail]>
Reviewed-by: Leon Romanovsky <[e-mail]>
---
 pyverbs/CMakeLists.txt |   1 +
 pyverbs/tests/cq.py    | 130 +++++++++++++++++++++++++++++++++++++++++
 2 files changed, 131 insertions(+)
 create mode 100644 pyverbs/tests/cq.py

Patch

diff --git a/pyverbs/CMakeLists.txt b/pyverbs/CMakeLists.txt
index ef5114594e20..d83f77317ecb 100644
--- a/pyverbs/CMakeLists.txt
+++ b/pyverbs/CMakeLists.txt
@@ -19,6 +19,7 @@  rdma_python_module(pyverbs
 
 rdma_python_test(pyverbs/tests
   tests/__init__.py
+  tests/cq.py
   tests/device.py
   tests/mr.py
   tests/pd.py
diff --git a/pyverbs/tests/cq.py b/pyverbs/tests/cq.py
new file mode 100644
index 000000000000..afe732e5e93c
--- /dev/null
+++ b/pyverbs/tests/cq.py
@@ -0,0 +1,130 @@ 
+# SPDX-License-Identifier: (GPL-2.0 OR Linux-OpenIB)
+# Copyright (c) 2019, Mellanox Technologies. All rights reserved.  See COPYING file
+"""
+Test module for pyverbs' cq module.
+"""
+import unittest
+import random
+
+from pyverbs.pyverbs_error import PyverbsError
+from pyverbs.cq import CompChannel, CQ
+import pyverbs.device as d
+import pyverbs.enums as e
+
+
+class CQTest(unittest.TestCase):
+    """
+    Test various functionalities of the CQ class.
+    """
+
+    @staticmethod
+    def test_create_cq():
+        """
+        Test ibv_create_cq()
+        """
+        lst = d.get_device_list()
+        for dev in lst:
+            with d.Context(name=dev.name.decode()) as ctx:
+                cqes = get_num_cqes(ctx)
+                comp_vector = random.randint(0, ctx.num_comp_vectors - 1)
+                if random.choice([True, False]):
+                    with CompChannel(ctx) as cc:
+                        with CQ(ctx, cqes, None, cc, comp_vector):
+                            pass
+                else:
+                    with CQ(ctx, cqes, None, None, comp_vector):
+                        pass
+
+    @staticmethod
+    def test_create_cq_bad_flow():
+        """
+        Test ibv_create_cq() with a wrong comp_vector / cqe number
+        """
+        lst = d.get_device_list()
+        for dev in lst:
+            with d.Context(name=dev.name.decode()) as ctx:
+                cqes = get_num_cqes(ctx)
+                comp_vector = random.randint(ctx.num_comp_vectors, 100)
+                try:
+                    if random.choice([True, False]):
+                        with CompChannel(ctx) as cc:
+                            with CQ(ctx, cqes, None, cc, comp_vector):
+                                pass
+                    else:
+                        with CQ(ctx, cqes, None, None, comp_vector):
+                            pass
+                except PyverbsError as e:
+                    assert 'Failed to create a CQ' in e.args[0]
+                    assert 'Invalid argument' in e.args[0]
+                else:
+                    raise PyverbsError(
+                        'Created a CQ with comp_vector={n} while device\'s num_comp_vectors={nc}'.
+                        format(n=comp_vector, nc=ctx.num_comp_vectors))
+                max_cqe = ctx.query_device().max_cqe
+                cqes = random.randint(max_cqe + 1, max_cqe + 100)
+                try:
+                    if random.choice([True, False]):
+                        with CompChannel(ctx) as cc:
+                            with CQ(ctx, cqes, None, cc, 0):
+                                pass
+                    else:
+                        with CQ(ctx, cqes, None, None, 0):
+                            pass
+                except PyverbsError as err:
+                    assert 'Failed to create a CQ' in err.args[0]
+                    assert 'Invalid argument' in err.args[0]
+                else:
+                    raise PyverbsError(
+                        'Created a CQ with cqe={n} while device\'s max_cqe={nc}'.
+                        format(n=cqes, nc=max_cqe))
+
+    @staticmethod
+    def test_destroy_cq():
+        """
+        Test ibv_destroy_cq()
+        """
+        lst = d.get_device_list()
+        for dev in lst:
+            with d.Context(name=dev.name.decode()) as ctx:
+                cqes = get_num_cqes(ctx)
+                comp_vector = random.randint(0, ctx.num_comp_vectors - 1)
+                if random.choice([True, False]):
+                    with CompChannel(ctx) as cc:
+                        cq = CQ(ctx, cqes, None, cc, comp_vector)
+                else:
+                    cq = CQ(ctx, cqes, None, None, comp_vector)
+                cq.close()
+
+
+class CCTest(unittest.TestCase):
+    """
+    Test various functionalities of the Completion Channel class.
+    """
+
+    @staticmethod
+    def test_create_comp_channel():
+        """
+        Test ibv_create_comp_channel()
+        """
+        lst = d.get_device_list()
+        for dev in lst:
+            with d.Context(name=dev.name.decode()) as ctx:
+                with CompChannel(ctx):
+                    pass
+
+    @staticmethod
+    def test_destroy_comp_channel():
+        """
+        Test ibv_destroy_comp_channel()
+        """
+        lst = d.get_device_list()
+        for dev in lst:
+            with d.Context(name=dev.name.decode()) as ctx:
+                cc = CompChannel(ctx)
+                cc.close()
+
+
+def get_num_cqes(ctx):
+    attr = ctx.query_device()
+    max_cqe = attr.max_cqe
+    return random.randint(0, max_cqe)