Quote ALTER SCHEMA SET values

snowcap/lifecycle.py

@@ -6,7 +6,7 @@
 from .builder import tidy_sql
 from .enums import GrantType, ResourceType
 from .identifiers import FQN, URN
-from .props import BoolProp, IntProp, Props, StringProp
+from .props import BoolProp, IntProp, Props, StringProp, quote_value
 from .resource_name import ResourceName
 
 __this__ = sys.modules[__name__]
@@ -50,7 +50,7 @@ def create__default(urn: URN, data: dict, props: Props, if_not_exists: bool = Fa
 def create_account_parameter(urn: URN, data: dict, props: Props, if_not_exists: bool = False) -> str:
     value = data["value"]
     if isinstance(value, str):
-        value = f"'{value}'"
+        value = quote_value(value)
     return tidy_sql(
         "ALTER",
         "ACCOUNT",
@@ -292,14 +292,7 @@ def create_role_grant(urn: URN, data: dict, props: Props, if_not_exists: bool =
 
 
 def create_scanner_package(urn: URN, data: dict, props: Props, if_not_exists: bool = False) -> str:
-    package_name = f"'{urn.fqn.name}'"
-    return tidy_sql(
-        "CALL SNOWFLAKE.TRUST_CENTER.SET_CONFIGURATION(",
-        "'ENABLED',",
-        "'TRUE',",
-        package_name,
-        ")",
-    )
+    return update_scanner_package(urn, {"ENABLED": "TRUE"}, props)
 
 
 def create_schema(urn: URN, data: dict, props: Props, if_not_exists: bool = False) -> str:
@@ -505,15 +498,15 @@ def update_role_grant(urn: URN, data: dict, props: Props) -> str:
 
 
 def update_scanner_package(urn: URN, data: dict, props: Props) -> str:
-    package_name = f"'{urn.fqn.name}'"
+    package_name = quote_value(urn.fqn.name.unquoted)
     attr, new_value = data.popitem()
     if attr == "schedule":
-        new_value = f"'USING CRON {new_value}'"
+        new_value = quote_value(f"USING CRON {new_value}")
     else:
-        new_value = f"'{new_value}'"
+        new_value = quote_value(new_value)
     return tidy_sql(
         "CALL SNOWFLAKE.TRUST_CENTER.SET_CONFIGURATION(",
-        f"'{attr}',",
+        f"{quote_value(attr)},",
         new_value,
         ",",
         package_name,
@@ -535,7 +528,7 @@ def update_schema(urn: URN, data: dict, props: Props) -> str:
     elif attr == "managed_access":
         return tidy_sql("ALTER SCHEMA", urn.fqn, "ENABLE" if new_value else "DISABLE", "MANAGED ACCESS")
     else:
-        new_value = f"'{new_value}'" if isinstance(new_value, str) else new_value
+        new_value = quote_value(new_value) if isinstance(new_value, str) else new_value
         return tidy_sql("ALTER SCHEMA", urn.fqn, "SET", attr, "=", new_value)
 
 
@@ -818,14 +811,7 @@ def drop_role_grant(urn: URN, data: dict, **kwargs):
 
 
 def drop_scanner_package(urn: URN, data: dict, **kwargs) -> str:
-    package_name = f"'{urn.fqn.name}'"
-    return tidy_sql(
-        "CALL SNOWFLAKE.TRUST_CENTER.SET_CONFIGURATION(",
-        "'ENABLED',",
-        "'FALSE',",
-        package_name,
-        ")",
-    )
+    return update_scanner_package(urn, {"ENABLED": "FALSE"}, Props())
 
 
 def drop_tag_masking_policy_reference(urn: URN, data: dict, **kwargs) -> str:

snowcap/parse.py

@@ -11,8 +11,22 @@
 Keyword = pp.CaselessKeyword
 Literal = pp.CaselessLiteral
 
-StringLiteral = pp.QuotedString("'", multiline=False, unquote_results=True) | pp.QuotedString(
-    "$$", multiline=True, unquote_results=True
+_SQL_ESCAPES = {"b": "\b", "f": "\f", "n": "\n", "r": "\r", "t": "\t", "0": "\0"}
+
+
+def _unescape_sql_string(quoted: str) -> str:
+    # pyparsing 3.0's own escape handling turns an escaped backslash before t/n into a tab/newline
+    return re.sub(
+        r"\\(u[0-9a-fA-F]{4}|.)",
+        lambda m: chr(int(m[1][1:], 16)) if len(m[1]) == 5 else _SQL_ESCAPES.get(m[1], m[1]),
+        quoted[1:-1].replace("''", "'"),
+    )
+
+
+StringLiteral = pp.QuotedString(
+    "'", esc_char="\\", esc_quote="''", multiline=False, unquote_results=False
+).set_parse_action(lambda t: _unescape_sql_string(t[0])) | pp.QuotedString(
+    "$$", multiline=True, unquote_results=True, convert_whitespace_escapes=False
 )
 Numeric = pp.Word(pp.nums + ".")
 

snowcap/props.py

@@ -24,9 +24,20 @@
 __this__ = sys.modules[__name__]
 
 
+def _single_quote(value) -> str:
+    # JSON and Snowflake share backslash escape syntax for control characters
+    escaped = json.dumps(str(value), ensure_ascii=False)[1:-1].replace("'", "''")
+    return f"'{escaped}'"
+
+
 def quote_value(value: str):
     if value is None or value == "":
         return "''"
+    text = str(value)
+    # Fail here rather than in the driver: lone surrogates cannot be encoded as UTF-8
+    text.encode("utf-8")
+    if "$$" in text or text.endswith("$"):
+        return _single_quote(text)
     return f"$${value}$$"
 
 
@@ -383,7 +394,7 @@ def typecheck(self, prop_value):
     def render(self, value: dict) -> str:
         if value is None:
             return ""
-        kv_pairs = ", ".join([f"'{key}' = '{value}'" for key, value in value.items()])
+        kv_pairs = ", ".join([f"{quote_value(key)} = {quote_value(value)}" for key, value in value.items()])
         eq = " = " if self.eq else " "
         return f"{self.label}{eq}({kv_pairs})"
 
@@ -440,7 +451,7 @@ def render(self, value):
             value = value.value
         eq = " = " if self.eq else " "
         if self.quoted:
-            value = f"'{value}'"
+            value = _single_quote(value)
         return f"{self.label}{eq}{value}"
 
 
@@ -538,7 +549,7 @@ def render(self, values):
 
         key, value = values.popitem()
         if key.upper() == "STREAM":
-            value = f"'{value}'"
+            value = quote_value(value)
         time_point = f"{key} => {value}"
         return f"{self.label} ({time_point})"
 

snowcap/resource_name.py

@@ -80,6 +80,10 @@ def __eq__(self, other: Any):
     def upper(self):
         return self
 
+    @property
+    def unquoted(self) -> str:
+        return self._name if self._quoted else self._name.upper()
+
     def startswith(self, prefix: str) -> bool:
         return self._name.startswith(prefix)
 

tests/test_lifecycle.py

@@ -214,7 +214,15 @@ def test_string_value(self):
         data = {"value": "America/New_York"}
         props = MockProps("")
         result = create_account_parameter(urn, data, props)
-        assert result == "ALTER ACCOUNT SET TIMEZONE = 'America/New_York'"
+        assert result == "ALTER ACCOUNT SET TIMEZONE = $$America/New_York$$"
+
+    def test_string_value_with_apostrophe(self):
+        """A value containing an apostrophe must not break out of the literal."""
+        urn = make_urn(ResourceType.ACCOUNT_PARAMETER, "QUERY_TAG")
+        data = {"value": "today's batch"}
+        props = MockProps("")
+        result = create_account_parameter(urn, data, props)
+        assert result == "ALTER ACCOUNT SET QUERY_TAG = $$today's batch$$"
 
     def test_numeric_value(self):
         """Test setting numeric parameter."""
@@ -576,9 +584,7 @@ def test_basic_create(self):
         props = MockProps("")
         result = create_scanner_package(urn, data, props)
         assert "CALL SNOWFLAKE.TRUST_CENTER.SET_CONFIGURATION" in result
-        assert "'ENABLED'" in result
-        assert "'TRUE'" in result
-        assert "'CIS_BENCHMARKS'" in result
+        assert result == ("CALL SNOWFLAKE.TRUST_CENTER.SET_CONFIGURATION( $$ENABLED$$, $$TRUE$$ , $$CIS_BENCHMARKS$$ )")
 
 
 class TestCreateSchema:
@@ -805,7 +811,7 @@ def test_update_delegates_to_create(self):
         data = {"value": "UTC"}
         props = MockProps("")
         result = update_account_parameter(urn, data, props)
-        assert "ALTER ACCOUNT SET TIMEZONE = 'UTC'" in result
+        assert "ALTER ACCOUNT SET TIMEZONE = $$UTC$$" in result
 
 
 class TestUpdateEventTable:
@@ -864,7 +870,7 @@ def test_update_schedule(self):
         props = MockProps("")
         result = update_scanner_package(urn, data, props)
         assert "CALL SNOWFLAKE.TRUST_CENTER.SET_CONFIGURATION" in result
-        assert "'schedule'" in result
+        assert "$$schedule$$" in result
         assert "USING CRON" in result
 
     def test_update_other_property(self):
@@ -873,8 +879,32 @@ def test_update_other_property(self):
         data = {"enabled": "TRUE"}
         props = MockProps("")
         result = update_scanner_package(urn, data, props)
-        assert "'enabled'" in result
-        assert "'TRUE'" in result
+        assert "$$enabled$$" in result
+        assert "$$TRUE$$" in result
+
+    def test_update_comment_with_apostrophe(self):
+        """A value containing an apostrophe must not break out of the literal."""
+        urn = make_urn(ResourceType.SCANNER_PACKAGE, "CIS_BENCHMARKS")
+        data = {"comment": "the account's weekly scan"}
+        props = MockProps("")
+        result = update_scanner_package(urn, data, props)
+        assert "$$the account's weekly scan$$" in result
+
+    def test_update_package_name_with_apostrophe(self):
+        """A package name containing an apostrophe must be escaped in the literal."""
+        urn = make_urn(ResourceType.SCANNER_PACKAGE, "CIS'BENCHMARKS")
+        data = {"enabled": "TRUE"}
+        props = MockProps("")
+        result = update_scanner_package(urn, data, props)
+        assert "$$CIS'BENCHMARKS$$" in result
+
+    def test_update_schedule_ending_in_dollar(self):
+        """A trailing $ would close the dollar quote early, so the value falls back to single quotes."""
+        urn = make_urn(ResourceType.SCANNER_PACKAGE, "CIS_BENCHMARKS")
+        data = {"schedule": "0 * * * * UTC$"}
+        props = MockProps("")
+        result = update_scanner_package(urn, data, props)
+        assert "'USING CRON 0 * * * * UTC$'" in result
 
 
 class TestUpdateSchema:
@@ -937,6 +967,22 @@ def test_set_other_property(self):
         result = update_schema(urn, data, props)
         assert "SET data_retention_time_in_days = 7" in result
 
+    def test_set_comment_with_apostrophe(self):
+        """A comment containing an apostrophe must not break out of the literal."""
+        urn = make_urn(ResourceType.SCHEMA, "MY_SCHEMA", database="MY_DB")
+        data = {"comment": "the database's two-limb test"}
+        props = MockProps("")
+        result = update_schema(urn, data, props)
+        assert result == "ALTER SCHEMA MY_DB.MY_SCHEMA SET comment = $$the database's two-limb test$$"
+
+    def test_set_comment_containing_dollar_quote(self):
+        """A comment containing $$ falls back to a single-quoted literal."""
+        urn = make_urn(ResourceType.SCHEMA, "MY_SCHEMA", database="MY_DB")
+        data = {"comment": "costs $$ and it's dear"}
+        props = MockProps("")
+        result = update_schema(urn, data, props)
+        assert result == "ALTER SCHEMA MY_DB.MY_SCHEMA SET comment = 'costs $$ and it''s dear'"
+
 
 class TestUpdateTable:
     """Tests for update_table function."""
@@ -1277,8 +1323,9 @@ def test_basic_drop(self):
         data = {}
         result = drop_scanner_package(urn, data)
         assert "CALL SNOWFLAKE.TRUST_CENTER.SET_CONFIGURATION" in result
-        assert "'ENABLED'" in result
-        assert "'FALSE'" in result
+        assert result == (
+            "CALL SNOWFLAKE.TRUST_CENTER.SET_CONFIGURATION( $$ENABLED$$, $$FALSE$$ , $$CIS_BENCHMARKS$$ )"
+        )
 
 
 # ============================================================================

tests/test_parse.py

@@ -24,8 +24,10 @@
     parse_region,
     parse_view_ddl,
     resolve_resource_class,
+    StringLiteral,
 )
 from snowcap.enums import ResourceType, Scope
+from snowcap.props import quote_value
 from snowcap.scope import DatabaseScope, SchemaScope
 
 # =============================================================================
@@ -930,3 +932,13 @@ def my_action(text):
         lex = Lexicon({"PREFIX": my_action})
         result = convert_match(lex, "PREFIX some text")
         assert "processed:" in result
+
+
+@pytest.mark.parametrize("value", ["$$ it's", '$$ say "hi"', "$$ a\nb\tc", "$$ C:\\tmp", "$$\0\x01", "it's $"])
+def test_string_literal_decodes_quote_value_fallback(value):
+    assert StringLiteral.parse_string(quote_value(value), parse_all=True)[0] == value
+
+
+@pytest.mark.parametrize("value", [r"C:\Users\foo", r"a\nb\tc\rd\fe", "plain value"])
+def test_string_literal_keeps_backslashes_in_dollar_quotes(value):
+    assert StringLiteral.parse_string(quote_value(value), parse_all=True)[0] == value

tests/test_props.py

@@ -124,28 +124,47 @@ def test_props_render(self):
 # ============================================================================
 
 
-class TestQuoteValue:
-    """Tests for the quote_value helper function."""
+@pytest.mark.parametrize(
+    "value, expected",
+    [
+        ("hello world", "$$hello world$$"),
+        ("", "''"),
+        (None, "''"),
+        ('it\'s a "test"', '$$it\'s a "test"$$'),
+        ("line1\nline2", "$$line1\nline2$$"),
+        # $$ in the value forces the single-quoted fallback
+        ("costs $$ and it's dear", "'costs $$ and it''s dear'"),
+        ('$$ say "hi"', "'$$ say \\\"hi\\\"'"),
+        ("$$ path C:\\tmp", "'$$ path C:\\\\tmp'"),
+        ("costs $$\nper line", "'costs $$\\nper line'"),
+        ("$$\r\tx", "'$$\\r\\tx'"),
+        ("$$\b\fx", "'$$\\b\\fx'"),
+        ("$$\0x", "'$$\\u0000x'"),
+        ("$$\x01x", "'$$\\u0001x'"),
+        # Single $ not at the end uses dollar quoting
+        ("costs $5", "$$costs $5$$"),
+        ("$100", "$$$100$$"),
+        # Trailing $ combines with the closing $$ so forces single-quoted fallback
+        ("abc$", "'abc$'"),
+        ("$", "'$'"),
+        ("it's dear and ends with $", "'it''s dear and ends with $'"),
+    ],
+)
+def test_quote_value(value, expected):
+    assert quote_value(value) == expected
 
-    def test_quote_value_normal_string(self):
-        result = quote_value("hello world")
-        assert result == "$$hello world$$"
 
-    def test_quote_value_empty_string(self):
-        result = quote_value("")
-        assert result == "''"
+def test_quote_value_rejects_lone_surrogate():
+    with pytest.raises(UnicodeEncodeError):
+        quote_value("bad \ud800$")
 
-    def test_quote_value_none(self):
-        result = quote_value(None)
-        assert result == "''"
 
-    def test_quote_value_with_quotes(self):
-        result = quote_value('it\'s a "test"')
-        assert result == '$$it\'s a "test"$$'
+def test_enum_prop_quoted_render_escapes_apostrophe():
+    class Kind(Enum):
+        APOSTROPHE = "it's"
 
-    def test_quote_value_multiline(self):
-        result = quote_value("line1\nline2")
-        assert result == "$$line1\nline2$$"
+    prop = EnumProp("type", [Kind.APOSTROPHE], quoted=True)
+    assert prop.render(Kind.APOSTROPHE) == "type = 'it''s'"
 
 
 class TestBoolPropExtended:
@@ -427,7 +446,12 @@ def test_parse_single_kv(self):
     def test_render_dict(self):
         prop = DictProp("HEADERS", parens=True)
         result = prop.render({"Content-Type": "application/json"})
-        assert result == "HEADERS = ('Content-Type' = 'application/json')"
+        assert result == "HEADERS = ($$Content-Type$$ = $$application/json$$)"
+
+    def test_render_dict_with_apostrophe(self):
+        prop = DictProp("HEADERS", parens=True)
+        result = prop.render({"X-Note": "it's fine"})
+        assert result == "HEADERS = ($$X-Note$$ = $$it's fine$$)"
 
     def test_render_none(self):
         prop = DictProp("HEADERS")
@@ -614,7 +638,7 @@ def test_render_timestamp(self):
     def test_render_stream(self):
         prop = TimeTravelProp("AT")
         result = prop.render({"STREAM": "my_stream"})
-        assert result == "AT (STREAM => 'my_stream')"
+        assert result == "AT (STREAM => $$my_stream$$)"
 
     def test_render_none(self):
         prop = TimeTravelProp("AT")