|
12 | 12 | # See the License for the specific language governing permissions and |
13 | 13 | # limitations under the License. |
14 | 14 | import unittest |
| 15 | +import warnings |
15 | 16 |
|
16 | 17 | from binascii import unhexlify |
17 | 18 | import logging |
|
21 | 22 | import uuid |
22 | 23 |
|
23 | 24 | import cassandra |
24 | | -from cassandra.cqltypes import strip_frozen |
| 25 | +from cassandra.cqltypes import cqltype_to_python, python_to_cqltype, strip_frozen |
25 | 26 | from cassandra.marshal import uint16_unpack, uint16_pack |
26 | 27 | from cassandra.metadata import (Murmur3Token, MD5Token, |
27 | 28 | BytesToken, ReplicationStrategy, |
@@ -846,3 +847,39 @@ 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 | + def test_cqltype_backslash_escape(self): |
| 852 | + """Verify that UDT names containing backslashes don't trigger SyntaxWarning (python-driver#750).""" |
| 853 | + udt_input = r'map<"!@#$%^&*()[]\ frozen >>>", int>' |
| 854 | + |
| 855 | + # Parsing must not emit SyntaxWarning |
| 856 | + with warnings.catch_warnings(record=True) as caught: |
| 857 | + warnings.simplefilter('always') |
| 858 | + result = cqltype_to_python(udt_input) |
| 859 | + syntax_warnings = [w for w in caught if issubclass(w.category, SyntaxWarning)] |
| 860 | + self.assertEqual(syntax_warnings, [], 'cqltype_to_python emitted SyntaxWarning') |
| 861 | + |
| 862 | + # Parsed result should preserve the quoted UDT name with backslash |
| 863 | + self.assertEqual(result[0], 'map') |
| 864 | + self.assertIsInstance(result[1], list) |
| 865 | + self.assertEqual(result[1][0], r'"!@#$%^&*()[]\ frozen >>>"') |
| 866 | + self.assertEqual(result[1][1], 'int') |
| 867 | + |
| 868 | + # Round-trip: python_to_cqltype(cqltype_to_python(x)) == x |
| 869 | + round_tripped = python_to_cqltype(result) |
| 870 | + self.assertEqual(round_tripped, udt_input) |
| 871 | + |
| 872 | + def test_cqltype_single_quote_in_identifier(self): |
| 873 | + """Verify that UDT names containing single quotes parse and round-trip correctly.""" |
| 874 | + udt_input = 'map<"it\'s", int>' |
| 875 | + |
| 876 | + result = cqltype_to_python(udt_input) |
| 877 | + |
| 878 | + self.assertEqual(result[0], 'map') |
| 879 | + self.assertIsInstance(result[1], list) |
| 880 | + self.assertEqual(result[1][0], '"it\'s"') |
| 881 | + self.assertEqual(result[1][1], 'int') |
| 882 | + |
| 883 | + # Round-trip: python_to_cqltype(cqltype_to_python(x)) == x |
| 884 | + round_tripped = python_to_cqltype(result) |
| 885 | + self.assertEqual(round_tripped, udt_input) |
0 commit comments