QT/superset2
1
0
Fork 0
mirror of https://github.com/apache/superset.git synced 2026-04-18 23:55:00 +00:00
Code Issues Packages Projects Releases Wiki Activity

chore: remove deprecated apis stop_query, queries, search_queries (#24360)

Browse Source
This commit is contained in:
Daniel Vaz Gaspar
2023-06-13 15:56:06 +01:00
committed by GitHub
parent ca478bd5c4
commit c8e958fb7e
6 changed files with 2 additions and 296 deletions
Download Patch File Download Diff File Expand all files Collapse all files

23
tests/integration_tests/core_tests.py
View File

@@ -1243,29 +1243,6 @@ class TestCore(SupersetTestCase):
data = self.get_resp(url)
self.assertIn("Error message", data)
@mock.patch("superset.sql_lab.cancel_query")
@mock.patch("superset.views.core.db.session")
def test_stop_query_not_implemented(
self, mock_superset_db_session, mock_sql_lab_cancel_query
):
"""
Handles stop query when the DB engine spec does not
have a cancel query method.
"""
form_data = {"client_id": "foo"}
query_mock = mock.Mock()
query_mock.client_id = "foo"
query_mock.status = QueryStatus.RUNNING
self.login(username="admin")
mock_superset_db_session.query().filter_by().one().return_value = query_mock
mock_sql_lab_cancel_query.return_value = False
rv = self.client.post(
"/superset/stop_query/",
data={"form_data": json.dumps(form_data)},
)
assert rv.status_code == 422
@pytest.mark.usefixtures("load_energy_table_with_slice")
@mock.patch("superset.explore.form_data.commands.create.CreateFormDataCommand.run")
def test_explore_redirect(self, mock_command: mock.Mock):

129
tests/integration_tests/sqllab_tests.py
View File

@@ -17,8 +17,7 @@
# isort:skip_file
"""Unit tests for Sql Lab"""
import json
from datetime import datetime, timedelta
from math import ceil, floor
from datetime import datetime
import pytest
from celery.exceptions import SoftTimeLimitExceeded
@@ -327,132 +326,6 @@ class TestSqlLab(SupersetTestCase):
engine.execute(f"DROP TABLE IF EXISTS {CTAS_SCHEMA_NAME}.test_table")
db.session.commit()
def test_queries_endpoint(self):
self.run_some_queries()
# Not logged in, should error out
resp = self.client.get("/superset/queries/0")
# Redirects to the login page
self.assertEqual(401, resp.status_code)
# Admin sees queries
self.login("admin")
data = self.get_json_resp("/superset/queries/0")
self.assertEqual(2, len(data))
data = self.get_json_resp("/superset/queries/0.0")
self.assertEqual(2, len(data))
# Run 2 more queries
self.run_sql("SELECT * FROM birth_names LIMIT 1", client_id="client_id_4")
self.run_sql("SELECT * FROM birth_names LIMIT 2", client_id="client_id_5")
self.login("admin")
data = self.get_json_resp("/superset/queries/0")
self.assertEqual(4, len(data))
now = datetime.now() + timedelta(days=1)
query = (
db.session.query(Query)
.filter_by(sql="SELECT * FROM birth_names LIMIT 1")
.first()
)
query.changed_on = now
db.session.commit()
data = self.get_json_resp(
f"/superset/queries/{float(datetime_to_epoch(now)) - 1000}"
)
self.assertEqual(1, len(data))
self.logout()
resp = self.client.get("/superset/queries/0")
# Redirects to the login page
self.assertEqual(401, resp.status_code)
def test_search_query_on_db_id(self):
self.run_some_queries()
self.login("admin")
examples_dbid = get_example_database().id
# Test search queries on database Id
data = self.get_json_resp(
f"/superset/search_queries?database_id={examples_dbid}"
)
self.assertEqual(3, len(data))
db_ids = [k["dbId"] for k in data]
self.assertEqual([examples_dbid for i in range(3)], db_ids)
resp = self.get_resp("/superset/search_queries?database_id=-1")
data = json.loads(resp)
self.assertEqual(0, len(data))
def test_search_query_on_user(self):
self.run_some_queries()
self.login("admin")
# Test search queries on user Id
user_id = security_manager.find_user("admin").id
data = self.get_json_resp(f"/superset/search_queries?user_id={user_id}")
self.assertEqual(2, len(data))
user_ids = {k["userId"] for k in data}
self.assertEqual({user_id}, user_ids)
user_id = security_manager.find_user("gamma_sqllab").id
resp = self.get_resp(f"/superset/search_queries?user_id={user_id}")
data = json.loads(resp)
self.assertEqual(1, len(data))
self.assertEqual(data[0]["userId"], user_id)
@pytest.mark.usefixtures("load_birth_names_dashboard_with_slices")
def test_search_query_on_status(self):
self.run_some_queries()
self.login("admin")
# Test search queries on status
resp = self.get_resp("/superset/search_queries?status=success")
data = json.loads(resp)
self.assertEqual(2, len(data))
states = [k["state"] for k in data]
self.assertEqual(["success", "success"], states)
resp = self.get_resp("/superset/search_queries?status=failed")
data = json.loads(resp)
self.assertEqual(1, len(data))
self.assertEqual(data[0]["state"], "failed")
def test_search_query_on_text(self):
self.run_some_queries()
self.login("admin")
url = "/superset/search_queries?search_text=birth"
data = self.get_json_resp(url)
self.assertEqual(2, len(data))
self.assertIn("birth", data[0]["sql"])
def test_search_query_filter_by_time(self):
self.run_some_queries()
self.login("admin")
from_time = floor(
(db.session.query(Query).filter_by(sql=QUERY_1).one()).start_time
)
to_time = ceil(
(db.session.query(Query).filter_by(sql=QUERY_2).one()).start_time
)
url = f"/superset/search_queries?from={from_time}&to={to_time}"
assert len(self.client.get(url).json) == 2
def test_search_query_only_owned(self) -> None:
"""
Test a search query with a user that does not have can_access_all_queries.
"""
# Test search_queries for Alpha user
self.run_some_queries()
self.login("gamma_sqllab")
user_id = security_manager.find_user("gamma_sqllab").id
data = self.get_json_resp("/superset/search_queries")
self.assertEqual(1, len(data))
user_ids = {k["userId"] for k in data}
self.assertEqual({user_id}, user_ids)
def test_alias_duplicate(self):
self.run_sql(
"SELECT name as col, gender as col FROM birth_names LIMIT 10",