Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def test_parse(test_case, parse_pattern):
# Get test ddl script
data = TEST_DATA[test_case]
# Create parse object instance
ddlparse = DdlParse()
# Set source database option
# Set source database option & Parse ddl
if parse_pattern == DDL_SET_PATTERN.method:
if data["database"] is not None:
table = ddlparse.parse(data["ddl"], data["database"])
else:
table = ddlparse.parse(data["ddl"])
else:
if data["database"] is not None:
ddlparse.source_database = data["database"]
ddlparse.ddl = data["ddl"]
table = ddlparse.parse()
def test_set_comment():
ddl = """
CREATE TABLE Sample_Table (
Col_01 integer
);
"""
# Parse DDL
table = DdlParse().parse(ddl)
# Check blank comment
assert table.columns["Col_01"].comment is None
assert table.columns["Col_01"].description is None
# Check comment setter
table.columns["Col_01"].comment = "comment_1"
assert table.columns["Col_01"].comment == "comment_1"
# Check description setter
table.columns["Col_01"].description = "comment_2"
assert table.columns["Col_01"].description == "comment_2"
def test_exception_ddl():
# Create parse object instance
ddlparse = DdlParse()
# Do not set DDL
# Error : DDL is not specified
with pytest.raises(ValueError):
ddlparse.parse()
def test_exception_bq_data_type():
ddl = """
CREATE TABLE Sample_Table (
Col_01 NG_DATA_TYPE,
);
"""
# Parse DDL
table = DdlParse().parse(ddl)
# Error: Unknown data type
with pytest.raises(ValueError):
print(table.columns["col_01"].bigquery_data_type)
def test_bq_ddl(test_case):
# Get test data
data = TEST_DATA_DDL[test_case]
# Parse ddl
table = DdlParse().parse(data["source_ddl"])
# Check generate BigQuery DDL statements of DdlParseTable
assert table.to_bigquery_ddl() == textwrap.dedent(data["bq_ddl"][DdlParse.NAME_CASE.original])
assert table.to_bigquery_ddl(DdlParse.NAME_CASE.original) == textwrap.dedent(data["bq_ddl"][DdlParse.NAME_CASE.original])
assert table.to_bigquery_ddl(DdlParse.NAME_CASE.lower) == textwrap.dedent(data["bq_ddl"][DdlParse.NAME_CASE.lower])
assert table.to_bigquery_ddl(DdlParse.NAME_CASE.upper) == textwrap.dedent(data["bq_ddl"][DdlParse.NAME_CASE.upper])