|
32 | 32 | _UnknownStrategy, ColumnMetadata, TableMetadata, |
33 | 33 | IndexMetadata, Function, Aggregate, |
34 | 34 | Metadata, TokenMap, ReplicationFactor, |
35 | | - SchemaParserDSE68) |
| 35 | + SchemaParserDSE68, |
| 36 | + _RowView, _row_factory) |
36 | 37 | from cassandra.policies import SimpleConvictionPolicy |
37 | 38 | from cassandra.pool import Host |
38 | 39 | from cassandra.protocol import QueryMessage |
@@ -846,3 +847,66 @@ def test_strip_frozen(self): |
846 | 847 | for argument, expected_result in argument_to_expected_results: |
847 | 848 | result = strip_frozen(argument) |
848 | 849 | assert result == expected_result, "strip_frozen() arg: {}".format(argument) |
| 850 | + |
| 851 | +class RowViewTest(unittest.TestCase): |
| 852 | + """Tests for the internal _RowView and _row_factory helpers.""" |
| 853 | + |
| 854 | + def test_getitem(self): |
| 855 | + rv = _RowView(("a_val", "b_val"), {"a": 0, "b": 1}) |
| 856 | + self.assertEqual(rv["a"], "a_val") |
| 857 | + self.assertEqual(rv["b"], "b_val") |
| 858 | + |
| 859 | + def test_getitem_missing_key(self): |
| 860 | + rv = _RowView(("a_val",), {"a": 0}) |
| 861 | + with self.assertRaises(KeyError): |
| 862 | + rv["missing"] |
| 863 | + |
| 864 | + def test_get_present(self): |
| 865 | + rv = _RowView(("a_val", "b_val"), {"a": 0, "b": 1}) |
| 866 | + self.assertEqual(rv.get("a"), "a_val") |
| 867 | + self.assertEqual(rv.get("b"), "b_val") |
| 868 | + |
| 869 | + def test_get_missing_returns_default(self): |
| 870 | + rv = _RowView(("a_val",), {"a": 0}) |
| 871 | + self.assertIsNone(rv.get("missing")) |
| 872 | + self.assertEqual(rv.get("missing", 42), 42) |
| 873 | + |
| 874 | + def test_contains(self): |
| 875 | + rv = _RowView(("a_val",), {"a": 0}) |
| 876 | + self.assertIn("a", rv) |
| 877 | + self.assertNotIn("b", rv) |
| 878 | + |
| 879 | + def test_repr(self): |
| 880 | + rv = _RowView(("a_val", "b_val"), {"a": 0, "b": 1}) |
| 881 | + r = repr(rv) |
| 882 | + self.assertIn("'a'", r) |
| 883 | + self.assertIn("'a_val'", r) |
| 884 | + |
| 885 | + def test_shared_index_map(self): |
| 886 | + """All _RowView objects from the same _row_factory call share one index map.""" |
| 887 | + rows = _row_factory(["x", "y"], [("x1", "y1"), ("x2", "y2")]) |
| 888 | + self.assertIs(rows[0]._index_map, rows[1]._index_map) |
| 889 | + |
| 890 | + def test_read_only(self): |
| 891 | + """_RowView must not allow item assignment or deletion.""" |
| 892 | + rv = _RowView(("val",), {"col": 0}) |
| 893 | + with self.assertRaises(TypeError): |
| 894 | + rv["col"] = "new" |
| 895 | + with self.assertRaises(TypeError): |
| 896 | + del rv["col"] |
| 897 | + |
| 898 | + def test_row_factory_empty(self): |
| 899 | + result = _row_factory(["a", "b"], []) |
| 900 | + self.assertEqual(result, []) |
| 901 | + |
| 902 | + def test_row_factory_single_column(self): |
| 903 | + rows = _row_factory(["only"], [("v1",), ("v2",)]) |
| 904 | + self.assertEqual(rows[0]["only"], "v1") |
| 905 | + self.assertEqual(rows[1]["only"], "v2") |
| 906 | + |
| 907 | + def test_row_factory_values(self): |
| 908 | + rows = _row_factory(["id", "name"], [(1, "alice"), (2, "bob")]) |
| 909 | + self.assertEqual(rows[0]["id"], 1) |
| 910 | + self.assertEqual(rows[0]["name"], "alice") |
| 911 | + self.assertEqual(rows[1]["id"], 2) |
| 912 | + self.assertEqual(rows[1]["name"], "bob") |
0 commit comments