Merge branch 'joetsoi-package-tests'

ckan/controllers/package.py

@@ -1054,11 +1054,6 @@ def delete(self, id):
         context = {'model': model, 'session': model.Session,
                    'user': c.user or c.author, 'auth_user_obj': c.userobj}
 
-        try:
-            check_access('package_delete', context, {'id': id})
-        except NotAuthorized:
-            abort(401, _('Unauthorized to delete package %s') % '')
-
         try:
             if request.method == 'POST':
                 get_action('package_delete')(context, {'id': id})

ckan/templates/package/confirm_delete.html

@@ -10,7 +10,7 @@
       {% block form %}
         <p>{{ _('Are you sure you want to delete dataset - {name}?').format(name=c.pkg_dict.name) }}</p>
         <p class="form-actions">
-          <form action="{% url_for controller='package', action='delete', id=c.pkg_dict.name %}" method="post">
+          <form id='confirm-dataset-delete-form' action="{% url_for controller='package', action='delete', id=c.pkg_dict.name %}" method="post">
             <button class="btn" type="submit" name="cancel" >{{ _('Cancel') }}</button>
             <button class="btn btn-primary" type="submit" name="delete" >{{ _('Confirm Delete') }}</button>
           </form>

ckan/templates/package/confirm_delete_resource.html

@@ -10,7 +10,7 @@
       {% block form %}
         <p>{{ _('Are you sure you want to delete resource - {name}?').format(name=h.resource_display_name(c.resource_dict)) }}</p>
         <p class="form-actions">
-          <form action="{% url_for controller='package', action='resource_delete', resource_id=c.resource_dict.id, id=c.pkg_id %}" method="post">
+          <form id='confirm-resource-delete-form' action="{% url_for controller='package', action='resource_delete', resource_id=c.resource_dict.id, id=c.pkg_id %}" method="post">
             <button class="btn" type="submit" name="cancel" >{{ _('Cancel') }}</button>
             <button class="btn btn-primary" type="submit" name="delete" >{{ _('Confirm Delete') }}</button>
           </form>

ckan/templates/user/new_user_form.html

@@ -1,6 +1,6 @@
 {% import "macros/form.html" as form %}
 
-<form class="form-horizontal" action="" method="post">
+<form id="user-register-form" class="form-horizontal" action="" method="post">
   {{ form.errors(error_summary) }}
   {{ form.input("name", id="field-username", label=_("Username"), placeholder=_("username"), value=data.name, error=errors.name, classes=["control-medium"]) }}
   {{ form.input("fullname", id="field-fullname", label=_("Full Name"), placeholder=_("Joe Bloggs"), value=data.fullname, error=errors.fullname, classes=["control-medium"]) }}

ckan/tests/controllers/test_package.py

@@ -1,4 +1,10 @@
-from nose.tools import assert_equal, assert_true, assert_not_equal
+from nose.tools import (
+    assert_equal,
+    assert_not_equal,
+    assert_raises,
+    assert_true,
+)
+
 
 from routes import url_for
 
