Test with migrations

ckanext/tracking/blueprints/csv.py

@@ -45,8 +45,10 @@ def most_accessed_dataset_with_token():
             'total': row['total'],
         })
 
+    headers = ['Dataset ID', 'Dataset title', 'Dataset url', 'total']
+
     buffer = StringIO()
-    writer = csv.DictWriter(buffer, fieldnames=rows[0].keys())
+    writer = csv.DictWriter(buffer, fieldnames=headers)
 
     writer.writeheader()
     for row in rows:
@@ -95,8 +97,9 @@ def most_accessed_token():
             'total': row['total'],
         })
 
+    headers = ['User ID', 'User fullname', 'User name', 'User url', 'total']
     buffer = StringIO()
-    writer = csv.DictWriter(buffer, fieldnames=rows[0].keys())
+    writer = csv.DictWriter(buffer, fieldnames=headers)
 
     writer.writeheader()
     for row in rows:

ckanext/tracking/tests/factories.py

@@ -0,0 +1,86 @@
+import factory
+from ckan import model
+from ckan.plugins import toolkit
+from ckantoolkit.tests import factories
+from ckanext.tracking.models import TrackingUsage
+
+
+class TrackingUsageF(factory.Factory):
+    class Meta:
+        model = TrackingUsage
+
+    user_id = factory.LazyAttribute(lambda obj: factories.UserWithToken()['id'])
+    # ui | api
+    tracking_type = factory.Iterator(['ui', 'api'])
+    # show | edit | home | download
+    tracking_sub_type = factory.Iterator(['show', 'edit', 'home', 'download'])
+    # get the token from the UserWithToken factory
+    token_name = factory.Sequence(lambda n: "token-{0:05d}".format(n))
+    # dataset | resource | organization
+    object_type = factory.Iterator(['dataset', 'resource', 'organization'])
+    object_id = factory.Sequence(lambda n: "object-id-{0:05d}".format(n))
+    # More information about the usage
+    extras = {}
+
+    # allow defining user_id and token name from an user dict
+    @factory.post_generation
+    def user(self, create, extracted, **kwargs):
+        if extracted:
+            self.user_id = extracted['id']
+            user_obj = model.User.get(self.user_id)
+            token_name = f'token-{user_obj.name}'
+            self.token_name = token_name
+
+            # Check if the token already exists in DB
+            token = model.Session.query(model.ApiToken).filter_by(name=token_name).first()
+            if token:
+                return
+
+            # Create the token
+            token_data = {"user": user_obj.name, "name": token_name}
+            toolkit.get_action(u"api_token_create")({"ignore_auth": True}, token_data)
+
+    @classmethod
+    def _create(cls, target_class, *args, **kwargs):
+        obj = target_class(**kwargs)
+        model.Session.add(obj)
+        model.Session.commit()
+        model.Session.remove()
+
+        return obj
+
+
+class TrackingUsageUIDataset(TrackingUsageF):
+    tracking_type = 'ui'
+    tracking_sub_type = 'show'
+    object_type = 'dataset'
+
+
+class TrackingUsageUIDatasetEdit(TrackingUsageF):
+    tracking_type = 'ui'
+    tracking_sub_type = 'edit'
+    object_type = 'dataset'
+
+
+class TrackingUsageUIDatasetHome(TrackingUsageF):
+    tracking_type = 'ui'
+    tracking_sub_type = 'home'
+    object_type = 'dataset'
+
+
+class TrackingUsageUIResourceDownload(TrackingUsageF):
+    tracking_type = 'ui'
+    tracking_sub_type = 'download'
+    object_type = 'resource'
+
+
+class TrackingUsageAPIDataset(TrackingUsageF):
+    tracking_type = 'api'
+    tracking_sub_type = 'show'
+    object_type = 'dataset'
+
+
+class TrackingUsageAPIResourceDownload(TrackingUsageF):
+    tracking_type = 'api'
+    tracking_sub_type = 'download'
+    object_type = 'resource'

ckanext/tracking/tests/fixtures.py

@@ -0,0 +1,7 @@
+import pytest
+
+
+@pytest.fixture
+def tracking_migrate(migrate_db_for):
+    """ Apply the tracking migrations """
+    migrate_db_for('tracking')

ckanext/tracking/tests/test_csv.py

@@ -0,0 +1,63 @@
+import pytest
+from types import SimpleNamespace
+from ckan.plugins import toolkit
+from ckan.lib.helpers import url_for
+from ckan.tests import factories
+
+from ckanext.tracking.tests import factories as tf
+
+
+@pytest.fixture
+def base_data():
+    obj = SimpleNamespace()
+    obj.user1 = factories.UserWithToken()
+    obj.user2 = factories.UserWithToken()
+    obj.dataset1 = factories.Dataset()
+    obj.dataset2 = factories.Dataset()
+    obj.trackings = []
+    for user in [obj.user1, obj.user1, obj.user2]:
+        for dataset in [obj.dataset1, obj.dataset1, obj.dataset2]:
+            new_tracking = tf.TrackingUsageAPIDataset(user=user, object_id=dataset['id'])
+            obj.trackings.append(new_tracking)
+
+    return obj
+
+
+@pytest.mark.usefixtures('clean_db', 'tracking_migrate')
+class TestTrackingCSVView:
+    """ Test basic tracking from requests """
+    def test_dataset_with_token_csv_no_auth(self, app):
+        url = url_for('tracking_csv.most_accessed_dataset_with_token')
+        with pytest.raises(toolkit.NotAuthorized):
+            app.get(url)
+
+    def test_dataset_with_token_csv(self, app, base_data):
+        url = url_for('tracking_csv.most_accessed_dataset_with_token')
+        # download the CSV
+        auth = {"Authorization": base_data.user1['token']}
+        response = app.get(url, extra_environ=auth)
+        assert response.status_code == 200
+        # save the response locally
+        full_response = response.body
+        with open('most-accessed-dataset-with-token.csv', 'w') as f:
+            f.write(full_response)
+
+        # check the CSV content
+        lines = full_response.splitlines()
+        header = lines[0].split(',')
+        assert header == ['Dataset ID', 'Dataset title', 'Dataset url', 'total']
+        rows = lines[1:]
+        # They are just two datasets
+        assert len(rows) == 2
+        for row in rows:
+            fields = row.split(',')
+            if fields[0] == base_data.dataset1['id']:
+                assert fields[1] == base_data.dataset1['title']
+                assert fields[2] == url_for('dataset.read', id=base_data.dataset1['id'], qualified=True)
+                assert fields[3] == '6'
+            elif fields[0] == base_data.dataset2['id']:
+                assert fields[1] == base_data.dataset2['title']
+                assert fields[2] == url_for('dataset.read', id=base_data.dataset2['id'], qualified=True)
+                assert fields[3] == '3'
+            else:
+                assert False, f"Unexpected dataset id: {fields[0]}"

conftest.py

@@ -0,0 +1,3 @@
+pytest_plugins = [
+    u'ckanext.tracking.tests.fixtures',
+]