74 lines
2.6 KiB
Python
74 lines
2.6 KiB
Python
import pathlib
|
|
|
|
from lolafect.lolaconfig import build_lolaconfig
|
|
from lolafect.connections import connect_to_trino, close_trino_connection, _temp_secret_file_from_s3
|
|
|
|
# __ __ _____ _ _ _____ _ _ _____ _
|
|
# \ \ / /\ | __ \| \ | |_ _| \ | |/ ____| |
|
|
# \ \ /\ / / \ | |__) | \| | | | | \| | | __| |
|
|
# \ \/ \/ / /\ \ | _ /| . ` | | | | . ` | | |_ | |
|
|
# \ /\ / ____ \| | \ \| |\ |_| |_| |\ | |__| |_|
|
|
# \/ \/_/ \_\_| \_\_| \_|_____|_| \_|\_____(_)
|
|
# 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_that_trino_connect_and_disconnect_works_properly():
|
|
|
|
connection = connect_to_trino.run(
|
|
trino_credentials=TEST_LOLACONFIG.TRINO_CREDENTIALS
|
|
)
|
|
|
|
connection.cursor().execute("SELECT 1")
|
|
|
|
close_trino_connection.run(trino_connection=connection)
|
|
|
|
|
|
def test_temporal_download_of_secret_file_works_properly_in_happy_path():
|
|
|
|
temp_file_name = "test_temp_file"
|
|
|
|
with _temp_secret_file_from_s3(
|
|
TEST_LOLACONFIG.S3_BUCKET_NAME,
|
|
s3_file_key="env/env_prd.json", # Not a secret file, but then again, this is a test,
|
|
local_temp_file_path=temp_file_name,
|
|
) as temp:
|
|
temp_file_found_when_in_context_manager = pathlib.Path(temp).exists()
|
|
|
|
temp_file_missing_when_outside_context_manager = not pathlib.Path(
|
|
temp_file_name
|
|
).exists()
|
|
|
|
assert (
|
|
temp_file_found_when_in_context_manager
|
|
and temp_file_missing_when_outside_context_manager
|
|
)
|
|
|
|
|
|
def test_temporal_download_of_secret_file_works_properly_even_with_exception():
|
|
temp_file_name = "test_temp_file"
|
|
|
|
try:
|
|
with _temp_secret_file_from_s3(
|
|
TEST_LOLACONFIG.S3_BUCKET_NAME,
|
|
s3_file_key="env/env_prd.json", # Not a secret file, but then again, this is a test,
|
|
local_temp_file_path=temp_file_name,
|
|
) as temp:
|
|
temp_file_found_when_in_context_manager = pathlib.Path(temp).exists()
|
|
raise Exception # Something nasty happens within the context manager
|
|
except:
|
|
pass # We go with the test, ignoring the forced exception
|
|
|
|
temp_file_missing_when_outside_context_manager = not pathlib.Path(
|
|
temp_file_name
|
|
).exists()
|
|
|
|
assert (
|
|
temp_file_found_when_in_context_manager
|
|
and temp_file_missing_when_outside_context_manager
|
|
)
|