counterweight/lolafect
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

Test working

Browse source
This commit is contained in:
Pablo Martin 2023-04-24 11:38:51 +02:00
parent 5b53b71fd8
commit 47a3e8b1e7
1 changed files with 116 additions and 18 deletions
Download patch file Download diff file Expand all files Collapse all files

134
tests/test_integration/test_utils.py
View file

@ -1,4 +1,12 @@
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
# __ __ _____ _ _ _____ _ _ _____ _
# \ \ / /\ | __ \| \ | |_ _| \ | |/ ____| |
@ -18,24 +26,114 @@ TEST_LOLACONFIG = build_lolaconfig(flow_name="testing-suite")
def test_sql_transaction_persists_changes_properly():
# Connect
# Create table in Sandbox
# Check that table is empty
# Start transaction
# Insert value in table
# Commit transaction
# Check that value is there
# assert that the table was initially empty and afterwards it got a record
pass
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
)
""")
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)
cursor.execute("""
INSERT INTO sandbox.lolafect_transaction_test_table (a_test_column)
VALUES (1)
""")
end_sql_transaction.run(connection=connection, dry_run=False)
cursor.execute("""
SELECT a_test_column
FROM sandbox.lolafect_transaction_test_table
""")
table_has_a_record_after_transaction = bool(cursor.fetchall()) # A non-empty tuple yields True
cursor.execute("""
DROP TABLE sandbox.lolafect_transaction_test_table
"""
)
close_mysql_connection.run(connection=connection)
close_ssh_tunnel.run(tunnel=tunnel)
assert table_is_empty_at_first and table_has_a_record_after_transaction
def test_sql_transaction_rollbacks_changes_properly():
# Connect
# Create table in Sandbox
# Check that table is empty
# Start transaction
# Insert value in table
# Commit transaction
# Check that the table is still empty
# assert that the table was initially empty and afterwards as well
pass
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
)
""")
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)
cursor.execute("""
INSERT INTO sandbox.lolafect_transaction_test_table (a_test_column)
VALUES (1)
""")
end_sql_transaction.run(connection=connection, dry_run=True)
cursor.execute("""
SELECT a_test_column
FROM sandbox.lolafect_transaction_test_table
""")
table_is_empty_after_rollback = not bool(cursor.fetchall()) # An tuple yields False
cursor.execute("""
DROP TABLE sandbox.lolafect_transaction_test_table
"""
)
close_mysql_connection.run(connection=connection)
close_ssh_tunnel.run(tunnel=tunnel)
assert table_is_empty_at_first and table_is_empty_after_rollback