diff --git a/pylib/cqlshlib/cql3handling.py b/pylib/cqlshlib/cql3handling.py index a83c6535b8..a966b90b11 100644 --- a/pylib/cqlshlib/cql3handling.py +++ b/pylib/cqlshlib/cql3handling.py @@ -116,43 +116,6 @@ class Cql3ParsingRuleSet(CqlParsingRuleSet): 'overlap_inclusion_method' ) - @classmethod - def escape_value(cls, value): - if value is None: - return 'NULL' # this totally won't work - if isinstance(value, bool): - value = str(value).lower() - elif isinstance(value, float): - return '%f' % value - elif isinstance(value, int): - return str(value) - return "'%s'" % value.replace("'", "''") - - @classmethod - def escape_name(cls, name): - if name is None: - return 'NULL' - return "'%s'" % name.replace("'", "''") - - @staticmethod - def dequote_name(name): - name = name.strip() - if name == '': - return name - if name[0] == '"' and name[-1] == '"': - return name[1:-1].replace('""', '"') - else: - return name.lower() - - @staticmethod - def dequote_value(cqlword): - cqlword = cqlword.strip() - if cqlword == '': - return cqlword - if cqlword[0] == "'" and cqlword[-1] == "'": - cqlword = cqlword[1:-1].replace("''", "'") - return cqlword - CqlRuleSet = Cql3ParsingRuleSet() diff --git a/pylib/cqlshlib/cqlhandling.py b/pylib/cqlshlib/cqlhandling.py index 90e552fc27..5d5cd396b5 100644 --- a/pylib/cqlshlib/cqlhandling.py +++ b/pylib/cqlshlib/cqlhandling.py @@ -132,6 +132,43 @@ class CqlParsingRuleSet(pylexotron.ParsingRuleSet): tokens = self.cql_massage_tokens(tokens) return self.parse(startsymbol, tokens, init_bindings={'*SRC*': text}) + @staticmethod + def dequote_value(cqlword): + cqlword = cqlword.strip() + if cqlword == '': + return cqlword + if cqlword[0] == "'" and cqlword[-1] == "'": + cqlword = cqlword[1:-1].replace("''", "'") + return cqlword + + @staticmethod + def dequote_name(name): + name = name.strip() + if name == '': + return name + if name[0] == '"' and name[-1] == '"': + return name[1:-1].replace('""', '"') + else: + return name.lower() + + @staticmethod + def escape_value(value): + if value is None: + return 'NULL' # this totally won't work + if isinstance(value, bool): + value = str(value).lower() + elif isinstance(value, float): + return '%f' % value + elif isinstance(value, int): + return str(value) + return "'%s'" % value.replace("'", "''") + + @staticmethod + def escape_name(name): + if name is None: + return 'NULL' + return '"%s"' % name.replace('"', '""') + def cql_whole_parse_tokens(self, toklist, srcstr=None, startsymbol='Start'): return self.whole_match(startsymbol, toklist, srcstr=srcstr) diff --git a/pylib/cqlshlib/saferscanner.py b/pylib/cqlshlib/saferscanner.py index 2c2b6100d9..f32c1afcb1 100644 --- a/pylib/cqlshlib/saferscanner.py +++ b/pylib/cqlshlib/saferscanner.py @@ -30,6 +30,7 @@ class SaferScannerBase(re.Scanner): @classmethod def subpat(cls, phrase, flags): + # pylint: disable=no-member return cls.scrub_sub(re.sre_parse.parse(phrase, flags), flags) @classmethod @@ -38,6 +39,7 @@ class SaferScannerBase(re.Scanner): seqtypes = (type(()), type([])) for op, arg in sub.data: if type(arg) in seqtypes: + # pylint: disable=no-member arg = [cls.scrub_sub(a, flags) if isinstance(a, re.sre_parse.SubPattern) else a for a in arg] if op in (BRANCH, SUBPATTERN): @@ -49,6 +51,7 @@ class SaferScannerBase(re.Scanner): raise ValueError("Named captures not allowed in SaferScanner lexicon") if sub.pattern.flags ^ flags: raise ValueError("RE flag setting not allowed in SaferScanner lexicon (%s)" % (bin(sub.pattern.flags),)) + # pylint: disable=no-member return re.sre_parse.SubPattern(sub.pattern, scrubbedsub) @@ -57,6 +60,7 @@ class Py36SaferScanner(SaferScannerBase): def __init__(self, lexicon, flags=0): self.lexicon = lexicon p = [] + # pylint: disable=no-member s = re.sre_parse.Pattern() s.flags = flags for phrase, action in lexicon: @@ -73,6 +77,7 @@ class Py38SaferScanner(SaferScannerBase): def __init__(self, lexicon, flags=0): self.lexicon = lexicon p = [] + # pylint: disable=no-member s = re.sre_parse.State() s.flags = flags for phrase, action in lexicon: @@ -89,6 +94,7 @@ class Py311SaferScanner(SaferScannerBase): def __init__(self, lexicon, flags=0): self.lexicon = lexicon p = [] + # pylint: disable=no-member s = re._parser.State() s.flags = flags for phrase, action in lexicon: diff --git a/pylib/cqlshlib/test/test_rule_set_utils.py b/pylib/cqlshlib/test/test_rule_set_utils.py new file mode 100644 index 0000000000..9f0070efea --- /dev/null +++ b/pylib/cqlshlib/test/test_rule_set_utils.py @@ -0,0 +1,44 @@ +# Licensed to the Apache Software Foundation (ASF) under one +# or more contributor license agreements. See the NOTICE file +# distributed with this work for additional information +# regarding copyright ownership. The ASF licenses this file +# to you under the Apache License, Version 2.0 (the +# "License"); you may not use this file except in compliance +# with the License. You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from .basecase import BaseTestCase +from cqlshlib.cql3handling import CqlRuleSet + + +class TestRuleSetUtils(BaseTestCase): + def test_dequote_value(self): + self.assertEqual(CqlRuleSet.dequote_value("'test'"), "test") + self.assertEqual(CqlRuleSet.dequote_value("'test''val'"), "test'val") + self.assertEqual(CqlRuleSet.dequote_value("unquoted"), "unquoted") + self.assertEqual(CqlRuleSet.dequote_value(""), "") + + def test_dequote_name(self): + self.assertEqual(CqlRuleSet.dequote_name('"TestName"'), "TestName") + self.assertEqual(CqlRuleSet.dequote_name('"Test""Name"'), 'Test"Name') + self.assertEqual(CqlRuleSet.dequote_name("LowercaseName"), "lowercasename") + self.assertEqual(CqlRuleSet.dequote_name(""), "") + + def test_escape_value(self): + self.assertEqual(CqlRuleSet.escape_value("val'with'quote"), "'val''with''quote'") + self.assertEqual(CqlRuleSet.escape_value(True), "'true'") + self.assertEqual(CqlRuleSet.escape_value(123), "123") + self.assertEqual(CqlRuleSet.escape_value(1.23), "1.230000") + self.assertEqual(CqlRuleSet.escape_value(None), "NULL") + + def test_escape_name(self): + self.assertEqual(CqlRuleSet.escape_name('name"with"quote'), '"name""with""quote"') + self.assertEqual(CqlRuleSet.dequote_name(CqlRuleSet.escape_name('name"with"quote')), 'name"with"quote') + self.assertEqual(CqlRuleSet.escape_name(None), "NULL")