# Copyright 2016 Google, Inc.

#

# Licensed under the Apache License, Version 2.0 (the "License");

# you may not use this file except in compliance with the License.

# You may obtain a copy of the License at

#

# http://www.apache.org/licenses/LICENSE-2.0

#

# Unless required by applicable law or agreed to in writing, software

# distributed under the License is distributed on an "AS IS" BASIS,

# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

# See the License for the specific language governing permissions and

# limitations under the License.

import time

import uuid

from google.api_core import exceptions

from google.cloud import spanner

from google.cloud.spanner_admin_database_v1.types.common import DatabaseDialect

import pytest

from test_utils.retry import RetryErrors

import snippets

CREATE_TABLE_SINGERS = """\

CREATE TABLE Singers (

SingerId INT64 NOT NULL,

FirstName STRING(1024),

LastName STRING(1024),

SingerInfo BYTES(MAX),

FullName STRING(2048) AS (

ARRAY_TO_STRING([FirstName, LastName], " ")

) STORED

) PRIMARY KEY (SingerId)

"""

CREATE_TABLE_ALBUMS = """\

CREATE TABLE Albums (

SingerId INT64 NOT NULL,

AlbumId INT64 NOT NULL,

AlbumTitle STRING(MAX)

) PRIMARY KEY (SingerId, AlbumId),

INTERLEAVE IN PARENT Singers ON DELETE CASCADE

"""

CREATE_TABLE_SINGERS_ = """\

CREATE TABLE Singers (

SingerId INT64 NOT NULL,

FirstName STRING(1024),

LastName STRING(1024),

) PRIMARY KEY (SingerId)

"""

retry_429 = RetryErrors(exceptions.ResourceExhausted, delay=15)

@pytest.fixture(scope="module")

def sample_name():

return "snippets"

@pytest.fixture(scope="module")

def database_dialect():

"""Spanner dialect to be used for this sample.

The dialect is used to initialize the dialect for the database.

It can either be GoogleStandardSql or PostgreSql.

"""

return DatabaseDialect.GOOGLE_STANDARD_SQL

@pytest.fixture(scope="module")

def create_instance_id():

"""Id for the low-cost instance."""

return f"create-instance-{uuid.uuid4().hex[:10]}"

@pytest.fixture(scope="module")

def lci_instance_id():

"""Id for the low-cost instance."""

return f"lci-instance-{uuid.uuid4().hex[:10]}"

@pytest.fixture(scope="module")

def instance_partition_instance_id():

"""Id for the instance that tests instance partitions."""

return f"instance-partition-test-{uuid.uuid4().hex[:10]}"

@pytest.fixture(scope="module")

def database_id():

return f"test-db-{uuid.uuid4().hex[:10]}"

@pytest.fixture(scope="module")

def create_database_id():

return f"create-db-{uuid.uuid4().hex[:10]}"

@pytest.fixture(scope="module")

def cmek_database_id():

return f"cmek-db-{uuid.uuid4().hex[:10]}"

@pytest.fixture(scope="module")

def default_leader_database_id():

return f"leader_db_{uuid.uuid4().hex[:10]}"

@pytest.fixture(scope="module")

def proto_columns_database_id():

return f"test-db-proto-{uuid.uuid4().hex[:10]}"

@pytest.fixture(scope="module")

def database_ddl():

"""Sequence of DDL statements used to set up the database.

Sample testcase modules can override as needed.

"""

return [CREATE_TABLE_SINGERS, CREATE_TABLE_ALBUMS]

@pytest.fixture(scope="module")

def proto_columns_database_ddl():

"""Sequence of DDL statements used to set up the database for proto columns.

Sample testcase modules can override as needed.

"""

return [CREATE_TABLE_SINGERS_, CREATE_TABLE_ALBUMS]

@pytest.fixture(scope="module")

def default_leader():

"""Default leader for multi-region instances."""

return "us-east4"

@pytest.fixture(scope="module")

def user_managed_instance_config_name(spanner_client):

