Add tests

ckan/tests/controllers/test_organization.py

@@ -1,11 +1,13 @@
 # encoding: utf-8
 
-from ckan.common import config
-from bs4 import BeautifulSoup
 from nose.tools import assert_equal, assert_true, assert_in
-from ckan.lib.helpers import url_for
+from bs4 import BeautifulSoup
 from mock import patch
 
+from ckan.common import config
+from ckan.lib.helpers import url_for
+from ckan import model
+
 from ckan.tests import factories, helpers
 from ckan.tests.helpers import webtest_submit, submit_and_follow
 
@@ -569,3 +571,90 @@ def test_anonymous_users_cannot_add_members(self):
                     'role': 'test'
                 },
                 status=403, )
+
+
+class TestActivity(helpers.FunctionalTestBase):
+    def test_simple(self):
+        '''Checking the template shows the activity stream.'''
+        app = self._get_test_app()
+        user = factories.User()
+        org = factories.Organization(user=user)
+
+        url = url_for('organization.activity',
+                      id=org['id'])
+        response = app.get(url)
+        assert_in('Mr. Test User', response)
+        assert_in('created the organization', response)
+
+    def test_create_organization(self):
+        app = self._get_test_app()
+        user = factories.User()
+        org = factories.Organization(user=user)
+
+        url = url_for('organization.activity',
+                      id=org['id'])
+        response = app.get(url)
+        assert_in('<a href="/user/{}">Mr. Test User'.format(user['name']),
+                  response)
+        assert_in('created the organization', response)
+        assert_in('<a href="/organization/{}">Test Organization'.format(
+                  org['name']), response)
+
+    def test_create_dataset(self):
+        app = self._get_test_app()
+        user = factories.User()
+        org = factories.Organization()
+        dataset = factories.Dataset(owner_org=org['id'], user=user)
+
+        url = url_for('organization.activity',
+                      id=org['id'])
+        response = app.get(url)
+        assert_in('<a href="/user/{}">Mr. Test User'.format(user['name']),
+                  response)
+        assert_in('created the dataset', response)
+        assert_in('<a href="/dataset/{}">Test Dataset'.format(dataset['name']),
+                  response)
+
+    def _clear_activities(self):
+        model.Session.query(model.ActivityDetail).delete()
+        model.Session.query(model.Activity).delete()
+        model.Session.flush()
+
+    def test_change_dataset(self):
+        app = self._get_test_app()
+        user = factories.User()
+        org = factories.Organization()
+        dataset = factories.Dataset(owner_org=org['id'], user=user)
+        self._clear_activities()
+        dataset['title'] = 'Dataset with changed title'
+        helpers.call_action(
+            'package_update', context={'user': user['name']}, **dataset)
+
+        url = url_for('organization.activity',
+                      id=org['id'])
+        response = app.get(url)
+        assert_in('<a href="/user/{}">Mr. Test User'.format(user['name']),
+                  response)
+        assert_in('updated the dataset', response)
+        assert_in('<a href="/dataset/{}">Dataset with changed title'
+                  .format(dataset['name']),
+                  response)
+
+    def test_delete_dataset(self):
+        app = self._get_test_app()
+        user = factories.User()
+        org = factories.Organization()
+        dataset = factories.Dataset(owner_org=org['id'], user=user)
+        self._clear_activities()
+        helpers.call_action(
+            'package_delete', context={'user': user['name']}, **dataset)
+
+        url = url_for('organization.activity',
+                      id=org['id'])
+        response = app.get(url)
+        assert_in('<a href="/user/{}">Mr. Test User'.format(user['name']),
+                  response)
+        assert_in('deleted the dataset', response)
+        assert_in('<a href="/dataset/{}">Test Dataset'
+                  .format(dataset['name']),
+                  response)

ckan/tests/controllers/test_package.py

@@ -14,6 +14,7 @@
 import ckan.model as model
 import ckan.plugins as p
 from ckan.logic import get_action
+import ckan.lib.dictization as dictization
 
 import ckan.tests.helpers as helpers
 import ckan.tests.factories as factories
@@ -1787,3 +1788,129 @@ def test_no_redirect_loop_when_name_is_the_same_as_the_id(self):
         app = helpers._get_test_app()
         app.get(url_for('dataset.read', id=dataset['id']),
                 status=200)  # ie no redirect
