DRIVER-153: tests for SCYLLA_USE_METADATA_ID extension

test_protocol_features.py:
- test_use_metadata_id_parsing: SCYLLA_USE_METADATA_ID parsed from SUPPORTED
- test_use_metadata_id_missing: use_metadata_id False when key absent
- test_use_metadata_id_startup_options: key present in STARTUP when negotiated
- test_use_metadata_id_not_in_startup_when_not_negotiated: absent otherwise

test_protocol.py:
- test_execute_message_skip_meta_flag: _SKIP_METADATA_FLAG (0x02) is written
- test_execute_message_scylla_metadata_id_v4: result_metadata_id written on
  v4 when use_metadata_id=True (Scylla extension)
- test_execute_message_scylla_metadata_id_none_writes_sentinel: extension
  active but result_metadata_id=None writes empty sentinel b'' (LWT / mixed
  cluster); frame layout preserved
- test_execute_message_v5_metadata_id_none_writes_sentinel: v5 with
  result_metadata_id=None writes empty sentinel instead of TypeError crash
- test_recv_results_prepared_scylla_extension_reads_metadata_id
- test_recv_results_prepared_no_extension_skips_metadata_id
- test_recv_results_metadata_changed_flag
- test_recv_results_metadata_no_metadata_flag_skips_metadata_id

test_response_future.py:
_set_result METADATA_CHANGED path:
- test_set_result_updates_metadata_when_metadata_changed
- test_set_result_does_not_update_metadata_when_metadata_id_absent
- test_set_result_warns_when_metadata_id_but_no_column_metadata

_query per-connection feature gating (5 scenarios):
- test_query_sets_skip_meta_with_scylla_extension
- test_query_no_skip_meta_without_extension
- test_query_no_skip_meta_when_prepared_statement_has_no_metadata_id
- test_query_sets_skip_meta_for_protocol_v5
- test_query_no_skip_meta_when_result_metadata_is_none (LWT guard)

Author: nikagra

tests/unit/test_protocol.py

@@ -12,6 +12,8 @@
 # See the License for the specific language governing permissions and
 # limitations under the License.
 
+import io
+import struct
 import unittest
 
 from unittest.mock import Mock
@@ -21,8 +23,11 @@
     PrepareMessage, QueryMessage, ExecuteMessage, UnsupportedOperation,
     _PAGING_OPTIONS_FLAG, _WITH_SERIAL_CONSISTENCY_FLAG,
     _PAGE_SIZE_FLAG, _WITH_PAGING_STATE_FLAG,
-    BatchMessage
+    _SKIP_METADATA_FLAG,
+    BatchMessage, ResultMessage,
+    RESULT_KIND_ROWS
 )
+from cassandra.protocol_features import ProtocolFeatures
 from cassandra.query import BatchType
 from cassandra.marshal import uint32_unpack
 from cassandra.cluster import ContinuousPagingOptions
@@ -68,6 +73,162 @@ def test_execute_message(self):
                                (b'\x00\x04',),
                                (b'\x00\x00\x00\x01',), (b'\x00\x00',)])
 