name = f"custom-python-samples-config-{uuid.uuid4().hex[:10]}"

yield name

snippets.delete_instance_config(

"{}/instanceConfigs/{}".format(spanner_client.project_name, name)

)

return

@pytest.fixture(scope="module")

def base_instance_config_id(spanner_client):

return "{}/instanceConfigs/{}".format(spanner_client.project_name, "nam7")

def test_create_and_update_instance_explicit(spanner_client, create_instance_id):

# Rather than re-use 'sample_isntance', we create a new instance, to

# ensure that the 'create_instance' snippet is tested.

retry_429(snippets.create_instance)(create_instance_id)

# Rather than re-use 'sample_isntance', we are using created instance, to

# ensure that the 'update_instance' snippet is tested.

retry_429(snippets.update_instance)(create_instance_id)

instance = spanner_client.instance(create_instance_id)

retry_429(instance.delete)()

def test_create_database_explicit(sample_instance, create_database_id):

# Rather than re-use 'sample_database', we create a new database, to

# ensure that the 'create_database' snippet is tested.

snippets.create_database(sample_instance.instance_id, create_database_id)

database = sample_instance.database(create_database_id)

database.drop()

def test_create_instance_with_processing_units(capsys, lci_instance_id):

processing_units = 500

retry_429(snippets.create_instance_with_processing_units)(

lci_instance_id,

processing_units,

)

out, _ = capsys.readouterr()

assert lci_instance_id in out

assert "{} processing units".format(processing_units) in out

spanner_client = spanner.Client()

instance = spanner_client.instance(lci_instance_id)

retry_429(instance.delete)()

def test_create_instance_with_autoscaling_config(capsys, lci_instance_id):

retry_429(snippets.create_instance_with_autoscaling_config)(

lci_instance_id,

)

out, _ = capsys.readouterr()

assert lci_instance_id in out

assert "autoscaling config" in out

spanner_client = spanner.Client()

instance = spanner_client.instance(lci_instance_id)

retry_429(instance.delete)()

def test_create_and_update_instance_default_backup_schedule_type(

capsys, lci_instance_id

):

retry_429(snippets.create_instance_without_default_backup_schedules)(

lci_instance_id,

)

create_out, _ = capsys.readouterr()

assert lci_instance_id in create_out

assert "without default backup schedules" in create_out

retry_429(snippets.update_instance_default_backup_schedule_type)(

lci_instance_id,

)

update_out, _ = capsys.readouterr()

assert lci_instance_id in update_out

assert "to have default backup schedules" in update_out

spanner_client = spanner.Client()

instance = spanner_client.instance(lci_instance_id)

retry_429(instance.delete)()

def test_create_instance_partition(capsys, instance_partition_instance_id):

# Unable to use create_instance since it has editions set where partitions are unsupported.

# The minimal requirement for editions is ENTERPRISE_PLUS for the paritions to get supported.

snippets.create_instance_with_processing_units(instance_partition_instance_id, 1000)

retry_429(snippets.create_instance_partition)(

instance_partition_instance_id, "my-instance-partition"

)

out, _ = capsys.readouterr()

assert "Created instance partition my-instance-partition" in out

spanner_client = spanner.Client()

instance = spanner_client.instance(instance_partition_instance_id)

retry_429(instance.delete)()

def test_update_database(capsys, instance_id, sample_database):