+
+
+class TestActivity(helpers.FunctionalTestBase):
+    def test_simple(self):
+        '''Checking the template shows the activity stream.'''
+        app = self._get_test_app()
+        user = factories.User()
+        dataset = factories.Dataset(user=user)
+
+        url = url_for('dataset.activity',
+                      id=dataset['id'])
+        response = app.get(url)
+        assert_in('Mr. Test User', response)
+        assert_in('created the dataset', response)
+
+    def test_create_dataset(self):
+        app = self._get_test_app()
+        user = factories.User()
+        dataset = factories.Dataset(user=user)
+
+        url = url_for('dataset.activity',
+                      id=dataset['id'])
+        response = app.get(url)
+        assert_in('<a href="/user/{}">Mr. Test User'.format(user['name']),
+                  response)
+        assert_in('created the dataset', response)
+        assert_in('<a href="/dataset/{}">Test Dataset'.format(dataset['name']),
+                  response)
+
+    def _clear_activities(self):
+        model.Session.query(model.ActivityDetail).delete()
+        model.Session.query(model.Activity).delete()
+        model.Session.flush()
+
+    def test_change_dataset(self):
+        app = self._get_test_app()
+        user = factories.User()
+        dataset = factories.Dataset(user=user)
+        self._clear_activities()
+        dataset['title'] = 'Dataset with changed title'
+        helpers.call_action(
+            'package_update', context={'user': user['name']}, **dataset)
+
+        url = url_for('dataset.activity',
+                      id=dataset['id'])
+        response = app.get(url)
+        assert_in('<a href="/user/{}">Mr. Test User'.format(user['name']),
+                  response)
+        assert_in('updated the dataset', response)
+        assert_in('<a href="/dataset/{}">Dataset with changed title'
+                  .format(dataset['name']),
+                  response)
+
+    def test_add_resource_to_dataset(self):
+        app = self._get_test_app()
+        user = factories.User()
+        dataset = factories.Dataset(user=user)
+        self._clear_activities()
+        resource = factories.Resource(package_id=dataset['id'], user=user)
+
+        url = url_for('dataset.activity',
+                      id=dataset['id'])
+        response = app.get(url)
+        assert_in('<a href="/user/{}">Mr. Test User'.format(user['name']),
+                  response)
+        assert_in('added the resource', response)
+        assert_in('<a href="/dataset/{}/resource/{}">test_resource_'.format(
+                  dataset['id'], resource['id']), response)
+        assert_in('to the dataset', response)
+        assert_in('<a href="/dataset/{}">Test Dataset'.format(dataset['name']),
+                  response)
+
+    def test_legacy_changed_package_activity(self):
+        '''Render a custom activity *** TODO ***
+        '''
+        app = self._get_test_app()
+
+        user = factories.User()
+        dataset = factories.Dataset(user=user)
+
+        # delete the modern Activity object that's been automatically created
+        modern_activity = model.Session.query(model.Activity) \
+            .filter_by(object_id=dataset['id']) \
+            .one()
+        revision_id = modern_activity.revision_id
+        modern_activity.delete()
+
+        # Create an Activity object as it was in earlier versions of CKAN.
+        # This code is based on:
+        # https://github.com/ckan/ckan/blob/b348bf2fe68db6704ea0a3e22d533ded3d8d4344/ckan/model/package.py#L508
+        activity_type = 'changed'
+        dataset_table_dict = dictization.table_dictize(
+            model.Package.get(dataset['id']), context={'model': model})
+        activity = model.Activity(
+            user_id=user['id'],
+            object_id=dataset['id'],
+            revision_id=revision_id,
+            activity_type="%s package" % activity_type,
+            data={
+                # "actor": a legacy activity had no "actor"
+                # "package": a legacy activity had just the package table,
+                # rather than the result of package_show
+                'package': dataset_table_dict,
+            }
+        )
+        model.Session.add(activity)
+        # a legacy activity had a ActivityDetail associated with the Activity
+        # This code is based on:
+        # https://github.com/ckan/ckan/blob/b348bf2fe68db6704ea0a3e22d533ded3d8d4344/ckan/model/package.py#L542
+        activity_detail = model.ActivityDetail(
+            activity_id=activity.id,
+            object_id=dataset['id'],
+            object_type=u"Package",
+            activity_type=activity_type,
+            data={u'package': dataset_table_dict})
+        model.Session.add(activity_detail)
+        model.Session.flush()
+
+        url = url_for('dataset.activity',
+                      id=dataset['id'])
+        response = app.get(url)
+        assert_in('<a href="/user/{}">Mr. Test User'.format(user['name']),
+                  response)
+        assert_in('updated the dataset', response)
+        assert_in('<a href="/dataset/{}">Test Dataset'.format(dataset['name']),
+                  response)