diff --git a/cassandra/numpy_parser.pyx b/cassandra/numpy_parser.pyx index 0ad34f66e2..ada59bca88 100644 --- a/cassandra/numpy_parser.pyx +++ b/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] @@ -97,7 +101,7 @@ cdef _parse_rows(BytesIOReader reader, ParseDesc desc, cdef Py_ssize_t i for i in range(rowcount): - unpack_row(reader, desc, arrs) + unpack_plain_row(reader, desc, arrs) ### Helper functions to create NumPy arrays and array descriptors @@ -144,7 +148,7 @@ def make_array(coltype, array_size): @cython.boundscheck(False) @cython.wraparound(False) -cdef inline int unpack_row( +cdef inline int unpack_plain_row( BytesIOReader reader, ParseDesc desc, ArrDesc *arrays) except -1: cdef Buffer buf cdef Py_ssize_t i, rowsize = desc.rowsize diff --git a/cassandra/obj_parser.pyx b/cassandra/obj_parser.pyx index cf43771dd7..357b078f3c 100644 --- a/cassandra/obj_parser.pyx +++ b/cassandra/obj_parser.pyx @@ -22,6 +22,7 @@ from cassandra.parsing cimport ParseDesc, ColumnParser, RowParser from cassandra.tuple cimport tuple_new, tuple_set from cpython.bytes cimport PyBytes_AsStringAndSize +cimport cython cdef class ListParser(ColumnParser): @@ -31,7 +32,10 @@ cdef class ListParser(ColumnParser): cdef Py_ssize_t i, rowcount rowcount = read_int(reader) cdef RowParser rowparser = TupleRowParser() - return [rowparser.unpack_row(reader, desc) for i in range(rowcount)] + if desc.column_encryption_policy: + return [rowparser.unpack_col_encrypted_row(reader, desc) for i in range(rowcount)] + else: + return [rowparser.unpack_plain_row(reader, desc) for i in range(rowcount)] cdef class LazyParser(ColumnParser): @@ -47,7 +51,10 @@ def parse_rows_lazy(BytesIOReader reader, ParseDesc desc): cdef Py_ssize_t i, rowcount rowcount = read_int(reader) cdef RowParser rowparser = TupleRowParser() - return (rowparser.unpack_row(reader, desc) for i in range(rowcount)) + if desc.column_encryption_policy: + return (rowparser.unpack_col_encrypted_row(reader, desc) for i in range(rowcount)) + else: + return (rowparser.unpack_plain_row(reader, desc) for i in range(rowcount)) cdef class TupleRowParser(RowParser): @@ -55,11 +62,13 @@ cdef class TupleRowParser(RowParser): Parse a single returned row into a tuple of objects: (obj1, ..., objN) + If CE (Column encryption) policy is enabled - use unpack_col_encrypted_row(), + otherwise use unpack_plain_row() """ - cpdef unpack_row(self, BytesIOReader reader, ParseDesc desc): - assert desc.rowsize >= 0 - + @cython.boundscheck(False) + @cython.wraparound(False) + cpdef unpack_col_encrypted_row(self, BytesIOReader reader, ParseDesc desc): cdef Buffer buf cdef Buffer newbuf cdef Py_ssize_t i, rowsize = desc.rowsize @@ -67,28 +76,53 @@ cdef class TupleRowParser(RowParser): cdef tuple res = tuple_new(desc.rowsize) ce_policy = desc.column_encryption_policy - for i in range(rowsize): - # Read the next few bytes - get_buf(reader, &buf) - - # Deserialize bytes to python object - deserializer = desc.deserializers[i] - coldesc = desc.coldescs[i] - uses_ce = ce_policy and ce_policy.contains_column(coldesc) - try: + try: + for i in range(rowsize): + # Read the next few bytes + get_buf(reader, &buf) + + # Deserialize bytes to python object + deserializer = desc.deserializers[i] + coldesc = desc.coldescs[i] + uses_ce = ce_policy.contains_column(coldesc) if uses_ce: 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) - except Exception as e: - raise DriverException('Failed decoding result column "%s" of type %s: %s' % (desc.colnames[i], - desc.coltypes[i].cql_parameterized_type(), - str(e))) - # Insert new object into tuple - tuple_set(res, i, val) + # Insert new object into tuple + tuple_set(res, i, val) + except Exception as e: + raise DriverException('Failed decoding result column "%s" of type %s: %s' % (desc.colnames[i], + desc.coltypes[i].cql_parameterized_type(), + str(e))) + + return res + + @cython.boundscheck(False) + @cython.wraparound(False) + cpdef unpack_plain_row(self, BytesIOReader reader, ParseDesc desc): + cdef Buffer buf + cdef Py_ssize_t i, rowsize = desc.rowsize + cdef Deserializer deserializer + cdef tuple res = tuple_new(desc.rowsize) + + try: + for i in range(rowsize): + # Read the next few bytes + get_buf(reader, &buf) + + # Deserialize bytes to python object + deserializer = desc.deserializers[i] + val = from_binary(deserializer, &buf, desc.protocol_version) + # Insert new object into tuple + tuple_set(res, i, val) + except Exception as e: + raise DriverException('Failed decoding result column "%s" of type %s: %s' % (desc.colnames[i], + desc.coltypes[i].cql_parameterized_type(), + str(e))) return res diff --git a/cassandra/parsing.pxd b/cassandra/parsing.pxd index 27dc368b07..3c8a916622 100644 --- a/cassandra/parsing.pxd +++ b/cassandra/parsing.pxd @@ -28,5 +28,6 @@ cdef class ColumnParser: cpdef parse_rows(self, BytesIOReader reader, ParseDesc desc) cdef class RowParser: - cpdef unpack_row(self, BytesIOReader reader, ParseDesc desc) + cpdef unpack_plain_row(self, BytesIOReader reader, ParseDesc desc) + cpdef unpack_col_encrypted_row(self, BytesIOReader reader, ParseDesc desc) diff --git a/cassandra/parsing.pyx b/cassandra/parsing.pyx index 954767d227..29146a1b9a 100644 --- a/cassandra/parsing.pyx +++ b/cassandra/parsing.pyx @@ -20,6 +20,18 @@ cdef class ParseDesc: """Description of what structure to parse""" def __init__(self, colnames, coltypes, column_encryption_policy, coldescs, deserializers, protocol_version): + if len(deserializers) != len(colnames): + # The row parsers (obj_parser.pyx TupleRowParser.unpack_plain_row / + # unpack_col_encrypted_row) index into `deserializers` with + # @cython.boundscheck(False), bounded by rowsize == len(colnames). + # A length mismatch here would turn into an out-of-bounds memory + # read at parse time instead of a clean, immediate error, so this + # invariant is validated once at construction time. Use a real + # exception (not `assert`) so the guard cannot be stripped by + # running Python with optimizations enabled (-O). + raise ValueError( + "deserializers must have the same length as colnames " + "(got %d deserializers for %d columns)" % (len(deserializers), len(colnames))) self.colnames = colnames self.coltypes = coltypes self.column_encryption_policy = column_encryption_policy @@ -39,8 +51,14 @@ cdef class ColumnParser: cdef class RowParser: """Parser for a single row""" - cpdef unpack_row(self, BytesIOReader reader, ParseDesc desc): + cpdef unpack_plain_row(self, BytesIOReader reader, ParseDesc desc): """ Unpack a single row of data in a ResultMessage. """ raise NotImplementedError + + cpdef unpack_col_encrypted_row(self, BytesIOReader reader, ParseDesc desc): + """ + Unpack a single row of data in a ResultMessage, with column encryption support. + """ + raise NotImplementedError diff --git a/cassandra/protocol.py b/cassandra/protocol.py index 9dfdbf3022..9911bc6670 100644 --- a/cassandra/protocol.py +++ b/cassandra/protocol.py @@ -766,28 +766,46 @@ def recv_results_rows(self, f, protocol_version, user_type_map, result_metadata, rows = [self.recv_row(f, len(column_metadata)) for _ in range(rowcount)] self.column_names = [c[2] for c in column_metadata] self.column_types = [c[3] for c in column_metadata] - col_descs = [ColDesc(md[0], md[1], md[2]) for md in column_metadata] - def decode_val(val, col_md, col_desc): - uses_ce = column_encryption_policy and column_encryption_policy.contains_column(col_desc) - col_type = column_encryption_policy.column_type(col_desc) if uses_ce else col_md[3] - raw_bytes = column_encryption_policy.decrypt(col_desc, val) if uses_ce else val - return col_type.from_binary(raw_bytes, protocol_version) + if column_encryption_policy: + col_descs = [ColDesc(md[0], md[1], md[2]) for md in column_metadata] - def decode_row(row): - return tuple(decode_val(val, col_md, col_desc) for val, col_md, col_desc in zip(row, column_metadata, col_descs)) + def decode_val(val, col_md, col_desc): + uses_ce = column_encryption_policy.contains_column(col_desc) + if uses_ce: + col_type = column_encryption_policy.column_type(col_desc) + raw_bytes = column_encryption_policy.decrypt(col_desc, val) + return col_type.from_binary(raw_bytes, protocol_version) + else: + return col_md[3].from_binary(val, protocol_version) + + def decode_row(row): + return tuple(decode_val(val, col_md, col_desc) for val, col_md, col_desc in zip(row, column_metadata, col_descs)) + else: + def decode_row(row): + return tuple(col_md[3].from_binary(val, protocol_version) for val, col_md in zip(row, column_metadata)) try: self.parsed_rows = [decode_row(row) for row in rows] except Exception: - for row in rows: - for val, col_md, col_desc in zip(row, column_metadata, col_descs): - try: - decode_val(val, col_md, col_desc) - except Exception as e: - raise DriverException('Failed decoding result column "%s" of type %s: %s' % (col_md[2], - col_md[3].cql_parameterized_type(), - str(e))) + if column_encryption_policy: + for row in rows: + for val, col_md, col_desc in zip(row, column_metadata, col_descs): + try: + decode_val(val, col_md, col_desc) + except Exception as e: + raise DriverException('Failed decoding result column "%s" of type %s: %s' % (col_md[2], + col_md[3].cql_parameterized_type(), + str(e))) + else: + for row in rows: + for val, col_md in zip(row, column_metadata): + try: + col_md[3].from_binary(val, protocol_version) + except Exception as e: + raise DriverException('Failed decoding result column "%s" of type %s: %s' % (col_md[2], + col_md[3].cql_parameterized_type(), + str(e))) def recv_results_prepared(self, f, protocol_version, protocol_features, user_type_map): self.query_id = read_binary_string(f) diff --git a/cassandra/query.py b/cassandra/query.py index 39b9fdb0ad..4a27929d2b 100644 --- a/cassandra/query.py +++ b/cassandra/query.py @@ -684,6 +684,23 @@ def bind(self, values): self.raw_values = values self.values = [] + + # The policy existence check happens once here (not per-value below) + # to select how column bytes get serialized; the per-value loop + # itself stays identical for both branches. + if ce_policy: + def _serialize(value, col_spec): + col_desc = ColDesc(col_spec.keyspace_name, col_spec.table_name, col_spec.name) + if ce_policy.contains_column(col_desc): + col_type = ce_policy.column_type(col_desc) + col_bytes = col_type.serialize(value, proto_version) + return ce_policy.encrypt(col_desc, col_bytes) + else: + return col_spec.type.serialize(value, proto_version) + else: + def _serialize(value, col_spec): + return col_spec.type.serialize(value, proto_version) + for value, col_spec in zip(values, col_meta): if value is None: self.values.append(None) @@ -694,13 +711,7 @@ def bind(self, values): raise ValueError("Attempt to bind UNSET_VALUE while using unsuitable protocol version (%d < 4)" % proto_version) else: try: - col_desc = ColDesc(col_spec.keyspace_name, col_spec.table_name, col_spec.name) - uses_ce = ce_policy and ce_policy.contains_column(col_desc) - col_type = ce_policy.column_type(col_desc) if uses_ce else col_spec.type - col_bytes = col_type.serialize(value, proto_version) - if uses_ce: - col_bytes = ce_policy.encrypt(col_desc, col_bytes) - self.values.append(col_bytes) + self.values.append(_serialize(value, col_spec)) except (TypeError, struct.error) as exc: actual_type = type(value) message = ('Received an argument of invalid type for column "%s". ' diff --git a/cassandra/row_parser.pyx b/cassandra/row_parser.pyx index 88277a4593..2c5f78253c 100644 --- a/cassandra/row_parser.pyx +++ b/cassandra/row_parser.pyx @@ -32,19 +32,30 @@ def make_recv_results_rows(ColumnParser colparser): self.column_names = [md[2] for md in column_metadata] self.column_types = [md[3] for md in column_metadata] + coldescs = [ColDesc(md[0], md[1], md[2]) for md in column_metadata] if column_encryption_policy else None desc = ParseDesc(self.column_names, self.column_types, column_encryption_policy, - [ColDesc(md[0], md[1], md[2]) for md in column_metadata], + coldescs, make_deserializers(self.column_types), protocol_version) reader = BytesIOReader(f.read()) try: self.parsed_rows = colparser.parse_rows(reader, desc) + except NotImplementedError: + # e.g. NumpyParser does not support column encryption. This + # signals an unsupported configuration, not a decoding failure, + # so it must propagate to the caller instead of being silently + # swallowed by the TupleRowParser fallback below. + raise except Exception as e: # Use explicitly the TupleRowParser to display better error messages for column decoding failures rowparser = TupleRowParser() reader.buf_ptr = reader.buf reader.pos = 0 rowcount = read_int(reader) - for i in range(rowcount): - rowparser.unpack_row(reader, desc) + if desc.column_encryption_policy: + for i in range(rowcount): + rowparser.unpack_col_encrypted_row(reader, desc) + else: + for i in range(rowcount): + rowparser.unpack_plain_row(reader, desc) return recv_results_rows diff --git a/tests/unit/test_protocol.py b/tests/unit/test_protocol.py index 75dc69bca5..9a57bacb13 100644 --- a/tests/unit/test_protocol.py +++ b/tests/unit/test_protocol.py @@ -20,6 +20,7 @@ from unittest.mock import Mock from cassandra import ConsistencyLevel, ProtocolVersion, UnsupportedOperation +from cassandra.cqltypes import Int32Type, UTF8Type from cassandra.protocol import ( PrepareMessage, QueryMessage, ExecuteMessage, BatchMessage, StartupMessage, OptionsMessage, RegisterMessage, @@ -28,8 +29,11 @@ ) from cassandra.protocol_features import ProtocolFeatures from cassandra.query import BatchType +from cassandra.marshal import int32_pack import pytest +from cassandra.policies import ColDesc +from tests.unit.cython.utils import cythontest, numpytest class MessageTest(unittest.TestCase): @@ -555,3 +559,350 @@ def test_frames_without_features(self): def test_frames_with_default_features(self): self._assert_frames(ProtocolFeatures()) + + +class _BoolCountingPolicy: + """ + Minimal column_encryption_policy stand-in whose truthiness (__bool__) + is instrumented with a counter. + + A plain Mock() is always truthy regardless of how many times it is + evaluated in a boolean context, so asserting on contains_column's call + count cannot distinguish the optimized "check policy truthiness once + per result message" code path from the old "column_encryption_policy + and ..." per-value check: both call contains_column the same number of + times. Counting __bool__ invocations directly proves which one runs. + """ + + def __init__(self): + self.bool_call_count = 0 + self.contains_column_call_count = 0 + + def __bool__(self): + self.bool_call_count += 1 + return True + + def contains_column(self, col_desc): + self.contains_column_call_count += 1 + return False + + +class ResultTest(unittest.TestCase): + """ + Tests to verify the optimization of column_encryption_policy checks + in recv_results_rows. The optimization checks if the policy exists once + per result message, avoiding the redundant 'column_encryption_policy and ...' + check for every value. + """ + + def _create_mock_result_metadata(self): + """Create mock result metadata for testing""" + return [ + ('keyspace1', 'table1', 'col1', Int32Type), + ('keyspace1', 'table1', 'col2', UTF8Type), + ] + + def _create_mock_result_message(self): + """Create a mock result message with data""" + msg = ResultMessage(kind=RESULT_KIND_ROWS) + msg.column_metadata = self._create_mock_result_metadata() + msg.recv_results_metadata = Mock() + msg.recv_row = Mock(side_effect=[ + [int32_pack(42), b'hello'], + [int32_pack(100), b'world'], + ]) + return msg + + def _create_mock_stream(self): + """Create a mock stream for reading rows""" + # Pack rowcount (2 rows) + data = int32_pack(2) + return io.BytesIO(data) + + def test_decode_without_encryption_policy(self): + """ + Test that decoding works correctly without column encryption policy. + This should use the optimized simple path. + """ + msg = self._create_mock_result_message() + f = self._create_mock_stream() + + msg.recv_results_rows(f, ProtocolVersion.V4, {}, None, None) + + # Verify results + self.assertEqual(len(msg.parsed_rows), 2) + self.assertEqual(msg.parsed_rows[0][0], 42) + self.assertEqual(msg.parsed_rows[0][1], 'hello') + self.assertEqual(msg.parsed_rows[1][0], 100) + self.assertEqual(msg.parsed_rows[1][1], 'world') + + def test_decode_with_encryption_policy_no_encrypted_columns(self): + """ + Test that decoding works with encryption policy when no columns are encrypted. + """ + msg = self._create_mock_result_message() + f = self._create_mock_stream() + + # Create mock encryption policy that has no encrypted columns + mock_policy = Mock() + mock_policy.contains_column = Mock(return_value=False) + + msg.recv_results_rows(f, ProtocolVersion.V4, {}, None, mock_policy) + + # Verify results + self.assertEqual(len(msg.parsed_rows), 2) + self.assertEqual(msg.parsed_rows[0][0], 42) + self.assertEqual(msg.parsed_rows[0][1], 'hello') + + # Verify contains_column was called for each value (but policy existence check happens once) + # Should be called 4 times (2 rows x 2 columns) + self.assertEqual(mock_policy.contains_column.call_count, 4) + + def test_decode_with_encryption_policy_with_encrypted_column(self): + """ + Test that decoding works with encryption policy when one column is encrypted. + """ + msg = self._create_mock_result_message() + f = self._create_mock_stream() + + # Create mock encryption policy where first column is encrypted + mock_policy = Mock() + def contains_column_side_effect(col_desc): + return col_desc.col == 'col1' + mock_policy.contains_column = Mock(side_effect=contains_column_side_effect) + mock_policy.column_type = Mock(return_value=Int32Type) + mock_policy.decrypt = Mock(side_effect=lambda col_desc, val: val) + + msg.recv_results_rows(f, ProtocolVersion.V4, {}, None, mock_policy) + + # Verify results + self.assertEqual(len(msg.parsed_rows), 2) + self.assertEqual(msg.parsed_rows[0][0], 42) + self.assertEqual(msg.parsed_rows[0][1], 'hello') + + # Verify contains_column was called for each value (but policy existence check happens once) + # Should be called 4 times (2 rows x 2 columns) + self.assertEqual(mock_policy.contains_column.call_count, 4) + + # Verify decrypt was called for each encrypted value (2 rows * 1 encrypted column) + self.assertEqual(mock_policy.decrypt.call_count, 2) + + def test_optimization_efficiency(self): + """ + Verify that the optimization checks policy existence once per result message. + The key optimization is checking 'if column_encryption_policy:' once, + rather than 'column_encryption_policy and ...' for every value. + + A plain Mock() is always truthy no matter how many times it is + evaluated, so counting contains_column calls alone cannot tell the + optimized code apart from the old per-value + 'column_encryption_policy and column_encryption_policy.contains_column(...)' + check: both call contains_column 200 times (100 rows * 2 columns) + either way. Using a policy whose __bool__ is instrumented catches a + regression back to the old hot-loop check, where truthiness would + be evaluated once per value instead of once per result message. + """ + msg = self._create_mock_result_message() + + # Create more rows to make the check pattern clear + msg.recv_row = Mock(side_effect=[ + [int32_pack(i), f'text{i}'.encode()] for i in range(100) + ]) + + # Create mock stream with 100 rows + f = io.BytesIO(int32_pack(100)) + + policy = _BoolCountingPolicy() + + msg.recv_results_rows(f, ProtocolVersion.V4, {}, None, policy) + + # With optimization: policy existence checked once, contains_column called per value + # = 100 rows * 2 columns = 200 calls to contains_column + # The key is we avoid checking 'column_encryption_policy and ...' 200 times + self.assertEqual(policy.contains_column_call_count, 200, + "contains_column should be called for each value when policy exists") + + # The actual optimization being verified: policy truthiness ('if + # column_encryption_policy:') is evaluated exactly once per result + # message, not once per value/row (which would be 200, or 100 if + # checked once per row). + self.assertEqual(policy.bool_call_count, 1, + "column_encryption_policy truthiness should be checked exactly once " + "per result message, not in the per-value/per-row hot loop") + + +@cythontest +class CythonParserTest(unittest.TestCase): + """ + Tests for the Cython fast-path parsers (ListParser, TupleRowParser) + to verify the column_encryption_policy optimization in obj_parser.pyx. + + Requires the Cython extensions (cassandra.bytesio, cassandra.obj_parser, + cassandra.parsing, ...) to be built, which is not the case e.g. on PyPy + wheels (see setup.py: try_cython is disabled for PyPy). Without this + guard these tests fail with ModuleNotFoundError instead of skipping. + """ + + 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_parse_desc_rejects_mismatched_deserializers_length(self): + """ + ParseDesc.__init__ must validate that deserializers has the same + length as colnames. + + TupleRowParser.unpack_plain_row / unpack_col_encrypted_row (in + obj_parser.pyx) index into desc.deserializers[i] for i in + range(desc.rowsize), where rowsize == len(colnames), under + @cython.boundscheck(False). If deserializers were ever shorter than + colnames, that indexing would become an out-of-bounds memory read + instead of a safe IndexError. Since that can't be constructed by + the normal production code path (row_parser.pyx always builds + colnames/coltypes/deserializers from the same column_metadata), + this test instead verifies the construction-time guard itself: + it must raise on a mismatch and stay silent when lengths agree. + See GH PR #630 review discussion. + """ + 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')] + + # Mismatched: only one deserializer for two columns. + short_deserializers = make_deserializers(coltypes[:1]) + with self.assertRaises(ValueError): + ParseDesc(colnames, coltypes, None, coldescs, short_deserializers, ProtocolVersion.V4) + + # Matching lengths must not raise. + deserializers = make_deserializers(coltypes) + desc = ParseDesc(colnames, coltypes, None, coldescs, deserializers, ProtocolVersion.V4) + self.assertEqual(desc.colnames, colnames) + + 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) + + @numpytest + def test_numpy_parser_rejects_encryption(self): + """ + NumPy result parsing + column_encryption_policy is an unsupported + combination and must raise NotImplementedError. + + This exercises the real production path - + row_parser.make_recv_results_rows(NumpyParser()), the same wrapper + used by NumpyProtocolHandler - rather than calling + NumpyParser().parse_rows() directly. That wrapper has a broad + 'except Exception' fallback to TupleRowParser for decoding + failures; it must not swallow NotImplementedError raised for this + unsupported configuration (see GH PR #630 review discussion). + """ + from cassandra.numpy_parser import NumpyParser + from cassandra.row_parser import make_recv_results_rows + + class _FastResultMessageForTest(ResultMessage): + recv_results_rows = make_recv_results_rows(NumpyParser()) + + mock_policy = Mock() + msg = _FastResultMessageForTest(kind=RESULT_KIND_ROWS) + msg.column_metadata = [ + ('ks', 'tbl', 'col1', Int32Type), + ('ks', 'tbl', 'col2', UTF8Type), + ] + msg.recv_results_metadata = Mock() + + data = self._build_binary_rows([[self._int32_bytes(1), b'x']]) + f = io.BytesIO(data) + + with self.assertRaises(NotImplementedError): + msg.recv_results_rows(f, ProtocolVersion.V4, {}, None, mock_policy)