diff --git a/snowcap/lifecycle.py b/snowcap/lifecycle.py index 701f4f5..05545af 100644 --- a/snowcap/lifecycle.py +++ b/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__] @@ -508,9 +508,9 @@ def update_scanner_package(urn: URN, data: dict, props: Props) -> str: package_name = f"'{urn.fqn.name}'" 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}',", @@ -535,7 +535,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) diff --git a/snowcap/props.py b/snowcap/props.py index d478bf1..acab577 100644 --- a/snowcap/props.py +++ b/snowcap/props.py @@ -27,6 +27,10 @@ def quote_value(value: str): if value is None or value == "": return "''" + if "$$" in str(value): + # JSON and Snowflake share backslash escape syntax for control characters + escaped = json.dumps(str(value), ensure_ascii=False)[1:-1].replace("'", "''") + return f"'{escaped}'" return f"$${value}$$" diff --git a/tests/test_lifecycle.py b/tests/test_lifecycle.py index 1d08006..eda88e3 100644 --- a/tests/test_lifecycle.py +++ b/tests/test_lifecycle.py @@ -874,7 +874,15 @@ def test_update_other_property(self): props = MockProps("") result = update_scanner_package(urn, data, props) assert "'enabled'" in result - assert "'TRUE'" 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 class TestUpdateSchema: @@ -937,6 +945,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.""" diff --git a/tests/test_props.py b/tests/test_props.py index e31ca48..9c77b56 100644 --- a/tests/test_props.py +++ b/tests/test_props.py @@ -124,28 +124,26 @@ def test_props_render(self): # ============================================================================ -class TestQuoteValue: - """Tests for the quote_value helper function.""" - - 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_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_quote_value_multiline(self): - result = quote_value("line1\nline2") - assert result == "$$line1\nline2$$" +@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'"), + ], +) +def test_quote_value(value, expected): + assert quote_value(value) == expected class TestBoolPropExtended: