Test the aliases

ckanext/datastore/db.py

@@ -209,11 +209,11 @@ def create_table(context, data_dict):
     alias = data_dict.get('alias', None)
     if alias:
         # create alias view
-        sql_alias_string = u'create view "{alias}" as select * from "{orig}"'.format(alias=alias, orig=data_dict['resource_id'])
+        sql_alias_string = u'create view "{alias}" as select * from "{main}"'.format(main=data_dict['resource_id'], alias=alias)
         context['connection'].execute(sql_alias_string)
 
         # add view to alias table
-        sql_add_alias = u'insert into "alias_mapping" values (\'{orig}\'::regclass, \'{alias}\'::regclass)'.format(alias=alias, orig=data_dict['resource_id'])
+        sql_add_alias = u'insert into "alias_mapping" values (\'{main}\', \'{alias}\')'.format(main=data_dict['resource_id'], alias=alias)
         context['connection'].execute(sql_add_alias)
 
 

ckanext/datastore/plugin.py

@@ -40,8 +40,6 @@ def configure(self, config):
             if 'ckan.datastore_read_url' in config:
                 self._check_read_permissions()
 
-        self._create_alias_table()
-
         # Make sure actions are cached
         resource_show = p.toolkit.get_action('resource_show')
 
@@ -55,7 +53,7 @@ def new_resource_show(context, data_dict):
             try:
                 connection = engine.connect()
                 result = connection.execute(
-                    'select 1 from pg_tables where tablename = %s',
+                    'select * from pg_tables where tablename = %s',
                     new_data_dict['id']
                 ).fetchone()
                 if result:
@@ -66,6 +64,8 @@ def new_resource_show(context, data_dict):
                 connection.close()
             return new_data_dict
 
+        self._create_alias_table()
+
         ## Make sure do not run many times if configure is called repeatedly
         ## as in tests.
         if not hasattr(resource_show, '_datastore_wrapped'):
@@ -130,7 +130,7 @@ def _check_read_permissions(self):
             write_connection.execute("DROP TABLE foo")
 
     def _create_alias_table(self):
-        create_alias_table_sql = u'create table if not exists alias_mapping (relation oid, alias oid)'
+        create_alias_table_sql = u'create table if not exists alias_mapping (main name, alias name)'
         connection = db._get_engine(None,
             {'connection_url': pylons.config['ckan.datastore_write_url']}).connect()
         connection.execute(create_alias_table_sql)

ckanext/datastore/tests/test_datastore.py

@@ -6,7 +6,7 @@
 import ckan.model as model
 import ckan.tests as tests
 import ckanext.datastore.db as db
-
+from pprint import pprint as pp
 
 class TestDatastoreCreate(tests.WsgiAppCase):
     sysadmin_user = None
@@ -339,6 +339,55 @@ 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='{}' and alias='{}'".format(resource.id, alias)
+        results = c.execute(sql)
+        assert results.rowcount == 1
+
 
 class TestDatastoreDelete(tests.WsgiAppCase):
     sysadmin_user = None