123 lines
4.2 KiB
Python
123 lines
4.2 KiB
Python
import pytest
|
|
|
|
from lolafect.lolaconfig import build_lolaconfig
|
|
from lolafect.connections import (
|
|
open_ssh_tunnel_with_s3_pkey,
|
|
get_local_bind_address_from_ssh_tunnel,
|
|
close_ssh_tunnel,
|
|
connect_to_mysql,
|
|
close_mysql_connection,
|
|
)
|
|
from lolafect.utils import begin_sql_transaction, end_sql_transaction
|
|
|
|
# __ __ _____ _ _ _____ _ _ _____ _
|
|
# \ \ / /\ | __ \| \ | |_ _| \ | |/ ____| |
|
|
# \ \ /\ / / \ | |__) | \| | | | | \| | | __| |
|
|
# \ \/ \/ / /\ \ | _ /| . ` | | | | . ` | | |_ | |
|
|
# \ /\ / ____ \| | \ \| |\ |_| |_| |\ | |__| |_|
|
|
# \/ \/_/ \_\_| \_\_| \_|_____|_| \_|\_____(_)
|
|
# This testing suite requires:
|
|
# - The calling shell to have permission in AWS
|
|
# - The calling shell to be within the Mercadão network
|
|
# - Do not use this tests as part of CI/CD pipelines since they are not idempotent and
|
|
# rely external resources. Instead, use them manually to check yourself that things
|
|
# are working properly.
|
|
|
|
|
|
TEST_LOLACONFIG = build_lolaconfig(flow_name="testing-suite")
|
|
|
|
|
|
@pytest.fixture
|
|
def connection_with_test_table():
|
|
"""
|
|
Connects to DW, creates a test table in the sandbox env, and yields the
|
|
connection to the test.
|
|
|
|
After the test, the table is dropped and the connection is closed.
|
|
"""
|
|
test_local_bind_host = "127.0.0.1"
|
|
test_local_bind_port = 12345
|
|
|
|
tunnel = open_ssh_tunnel_with_s3_pkey.run(
|
|
s3_bucket_name=TEST_LOLACONFIG.S3_BUCKET_NAME,
|
|
ssh_tunnel_credentials=TEST_LOLACONFIG.SSH_TUNNEL_CREDENTIALS,
|
|
remote_target_host=TEST_LOLACONFIG.DW_CREDENTIALS["host"],
|
|
remote_target_port=TEST_LOLACONFIG.DW_CREDENTIALS["port"],
|
|
local_bind_host=test_local_bind_host,
|
|
local_bind_port=test_local_bind_port,
|
|
)
|
|
|
|
connection = connect_to_mysql.run(
|
|
mysql_credentials=TEST_LOLACONFIG.DW_CREDENTIALS,
|
|
overriding_host_and_port=get_local_bind_address_from_ssh_tunnel.run(
|
|
tunnel=tunnel
|
|
),
|
|
)
|
|
cursor = connection.cursor()
|
|
cursor.execute("""
|
|
CREATE TABLE sandbox.lolafect_transaction_test_table
|
|
(
|
|
a_test_column INT
|
|
)
|
|
""")
|
|
|
|
# Connection and table ready for tests
|
|
yield connection # Test happens now
|
|
# Test finished, time to remove stuff and close connection
|
|
|
|
cursor.execute("""
|
|
DROP TABLE sandbox.lolafect_transaction_test_table
|
|
"""
|
|
)
|
|
close_mysql_connection.run(connection=connection)
|
|
close_ssh_tunnel.run(tunnel=tunnel)
|
|
|
|
|
|
def test_sql_transaction_persists_changes_properly(connection_with_test_table):
|
|
cursor = connection_with_test_table.cursor()
|
|
|
|
cursor.execute("""
|
|
SELECT a_test_column
|
|
FROM sandbox.lolafect_transaction_test_table
|
|
""")
|
|
table_is_empty_at_first = not bool(cursor.fetchall()) # An empty tuple yields False
|
|
|
|
begin_sql_transaction.run(connection=connection_with_test_table)
|
|
cursor.execute("""
|
|
INSERT INTO sandbox.lolafect_transaction_test_table (a_test_column)
|
|
VALUES (1)
|
|
""")
|
|
end_sql_transaction.run(connection=connection_with_test_table, dry_run=False)
|
|
|
|
cursor.execute("""
|
|
SELECT a_test_column
|
|
FROM sandbox.lolafect_transaction_test_table
|
|
""")
|
|
table_has_a_record_after_commit = bool(cursor.fetchall()) # A non-empty tuple yields True
|
|
|
|
assert table_is_empty_at_first and table_has_a_record_after_commit
|
|
|
|
|
|
def test_sql_transaction_rollbacks_changes_properly(connection_with_test_table):
|
|
cursor = connection_with_test_table.cursor()
|
|
|
|
cursor.execute("""
|
|
SELECT a_test_column
|
|
FROM sandbox.lolafect_transaction_test_table
|
|
""")
|
|
table_is_empty_at_first = not bool(cursor.fetchall()) # An empty tuple yields False
|
|
|
|
begin_sql_transaction.run(connection=connection_with_test_table)
|
|
cursor.execute("""
|
|
INSERT INTO sandbox.lolafect_transaction_test_table (a_test_column)
|
|
VALUES (1)
|
|
""")
|
|
end_sql_transaction.run(connection=connection_with_test_table, dry_run=True)
|
|
|
|
cursor.execute("""
|
|
SELECT a_test_column
|
|
FROM sandbox.lolafect_transaction_test_table
|
|
""")
|
|
table_is_still_empty_after_rollback = not bool(cursor.fetchall()) # A tuple yields False
|
|
|
|
assert table_is_empty_at_first and table_is_still_empty_after_rollback
|