117 lines
4.3 KiB
Python
117 lines
4.3 KiB
Python
from great_expectations.core.expectation_configuration import ExpectationConfiguration
|
|
|
|
from lolafect.lolaconfig import build_lolaconfig
|
|
from lolafect.data_testing import run_data_test_on_mysql
|
|
from lolafect.connections import open_ssh_tunnel_with_s3_pkey, close_ssh_tunnel
|
|
|
|
# __ __ _____ _ _ _____ _ _ _____ _
|
|
# \ \ / /\ | __ \| \ | |_ _| \ | |/ ____| |
|
|
# \ \ /\ / / \ | |__) | \| | | | | \| | | __| |
|
|
# \ \/ \/ / /\ \ | _ /| . ` | | | | . ` | | |_ | |
|
|
# \ /\ / ____ \| | \ \| |\ |_| |_| |\ | |__| |_|
|
|
# \/ \/_/ \_\_| \_\_| \_|_____|_| \_|\_____(_)
|
|
# 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")
|
|
|
|
|
|
def test_validation_on_mysql_succeeds():
|
|
|
|
test_query = """
|
|
SELECT 1 AS a_one,
|
|
"lol" AS a_string,
|
|
NULL AS a_null
|
|
"""
|
|
test_expectations = [
|
|
ExpectationConfiguration(
|
|
expectation_type="expect_column_values_to_be_between",
|
|
kwargs={"column": "a_one", "min_value": 1, "max_value": 1},
|
|
),
|
|
ExpectationConfiguration(
|
|
expectation_type="expect_column_values_to_match_like_pattern",
|
|
kwargs={"column": "a_string", "like_pattern": "%lol%"},
|
|
),
|
|
ExpectationConfiguration(
|
|
expectation_type="expect_column_values_to_be_null",
|
|
kwargs={"column": "a_null"},
|
|
),
|
|
]
|
|
|
|
ssh_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"],
|
|
)
|
|
|
|
validation_result = run_data_test_on_mysql.run(
|
|
name="lolafect-testing-test_validation_on_mysql_succeeds",
|
|
mysql_credentials={
|
|
"host": ssh_tunnel.local_bind_address[0],
|
|
"port": ssh_tunnel.local_bind_address[1],
|
|
"user": TEST_LOLACONFIG.DW_CREDENTIALS["user"],
|
|
"password": TEST_LOLACONFIG.DW_CREDENTIALS["password"],
|
|
"db": TEST_LOLACONFIG.DW_CREDENTIALS["default_db"]
|
|
},
|
|
query=test_query,
|
|
expectations=test_expectations,
|
|
)
|
|
|
|
closed_tunnel = close_ssh_tunnel.run(ssh_tunnel)
|
|
|
|
data_test_passed = validation_result["success"] == True
|
|
|
|
assert data_test_passed
|
|
|
|
|
|
def test_validation_on_mysql_fails():
|
|
test_query = """
|
|
SELECT 1 AS a_one,
|
|
"lol" AS a_string,
|
|
NULL AS a_null
|
|
"""
|
|
test_expectations = [
|
|
ExpectationConfiguration(
|
|
expectation_type="expect_column_values_to_be_between",
|
|
kwargs={"column": "a_one", "min_value": 2, "max_value": 2},
|
|
),
|
|
ExpectationConfiguration(
|
|
expectation_type="expect_column_values_to_match_like_pattern",
|
|
kwargs={"column": "a_string", "like_pattern": "%xD%"},
|
|
),
|
|
ExpectationConfiguration(
|
|
expectation_type="expect_column_values_to_not_be_null",
|
|
kwargs={"column": "a_null"},
|
|
),
|
|
]
|
|
|
|
ssh_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"],
|
|
)
|
|
|
|
validation_result = run_data_test_on_mysql.run(
|
|
name="lolafect-testing-test_validation_on_mysql_fails",
|
|
mysql_credentials={
|
|
"host": ssh_tunnel.local_bind_address[0],
|
|
"port": ssh_tunnel.local_bind_address[1],
|
|
"user": TEST_LOLACONFIG.DW_CREDENTIALS["user"],
|
|
"password": TEST_LOLACONFIG.DW_CREDENTIALS["password"],
|
|
"db": TEST_LOLACONFIG.DW_CREDENTIALS["default_db"]
|
|
},
|
|
query=test_query,
|
|
expectations=test_expectations,
|
|
)
|
|
|
|
closed_tunnel = close_ssh_tunnel.run(ssh_tunnel)
|
|
|
|
data_test_failed = validation_result["success"] == False
|
|
|
|
assert data_test_failed
|