NumpyParser: reject column encryption; fix double column_type() call; add Cython parser tests

- NumpyParser.parse_rows() now raises NotImplementedError when
  column_encryption_policy is set, as NumpyParser does not support
  column encryption (reviewer feedback from dkropachev).
- Fix redundant ce_policy.column_type(coldesc) call in
  TupleRowParser.unpack_col_encrypted_row(); reuse the already-assigned
  col_type variable.
- Add CythonParserTest covering ListParser and NumpyParser paths.

Author: mykaul

cassandra/numpy_parser.pyx

@@ -81,6 +81,10 @@ cdef class NumpyParser(ColumnParser):
         cdef ArrDesc[::1] array_descs
         cdef ArrDesc *arrs
 
+        if desc.column_encryption_policy:
+            raise NotImplementedError(
+                "NumpyParser does not support column encryption")
+
         rowcount = read_int(reader)
         array_descs, arrays = make_arrays(desc, rowcount)
         arrs = &array_descs[0]

cassandra/obj_parser.pyx

@@ -89,7 +89,7 @@ cdef class TupleRowParser(RowParser):
                     col_type = ce_policy.column_type(coldesc)
                     decrypted_bytes = ce_policy.decrypt(coldesc, to_bytes(&buf))
                     PyBytes_AsStringAndSize(decrypted_bytes, &newbuf.ptr, &newbuf.size)
-                    deserializer = find_deserializer(ce_policy.column_type(coldesc))
+                    deserializer = find_deserializer(col_type)
                     val = from_binary(deserializer, &newbuf, desc.protocol_version)
                 else:
                     val = from_binary(deserializer, &buf, desc.protocol_version)

tests/unit/test_protocol.py

@@ -320,3 +320,122 @@ def test_optimization_efficiency(self):
         # The key is we avoid checking 'column_encryption_policy and ...' 200 times
         self.assertEqual(mock_policy.contains_column.call_count, 200,
                         "contains_column should be called for each value when policy exists")
+
+
+class CythonParserTest(unittest.TestCase):
+    """
+    Tests for the Cython fast-path parsers (ListParser, TupleRowParser)
+    to verify the column_encryption_policy optimization in obj_parser.pyx.
+    """
+
+    def _build_binary_rows(self, rows):
+        """
+        Build a binary buffer containing encoded rows.
+
+        Each row is a list of (size, raw_bytes) pairs.
+        Prepends a 4-byte big-endian row count.
+        """
+        import struct
+        data = struct.pack('>i', len(rows))
+        for row in rows:
+            for raw in row:
+                if raw is None:
+                    data += struct.pack('>i', -1)  # NULL
+                else:
+                    data += struct.pack('>i', len(raw)) + raw
+        return data
+
+    def _make_parse_desc(self, column_encryption_policy=None):
+        from cassandra.parsing import ParseDesc
+        from cassandra.deserializers import make_deserializers
+        from cassandra.policies import ColDesc
+
+        colnames = ['col1', 'col2']
+        coltypes = [Int32Type, UTF8Type]
+        coldescs = [ColDesc('ks', 'tbl', 'col1'), ColDesc('ks', 'tbl', 'col2')]
+        deserializers = make_deserializers(coltypes)
+        return ParseDesc(colnames, coltypes, column_encryption_policy,
+                         coldescs, deserializers, ProtocolVersion.V4)
+
+    def _int32_bytes(self, val):
+        import struct
+        return struct.pack('>i', val)
+
+    def test_list_parser_without_encryption(self):
+        """ListParser decodes rows correctly without encryption policy."""
+        from cassandra.bytesio import BytesIOReader
+        from cassandra.obj_parser import ListParser
+
+        desc = self._make_parse_desc(column_encryption_policy=None)
+        data = self._build_binary_rows([
+            [self._int32_bytes(42), b'hello'],
+            [self._int32_bytes(100), b'world'],
+        ])
+        reader = BytesIOReader(data)
+        result = ListParser().parse_rows(reader, desc)
+
+        self.assertEqual(len(result), 2)
+        self.assertEqual(result[0], (42, 'hello'))
+        self.assertEqual(result[1], (100, 'world'))
+
+    def test_list_parser_with_encryption_no_encrypted_cols(self):
+        """ListParser decodes rows correctly when policy exists but no columns are encrypted."""
+        from cassandra.bytesio import BytesIOReader
+        from cassandra.obj_parser import ListParser
+
+        mock_policy = Mock()
+        mock_policy.contains_column = Mock(return_value=False)
+
+        desc = self._make_parse_desc(column_encryption_policy=mock_policy)
+        data = self._build_binary_rows([
+            [self._int32_bytes(42), b'hello'],
+        ])
+        reader = BytesIOReader(data)
+        result = ListParser().parse_rows(reader, desc)
+
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0], (42, 'hello'))
+        # 1 row * 2 columns = 2 calls
+        self.assertEqual(mock_policy.contains_column.call_count, 2)
+
+    def test_list_parser_with_encrypted_column(self):
+        """ListParser decodes rows with an encrypted column (mock decrypt is identity)."""
+        from cassandra.bytesio import BytesIOReader
+        from cassandra.obj_parser import ListParser
+        from cassandra.deserializers import find_deserializer
+
+        mock_policy = Mock()
+        mock_policy.contains_column = Mock(
+            side_effect=lambda cd: cd.col == 'col1')
+        mock_policy.column_type = Mock(return_value=Int32Type)
+        # decrypt returns the raw bytes unchanged (identity)
+        mock_policy.decrypt = Mock(side_effect=lambda cd, val: val)
+
+        desc = self._make_parse_desc(column_encryption_policy=mock_policy)
+        data = self._build_binary_rows([
+            [self._int32_bytes(7), b'test'],
+        ])
+        reader = BytesIOReader(data)
+        result = ListParser().parse_rows(reader, desc)
+
+        self.assertEqual(len(result), 1)
+        self.assertEqual(result[0], (7, 'test'))
+        self.assertEqual(mock_policy.decrypt.call_count, 1)
+        self.assertEqual(mock_policy.column_type.call_count, 1)
+
+    def test_numpy_parser_rejects_encryption(self):
+        """NumpyParser raises NotImplementedError when column_encryption_policy is set."""
+        try:
+            from cassandra.numpy_parser import NumpyParser
+        except ImportError:
+            self.skipTest("NumPy or numpy_parser not available")
+
+        from cassandra.bytesio import BytesIOReader
+
+        mock_policy = Mock()
+        desc = self._make_parse_desc(column_encryption_policy=mock_policy)
+        data = self._build_binary_rows([[self._int32_bytes(1), b'x']])
+        reader = BytesIOReader(data)
+
+        with self.assertRaises(NotImplementedError):
+            NumpyParser().parse_rows(reader, desc)