This is an automated email from the ASF dual-hosted git repository. hugh pushed a commit to branch test-ssh-tunnel-1 in repository https://gitbox.apache.org/repos/asf/superset.git
commit 6a210562d49033be95e0951613d8e5aa6c96d084 Author: Antonio Rivero <[email protected]> AuthorDate: Sun Nov 27 12:12:22 2022 -0300 SSH Tunnel: - Add not found test when deleting the tunnel --- tests/unit_tests/databases/api_test.py | 72 ++++++++++++++++++++++++++++++++++ 1 file changed, 72 insertions(+) diff --git a/tests/unit_tests/databases/api_test.py b/tests/unit_tests/databases/api_test.py index 73e79f3a82..fe4211289c 100644 --- a/tests/unit_tests/databases/api_test.py +++ b/tests/unit_tests/databases/api_test.py @@ -263,3 +263,75 @@ def test_delete_ssh_tunnel( response_tunnel = DatabaseDAO.get_ssh_tunnel(1) assert response_tunnel is None + + +def test_delete_ssh_tunnel_not_found( + mocker: MockFixture, + app: Any, + session: Session, + client: Any, + full_api_access: None, +) -> None: + """ + Test that we cannot delete a tunnel that does not exist + """ + with app.app_context(): + from superset.databases.api import DatabaseRestApi + from superset.databases.dao import DatabaseDAO + from superset.databases.ssh_tunnel.models import SSHTunnel + from superset.models.core import Database + + DatabaseRestApi.datamodel.session = session + + # create table for databases + Database.metadata.create_all(session.get_bind()) # pylint: disable=no-member + + # Create our Database + database = Database( + database_name="my_database", + sqlalchemy_uri="gsheets://", + encrypted_extra=json.dumps( + { + "service_account_info": { + "type": "service_account", + "project_id": "black-sanctum-314419", + "private_key_id": "259b0d419a8f840056158763ff54d8b08f7b8173", + "private_key": "SECRET", + "client_email": "google-spreadsheets-demo-se...@black-sanctum-314419.iam.gserviceaccount.com", + "client_id": "SSH_TUNNEL_CREDENTIALS_CLIENT", + "auth_uri": "https://accounts.google.com/o/oauth2/auth";, + "token_uri": "https://oauth2.googleapis.com/token";, + "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs";, + "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/google-spreadsheets-demo-servi%40black-sanctum-314419.iam.gserviceaccount.com";, + }, + } + ), + ) + session.add(database) + session.commit() + + # mock the lookup so that we don't need to include the driver + mocker.patch("sqlalchemy.engine.URL.get_driver_name", return_value="gsheets") + mocker.patch("superset.utils.log.DBEventLogger.log") + + # Create our SSHTunnel + tunnel = SSHTunnel( + database_id=1, + database=database, + ) + + session.add(tunnel) + session.commit() + + # Delete the recently created SSHTunnel + response_delete_tunnel = client.delete("/api/v1/database/2/ssh_tunnel/") + assert response_delete_tunnel.json["message"] == "Not found" + + # Get our recently created SSHTunnel + response_tunnel = DatabaseDAO.get_ssh_tunnel(1) + assert response_tunnel + assert isinstance(response_tunnel, SSHTunnel) + assert 1 == response_tunnel.database_id + + response_tunnel = DatabaseDAO.get_ssh_tunnel(2) + assert response_tunnel is None
