110 lines
3.1 KiB
Python
110 lines
3.1 KiB
Python
from types import SimpleNamespace
|
|
|
|
from lolafect.lolaconfig import LolaConfig
|
|
|
|
|
|
def test_lolaconfig_instantiates_with_defaults_properly():
|
|
lolaconfig = LolaConfig(flow_name="some-flow")
|
|
|
|
|
|
def test_lolaconfig_instantiates_without_defaults_proplery():
|
|
lolaconfig = LolaConfig(
|
|
flow_name="some-flow",
|
|
env_s3_bucket="bucket",
|
|
kubernetes_labels=["some_label"],
|
|
kubernetes_image="loliloli:latest",
|
|
slack_webhooks_file="the_file/is/here.json",
|
|
)
|
|
|
|
|
|
def test_lolaconfig_fetches_webhooks_properly():
|
|
lolaconfig = LolaConfig(flow_name="some-flow")
|
|
|
|
fake_s3_reader = SimpleNamespace()
|
|
|
|
def mock_read_json_from_s3_file(bucket, key):
|
|
return {"a-channel-name": "a-channel-url.com"}
|
|
|
|
fake_s3_reader.read_json_from_s3_file = mock_read_json_from_s3_file
|
|
|
|
lolaconfig.fetch_slack_webhooks(s3_reader=fake_s3_reader)
|
|
|
|
assert type(lolaconfig.SLACK_WEBHOOKS) is dict
|
|
|
|
|
|
def test_lolaconfig_fetches_trino_creds_properly():
|
|
lolaconfig = LolaConfig(flow_name="some-flow")
|
|
|
|
fake_s3_reader = SimpleNamespace()
|
|
|
|
def mock_read_json_from_s3_file(bucket, key):
|
|
return {
|
|
"trino_host": "some_host",
|
|
"trino_user": "some_user",
|
|
"trino_pass": "some_password",
|
|
"trino_port": "some_port",
|
|
}
|
|
|
|
fake_s3_reader.read_json_from_s3_file = mock_read_json_from_s3_file
|
|
|
|
lolaconfig.fetch_trino_credentials(s3_reader=fake_s3_reader)
|
|
|
|
assert type(lolaconfig.TRINO_CREDENTIALS) is dict
|
|
|
|
|
|
def test_lolaconfig_fetches_ssh_tunnel_creds_properly():
|
|
lolaconfig = LolaConfig(flow_name="some-flow")
|
|
|
|
fake_s3_reader = SimpleNamespace()
|
|
|
|
def mock_read_json_from_s3_file(bucket, key):
|
|
return {
|
|
"pt_ssh_pkey_path": "some-path",
|
|
"pt_ssh_pkey_passphrase": "some-password",
|
|
"pt_ssh_username": "some-username",
|
|
"pt_ssh_jumphost_port": "some-port",
|
|
"pt_ssh_jumphost": "some-jumphost",
|
|
}
|
|
|
|
fake_s3_reader.read_json_from_s3_file = mock_read_json_from_s3_file
|
|
|
|
lolaconfig.fetch_ssh_tunnel_credentials(s3_reader=fake_s3_reader)
|
|
|
|
assert type(lolaconfig.SSH_TUNNEL_CREDENTIALS) is dict
|
|
|
|
|
|
def test_lolaconfig_fetches_dw_creds_properly():
|
|
lolaconfig = LolaConfig(flow_name="some-flow")
|
|
|
|
fake_s3_reader = SimpleNamespace()
|
|
|
|
def mock_read_json_from_s3_file(bucket, key):
|
|
return {
|
|
"datadw_host": "some_host",
|
|
"datadw_user": "some_user",
|
|
"datadw_pass": "some_password",
|
|
"datadw_port": "some_port",
|
|
}
|
|
|
|
fake_s3_reader.read_json_from_s3_file = mock_read_json_from_s3_file
|
|
|
|
lolaconfig.fetch_dw_credentials(s3_reader=fake_s3_reader)
|
|
|
|
assert type(lolaconfig.DW_CREDENTIALS) is dict
|
|
|
|
|
|
def test_lolaconfig_fetches_prefect_host_properly():
|
|
lolaconfig = LolaConfig(flow_name="some-flow")
|
|
|
|
fake_s3_reader = SimpleNamespace()
|
|
|
|
def mock_read_json_from_s3_file(bucket, key):
|
|
return {
|
|
"prefect_host": "some_host",
|
|
}
|
|
|
|
fake_s3_reader.read_json_from_s3_file = mock_read_json_from_s3_file
|
|
|
|
lolaconfig.fetch_prefect_host(s3_reader=fake_s3_reader)
|
|
|
|
assert type(lolaconfig.PREFECT_HOST) is str
|