+    def test_execute_message_skip_meta_flag(self):
+        """skip_meta=True must set _SKIP_METADATA_FLAG (0x02) in the flags byte."""
+        message = ExecuteMessage('1', [], 4, skip_meta=True)
+        mock_io = Mock()
+
+        message.send_body(mock_io, 4)
+        # flags byte should be VALUES_FLAG | SKIP_METADATA_FLAG = 0x01 | 0x02 = 0x03
+        self._check_calls(mock_io, [(b'\x00\x01',), (b'1',), (b'\x00\x04',), (b'\x03',), (b'\x00\x00',)])
+
+    def test_execute_message_scylla_metadata_id_v4(self):
+        """result_metadata_id should be written on protocol v4 when use_metadata_id=True (Scylla extension)."""
+        message = ExecuteMessage('1', [], 4)
+        message.result_metadata_id = b'foo'
+        message.use_metadata_id = True
+        mock_io = Mock()
+
+        message.send_body(mock_io, 4)
+        # metadata_id written before query params (same position as v5)
+        self._check_calls(mock_io, [(b'\x00\x01',), (b'1',),
+                                    (b'\x00\x03',), (b'foo',),
+                                    (b'\x00\x04',), (b'\x01',), (b'\x00\x00',)])
+
+    def test_execute_message_scylla_metadata_id_none_writes_sentinel(self):
+        """
+        When use_metadata_id=True but result_metadata_id is None (e.g. LWT statement or
+        mixed cluster), send_body must still write the field as an empty string sentinel
+        (\\x00\\x00) so the frame layout matches what the server expects.
+        """
+        message = ExecuteMessage('1', [], 4)
+        message.use_metadata_id = True
+        # result_metadata_id intentionally left as None
+        mock_io = Mock()
+
+        message.send_body(mock_io, 4)
+        # empty sentinel: \x00\x00 (zero-length short) + b'' (zero bytes), then normal query params
+        self._check_calls(mock_io, [(b'\x00\x01',), (b'1',),
+                                    (b'\x00\x00',), (b'',),
+                                    (b'\x00\x04',), (b'\x01',), (b'\x00\x00',)])
+
+    def test_execute_message_v5_metadata_id_none_writes_sentinel(self):
+        """
+        On protocol v5, result_metadata_id is always written (uses_prepared_metadata).
+        When result_metadata_id is None (e.g. LWT statement or mixed cluster where the
+        statement was prepared before the extension was active), send_body must write an
+        empty sentinel instead of crashing with TypeError.
+        """
+        message = ExecuteMessage('1', [], 4)
+        # result_metadata_id intentionally left as None; use_metadata_id stays False (v5 native path)
+        mock_io = Mock()
+
+        message.send_body(mock_io, 5)
+        # v5 always writes metadata_id: None → empty sentinel \x00\x00 + b'', then query params
+        # v5 uses 4-byte flags: VALUES_FLAG = \x00\x00\x00\x01
+        self._check_calls(mock_io, [(b'\x00\x01',), (b'1',),
+                                    (b'\x00\x00',), (b'',),
+                                    (b'\x00\x04',),
+                                    (b'\x00\x00\x00\x01',), (b'\x00\x00',)])
+
+    def test_recv_results_prepared_scylla_extension_reads_metadata_id(self):
+        """
+        When use_metadata_id is True (Scylla extension), result_metadata_id must be
+        read from the PREPARE response even for protocol v4.
+        """
+        # Build a minimal valid PREPARE response binary (no bind/result columns):
+        #   query_id:          short(2) + b'ab'
+        #   result_metadata_id: short(3) + b'xyz'  <-- only present when extension active
+        #   prepared flags:    int(1) = global_tables_spec
+        #   colcount:          int(0)
+        #   num_pk_indexes:    int(0)
+        #   ksname:            short(2) + b'ks'
+        #   cfname:            short(2) + b'tb'
+        #   result flags:      int(4) = no_metadata
+        #   result colcount:   int(0)
+        buf = io.BytesIO(
+            struct.pack('>H', 2) + b'ab'   # query_id
+            + struct.pack('>H', 3) + b'xyz'  # result_metadata_id
+            + struct.pack('>i', 1)           # prepared flags: global_tables_spec
+            + struct.pack('>i', 0)           # colcount = 0
+            + struct.pack('>i', 0)           # num_pk_indexes = 0
+            + struct.pack('>H', 2) + b'ks'  # ksname
+            + struct.pack('>H', 2) + b'tb'  # cfname
+            + struct.pack('>i', 4)           # result flags: no_metadata
+            + struct.pack('>i', 0)           # result colcount = 0
+        )
+
+        features_with_extension = ProtocolFeatures(use_metadata_id=True)
+        msg = ResultMessage(kind=4)  # RESULT_KIND_PREPARED = 4
+        msg.recv_results_prepared(buf, protocol_version=4,
+                                  protocol_features=features_with_extension,
+                                  user_type_map={})
+        assert msg.query_id == b'ab'
+        assert msg.result_metadata_id == b'xyz'
+
+    def test_recv_results_prepared_no_extension_skips_metadata_id(self):
+        """
+        Without use_metadata_id, result_metadata_id must NOT be read on protocol v4.
+        The buffer must NOT contain a metadata_id field.
+        """
+        buf = io.BytesIO(
+            struct.pack('>H', 2) + b'ab'   # query_id
+            # no result_metadata_id
+            + struct.pack('>i', 1)           # prepared flags: global_tables_spec
+            + struct.pack('>i', 0)           # colcount = 0
+            + struct.pack('>i', 0)           # num_pk_indexes = 0
+            + struct.pack('>H', 2) + b'ks'  # ksname
+            + struct.pack('>H', 2) + b'tb'  # cfname
+            + struct.pack('>i', 4)           # result flags: no_metadata
+            + struct.pack('>i', 0)           # result colcount = 0
+        )
+
+        features_without_extension = ProtocolFeatures(use_metadata_id=False)
+        msg = ResultMessage(kind=4)
+        msg.recv_results_prepared(buf, protocol_version=4,
+                                  protocol_features=features_without_extension,
+                                  user_type_map={})
+        assert msg.query_id == b'ab'
+        assert msg.result_metadata_id is None
+
+    def test_recv_results_metadata_changed_flag(self):
+        """
+        When _METADATA_ID_FLAG (0x0008) is set in a ROWS result,
+        recv_results_metadata must read and store the new result_metadata_id
+        sent by the server (METADATA_CHANGED signal), and still populate
+        column_metadata normally.
+        """
+        # Wire layout for a ROWS result with METADATA_CHANGED:
+        #   flags:             int(0x0008)  = _METADATA_ID_FLAG
+        #   colcount:          int(0)
+        #   result_metadata_id: short(4) + b'new1'
+        # (no columns — colcount=0 — to keep the buffer minimal)
+        buf = io.BytesIO(
+            struct.pack('>i', 0x0008)          # flags: METADATA_ID_FLAG
+            + struct.pack('>i', 0)             # colcount = 0
+            + struct.pack('>H', 4) + b'new1'   # result_metadata_id = b'new1'
+        )
+        msg = ResultMessage(kind=RESULT_KIND_ROWS)
+        msg.recv_results_metadata(buf, user_type_map={})
+        assert msg.result_metadata_id == b'new1'
+        assert msg.column_metadata == []
+
+    def test_recv_results_metadata_no_metadata_flag_skips_metadata_id(self):
+        """
+        When _NO_METADATA_FLAG (0x0004) is set, recv_results_metadata returns
+        early and must NOT read or set result_metadata_id, even if the caller
+        mistakenly sets _METADATA_ID_FLAG alongside it.
+        """
+        # flags = _NO_METADATA_FLAG (0x0004), colcount = 0
+        buf = io.BytesIO(
+            struct.pack('>i', 0x0004)  # flags: NO_METADATA
+            + struct.pack('>i', 0)     # colcount = 0
+        )
+        msg = ResultMessage(kind=RESULT_KIND_ROWS)
+        msg.recv_results_metadata(buf, user_type_map={})
+        assert not hasattr(msg, 'result_metadata_id') or msg.result_metadata_id is None
+        assert not hasattr(msg, 'column_metadata') or msg.column_metadata is None
+
     def test_query_message(self):
         """
         Test to check the appropriate calls are made

tests/unit/test_protocol_features.py

@@ -22,3 +22,38 @@ class OptionsHolder(object):
         assert protocol_features.rate_limit_error == 123
         assert protocol_features.shard_id == 0
         assert protocol_features.sharding_info is None
+
+    def test_use_metadata_id_parsing(self):
+        """
+        Test that SCYLLA_USE_METADATA_ID is parsed from SUPPORTED options.
+        """
+        options = {'SCYLLA_USE_METADATA_ID': ['']}
+        protocol_features = ProtocolFeatures.parse_from_supported(options)
+        assert protocol_features.use_metadata_id is True
+
+    def test_use_metadata_id_missing(self):
+        """
+        Test that use_metadata_id is False when SCYLLA_USE_METADATA_ID is absent.
+        """
+        options = {'SCYLLA_RATE_LIMIT_ERROR': ['ERROR_CODE=1']}
+        protocol_features = ProtocolFeatures.parse_from_supported(options)
+        assert protocol_features.use_metadata_id is False
+
+    def test_use_metadata_id_startup_options(self):
+        """
+        Test that SCYLLA_USE_METADATA_ID is included in STARTUP options when negotiated.
+        """
+        options = {'SCYLLA_USE_METADATA_ID': ['']}
+        protocol_features = ProtocolFeatures.parse_from_supported(options)
+        startup = {}
+        protocol_features.add_startup_options(startup)
+        assert 'SCYLLA_USE_METADATA_ID' in startup
+
+    def test_use_metadata_id_not_in_startup_when_not_negotiated(self):
+        """
+        Test that SCYLLA_USE_METADATA_ID is NOT included in STARTUP when not negotiated.
+        """
+        protocol_features = ProtocolFeatures.parse_from_supported({})
+        startup = {}
+        protocol_features.add_startup_options(startup)
+        assert 'SCYLLA_USE_METADATA_ID' not in startup