more tests for aliases

ckanext/datastore/logic/action.py

@@ -107,15 +107,26 @@ def datastore_search(context, data_dict):
     model = _get_or_bust(context, 'model')
     id = _get_or_bust(data_dict, 'resource_id')
 
-    if not model.Resource.get(id):
+    data_dict['connection_url'] = pylons.config['ckan.datastore_write_url']
+
+    res_exists = model.Resource.get(id)
+
+    alias_exists = False
+    if not res_exists:
+        # assume id is an alias
+        alias_sql = "select mainname from alias_mapping \
+            where aliasname = '{}'".format(id)
+        result = db._get_engine(None, data_dict).execute(alias_sql).fetchone()
+        if result:
+            alias_exists = model.Resource.get(result[0].strip('"'))
+
+    if not (res_exists or alias_exists):
         raise p.toolkit.ObjectNotFound(p.toolkit._(
             'Resource "{}" was not found.'.format(id)
         ))
 
     p.toolkit.check_access('datastore_search', context, data_dict)
 
-    data_dict['connection_url'] = pylons.config['ckan.datastore_write_url']
-
     result = db.search(context, data_dict)
     result.pop('id', None)
     result.pop('connection_url')
@@ -139,7 +150,8 @@ def data_search_sql(context, data_dict):
     if not db.is_single_statement(sql):
         raise p.toolkit.ValidationError({
             'query': ['Query is not a single statement or contains semicolons.'],
-            'hint': ['If you want to use semicolons, use character encoding (; equals chr(59)) and string concatenation (||). ']
+            'hint': ['If you want to use semicolons, use character encoding \
+                (; equals chr(59)) and string concatenation (||). ']
         })
 
     p.toolkit.check_access('datastore_search', context, data_dict)

ckanext/datastore/plugin.py

@@ -131,7 +131,9 @@ def _create_alias_table(self):
         mapping_sql = '''
             SELECT distinct
                 d.refobjid::regclass AS main,
-                r.ev_class::regclass AS alias
+                dependent.relname AS mainname,
+                r.ev_class::regclass AS alias,
+                dependee.relname AS aliasname
             FROM
                 pg_attribute    as a
                 JOIN pg_depend  as d on d.refobjid = a.attrelid AND d.refobjsubid = a.attnum

ckanext/datastore/tests/test_datastore.py

@@ -1,6 +1,6 @@
-
 import json
 import sqlalchemy
+import pylons
 import ckan.plugins as p
 import ckan.lib.create_test_data as ctd
 import ckan.model as model
@@ -132,8 +132,10 @@ def test_bad_records(self):
 
     def test_create_basic(self):
         resource = model.Package.get('annakarenina').resources[0]
+        alias = u'books1'
         data = {
             'resource_id': resource.id,
+            'alias' : alias,
             'fields': [{'id': 'book', 'type': 'text'},
                        {'id': 'author', 'type': '_json'}],
             'records': [
@@ -176,6 +178,21 @@ def test_create_basic(self):
         assert results.rowcount == 2
         model.Session.remove()
 
+        # check alias for resource
+        c = model.Session.connection()
+
+        results = [row for row in c.execute('select * from "{0}"'.format(resource.id))]
+        results_alias = [row for row in c.execute('select * from "{0}"'.format(alias))]
+
+        pp(results_alias)
+        pp(results)
+
+        assert results == results_alias
+
+        sql = "select * from alias_mapping where main='{}'::regclass and alias='{}'::regclass".format(resource.id, alias)
+        results = c.execute(sql)
+        assert results.rowcount == 1
+
         # check to test to see if resource now has a datastore table
         postparams = '%s=1' % json.dumps({'id': resource.id})
         auth = {'Authorization': str(self.sysadmin_user.apikey)}
@@ -239,6 +256,7 @@ def test_create_basic(self):
 
         results = c.execute('''select * from "{0}" where _full_text @@ to_tsquery('dostoevsky') '''.format(resource.id))
         assert results.rowcount == 2
+
         model.Session.remove()
 
     def test_guess_types(self):
@@ -339,55 +357,6 @@ def test_guess_types(self):
 
         assert res_dict['success'] is False
 
-class TestDatastoreCreate2(tests.WsgiAppCase):
-    sysadmin_user = None
-    normal_user = None
-    p.load('datastore')
-
-    @classmethod
-    def setup_class(cls):
-        ctd.CreateTestData.create()
-        cls.sysadmin_user = model.User.get('testsysadmin')
-        cls.normal_user = model.User.get('annafan')
-
-    @classmethod
-    def teardown_class(cls):
-        model.repo.rebuild_db()
-
-    def test_alias(self):
-        resource = model.Package.get('annakarenina').resources[0]
-        alias = u'books'
-        data = {
-            'resource_id': resource.id,
-            'alias' : alias,
-            'fields': [{'id': 'book', 'type': 'text'},
-                       {'id': 'author', 'type': '_json'}],
-            'records': [
-                        {'book': 'crime', 'author': ['tolstoy', 'dostoevsky']},
-                        {'book': 'annakarenina', 'author': ['tolstoy', 'putin']},
-                        {'book': 'warandpeace'}]  # treat author as null
-        }
-
-        postparams = '%s=1' % json.dumps(data)
-        auth = {'Authorization': str(self.sysadmin_user.apikey)}
-        res = self.app.post('/api/action/datastore_create', params=postparams,
-                            extra_environ=auth)
-        res_dict = json.loads(res.body)
-
-        assert res_dict['success'] is True
-
-        c = model.Session.connection()
-        results = [row for row in c.execute('select * from "{0}"'.format(resource.id))]
-        results2 = [row for row in c.execute('select * from "{0}"'.format(alias))]
-
-        # results from alias and main view should be the same
-        assert results == results2
-
-        # check whether the pair is in the alias table
-        sql = "select * from alias_mapping where main='{}'::regclass and alias='{}'::regclass".format(resource.id, alias)
-        results = c.execute(sql)
-        assert results.rowcount == 1
-
 
 class TestDatastoreDelete(tests.WsgiAppCase):
     sysadmin_user = None
@@ -402,6 +371,7 @@ def setup_class(cls):
         resource = model.Package.get('annakarenina').resources[0]
         cls.data = {
             'resource_id': resource.id,
+            'alias': 'books2',
             'fields': [{'id': 'book', 'type': 'text'},
                        {'id': 'author', 'type': 'text'}],
             'records': [{'book': 'annakarenina', 'author': 'tolstoy'},
@@ -438,6 +408,10 @@ def test_delete_basic(self):
         resource_id = self.data['resource_id']
         c = model.Session.connection()
 
+        # alias should be deleted
+        results = c.execute("select 1 from pg_views where viewname = '{}'".format(self.data['alias']))
+        assert results.rowcount == 0
+
         try:
             # check that data was actually deleted: this should raise a
             # ProgrammingError as the table should not exist any more
@@ -527,6 +501,7 @@ def setup_class(cls):
         resource = model.Package.get('annakarenina').resources[0]
         cls.data = {
             'resource_id': resource.id,
+            'alias': 'books3',
             'fields': [{'id': u'b\xfck', 'type': 'text'},
                        {'id': 'author', 'type': 'text'},
                        {'id': 'published'}],
@@ -769,6 +744,7 @@ def setup_class(cls):
         resource = model.Package.get('annakarenina').resources[0]
         cls.data = {
             'resource_id': resource.id,
+            'alias': 'books4',
             'fields': [{'id': u'b\xfck', 'type': 'text'},
                        {'id': 'author', 'type': 'text'},
                        {'id': 'published'}],
@@ -829,6 +805,16 @@ def test_select_basic(self):
         result = res_dict['result']
         assert result['records'] == self.expected_records
 
+        # test alias search
+        query = 'SELECT * FROM public."{}"'.format(self.data['alias'])
+        data = {'sql': query}
+        postparams = json.dumps(data)
+        res = self.app.post('/api/action/data_search_sql', params=postparams,
+                            extra_environ=auth)
+        res_dict_alias = json.loads(res.body)
+
+        assert result['records'] == res_dict_alias['result']['records']
+
     def test_self_join(self):
         query = 'SELECT a._id as first, b._id as second \
                 FROM "{0}" AS a, \