Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
4 changes: 2 additions & 2 deletions cassandra/cqltypes.py
Original file line number Diff line number Diff line change
Expand Up @@ -130,7 +130,7 @@ def cqltype_to_python(cql_string):
(r'<', lambda s, t: ', ['),
(r'>', lambda s, t: ']'),
(r'[, ]', lambda s, t: t),
(r'".*?"', lambda s, t: "'{}'".format(t)),
(r'".*?"', lambda s, t: "'{}'".format(t.replace('\\', '\\\\').replace("'", "\\'"))),
))

scanned_tokens = scanner.scan(cql_string)[0]
Expand All @@ -156,7 +156,7 @@ def python_to_cqltype(types):
))

scanned_tokens = scanner.scan(repr(types))[0]
cql = ''.join(scanned_tokens).replace('\\\\', '\\')
cql = ''.join(scanned_tokens).replace('\\\\', '\\').replace("\\'", "'")
return cql


Expand Down
39 changes: 38 additions & 1 deletion tests/unit/test_metadata.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
# See the License for the specific language governing permissions and
# limitations under the License.
import unittest
import warnings

from binascii import unhexlify
import logging
Expand All @@ -21,7 +22,7 @@
import uuid

import cassandra
from cassandra.cqltypes import strip_frozen
from cassandra.cqltypes import cqltype_to_python, python_to_cqltype, strip_frozen
from cassandra.marshal import uint16_unpack, uint16_pack
from cassandra.metadata import (Murmur3Token, MD5Token,
BytesToken, ReplicationStrategy,
Expand Down Expand Up @@ -846,3 +847,39 @@ 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)

def test_cqltype_backslash_escape(self):
"""Verify that UDT names containing backslashes don't trigger SyntaxWarning (python-driver#750)."""
udt_input = r'map<"!@#$%^&*()[]\ frozen >>>", int>'

# Parsing must not emit SyntaxWarning
with warnings.catch_warnings(record=True) as caught:
warnings.simplefilter('always')
result = cqltype_to_python(udt_input)
syntax_warnings = [w for w in caught if issubclass(w.category, SyntaxWarning)]
self.assertEqual(syntax_warnings, [], 'cqltype_to_python emitted SyntaxWarning')

# Parsed result should preserve the quoted UDT name with backslash
self.assertEqual(result[0], 'map')
self.assertIsInstance(result[1], list)
self.assertEqual(result[1][0], r'"!@#$%^&*()[]\ frozen >>>"')
self.assertEqual(result[1][1], 'int')

# Round-trip: python_to_cqltype(cqltype_to_python(x)) == x
round_tripped = python_to_cqltype(result)
self.assertEqual(round_tripped, udt_input)

def test_cqltype_single_quote_in_identifier(self):
"""Verify that UDT names containing single quotes parse and round-trip correctly."""
udt_input = 'map<"it\'s", int>'

result = cqltype_to_python(udt_input)

self.assertEqual(result[0], 'map')
self.assertIsInstance(result[1], list)
self.assertEqual(result[1][0], '"it\'s"')
self.assertEqual(result[1][1], 'int')

# Round-trip: python_to_cqltype(cqltype_to_python(x)) == x
round_tripped = python_to_cqltype(result)
self.assertEqual(round_tripped, udt_input)
Loading