tests: add _RowView and _row_factory unit tests

Cover __getitem__, get(), __contains__, __repr__, shared index map,
read-only enforcement, empty input, single-column, and multi-row
scenarios.

Author: mykaul

cassandra/metadata.py

@@ -58,7 +58,7 @@ class _RowView(Mapping):
     ``values``, ``items``, and ``__contains__`` for free.
     """
 
-    __slots__ = ("_row", "_index_map")
+    __slots__ = ("_index_map", "_row")
 
     def __init__(self, row, index_map):
         if len(row) < max(index_map.values(), default=-1) + 1:
@@ -78,12 +78,6 @@ def __iter__(self):
     def __len__(self):
         return len(self._index_map)
 
-    def values(self):
-        return (self._row[i] for i in self._index_map.values())
-
-    def items(self):
-        return ((k, self._row[i]) for k, i in self._index_map.items())
-
     def get(self, key, default=None):
         idx = self._index_map.get(key)
         if idx is not None:

tests/unit/test_metadata.py

@@ -32,7 +32,8 @@
                                 _UnknownStrategy, ColumnMetadata, TableMetadata,
                                 IndexMetadata, Function, Aggregate,
                                 Metadata, TokenMap, ReplicationFactor,
-                                SchemaParserDSE68)
+                                SchemaParserDSE68,
+                                _RowView, _row_factory)
 from cassandra.policies import SimpleConvictionPolicy
 from cassandra.pool import Host
 from cassandra.protocol import QueryMessage
@@ -846,3 +847,94 @@ def test_strip_frozen(self):
         for argument, expected_result in argument_to_expected_results:
             result = strip_frozen(argument)
             assert result == expected_result, "strip_frozen() arg: {}".format(argument)
+
+class RowViewTest(unittest.TestCase):
+    """Tests for the internal _RowView and _row_factory helpers."""
+
+    def test_getitem(self):
+        rv = _RowView(("a_val", "b_val"), {"a": 0, "b": 1})
+        self.assertEqual(rv["a"], "a_val")
+        self.assertEqual(rv["b"], "b_val")
+
+    def test_getitem_missing_key(self):
+        rv = _RowView(("a_val",), {"a": 0})
+        with self.assertRaises(KeyError):
+            rv["missing"]
+
+    def test_get_present(self):
+        rv = _RowView(("a_val", "b_val"), {"a": 0, "b": 1})
+        self.assertEqual(rv.get("a"), "a_val")
+        self.assertEqual(rv.get("b"), "b_val")
+
+    def test_get_missing_returns_default(self):
+        rv = _RowView(("a_val",), {"a": 0})
+        self.assertIsNone(rv.get("missing"))
+        self.assertEqual(rv.get("missing", 42), 42)
+
+    def test_contains(self):
+        rv = _RowView(("a_val",), {"a": 0})
+        self.assertIn("a", rv)
+        self.assertNotIn("b", rv)
+
+    def test_repr(self):
+        rv = _RowView(("a_val", "b_val"), {"a": 0, "b": 1})
+        r = repr(rv)
+        self.assertIn("'a'", r)
+        self.assertIn("'a_val'", r)
+
+    def test_shared_index_map(self):
+        """All _RowView objects from the same _row_factory call share one index map."""
+        rows = _row_factory(["x", "y"], [("x1", "y1"), ("x2", "y2")])
+        self.assertIs(rows[0]._index_map, rows[1]._index_map)
+
+    def test_read_only(self):
+        """_RowView must not allow item assignment or deletion."""
+        rv = _RowView(("val",), {"col": 0})
+        with self.assertRaises(TypeError):
+            rv["col"] = "new"
+        with self.assertRaises(TypeError):
+            del rv["col"]
+
+    def test_row_factory_empty(self):
+        result = _row_factory(["a", "b"], [])
+        self.assertEqual(result, [])
+
+    def test_row_factory_single_column(self):
+        rows = _row_factory(["only"], [("v1",), ("v2",)])
+        self.assertEqual(rows[0]["only"], "v1")
+        self.assertEqual(rows[1]["only"], "v2")
+
+    def test_row_factory_values(self):
+        rows = _row_factory(["id", "name"], [(1, "alice"), (2, "bob")])
+        self.assertEqual(rows[0]["id"], 1)
+        self.assertEqual(rows[0]["name"], "alice")
+        self.assertEqual(rows[1]["id"], 2)
+        self.assertEqual(rows[1]["name"], "bob")
+
+    def test_len(self):
+        rv = _RowView(("a", "b", "c"), {"x": 0, "y": 1, "z": 2})
+        self.assertEqual(len(rv), 3)
+        rv2 = _RowView((), {})
+        self.assertEqual(len(rv2), 0)
+
+    def test_keys(self):
+        rv = _RowView(("a", "b"), {"x": 0, "y": 1})
+        self.assertEqual(set(rv.keys()), {"x", "y"})
+
+    def test_values(self):
+        rv = _RowView(("a", "b"), {"x": 0, "y": 1})
+        self.assertEqual(list(rv.values()), ["a", "b"])
+
+    def test_items(self):
+        rv = _RowView(("a", "b"), {"x": 0, "y": 1})
+        self.assertEqual(set(rv.items()), {("x", "a"), ("y", "b")})
+
+    def test_init_raises_on_short_row(self):
+        with self.assertRaises(ValueError):
+            _RowView(("val",), {"a": 0, "b": 1})
+
+    def test_init_accepts_exact_row(self):
+        _RowView(("a", "b"), {"a": 0, "b": 1})
+
+    def test_init_accepts_empty(self):
+        _RowView((), {})