snippets.update_database(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Updated database {}.".format(sample_database.name) in out

# Cleanup

sample_database.enable_drop_protection = False

op = sample_database.update(["enable_drop_protection"])

op.result()

def test_create_database_with_encryption_config(

capsys, instance_id, cmek_database_id, kms_key_name

):

snippets.create_database_with_encryption_key(

instance_id, cmek_database_id, kms_key_name

)

out, _ = capsys.readouterr()

assert cmek_database_id in out

assert kms_key_name in out

@pytest.mark.skip(reason="skipped since the KMS keys are not added on test " "project")

def test_create_database_with_multiple_kms_keys(

capsys,

multi_region_instance,

multi_region_instance_id,

cmek_database_id,

kms_key_names,

):

snippets.create_database_with_multiple_kms_keys(

multi_region_instance_id, cmek_database_id, kms_key_names

)

out, _ = capsys.readouterr()

assert cmek_database_id in out

assert kms_key_names[0] in out

assert kms_key_names[1] in out

assert kms_key_names[2] in out

def test_get_instance_config(capsys):

instance_config = "nam6"

snippets.get_instance_config(instance_config)

out, _ = capsys.readouterr()

assert instance_config in out

def test_list_instance_config(capsys):

snippets.list_instance_config()

out, _ = capsys.readouterr()

assert "regional-us-central1" in out

@pytest.mark.dependency(name="create_instance_config")

def test_create_instance_config(

capsys, user_managed_instance_config_name, base_instance_config_id

):

snippets.create_instance_config(

user_managed_instance_config_name, base_instance_config_id

)

out, _ = capsys.readouterr()

assert "Created instance configuration" in out

@pytest.mark.dependency(depends=["create_instance_config"])

def test_update_instance_config(capsys, user_managed_instance_config_name):

snippets.update_instance_config(user_managed_instance_config_name)

out, _ = capsys.readouterr()

assert "Updated instance configuration" in out

@pytest.mark.dependency(depends=["create_instance_config"])

def test_delete_instance_config(capsys, user_managed_instance_config_name):

spanner_client = spanner.Client()

snippets.delete_instance_config(

"{}/instanceConfigs/{}".format(

spanner_client.project_name, user_managed_instance_config_name

)

)

out, _ = capsys.readouterr()

assert "successfully deleted" in out

def test_list_instance_config_operations(capsys):

snippets.list_instance_config_operations()

out, _ = capsys.readouterr()

assert "List instance config operations" in out

def test_list_databases(capsys, instance_id):

snippets.list_databases(instance_id)

out, _ = capsys.readouterr()

assert "has default leader" in out

def test_create_database_with_default_leader(

capsys,

multi_region_instance,

multi_region_instance_id,

default_leader_database_id,

default_leader,

):

retry_429 = RetryErrors(exceptions.ResourceExhausted, delay=15)

retry_429(snippets.create_database_with_default_leader)(

multi_region_instance_id, default_leader_database_id, default_leader

)

out, _ = capsys.readouterr()

assert default_leader_database_id in out

assert default_leader in out

def test_update_database_with_default_leader(

capsys,

multi_region_instance,

multi_region_instance_id,

default_leader_database_id,

default_leader,

):

retry_429 = RetryErrors(exceptions.ResourceExhausted, delay=15)

retry_429(snippets.update_database_with_default_leader)(

multi_region_instance_id, default_leader_database_id, default_leader

)

out, _ = capsys.readouterr()

assert default_leader_database_id in out

assert default_leader in out

def test_get_database_ddl(capsys, instance_id, sample_database):

snippets.get_database_ddl(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert sample_database.database_id in out

def test_query_information_schema_database_options(

capsys,

multi_region_instance,

multi_region_instance_id,

default_leader_database_id,

default_leader,

):

snippets.query_information_schema_database_options(

multi_region_instance_id, default_leader_database_id

)

out, _ = capsys.readouterr()

assert default_leader in out

@pytest.mark.dependency(name="insert_data")

def test_insert_data(capsys, instance_id, sample_database):

snippets.insert_data(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Inserted data" in out

@pytest.mark.dependency(name="batch_write")

def test_batch_write(capsys, instance_id, sample_database):

snippets.batch_write(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "could not be applied with error" not in out

@pytest.mark.dependency(depends=["insert_data"])

def test_delete_data(capsys, instance_id, sample_database):

snippets.delete_data(instance_id, sample_database.database_id)

# put it back for other tests

snippets.insert_data(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Deleted data" in out

@pytest.mark.dependency(depends=["insert_data"])

def test_query_data(capsys, instance_id, sample_database):

snippets.query_data(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "SingerId: 1, AlbumId: 1, AlbumTitle: Total Junk" in out

@pytest.mark.dependency(name="add_column", depends=["insert_data"])

def test_add_column(capsys, instance_id, sample_database):

snippets.add_column(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Added the MarketingBudget column." in out

@pytest.mark.dependency(depends=["insert_data"])

def test_read_data(capsys, instance_id, sample_database):

snippets.read_data(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "SingerId: 1, AlbumId: 1, AlbumTitle: Total Junk" in out

@pytest.mark.dependency(name="update_data", depends=["add_column"])

def test_update_data(capsys, instance_id, sample_database):

# Sleep for 15 seconds to ensure previous inserts will be

# 'stale' by the time test_read_stale_data is run.

time.sleep(15)

snippets.update_data(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Updated data." in out

@pytest.mark.dependency(depends=["update_data"])

def test_read_stale_data(capsys, instance_id, sample_database):

# This snippet relies on test_update_data inserting data

# at least 15 seconds after the previous insert

snippets.read_stale_data(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "SingerId: 1, AlbumId: 1, MarketingBudget: None" in out

@pytest.mark.dependency(depends=["add_column"])

def test_read_write_transaction(capsys, instance_id, sample_database):

snippets.read_write_transaction(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Transaction complete" in out

@pytest.mark.dependency(depends=["add_column"])

def test_query_data_with_new_column(capsys, instance_id, sample_database):

snippets.query_data_with_new_column(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "SingerId: 1, AlbumId: 1, MarketingBudget: 300000" in out

assert "SingerId: 2, AlbumId: 2, MarketingBudget: 300000" in out

@pytest.mark.dependency(name="add_index", depends=["insert_data"])

def test_add_index(capsys, instance_id, sample_database):

snippets.add_index(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Added the AlbumsByAlbumTitle index" in out

@pytest.mark.dependency(depends=["add_index"])

def test_query_data_with_index(capsys, instance_id, sample_database):

snippets.query_data_with_index(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Go, Go, Go" in out

assert "Forever Hold Your Peace" in out

assert "Green" not in out

@pytest.mark.dependency(depends=["add_index"])

def test_read_data_with_index(capsys, instance_id, sample_database):

snippets.read_data_with_index(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Go, Go, Go" in out

assert "Forever Hold Your Peace" in out

assert "Green" in out

@pytest.mark.dependency(name="add_storing_index", depends=["insert_data"])

def test_add_storing_index(capsys, instance_id, sample_database):

snippets.add_storing_index(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Added the AlbumsByAlbumTitle2 index." in out

@pytest.mark.dependency(depends=["add_storing_index"])

def test_read_data_with_storing_index(capsys, instance_id, sample_database):

snippets.read_data_with_storing_index(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "300000" in out

@pytest.mark.dependency(depends=["insert_data"])

def test_read_only_transaction(capsys, instance_id, sample_database):

snippets.read_only_transaction(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

# Snippet does two reads, so entry should be listed twice

assert out.count("SingerId: 1, AlbumId: 1, AlbumTitle: Total Junk") == 2

@pytest.mark.dependency(name="add_timestamp_column", depends=["insert_data"])

def test_add_timestamp_column(capsys, instance_id, sample_database):

snippets.add_timestamp_column(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert 'Altered table "Albums" on database ' in out

@pytest.mark.dependency(depends=["add_timestamp_column"])

def test_update_data_with_timestamp(capsys, instance_id, sample_database):

snippets.update_data_with_timestamp(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Updated data" in out

@pytest.mark.dependency(depends=["add_timestamp_column"])

def test_query_data_with_timestamp(capsys, instance_id, sample_database):

snippets.query_data_with_timestamp(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "SingerId: 1, AlbumId: 1, MarketingBudget: 1000000" in out

assert "SingerId: 2, AlbumId: 2, MarketingBudget: 750000" in out

@pytest.mark.dependency(name="create_table_with_timestamp")

def test_create_table_with_timestamp(capsys, instance_id, sample_database):

snippets.create_table_with_timestamp(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Created Performances table on database" in out

@pytest.mark.dependency(depends=["create_table_with_timestamp"])

def test_insert_data_with_timestamp(capsys, instance_id, sample_database):

snippets.insert_data_with_timestamp(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Inserted data." in out

@pytest.mark.dependency(name="write_struct_data")

def test_write_struct_data(capsys, instance_id, sample_database):

snippets.write_struct_data(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Inserted sample data for STRUCT queries" in out

@pytest.mark.dependency(depends=["write_struct_data"])

def test_query_with_struct(capsys, instance_id, sample_database):

snippets.query_with_struct(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "SingerId: 6" in out

@pytest.mark.dependency(depends=["write_struct_data"])

def test_query_with_array_of_struct(capsys, instance_id, sample_database):

snippets.query_with_array_of_struct(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "SingerId: 8" in out

assert "SingerId: 7" in out

assert "SingerId: 6" in out

@pytest.mark.dependency(depends=["write_struct_data"])

def test_query_struct_field(capsys, instance_id, sample_database):

snippets.query_struct_field(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "SingerId: 6" in out

@pytest.mark.dependency(depends=["write_struct_data"])

def test_query_nested_struct_field(capsys, instance_id, sample_database):

snippets.query_nested_struct_field(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "SingerId: 6 SongName: Imagination" in out

assert "SingerId: 9 SongName: Imagination" in out

@pytest.mark.dependency(name="insert_data_with_dml")

def test_insert_data_with_dml(capsys, instance_id, sample_database):

snippets.insert_data_with_dml(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "1 record(s) inserted." in out

@pytest.mark.dependency(name="log_commit_stats")

def test_log_commit_stats(capsys, instance_id, sample_database):

snippets.log_commit_stats(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "1 record(s) inserted." in out

# SingerId, FirstName, and LastName plus FullName which is generated.

assert "4 mutation(s) in transaction." in out

@pytest.mark.dependency(name="set_max_commit_delay")

def test_set_max_commit_delay(capsys, instance_id, sample_database):

snippets.set_max_commit_delay(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "1 record(s) inserted." in out

@pytest.mark.dependency(depends=["insert_data"])

def test_update_data_with_dml(capsys, instance_id, sample_database):

snippets.update_data_with_dml(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "1 record(s) updated." in out

@pytest.mark.dependency(depends=["add_column"])

def test_update_data_with_dml_returning(capsys, instance_id, sample_database):

snippets.update_data_with_dml_returning(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "1 record(s) updated." in out

@pytest.mark.dependency(depends=["insert_data"])

def test_delete_data_with_dml(capsys, instance_id, sample_database):

snippets.delete_data_with_dml(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "1 record(s) deleted." in out

@pytest.mark.dependency(depends=["insert_data"])

def test_delete_data_with_dml_returning(capsys, instance_id, sample_database):

snippets.delete_data_with_dml_returning(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "1 record(s) deleted." in out

@pytest.mark.dependency(depends=["add_timestamp_column"])

def test_update_data_with_dml_timestamp(capsys, instance_id, sample_database):

snippets.update_data_with_dml_timestamp(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "2 record(s) updated." in out

@pytest.mark.dependency(name="dml_write_read_transaction")

def test_dml_write_read_transaction(capsys, instance_id, sample_database):

snippets.dml_write_read_transaction(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "1 record(s) inserted." in out

assert "FirstName: Timothy, LastName: Campbell" in out

@pytest.mark.dependency(depends=["dml_write_read_transaction"])

def test_update_data_with_dml_struct(capsys, instance_id, sample_database):

snippets.update_data_with_dml_struct(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "1 record(s) updated" in out

@pytest.mark.dependency(name="insert_with_dml")

def test_insert_with_dml(capsys, instance_id, sample_database):

snippets.insert_with_dml(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "4 record(s) inserted" in out

@pytest.mark.dependency(depends=[""])

def test_insert_with_dml_returning(capsys, instance_id, sample_database):

snippets.insert_with_dml_returning(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "4 record(s) inserted" in out

@pytest.mark.dependency(depends=["insert_with_dml"])

def test_query_data_with_parameter(capsys, instance_id, sample_database):

snippets.query_data_with_parameter(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "SingerId: 12, FirstName: Melissa, LastName: Garcia" in out

@pytest.mark.dependency(depends=["add_column"])

def test_write_with_dml_transaction(capsys, instance_id, sample_database):

snippets.write_with_dml_transaction(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Transferred 200000 from Album2's budget to Album1's" in out

@pytest.mark.dependency(depends=["add_column"])

def test_update_data_with_partitioned_dml(capsys, instance_id, sample_database):

snippets.update_data_with_partitioned_dml(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "3 records updated" in out

@pytest.mark.dependency(

depends=[

"insert_with_dml",

"dml_write_read_transaction",

"log_commit_stats",

"set_max_commit_delay",

]

)

def test_delete_data_with_partitioned_dml(capsys, instance_id, sample_database):

snippets.delete_data_with_partitioned_dml(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "7 record(s) deleted" in out

@pytest.mark.dependency(depends=["add_column"])

def test_update_with_batch_dml(capsys, instance_id, sample_database):

snippets.update_with_batch_dml(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Executed 2 SQL statements using Batch DML" in out

@pytest.mark.dependency(name="create_table_with_datatypes")

def test_create_table_with_datatypes(capsys, instance_id, sample_database):

snippets.create_table_with_datatypes(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Created Venues table on database" in out

@pytest.mark.dependency(

name="insert_datatypes_data",

depends=["create_table_with_datatypes"],

)

def test_insert_datatypes_data(capsys, instance_id, sample_database):

snippets.insert_datatypes_data(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Inserted data." in out

@pytest.mark.dependency(depends=["insert_datatypes_data"])

def test_query_data_with_array(capsys, instance_id, sample_database):

snippets.query_data_with_array(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "VenueId: 19, VenueName: Venue 19, AvailableDate: 2020-11-01" in out

assert "VenueId: 42, VenueName: Venue 42, AvailableDate: 2020-10-01" in out

@pytest.mark.dependency(depends=["insert_datatypes_data"])

def test_query_data_with_bool(capsys, instance_id, sample_database):

snippets.query_data_with_bool(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "VenueId: 19, VenueName: Venue 19, OutdoorVenue: True" in out

@pytest.mark.dependency(depends=["insert_datatypes_data"])

def test_query_data_with_bytes(capsys, instance_id, sample_database):

snippets.query_data_with_bytes(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "VenueId: 4, VenueName: Venue 4" in out

@pytest.mark.dependency(depends=["insert_datatypes_data"])

def test_query_data_with_date(capsys, instance_id, sample_database):

snippets.query_data_with_date(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "VenueId: 4, VenueName: Venue 4, LastContactDate: 2018-09-02" in out

assert "VenueId: 42, VenueName: Venue 42, LastContactDate: 2018-10-01" in out

@pytest.mark.dependency(depends=["insert_datatypes_data"])

def test_query_data_with_float(capsys, instance_id, sample_database):

snippets.query_data_with_float(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "VenueId: 4, VenueName: Venue 4, PopularityScore: 0.8" in out

assert "VenueId: 19, VenueName: Venue 19, PopularityScore: 0.9" in out

@pytest.mark.dependency(depends=["insert_datatypes_data"])

def test_query_data_with_int(capsys, instance_id, sample_database):

snippets.query_data_with_int(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "VenueId: 19, VenueName: Venue 19, Capacity: 6300" in out

assert "VenueId: 42, VenueName: Venue 42, Capacity: 3000" in out

@pytest.mark.dependency(depends=["insert_datatypes_data"])

def test_query_data_with_string(capsys, instance_id, sample_database):

snippets.query_data_with_string(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "VenueId: 42, VenueName: Venue 42" in out

@pytest.mark.dependency(

name="add_numeric_column",

depends=["create_table_with_datatypes"],

)

def test_add_numeric_column(capsys, instance_id, sample_database):

snippets.add_numeric_column(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert 'Altered table "Venues" on database ' in out

@pytest.mark.dependency(depends=["add_numeric_column", "insert_datatypes_data"])

def test_update_data_with_numeric(capsys, instance_id, sample_database):

snippets.update_data_with_numeric(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Updated data" in out

@pytest.mark.dependency(depends=["add_numeric_column"])

def test_query_data_with_numeric_parameter(capsys, instance_id, sample_database):

snippets.query_data_with_numeric_parameter(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "VenueId: 4, Revenue: 35000" in out

@pytest.mark.dependency(

name="add_json_column",

depends=["create_table_with_datatypes"],

)

def test_add_json_column(capsys, instance_id, sample_database):

snippets.add_json_column(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert 'Altered table "Venues" on database ' in out

@pytest.mark.dependency(depends=["add_json_column", "insert_datatypes_data"])

def test_update_data_with_json(capsys, instance_id, sample_database):

snippets.update_data_with_json(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Updated data" in out

@pytest.mark.dependency(depends=["add_json_column"])

def test_query_data_with_json_parameter(capsys, instance_id, sample_database):

snippets.query_data_with_json_parameter(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "VenueId: 19, VenueDetails: {'open': True, 'rating': 9}" in out

@pytest.mark.dependency(depends=["insert_datatypes_data"])

def test_query_data_with_timestamp_parameter(capsys, instance_id, sample_database):

snippets.query_data_with_timestamp_parameter(

instance_id, sample_database.database_id

)

out, _ = capsys.readouterr()

assert "VenueId: 4, VenueName: Venue 4, LastUpdateTime:" in out

assert "VenueId: 19, VenueName: Venue 19, LastUpdateTime:" in out

assert "VenueId: 42, VenueName: Venue 42, LastUpdateTime:" in out

@pytest.mark.dependency(depends=["insert_datatypes_data"])

def test_query_data_with_query_options(capsys, instance_id, sample_database):

snippets.query_data_with_query_options(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "VenueId: 4, VenueName: Venue 4, LastUpdateTime:" in out

assert "VenueId: 19, VenueName: Venue 19, LastUpdateTime:" in out

assert "VenueId: 42, VenueName: Venue 42, LastUpdateTime:" in out

@pytest.mark.dependency(depends=["insert_datatypes_data"])

def test_create_client_with_query_options(capsys, instance_id, sample_database):

snippets.create_client_with_query_options(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "VenueId: 4, VenueName: Venue 4, LastUpdateTime:" in out

assert "VenueId: 19, VenueName: Venue 19, LastUpdateTime:" in out

assert "VenueId: 42, VenueName: Venue 42, LastUpdateTime:" in out

@pytest.mark.dependency(depends=["insert_datatypes_data"])

def test_set_transaction_tag(capsys, instance_id, sample_database):

snippets.set_transaction_tag(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Venue capacities updated." in out

assert "New venue inserted." in out

@pytest.mark.dependency(depends=["insert_datatypes_data"])

def test_set_transaction_timeout(capsys, instance_id, sample_database):

snippets.set_transaction_timeout(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "1 record(s) inserted." in out

@pytest.mark.dependency(depends=["insert_datatypes_data"])

def test_set_statement_timeout(capsys, instance_id, sample_database):

snippets.set_statement_timeout(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "1 record(s) inserted." in out

@pytest.mark.dependency(depends=["insert_data"])

def test_set_request_tag(capsys, instance_id, sample_database):

snippets.set_request_tag(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "SingerId: 1, AlbumId: 1, AlbumTitle: Total Junk" in out

@pytest.mark.dependency(name="add_and_drop_database_roles", depends=["insert_data"])

def test_add_and_drop_database_roles(capsys, instance_id, sample_database):

snippets.add_and_drop_database_roles(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "Created roles new_parent and new_child and granted privileges" in out

assert "Revoked privileges and dropped role new_child" in out

@pytest.mark.dependency(depends=["add_and_drop_database_roles"])

def test_read_data_with_database_role(capsys, instance_id, sample_database):

snippets.read_data_with_database_role(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "ingerId: 1, FirstName: Marc, LastName: Richards" in out

@pytest.mark.dependency(depends=["add_and_drop_database_roles"])

def test_list_database_roles(capsys, instance_id, sample_database):

snippets.list_database_roles(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "new_parent" in out

@pytest.mark.dependency(name="create_table_with_foreign_key_delete_cascade")

def test_create_table_with_foreign_key_delete_cascade(

capsys, instance_id, sample_database

):

snippets.create_table_with_foreign_key_delete_cascade(

instance_id, sample_database.database_id

)

out, _ = capsys.readouterr()

assert (

"Created Customers and ShoppingCarts table with FKShoppingCartsCustomerId"

in out

)

@pytest.mark.dependency(

name="alter_table_with_foreign_key_delete_cascade",

depends=["create_table_with_foreign_key_delete_cascade"],

)

def test_alter_table_with_foreign_key_delete_cascade(

capsys, instance_id, sample_database

):

snippets.alter_table_with_foreign_key_delete_cascade(

instance_id, sample_database.database_id

)

out, _ = capsys.readouterr()

assert "Altered ShoppingCarts table with FKShoppingCartsCustomerName" in out

@pytest.mark.dependency(depends=["alter_table_with_foreign_key_delete_cascade"])

def test_drop_foreign_key_contraint_delete_cascade(

capsys, instance_id, sample_database

):

snippets.drop_foreign_key_constraint_delete_cascade(

instance_id, sample_database.database_id

)

out, _ = capsys.readouterr()

assert "Altered ShoppingCarts table to drop FKShoppingCartsCustomerName" in out

def test_create_sequence(capsys, instance_id, bit_reverse_sequence_database):

snippets.create_sequence(instance_id, bit_reverse_sequence_database.database_id)

out, _ = capsys.readouterr()

assert (

"Created Seq sequence and Customers table, where the key column CustomerId uses the sequence as a default value on database"

in out

)

assert "Number of customer records inserted is 3" in out

assert "Inserted customer record with Customer Id:" in out

@pytest.mark.dependency(depends=["create_sequence"])

def test_alter_sequence(capsys, instance_id, bit_reverse_sequence_database):

snippets.alter_sequence(instance_id, bit_reverse_sequence_database.database_id)

out, _ = capsys.readouterr()

assert (

"Altered Seq sequence to skip an inclusive range between 1000 and 5000000 on database"

in out

)

assert "Number of customer records inserted is 3" in out

assert "Inserted customer record with Customer Id:" in out

@pytest.mark.dependency(depends=["alter_sequence"])

def test_drop_sequence(capsys, instance_id, bit_reverse_sequence_database):

snippets.drop_sequence(instance_id, bit_reverse_sequence_database.database_id)

out, _ = capsys.readouterr()

assert (

"Altered Customers table to drop DEFAULT from CustomerId column and dropped the Seq sequence on database"

in out

)

@pytest.mark.dependency(depends=["insert_data"])

def test_directed_read_options(capsys, instance_id, sample_database):

snippets.directed_read_options(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "SingerId: 1, AlbumId: 1, AlbumTitle: Total Junk" in out

@pytest.mark.dependency(depends=["insert_data"])

def test_set_custom_timeout_and_retry(capsys, instance_id, sample_database):

snippets.set_custom_timeout_and_retry(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "SingerId: 1, AlbumId: 1, AlbumTitle: Total Junk" in out

@pytest.mark.dependency(depends=["insert_data"])

def test_isolation_level_options(capsys, instance_id, sample_database):

snippets.isolation_level_options(instance_id, sample_database.database_id)

out, _ = capsys.readouterr()

assert "1 record(s) updated." in out