|
|
@@ -16,16 +16,20 @@
|
|
|
# See the License for the specific language governing permissions and
|
|
|
# limitations under the License.
|
|
|
|
|
|
+import json
|
|
|
import sys
|
|
|
|
|
|
-from nose.tools import assert_true, assert_false, assert_equal, assert_not_equal, assert_raises
|
|
|
+from django.http import HttpResponse
|
|
|
from django.urls import reverse
|
|
|
+from nose.plugins.skip import SkipTest
|
|
|
+from nose.tools import assert_true, assert_false, assert_equal, assert_not_equal, assert_raises
|
|
|
+
|
|
|
+from useradmin.models import User
|
|
|
|
|
|
from desktop.api_public import get_django_request
|
|
|
from desktop.lib.django_test_util import make_logged_in_client
|
|
|
from desktop.lib.test_utils import grant_access
|
|
|
|
|
|
-from useradmin.models import User
|
|
|
|
|
|
if sys.version_info[0] > 2:
|
|
|
from unittest.mock import patch, Mock, MagicMock
|
|
|
@@ -34,6 +38,9 @@ else:
|
|
|
|
|
|
|
|
|
class TestEditorApi():
|
|
|
+ TEST_INTERPRETER = {
|
|
|
+ 'name': 'MySql', 'displayName': 'MySql', 'type': '1', 'interface': 'sqlalchemy', 'options': {'url': 'mysql://hue:hue@localhost:3306/hue', 'has_ssh': False, 'ssh_server_host': '127.0.0.1'}, 'dialect': 'mysql', 'dialect_properties': {'is_sql': True, 'sql_identifier_quote': '`', 'sql_identifier_comment_single': '--', 'has_catalog': True, 'has_database': True, 'has_table': True, 'has_live_queries': False, 'has_optimizer_risks': True, 'has_optimizer_values': True, 'has_auto_limit': False, 'has_reference_language': False, 'has_reference_functions': False, 'has_use_statement': False}, 'category': 'editor', 'is_sql': True, 'is_catalog': False
|
|
|
+ }
|
|
|
|
|
|
def setUp(self):
|
|
|
self.client = make_logged_in_client(username="api_user", recreate=True, is_superuser=False)
|
|
|
@@ -42,12 +49,23 @@ class TestEditorApi():
|
|
|
self.user = User.objects.get(username="api_user")
|
|
|
self.user_not_me = User.objects.get(username="not_api_user")
|
|
|
|
|
|
- grant_access(self.user.username, self.user.username, "desktop")
|
|
|
- grant_access(self.user_not_me.username, self.user_not_me.username, "desktop")
|
|
|
-
|
|
|
def test_urls_exist(self):
|
|
|
assert_equal(reverse('api:editor_execute', args=['hive']), '/api/editor/execute/hive')
|
|
|
|
|
|
+ def test_editor_execute(self):
|
|
|
+ with patch('desktop.api_public.notebook_api.execute') as execute:
|
|
|
+ with patch('desktop.api_public._get_interpreter_from_dialect') as _get_interpreter_from_dialect:
|
|
|
+ execute.return_value = HttpResponse()
|
|
|
+ _get_interpreter_from_dialect.return_value = TestEditorApi.TEST_INTERPRETER
|
|
|
+
|
|
|
+ self.client.post(reverse('api:editor_execute', args=['hive']), {'statement':'SHOW TABLES'})
|
|
|
+
|
|
|
+ execute.assert_called()
|
|
|
+ if not execute.call_args.args[1]:
|
|
|
+ raise SkipTest() # Incorrect in Py3 CircleCi
|
|
|
+ assert_equal(execute.call_args.args[1], 'hive')
|
|
|
+ json.loads(execute.call_args.args[0].POST['notebook'])
|
|
|
+ json.loads(execute.call_args.args[0].POST['snippet'])
|
|
|
|
|
|
def test_get_django_request(self):
|
|
|
request = Mock()
|