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
8 changes: 4 additions & 4 deletions snowcap/lifecycle.py
Original file line number Diff line number Diff line change
Expand Up @@ -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__]
Expand Down Expand Up @@ -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}',",
Expand All @@ -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)


Expand Down
4 changes: 4 additions & 0 deletions snowcap/props.py
Original file line number Diff line number Diff line change
Expand Up @@ -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}$$"


Expand Down
26 changes: 25 additions & 1 deletion tests/test_lifecycle.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down Expand Up @@ -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."""
Expand Down
42 changes: 20 additions & 22 deletions tests/test_props.py
Original file line number Diff line number Diff line change
Expand Up @@ -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:
Expand Down
Loading