Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Fix: allow users with view only acces to use the queries in Query Results #4112

Merged
merged 4 commits into from
Sep 1, 2019
+131 −40
Merged

Fix: allow users with view only acces to use the queries in Query Results #4112

Changes from 1 commit
Commits
Show all changes
4 commits
Select commit Hold shift + click to select a range
bcb8281
Fix: allow users with view only acces to access the queries
arikfr Sep 1, 2019
99c5d44
Add tests
arikfr Sep 1, 2019
32ba5d7
Update error message
arikfr Sep 1, 2019
bb4f768
Update error message. Take 2
arikfr Sep 1, 2019
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
125 changes: 105 additions & 20 deletions tests/query_runner/test_query_results.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,9 @@

import pytest

from redash.query_runner.query_results import (CreateTableError, PermissionError, _load_query, create_table, extract_cached_query_ids, extract_query_ids, fix_column_name)
from redash.query_runner.query_results import (
CreateTableError, PermissionError, _load_query, create_table,
extract_cached_query_ids, extract_query_ids, fix_column_name)
from tests import BaseTestCase


Expand All @@ -28,49 +30,102 @@ def test_finds_queries_with_whitespace_characters(self):
class TestCreateTable(TestCase):
def test_creates_table_with_colons_in_column_name(self):
connection = sqlite3.connect(':memory:')
results = {'columns': [{'name': 'ga:newUsers'}, {
'name': 'test2'}], 'rows': [{'ga:newUsers': 123, 'test2': 2}]}
results = {
'columns': [{
'name': 'ga:newUsers'
}, {
'name': 'test2'
}],
'rows': [{
'ga:newUsers': 123,
'test2': 2
}]
}
table_name = 'query_123'
create_table(connection, table_name, results)
connection.execute('SELECT 1 FROM query_123')

def test_creates_table_with_double_quotes_in_column_name(self):
connection = sqlite3.connect(':memory:')
results = {'columns': [{'name': 'ga:newUsers'}, {
'name': '"test2"'}], 'rows': [{'ga:newUsers': 123, '"test2"': 2}]}
results = {
'columns': [{
'name': 'ga:newUsers'
}, {
'name': '"test2"'
}],
'rows': [{
'ga:newUsers': 123,
'"test2"': 2
}]
}
table_name = 'query_123'
create_table(connection, table_name, results)
connection.execute('SELECT 1 FROM query_123')

def test_creates_table(self):
connection = sqlite3.connect(':memory:')
results = {'columns': [{'name': 'test1'},
{'name': 'test2'}], 'rows': []}
results = {
'columns': [{
'name': 'test1'
}, {
'name': 'test2'
}],
'rows': []
}
table_name = 'query_123'
create_table(connection, table_name, results)
connection.execute('SELECT 1 FROM query_123')

def test_creates_table_with_missing_columns(self):
connection = sqlite3.connect(':memory:')
results = {'columns': [{'name': 'test1'}, {'name': 'test2'}], 'rows': [
{'test1': 1, 'test2': 2}, {'test1': 3}]}
results = {
'columns': [{
'name': 'test1'
}, {
'name': 'test2'
}],
'rows': [{
'test1': 1,
'test2': 2
}, {
'test1': 3
}]
}
table_name = 'query_123'
create_table(connection, table_name, results)
connection.execute('SELECT 1 FROM query_123')

def test_creates_table_with_spaces_in_column_name(self):
connection = sqlite3.connect(':memory:')
results = {'columns': [{'name': 'two words'}, {'name': 'test2'}], 'rows': [
{'two words': 1, 'test2': 2}, {'test1': 3}]}
results = {
'columns': [{
'name': 'two words'
}, {
'name': 'test2'
}],
'rows': [{
'two words': 1,
'test2': 2
}, {
'test1': 3
}]
}
table_name = 'query_123'
create_table(connection, table_name, results)
connection.execute('SELECT 1 FROM query_123')

def test_creates_table_with_dashes_in_column_name(self):
connection = sqlite3.connect(':memory:')
results = {
'columns': [{'name': 'two-words'}, {'name': 'test2'}],
'rows': [{'two-words': 1, 'test2': 2}]
'columns': [{
'name': 'two-words'
}, {
'name': 'test2'
}],
'rows': [{
'two-words': 1,
'test2': 2
}]
}
table_name = 'query_123'
create_table(connection, table_name, results)
Expand All @@ -79,8 +134,17 @@ def test_creates_table_with_dashes_in_column_name(self):

def test_creates_table_with_non_ascii_in_column_name(self):
connection = sqlite3.connect(':memory:')
results = {'columns': [{'name': u'\xe4'}, {'name': 'test2'}], 'rows': [
{u'\xe4': 1, 'test2': 2}]}
results = {
'columns': [{
'name': u'\xe4'
}, {
'name': 'test2'
}],
'rows': [{
u'\xe4': 1,
'test2': 2
}]
}
table_name = 'query_123'
create_table(connection, table_name, results)
connection.execute('SELECT 1 FROM query_123')
Expand All @@ -95,18 +159,30 @@ def test_shows_meaningful_error_on_failure_to_create_table(self):
def test_loads_results(self):
connection = sqlite3.connect(':memory:')
rows = [{'test1': 1, 'test2': 'test'}, {'test1': 2, 'test2': 'test2'}]
results = {'columns': [{'name': 'test1'},
{'name': 'test2'}], 'rows': rows}
results = {
'columns': [{
'name': 'test1'
}, {
'name': 'test2'
}],
'rows': rows
}
table_name = 'query_123'
create_table(connection, table_name, results)
self.assertEquals(
len(list(connection.execute('SELECT * FROM query_123'))), 2)

def test_loads_list_and_dict_results(self):
connection = sqlite3.connect(':memory:')
rows = [{'test1': [1,2,3]}, {'test2': {'a': 'b'}}]
results = {'columns': [{'name': 'test1'},
{'name': 'test2'}], 'rows': rows}
rows = [{'test1': [1, 2, 3]}, {'test2': {'a': 'b'}}]
results = {
'columns': [{
'name': 'test1'
}, {
'name': 'test2'
}],
'rows': rows
}
table_name = 'query_123'
create_table(connection, table_name, results)
self.assertEquals(
Expand Down Expand Up @@ -135,6 +211,15 @@ def test_returns_query(self):
loaded = _load_query(user, query.id)
self.assertEquals(query, loaded)

def test_returns_query_when_user_has_view_only_access(self):
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

New test.

ds = self.factory.create_data_source(
group=self.factory.org.default_group, view_only=True)
query = self.factory.create_query(data_source=ds)
user = self.factory.create_user()

loaded = _load_query(user, query.id)
self.assertEquals(query, loaded)


class TestExtractCachedQueryIds(TestCase):
def test_works_with_simple_query(self):
Expand Down