Add unittests for extended CQ control path:
- Creation and deletion of an extended 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 <noaos@xxxxxxxxxxxx>
Reviewed-by: Maor Gottlieb <maorg@xxxxxxxxxxxx>
Reviewed-by: Leon Romanovsky <leonro@xxxxxxxxxxxx>
---
pyverbs/tests/cq.py | 81 ++++++++++++++++++++++++++++++++++++++++++++-
1 file changed, 80 insertions(+), 1 deletion(-)
diff --git a/pyverbs/tests/cq.py b/pyverbs/tests/cq.py
index afe732e5e93c..4c1cea2b5296 100644
--- a/pyverbs/tests/cq.py
+++ b/pyverbs/tests/cq.py
@@ -6,8 +6,8 @@ Test module for pyverbs' cq module.
import unittest
import random
+from pyverbs.cq import CompChannel, CQ, CqInitAttrEx, CQEX
from pyverbs.pyverbs_error import PyverbsError
-from pyverbs.cq import CompChannel, CQ
import pyverbs.device as d
import pyverbs.enums as e
@@ -124,7 +124,86 @@ class CCTest(unittest.TestCase):
cc.close()
+class CQEXTest(unittest.TestCase):
+ """
+ Test various functionalities of the CQEX class.
+ """
+
+ @staticmethod
+ def test_create_cq_ex():
+ """
+ Test ibv_create_cq_ex()
+ """
+ lst = d.get_device_list()
+ for dev in lst:
+ with d.Context(name=dev.name.decode()) as ctx:
+ with CQEX(ctx, get_attrs_ex(ctx)):
+ pass
+
+ @staticmethod
+ def test_create_cq_ex_bad_flow():
+ """
+ Test ibv_create_cq_ex() with wrong comp_vector / number of cqes
+ """
+ lst = d.get_device_list()
+ for dev in lst:
+ with d.Context(name=dev.name.decode()) as ctx:
+ attrs_ex = get_attrs_ex(ctx)
+ max_cqe = ctx.query_device().max_cqe
+ attrs_ex.cqe = max_cqe + random.randint(1, 100)
+ try:
+ CQEX(ctx, attrs_ex)
+ except PyverbsError as e:
+ assert 'Failed to create extended CQ' in e.args[0]
+ assert ' Errno: 22' in e.args[0]
+ else:
+ raise PyverbsError(
+ 'Created a CQEX with {c} CQEs while device\'s max CQE={dc}'.
+ format(c=attrs_ex.cqe, dc=max_cqe))
+ comp_channel = random.randint(ctx.num_comp_vectors, 100)
+ attrs_ex.comp_vector = comp_channel
+ attrs_ex.cqe = get_num_cqes(ctx)
+ try:
+ CQEX(ctx, attrs_ex)
+ except PyverbsError as e:
+ assert 'Failed to create extended CQ' in e.args[0]
+ assert ' Errno: 22' in e.args[0]
+ else:
+ raise PyverbsError(
+ 'Created a CQEX with comp_vector={c} while device\'s num_comp_vectors={dc}'.
+ format(c=comp_channel, dc=ctx.num_comp_vectors))
+
+ @staticmethod
+ def test_destroy_cq_ex():
+ """
+ Test ibv_destroy_cq() for extended CQs
+ """
+ lst = d.get_device_list()
+ for dev in lst:
+ with d.Context(name=dev.name.decode()) as ctx:
+ with CQEX(ctx, get_attrs_ex(ctx)) as cq:
+ cq.close()
+
+
def get_num_cqes(ctx):
attr = ctx.query_device()
max_cqe = attr.max_cqe
return random.randint(0, max_cqe)
+
+
+def get_attrs_ex(ctx):
+ cqe = get_num_cqes(ctx)
+ sample = random.sample(list(e.ibv_create_cq_wc_flags),
+ random.randint(0, 11))
+ wc_flags = 0
+ for flag in sample:
+ wc_flags |= flag
+ comp_mask = random.choice([0, e.IBV_CQ_INIT_ATTR_MASK_FLAGS])
+ flags = 0
+ if comp_mask is not 0:
+ sample = random.sample(list(e.ibv_create_cq_attr_flags),
+ random.randint(0, 2))
+ for flag in sample:
+ flags |= flag
+ return CqInitAttrEx(cqe=cqe, wc_flags=wc_flags, comp_mask=comp_mask,
+ flags=flags)
--
2.17.2
