From ce7f519adb08b3c7b4bac2de5084590b20749b85 Mon Sep 17 00:00:00 2001 From: avdata99 Date: Tue, 29 Oct 2024 16:43:48 -0300 Subject: [PATCH] Test with migrations --- ckanext/tracking/blueprints/csv.py | 7 ++- ckanext/tracking/tests/factories.py | 86 +++++++++++++++++++++++++++++ ckanext/tracking/tests/fixtures.py | 7 +++ ckanext/tracking/tests/test_csv.py | 63 +++++++++++++++++++++ conftest.py | 3 + 5 files changed, 164 insertions(+), 2 deletions(-) create mode 100644 ckanext/tracking/tests/factories.py create mode 100644 ckanext/tracking/tests/fixtures.py create mode 100644 ckanext/tracking/tests/test_csv.py create mode 100644 conftest.py diff --git a/ckanext/tracking/blueprints/csv.py b/ckanext/tracking/blueprints/csv.py index 8b54bd0..8e1ec9b 100644 --- a/ckanext/tracking/blueprints/csv.py +++ b/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: diff --git a/ckanext/tracking/tests/factories.py b/ckanext/tracking/tests/factories.py new file mode 100644 index 0000000..2516ac8 --- /dev/null +++ b/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' diff --git a/ckanext/tracking/tests/fixtures.py b/ckanext/tracking/tests/fixtures.py new file mode 100644 index 0000000..7212e11 --- /dev/null +++ b/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') diff --git a/ckanext/tracking/tests/test_csv.py b/ckanext/tracking/tests/test_csv.py new file mode 100644 index 0000000..f9f4529 --- /dev/null +++ b/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]}" diff --git a/conftest.py b/conftest.py new file mode 100644 index 0000000..628b203 --- /dev/null +++ b/conftest.py @@ -0,0 +1,3 @@ +pytest_plugins = [ + u'ckanext.tracking.tests.fixtures', +]