from redash import models from redash.models import db from redash.permissions import ACCESS_TYPE_MODIFY from redash.serializers import serialize_query from tests import BaseTestCase class TestQueryResourceGet(BaseTestCase): def test_get_query(self): query = self.factory.create_query() rv = self.make_request("get", "/api/queries/{0}".format(query.id)) self.assertEqual(rv.status_code, 200) expected = serialize_query(query, with_visualizations=True) expected["can_edit"] = True expected["is_favorite"] = False self.assertResponseEqual(expected, rv.json) def test_get_all_queries(self): [self.factory.create_query() for _ in range(10)] rv = self.make_request("get", "/api/queries") self.assertEqual(rv.status_code, 200) self.assertEqual(len(rv.json["results"]), 10) def test_query_without_data_source_should_be_available_only_by_admin(self): query = self.factory.create_query() query.data_source = None db.session.add(query) rv = self.make_request("get", "/api/queries/{}".format(query.id)) self.assertEqual(rv.status_code, 403) rv = self.make_request("get", "/api/queries/{}".format(query.id), user=self.factory.create_admin()) self.assertEqual(rv.status_code, 200) def test_query_only_accessible_to_users_from_its_organization(self): second_org = self.factory.create_org() second_org_admin = self.factory.create_admin(org=second_org) query = self.factory.create_query() query.data_source = None db.session.add(query) rv = self.make_request("get", "/api/queries/{}".format(query.id), user=second_org_admin) self.assertEqual(rv.status_code, 404) rv = self.make_request("get", "/api/queries/{}".format(query.id), user=self.factory.create_admin()) self.assertEqual(rv.status_code, 200) def test_query_search(self): names = ["Harder", "Better", "Faster", "Stronger"] for name in names: self.factory.create_query(name=name) rv = self.make_request("get", "/api/queries?q=better") self.assertEqual(rv.status_code, 200) self.assertEqual(len(rv.json["results"]), 1) rv = self.make_request("get", "/api/queries?q=better or faster") self.assertEqual(rv.status_code, 200) self.assertEqual(len(rv.json["results"]), 2) # test the old search API and that it redirects to the new one rv = self.make_request("get", "/api/queries/search?q=stronger") self.assertEqual(rv.status_code, 301) self.assertIn("/api/queries?q=stronger", rv.headers["Location"]) rv = self.make_request("get", "/api/queries/search?q=stronger", follow_redirects=True) self.assertEqual(rv.status_code, 200) self.assertEqual(len(rv.json["results"]), 1) class TestQueryResourcePost(BaseTestCase): def test_update_query(self): admin = self.factory.create_admin() query = self.factory.create_query() new_ds = self.factory.create_data_source() new_qr = self.factory.create_query_result() data = { "name": "Testing", "query": "select 2", "latest_query_data_id": new_qr.id, "data_source_id": new_ds.id, } rv = self.make_request("post", "/api/queries/{0}".format(query.id), data=data, user=admin) self.assertEqual(rv.status_code, 200) self.assertEqual(rv.json["name"], data["name"]) self.assertEqual(rv.json["last_modified_by"]["id"], admin.id) self.assertEqual(rv.json["query"], data["query"]) self.assertEqual(rv.json["data_source_id"], data["data_source_id"]) self.assertEqual(rv.json["latest_query_data_id"], data["latest_query_data_id"]) def test_raises_error_in_case_of_conflict(self): q = self.factory.create_query() q.name = "Another Name" db.session.add(q) rv = self.make_request( "post", "/api/queries/{0}".format(q.id), data={"name": "Testing", "version": q.version - 1}, user=self.factory.user, ) self.assertEqual(rv.status_code, 409) def test_prevents_association_with_view_only_data_sources(self): view_only_data_source = self.factory.create_data_source(view_only=True) my_data_source = self.factory.create_data_source() my_query = self.factory.create_query(data_source=my_data_source) db.session.add(my_query) rv = self.make_request( "post", "/api/queries/{0}".format(my_query.id), data={"data_source_id": view_only_data_source.id}, user=self.factory.user, ) self.assertEqual(rv.status_code, 403) def test_allows_association_with_authorized_dropdown_queries(self): data_source = self.factory.create_data_source(group=self.factory.default_group) other_query = self.factory.create_query(data_source=data_source) db.session.add(other_query) my_query = self.factory.create_query(data_source=data_source) db.session.add(my_query) options = { "parameters": [ {"name": "foo", "type": "query", "queryId": other_query.id}, {"name": "bar", "type": "query", "queryId": other_query.id}, ] } rv = self.make_request( "post", "/api/queries/{0}".format(my_query.id), data={"options": options}, user=self.factory.user, ) self.assertEqual(rv.status_code, 200) def test_prevents_association_with_unauthorized_dropdown_queries(self): other_data_source = self.factory.create_data_source(group=self.factory.create_group()) other_query = self.factory.create_query(data_source=other_data_source) db.session.add(other_query) my_data_source = self.factory.create_data_source(group=self.factory.create_group()) my_query = self.factory.create_query(data_source=my_data_source) db.session.add(my_query) options = {"parameters": [{"type": "query", "queryId": other_query.id}]} rv = self.make_request( "post", "/api/queries/{0}".format(my_query.id), data={"options": options}, user=self.factory.user, ) self.assertEqual(rv.status_code, 403) def test_prevents_association_with_non_existing_dropdown_queries(self): my_data_source = self.factory.create_data_source(group=self.factory.create_group()) my_query = self.factory.create_query(data_source=my_data_source) db.session.add(my_query) options = {"parameters": [{"type": "query", "queryId": 100000}]} rv = self.make_request( "post", "/api/queries/{0}".format(my_query.id), data={"options": options}, user=self.factory.user, ) self.assertEqual(rv.status_code, 400) def test_overrides_existing_if_no_version_specified(self): q = self.factory.create_query() q.name = "Another Name" db.session.add(q) rv = self.make_request( "post", "/api/queries/{0}".format(q.id), data={"name": "Testing"}, user=self.factory.user, ) self.assertEqual(rv.status_code, 200) def test_works_for_non_owner_with_permission(self): query = self.factory.create_query() user = self.factory.create_user() rv = self.make_request( "post", "/api/queries/{0}".format(query.id), data={"name": "Testing"}, user=user, ) self.assertEqual(rv.status_code, 403) models.AccessPermission.grant(obj=query, access_type=ACCESS_TYPE_MODIFY, grantee=user, grantor=query.user) rv = self.make_request( "post", "/api/queries/{0}".format(query.id), data={"name": "Testing"}, user=user, ) self.assertEqual(rv.status_code, 200) self.assertEqual(rv.json["name"], "Testing") self.assertEqual(rv.json["last_modified_by"]["id"], user.id) class TestQueryListResourceGet(BaseTestCase): def test_returns_queries(self): q1 = self.factory.create_query() q2 = self.factory.create_query() q3 = self.factory.create_query() rv = self.make_request("get", "/api/queries") assert len(rv.json["results"]) == 3 assert set([result["id"] for result in rv.json["results"]]) == {q1.id, q2.id, q3.id} def test_filters_with_tags(self): q1 = self.factory.create_query(tags=["test"]) self.factory.create_query() self.factory.create_query() rv = self.make_request("get", "/api/queries?tags=test") assert len(rv.json["results"]) == 1 assert set([result["id"] for result in rv.json["results"]]) == {q1.id} def test_search_term(self): q1 = self.factory.create_query(name="Sales") q2 = self.factory.create_query(name="Q1 sales") self.factory.create_query(name="Ops") rv = self.make_request("get", "/api/queries?q=sales") assert len(rv.json["results"]) == 2 assert set([result["id"] for result in rv.json["results"]]) == {q1.id, q2.id} class TestQueryListResourcePost(BaseTestCase): def test_create_query(self): query_data = { "name": "Testing", "query": "SELECT 1", "schedule": {"interval": "3600"}, "data_source_id": self.factory.data_source.id, } rv = self.make_request("post", "/api/queries", data=query_data) self.assertEqual(rv.status_code, 200) self.assertLessEqual(query_data.items(), rv.json.items()) self.assertEqual(rv.json["user"]["id"], self.factory.user.id) self.assertIsNotNone(rv.json["api_key"]) self.assertIsNotNone(rv.json["query_hash"]) query = models.Query.query.get(rv.json["id"]) self.assertEqual(len(list(query.visualizations)), 1) self.assertTrue(query.is_draft) def test_allows_association_with_authorized_dropdown_queries(self): data_source = self.factory.create_data_source(group=self.factory.default_group) other_query = self.factory.create_query(data_source=data_source) db.session.add(other_query) query_data = { "name": "Testing", "query": "SELECT 1", "schedule": {"interval": "3600"}, "data_source_id": self.factory.data_source.id, "options": { "parameters": [ {"name": "foo", "type": "query", "queryId": other_query.id}, {"name": "bar", "type": "query", "queryId": other_query.id}, ] }, } rv = self.make_request("post", "/api/queries", data=query_data) self.assertEqual(rv.status_code, 200) def test_prevents_association_with_unauthorized_dropdown_queries(self): other_data_source = self.factory.create_data_source(group=self.factory.create_group()) other_query = self.factory.create_query(data_source=other_data_source) db.session.add(other_query) my_data_source = self.factory.create_data_source(group=self.factory.create_group()) query_data = { "name": "Testing", "query": "SELECT 1", "schedule": {"interval": "3600"}, "data_source_id": my_data_source.id, "options": {"parameters": [{"type": "query", "queryId": other_query.id}]}, } rv = self.make_request("post", "/api/queries", data=query_data) self.assertEqual(rv.status_code, 403) def test_prevents_association_with_non_existing_dropdown_queries(self): query_data = { "name": "Testing", "query": "SELECT 1", "schedule": {"interval": "3600"}, "data_source_id": self.factory.data_source.id, "options": {"parameters": [{"type": "query", "queryId": 100000}]}, } rv = self.make_request("post", "/api/queries", data=query_data) self.assertEqual(rv.status_code, 400) class TestQueryArchiveResourceGet(BaseTestCase): def test_returns_queries(self): q1 = self.factory.create_query(is_archived=True) q2 = self.factory.create_query(is_archived=True) self.factory.create_query() rv = self.make_request("get", "/api/queries/archive") assert len(rv.json["results"]) == 2 assert set([result["id"] for result in rv.json["results"]]) == {q1.id, q2.id} def test_search_term(self): q1 = self.factory.create_query(name="Sales", is_archived=True) q2 = self.factory.create_query(name="Q1 sales", is_archived=True) self.factory.create_query(name="Q2 sales") rv = self.make_request("get", "/api/queries/archive?q=sales") assert len(rv.json["results"]) == 2 assert set([result["id"] for result in rv.json["results"]]) == {q1.id, q2.id} class QueryRefreshTest(BaseTestCase): def setUp(self): super(QueryRefreshTest, self).setUp() self.query = self.factory.create_query() self.path = "/api/queries/{}/refresh".format(self.query.id) def test_refresh_regular_query(self): response = self.make_request("post", self.path) self.assertEqual(200, response.status_code) def test_refresh_of_query_with_parameters(self): self.query.query_text = "SELECT {{param}}" db.session.add(self.query) response = self.make_request("post", "{}?p_param=1".format(self.path)) self.assertEqual(200, response.status_code) def test_refresh_of_query_with_parameters_without_parameters(self): self.query.query_text = "SELECT {{param}}" db.session.add(self.query) response = self.make_request("post", "{}".format(self.path)) self.assertEqual(400, response.status_code) def test_refresh_query_you_dont_have_access_to(self): group = self.factory.create_group() db.session.add(group) db.session.commit() user = self.factory.create_user(group_ids=[group.id]) response = self.make_request("post", self.path, user=user) self.assertEqual(403, response.status_code) def test_refresh_forbiden_with_query_api_key(self): response = self.make_request("post", "{}?api_key={}".format(self.path, self.query.api_key), user=False) self.assertEqual(403, response.status_code) response = self.make_request( "post", "{}?api_key={}".format(self.path, self.factory.user.api_key), user=False, ) self.assertEqual(200, response.status_code) class TestQueryRegenerateApiKey(BaseTestCase): def test_non_admin_cannot_regenerate_api_key_of_other_user(self): query_creator = self.factory.create_user() query = self.factory.create_query(user=query_creator) other_user = self.factory.create_user() orig_api_key = query.api_key rv = self.make_request( "post", "/api/queries/{}/regenerate_api_key".format(query.id), user=other_user, ) self.assertEqual(rv.status_code, 403) reloaded_query = models.Query.query.get(query.id) self.assertEqual(orig_api_key, reloaded_query.api_key) def test_admin_can_regenerate_api_key_of_other_user(self): query_creator = self.factory.create_user() query = self.factory.create_query(user=query_creator) admin_user = self.factory.create_admin() orig_api_key = query.api_key rv = self.make_request( "post", "/api/queries/{}/regenerate_api_key".format(query.id), user=admin_user, ) self.assertEqual(rv.status_code, 200) reloaded_query = models.Query.query.get(query.id) self.assertNotEqual(orig_api_key, reloaded_query.api_key) def test_admin_can_regenerate_api_key_of_myself(self): query_creator = self.factory.create_user() admin_user = self.factory.create_admin() query = self.factory.create_query(user=query_creator) orig_api_key = query.api_key rv = self.make_request( "post", "/api/queries/{}/regenerate_api_key".format(query.id), user=admin_user, ) self.assertEqual(rv.status_code, 200) updated_query = models.Query.query.get(query.id) self.assertNotEqual(orig_api_key, updated_query.api_key) def test_user_can_regenerate_api_key_of_myself(self): user = self.factory.create_user() query = self.factory.create_query(user=user) orig_api_key = query.api_key rv = self.make_request("post", "/api/queries/{}/regenerate_api_key".format(query.id), user=user) self.assertEqual(rv.status_code, 200) updated_query = models.Query.query.get(query.id) self.assertNotEqual(orig_api_key, updated_query.api_key) class TestQueryForkResourcePost(BaseTestCase): def test_forks_a_query(self): ds = self.factory.create_data_source(group=self.factory.org.default_group, view_only=False) query = self.factory.create_query(data_source=ds) rv = self.make_request("post", "/api/queries/{}/fork".format(query.id)) self.assertEqual(rv.status_code, 200) def test_must_have_full_access_to_data_source(self): ds = self.factory.create_data_source(group=self.factory.org.default_group, view_only=True) query = self.factory.create_query(data_source=ds) rv = self.make_request("post", "/api/queries/{}/fork".format(query.id)) self.assertEqual(rv.status_code, 403) class TestFormatSQLQueryAPI(BaseTestCase): def test_format_sql_query(self): admin = self.factory.create_admin() query = "select a,b,c FROM foobar Where x=1 and y=2;" expected = """SELECT a, b, c FROM foobar WHERE x=1 AND y=2;""" rv = self.make_request("post", "/api/queries/format", user=admin, data={"query": query}) self.assertEqual(rv.json["query"], expected)