added tests and fixed issues with existing tests

Author: Martin Zoellner

simple_ddl_parser/dialects/sql.py

@@ -971,7 +971,9 @@ def p_expression_comment_on(self, p: List):
         p[0] = {"comment_on": comment_on}
         p_list = list(p)
         obj_type = p_list[3]
-        comment_on["comment"] = p_list[-1]
+
+        # Cleanse comment quotes and handle escaped quotes
+        comment_on["comment"] = p_list[-1][1:-1].replace("''", "'")
         comment_on["object_type"] = obj_type
 
         if obj_type == "COLUMN":
@@ -1055,7 +1057,7 @@ def p_id_equals(self, p: List) -> None:
         """
         p_list = list(p)
 
-        if not p_list[-1] in [")", "]"]:
+        if p_list[-1] not in [")", "]"]:
             p[0] = {p[1]: p_list[-1]}
         else:
             if len(p_list) > 6 and isinstance(p_list[5], list):

simple_ddl_parser/output/base_data.py

@@ -285,13 +285,13 @@ def set_object_comment(self, statement: Dict) -> None:
 
     def set_table_comment(self, statement: Dict) -> None:
         comment = statement["comment_on"]
-        self.comment = comment["comment"][1:-1]
+        self.comment = comment["comment"]
 
     def set_column_comments(self, statement: Dict) -> None:
         comment = statement["comment_on"]
         for column in self.columns:
             if column["name"] == comment["column_name"]:
-                column["comment"] = comment["comment"][1:-1]
+                column["comment"] = comment["comment"]
                 break
 
     def set_default_columns_from_alter(self, statement: Dict) -> None:

simple_ddl_parser/parser.py

@@ -2,7 +2,7 @@
 import logging
 import os
 import re
-from typing import Dict, List, Optional, Tuple, Union
+from typing import Dict, List, Optional, Tuple, Union, cast
 
 from ply import lex, yacc
 
@@ -19,6 +19,8 @@
 IN_COM = "--"
 MYSQL_COM = "#"
 
+LF_IN_QUOTE = r"\N"
+
 
 def set_logging_config(
     log_level: Union[str, int], log_file: Optional[str] = None
@@ -173,7 +175,7 @@ def process_regex_input(self, data):
         return data
 
     def pre_process_data(self, data):
-        data = data.decode("utf-8")
+        data = cast(str, data.decode("utf-8"))
         # todo: not sure how to workaround ',' normal way
         if "input.regex" in data:
             data = self.process_regex_input(data)
@@ -182,37 +184,44 @@ def pre_process_data(self, data):
         result = []
         in_quote = False
         i = 0
-        symbol_spacing_map = {
-            ",": " , ",
-            "(": " ( ",
-            ")": " ) ",
-        }
+        symbol_spacing_map = {",", "(", ")"}
+
+        # Special handling for odd number of single quotes
+        if data.count("'") % 2 != 0:
+            data = data.replace("\\'", "pars_m_single")
+
         while i < len(data):
             char = data[i]
+            startswith = data[i:].startswith
 
             # Handle quote start/end
-            if char == "'" and (i == 0 or data[i - 1] != "\\"):
+            if char == "'":
                 in_quote = not in_quote
                 result.append(char)
+
+            # Handle line feeds in quotes
+            elif in_quote and startswith("\\n"):
+                result.append(LF_IN_QUOTE)
+                i += 1
+
+            # Handle special unicode quotes
+            elif not in_quote and (startswith(r"\u2018") or startswith(r"\u2019")):
+                result.append("'")
+                i += 5
+
             # Handle symbols that need spacing
             elif not in_quote and char in symbol_spacing_map:
-                result.append(symbol_spacing_map[char])
+                result.append(f" {char} ")
+
             # Keep all other characters as-is
             else:
                 result.append(char)
 
             i += 1
 
         data = "".join(result)
-
-        if data.count("'") % 2 != 0:
-            data = data.replace("\\'", "pars_m_single")
         data = (
             data.replace("\\x", "\\0")
-            .replace("‘", "'")
-            .replace("’", "'")
-            .replace("\\u2018", "'")
-            .replace("\\u2019", "'")
             .replace("'\\t'", "'pars_m_t'")
             .replace("\\t", " ")
         )
@@ -296,7 +305,13 @@ def process_line(
     ) -> Tuple[Optional[str], bool]:
         self.pre_process_line()
 
-        self.line = self.line.strip().replace("\n", "").replace("\t", "")
+        # Remove whitespace, while preserving newlines in quotes
+        self.line = (
+            self.line.strip()
+            .replace("\n", "")
+            .replace("\t", "")
+            .replace(LF_IN_QUOTE, "\\n")
+        )
         self.skip = self.check_line_on_skip_words()
 
         self.parse_set_statement()