test_connection.py

""" Given a connection name, obtain all supported tests, and run these test

 $ python test_connection.py <connection_name> 

Example:
 $ python test_connection.py thelook 

Notes: Connections to Looker's internal database cannot be tested. 

Last modified: Feb 27 2024
"""

fromfunctoolsimportreduce
importsys
fromtypingimportcast, MutableSequence, Sequence

importlooker_sdk
fromlooker_sdkimportmodels40asmodels

sdk=looker_sdk.init40("../../looker.ini")

defmain():
connection_name=sys.argv[1] iflen(sys.argv) >1else""

ifnotconnection_name:
raiseException("Please provide a connection name")
elifconnection_namein ["looker", "looker__internal__analytics"]:
raiseException(
f"Connection '{connection_name}' is internal and cannot be tested."
 )

connection=get_connections(connection_name)

results=test_connection(connection)

output_results(cast(str, connection.name), results)


defget_connections(name: str) ->models.DBConnection:
connection=sdk.connection(name, fields="name, dialect")
returnconnection


deftest_connection(
connection: models.DBConnection,
) ->Sequence[models.DBConnectionTestResult]:
"""Run supported tests against a given connection."""
assertconnection.name
assertconnection.dialectandconnection.dialect.connection_tests
supported_tests: MutableSequence[str] =list(connection.dialect.connection_tests)
test_results=sdk.test_connection(
connection.name, models.DelimSequence(supported_tests)
 )
returntest_results


defoutput_results(
connection_name: str, test_results: Sequence[models.DBConnectionTestResult]
):
"""Prints connection test results."""
errors=list(filter(lambdatest: cast(str, test.status) =="error", test_results))
iferrors:
report=reduce(
lambdafailures, error: failures+f"\n - {error.message}",
errors,
f"{connection_name}:",
 )
else:
report=f"All tests for connection '{connection_name}' were successful."
print(report)


main()