@@ -329,7 +335,157 @@ def test_dataset_edit_org_dropdown_visible_to_sysadmin_with_no_orgs_available(se
         assert 'value="{0}"'.format(org['id']) in pkg_edit_response
 
 
-class TestPackageResourceRead(helpers.FunctionalTestBase):
+class TestPackageRead(helpers.FunctionalTestBase):
+    @classmethod
+    def setup_class(cls):
+        super(cls, cls).setup_class()
+        helpers.reset_db()
+
+    def setup(self):
+        model.repo.rebuild_db()
+
+    def test_read(self):
+        dataset = factories.Dataset()
+        app = helpers._get_test_app()
+        response = app.get(url_for(controller='package', action='read',
+                                   id=dataset['name']))
+        response.mustcontain('Test Dataset')
+        response.mustcontain('Just another test dataset')
+
+    def test_read_rdf(self):
+        dataset1 = factories.Dataset()
+
+        offset = url_for(controller='package', action='read',
+                         id=dataset1['name']) + ".rdf"
+        app = self._get_test_app()
+        res = app.get(offset, status=200)
+
+        assert 'dcat' in res, res
+        assert '{{' not in res, res
+
+    def test_read_n3(self):
+        dataset1 = factories.Dataset()
+
+        offset = url_for(controller='package', action='read',
+                         id=dataset1['name']) + ".n3"
+        app = self._get_test_app()
+        res = app.get(offset, status=200)
+
+        assert 'dcat' in res, res
+        assert '{{' not in res, res
+
+
+class TestPackageDelete(helpers.FunctionalTestBase):
+    def test_owner_delete(self):
+        user = factories.User()
+        owner_org = factories.Organization(
+            users=[{'name': user['id'], 'capacity': 'admin'}]
+        )
+        dataset = factories.Dataset(owner_org=owner_org['id'])
+
+        app = helpers._get_test_app()
+        env = {'REMOTE_USER': user['name'].encode('ascii')}
+        response = app.post(
+            url_for(controller='package', action='delete', id=dataset['name']),
+            extra_environ=env,
+        )
+        response = response.follow()
+        assert_equal(200, response.status_int)
+
+        deleted = helpers.call_action('package_show', id=dataset['id'])
+        assert_equal('deleted', deleted['state'])
+
+    def test_delete_on_non_existing_dataset(self):
+        app = helpers._get_test_app()
+        response = app.post(
+            url_for(controller='package', action='delete',
+                    id='schrodingersdatset'),
+            expect_errors=True,
+        )
+        assert_equal(404, response.status_int)
+
+    def test_sysadmin_can_delete_any_dataset(self):
+        owner_org = factories.Organization()
+        dataset = factories.Dataset(owner_org=owner_org['id'])
+        app = helpers._get_test_app()
+
+        user = factories.Sysadmin()
+        env = {'REMOTE_USER': user['name'].encode('ascii')}
+
+        response = app.post(
+            url_for(controller='package', action='delete', id=dataset['name']),
+            extra_environ=env,
+        )
+        response = response.follow()
+        assert_equal(200, response.status_int)
+
+        deleted = helpers.call_action('package_show', id=dataset['id'])
+        assert_equal('deleted', deleted['state'])
+
+    def test_anon_user_cannot_delete_owned_dataset(self):
+        user = factories.User()
+        owner_org = factories.Organization(
+            users=[{'name': user['id'], 'capacity': 'admin'}]
+        )
+        dataset = factories.Dataset(owner_org=owner_org['id'])
+
+        app = helpers._get_test_app()
+        response = app.post(
+            url_for(controller='package', action='delete', id=dataset['name']),
+        )
+        response = response.follow()
+        assert_equal(200, response.status_int)
+        response.mustcontain('Unauthorized to delete package')
+
+        deleted = helpers.call_action('package_show', id=dataset['id'])
+        assert_equal('active', deleted['state'])
+
+    def test_logged_in_user_cannot_delete_owned_dataset(self):
+        owner = factories.User()
+        owner_org = factories.Organization(
+            users=[{'name': owner['id'], 'capacity': 'admin'}]
+        )
+        dataset = factories.Dataset(owner_org=owner_org['id'])
+
+        app = helpers._get_test_app()
+        user = factories.User()
+        env = {'REMOTE_USER': user['name'].encode('ascii')}
+        response = app.post(
+            url_for(controller='package', action='delete', id=dataset['name']),
+            extra_environ=env,
+            expect_errors=True
+        )
+        assert_equal(401, response.status_int)
+        response.mustcontain('Unauthorized to delete package')
+
+    def test_confirm_cancel_delete(self):
+        '''Test confirmation of deleting datasets
+
+        When package_delete is made as a get request, it should return a
+        'do you want to delete this dataset? confirmation page'''
+        user = factories.User()
+        owner_org = factories.Organization(
+            users=[{'name': user['id'], 'capacity': 'admin'}]
+        )
+        dataset = factories.Dataset(owner_org=owner_org['id'])
+
+        app = helpers._get_test_app()
+        env = {'REMOTE_USER': user['name'].encode('ascii')}
+        response = app.get(
+            url_for(controller='package', action='delete', id=dataset['name']),
+            extra_environ=env,
+        )
+        assert_equal(200, response.status_int)
+        message = 'Are you sure you want to delete dataset - {name}?'
+        response.mustcontain(message.format(name=dataset['name']))
+
+        form = response.forms['confirm-dataset-delete-form']
+        response = form.submit('cancel')
+        response = helpers.webtest_maybe_follow(response)
+        assert_equal(200, response.status_int)
+
+
+class TestResourceRead(helpers.FunctionalTestBase):
     @classmethod
     def setup_class(cls):
         super(cls, cls).setup_class()
@@ -446,36 +602,130 @@ def test_resource_read_sysadmin(self):
         app.get(url, status=200, extra_environ=env)
 
 
-class TestPackageRead(helpers.FunctionalTestBase):
-    @classmethod
-    def setup_class(cls):
-        super(cls, cls).setup_class()
-        helpers.reset_db()
+class TestResourceDelete(helpers.FunctionalTestBase):
+    def test_dataset_owners_can_delete_resources(self):
+        user = factories.User()
+        owner_org = factories.Organization(
+            users=[{'name': user['id'], 'capacity': 'admin'}]
+        )
+        dataset = factories.Dataset(owner_org=owner_org['id'])
+        resource = factories.Resource(package_id=dataset['id'])
+        app = helpers._get_test_app()
+        env = {'REMOTE_USER': user['name'].encode('ascii')}
+        response = app.post(
+            url_for(controller='package', action='resource_delete',
+                    id=dataset['name'], resource_id=resource['id']),
+            extra_environ=env,
+        )
+        response = response.follow()
+        assert_equal(200, response.status_int)
+        response.mustcontain('This dataset has no data')
 
-    def setup(self):
-        model.repo.rebuild_db()
+        assert_raises(p.toolkit.ObjectNotFound, helpers.call_action,
+                      'resource_show', id=resource['id'])
 
-    def test_read_rdf(self):
-        dataset1 = factories.Dataset()
+    def test_deleting_non_existing_resource_404s(self):
+        user = factories.User()
+        owner_org = factories.Organization(
+            users=[{'name': user['id'], 'capacity': 'admin'}]
+        )
+        dataset = factories.Dataset(owner_org=owner_org['id'])
+        env = {'REMOTE_USER': user['name'].encode('ascii')}
+        app = helpers._get_test_app()
+        response = app.post(
+            url_for(controller='package', action='resource_delete',
+                    id=dataset['name'], resource_id='doesnotexist'),
+            extra_environ=env,
+            expect_errors=True
+        )
+        assert_equal(404, response.status_int)
 
-        offset = url_for(controller='package', action='read',
-                         id=dataset1['name']) + ".rdf"
-        app = self._get_test_app()
-        res = app.get(offset, status=200)
+    def test_anon_users_cannot_delete_owned_resources(self):
+        user = factories.User()
+        owner_org = factories.Organization(
+            users=[{'name': user['id'], 'capacity': 'admin'}]
+        )
+        dataset = factories.Dataset(owner_org=owner_org['id'])
+        resource = factories.Resource(package_id=dataset['id'])
 
-        assert 'dcat' in res, res
-        assert '{{' not in res, res
+        app = helpers._get_test_app()
+        response = app.post(
+            url_for(controller='package', action='resource_delete',
+                    id=dataset['name'], resource_id=resource['id']),
+        )
+        response = response.follow()
+        assert_equal(200, response.status_int)
+        response.mustcontain('Unauthorized to delete package')
+
+    def test_logged_in_users_cannot_delete_resources_they_do_not_own(self):
+        # setup our dataset
+        owner = factories.User()
+        owner_org = factories.Organization(
+            users=[{'name': owner['id'], 'capacity': 'admin'}]
+        )
+        dataset = factories.Dataset(owner_org=owner_org['id'])
+        resource = factories.Resource(package_id=dataset['id'])
 
-    def test_read_n3(self):
-        dataset1 = factories.Dataset()
+        # access as another user
+        user = factories.User()
+        env = {'REMOTE_USER': user['name'].encode('ascii')}
+        app = helpers._get_test_app()
+        response = app.post(
+            url_for(controller='package', action='resource_delete',
+                    id=dataset['name'], resource_id=resource['id']),
+            extra_environ=env,
+            expect_errors=True
+        )
+        assert_equal(401, response.status_int)
+        response.mustcontain('Unauthorized to delete package')
 
-        offset = url_for(controller='package', action='read',
-                         id=dataset1['name']) + ".n3"
-        app = self._get_test_app()
-        res = app.get(offset, status=200)
+    def test_sysadmins_can_delete_any_resource(self):
+        owner_org = factories.Organization()
+        dataset = factories.Dataset(owner_org=owner_org['id'])
+        resource = factories.Resource(package_id=dataset['id'])
 
-        assert 'dcat' in res, res
-        assert '{{' not in res, res
+        sysadmin = factories.Sysadmin()
+        app = helpers._get_test_app()
+        env = {'REMOTE_USER': sysadmin['name'].encode('ascii')}
+        response = app.post(
+            url_for(controller='package', action='resource_delete',
+                    id=dataset['name'], resource_id=resource['id']),
+            extra_environ=env,
+        )
+        response = response.follow()
+        assert_equal(200, response.status_int)
+        response.mustcontain('This dataset has no data')
+
+        assert_raises(p.toolkit.ObjectNotFound, helpers.call_action,
+                      'resource_show', id=resource['id'])
+
+    def test_confirm_and_cancel_deleting_a_resource(self):
+        '''Test confirmation of deleting resources
+
+        When resource_delete is made as a get request, it should return a
+        'do you want to delete this reource? confirmation page'''
+        user = factories.User()
+        owner_org = factories.Organization(
+            users=[{'name': user['id'], 'capacity': 'admin'}]
+        )
+        dataset = factories.Dataset(owner_org=owner_org['id'])
+        resource = factories.Resource(package_id=dataset['id'])
+        app = helpers._get_test_app()
+        env = {'REMOTE_USER': user['name'].encode('ascii')}
+        response = app.get(
+            url_for(controller='package', action='resource_delete',
+                    id=dataset['name'], resource_id=resource['id']),
+            extra_environ=env,
+        )
+        assert_equal(200, response.status_int)
+        message = 'Are you sure you want to delete resource - {name}?'
+        response.mustcontain(message.format(name=resource['name']))
+
+        # cancelling sends us back to the resource edit page
+        form = response.forms['confirm-resource-delete-form']
+        response = form.submit('cancel')
+        response = response.follow()
+        assert_equal(200, response.status_int)
 
 
 class TestSearch(helpers.FunctionalTestBase):