diff --git a/ckanext/versions/helpers.py b/ckanext/versions/helpers.py index 3604e8a..2478953 100644 --- a/ckanext/versions/helpers.py +++ b/ckanext/versions/helpers.py @@ -78,3 +78,25 @@ def download_url(resource_url, version_id): ) return url + + +def dataset_version_for_activity_id(dataset_id, activity_id): + """Return dataset version created for given activity. + + :param dataset_id: the id or name of the dataset + :type dataset_id: string + :param activity_id: the id of the activity + :type activity_id: string + :returns: version, None if no version created for the given activity. + :rtype: dictionary + """ + # TODO: Think if instead `IPackageController.after_show` could be + # used to include version dict in package_dict + context = {'user': toolkit.g.user} + versions = toolkit.get_action('dataset_version_list')( + context, {'dataset_id': dataset_id} + ) + for version in versions: + if version['activity_id'] == activity_id: + return version + return None diff --git a/ckanext/versions/logic/action.py b/ckanext/versions/logic/action.py index 149d7fa..1559543 100644 --- a/ckanext/versions/logic/action.py +++ b/ckanext/versions/logic/action.py @@ -18,8 +18,6 @@ def version_update(context, data_dict): """Update a version from the current dataset. - :param package_id: the id the dataset - :type package_id: string :param version_id: the id of the version :type version_id: string :param name: A short name for the version @@ -30,7 +28,7 @@ def version_update(context, data_dict): :rtype: dictionary """ model = context.get('model', core_model) - version_id, name = toolkit.get_or_bust(data_dict, ['version', 'name']) + version_id, name = toolkit.get_or_bust(data_dict, ['version_id', 'name']) # I'll create my own session! With Blackjack! And H**kers! session = model.meta.create_local_session() @@ -42,11 +40,11 @@ def version_update(context, data_dict): if not version: raise toolkit.ObjectNotFound('Version not found') - toolkit.check_access('dataset_version_create', context, data_dict) + toolkit.check_access('version_create', context, data_dict) assert context.get('auth_user_obj') # Should be here after `check_access` version.name = name - version.description = data_dict.get('description', None) + version.notes = data_dict.get('notes', None) session.add(version) @@ -228,14 +226,26 @@ def version_delete(context, data_dict): def version_show(context, data_dict): """Show a specific version object - :param version_id: the id of the version + :param version_id: the id or name of the version :type version_id: string + :param dataset_id: [Optional] the id or name of a dataset. Mandatory + if version name provided as version_id + :type dataset_id: string :returns: the version dictionary :rtype: dict """ model = context.get('model', core_model) - version_id = toolkit.get_or_bust(data_dict, ['version_id']) - version = model.Session.query(Version).get(version_id) + version_name_or_id = toolkit.get_or_bust(data_dict, ['version_id']) + version = model.Session.query(Version).get(version_name_or_id) + if not version: + version_name = version_name_or_id + dataset_name_or_id = data_dict.get('dataset_id') + dataset = model.Package.get(dataset_name_or_id) + if dataset: + dataset_id = dataset.id + version = model.Session.query(Version). \ + filter(Version.package_id == dataset_id). \ + filter(Version.name == version_name).one_or_none() if not version: raise toolkit.ObjectNotFound('Version not found') diff --git a/ckanext/versions/logic/dataset_version_action.py b/ckanext/versions/logic/dataset_version_action.py new file mode 100644 index 0000000..44f9dc7 --- /dev/null +++ b/ckanext/versions/logic/dataset_version_action.py @@ -0,0 +1,226 @@ +import logging + +from datetime import datetime +from sqlalchemy.exc import IntegrityError + +from ckan import model as core_model +from ckan.plugins import toolkit +from ckanext.versions.logic.action import version_show +from ckanext.versions.model import Version + +log = logging.getLogger(__name__) + + +def dataset_version_create(context, data_dict): + """Create a new version from the current dataset's activity_id + + Currently you must have editor level access on the dataset + to create a version. If creator_user_id is not present, it will be set as + the logged it user. + + :param dataset_id: the id of the dataset + :type dataset_id: string + :param name: A short name for the version, e.g. v1.0 + :type name: string + :param notes optional: Notes about the version + :type notes: string + :returns: the newly created version + :rtype: dictionary + """ + model = context.get('model', core_model) + dataset_name_or_id, name = toolkit.get_or_bust( + data_dict, ['dataset_id', 'name']) + activity_id = data_dict.get('activity_id') + + dataset = model.Package.get(dataset_name_or_id) + if not dataset: + raise toolkit.ObjectNotFound("Dataset not found") + dataset_id = dataset.id + toolkit.check_access('version_create', + context, + {"package_id": dataset_id}) + creator_user_id = context['auth_user_obj'].id + + if activity_id: + activity = model.Activity.get(activity_id) + else: + activity = model.Session.query(model.Activity). \ + filter_by(object_id=dataset_id). \ + order_by(model.Activity.timestamp.desc()). \ + first() + if not activity: + raise toolkit.ObjectNotFound('Activity not found') + + version_for_activity = model.Session.query(Version). \ + filter_by(activity_id=activity.id). \ + first() + if version_for_activity: + raise toolkit.ValidationError("Version already exists for this activity") + + version = Version( + package_id=dataset_id, + activity_id=activity.id, + name=name, + notes=data_dict.get('notes', None), + created=datetime.utcnow(), + creator_user_id=creator_user_id) + + model.Session.add(version) + try: + model.Session.commit() + except IntegrityError as e: + # Name not unique, or foreign key constraint violated + model.Session.rollback() + log.debug("DB integrity error (version name not unique?): %s", e) + raise toolkit.ValidationError( + 'Version names must be unique per dataset' + ) + + log.info( + 'Version "%s" created for dataset %s', + name, + dataset_id + ) + + return version.as_dict() + + +def dataset_version_restore(context, data_dict): + """Restores dataset version by restoring dataset + metadata and creating a new version with it + + :param dataset_id: the id or name of the dataset + :type dataset_id: string + :param version_id: the id or name of the dataset + :type version_id: string + :returns: restored dataset + :rtype: dict + """ + model = context.get('model', core_model) + dataset_name_or_id, version_name_or_id = toolkit.get_or_bust( + data_dict, ['dataset_id', 'version_id']) + + dataset = model.Package.get(dataset_name_or_id) + if not dataset: + raise toolkit.ObjectNotFound("Dataset not found") + dataset_id = dataset.id + toolkit.check_access('version_create', + context, + {"package_id": dataset_id}) + + version = version_show(context, data_dict) + if not version: + raise toolkit.ObjectNotFound("Version not found") + v_name = "restored_{}".format(version['name']) + v_notes = "Restored from version: {}".format(version['name']) + old_dataset = activity_dataset_show( + context, + { + 'activity_id': version['activity_id'], + 'dataset_id': dataset_id + } + ) + restored_dataset = toolkit.get_action('package_update')(context, old_dataset) + dataset_version_create(context, {'dataset_id': dataset.id, 'name': v_name, 'notes': v_notes}) + + return restored_dataset + + +@toolkit.side_effect_free +def dataset_version_list(context, data_dict): + """List versions of a given dataset + + :param dataset_id: the id of the dataset + :type dataset_id: string + :returns: list of versions created for the dataset + :rtype: list + """ + model = context.get('model', core_model) + dataset_id = toolkit.get_or_bust(data_dict, ['dataset_id']) + dataset = model.Package.get(dataset_id) + if not dataset: + raise toolkit.ObjectNotFound('Dataset not found') + + toolkit.check_access('version_list', context, + {"package_id": dataset_id}) + + versions = model.Session.query(Version). \ + filter(Version.package_id == dataset.id). \ + order_by(Version.created.desc()) + + return [v.as_dict() for v in versions] + + +@toolkit.side_effect_free +def dataset_version_latest(context, data_dict): + ''' Show the latest version for a dataset + + :param dataset_id: the if of the dataset + :type dataset_id: string + :returns the version dictionary + :rtype dict + ''' + version_list = dataset_version_list(context, data_dict) + if len(version_list) < 1: + raise toolkit.ObjectNotFound("Versions not found for this dataset") + return version_list[0] + + +def activity_dataset_show(context, data_dict): + ''' Returns a dataset from the activity object. + + :param activity_id: the id of the activity + :type activity_id: string + :param dataset_id: the id of the resource + :type dataset_id: string + :returns: The dataset in the activity + :rtype: dict + ''' + activity_id, dataset_id = toolkit.get_or_bust( + data_dict, + ['activity_id', 'dataset_id'] + ) + dataset = toolkit.get_action('activity_data_show')( + context, + {'id': activity_id, 'object_type': 'package'} + ) + if not dataset or dataset['id'] != dataset_id: + raise toolkit.ObjectNotFound('Dataset not found in the activity object.') + + return dataset + + +def get_activity_id_from_dataset_version_name(context, data_dict): + ''' Returns the activity_id for the dataset version + + :param dataset_id: the id of the dataset + :type dataset_id: string + :param version: the name or id of the version + :type version: string + :returns: The activity_id of the version + :rtype: string + ''' + version_name, dataset_id = toolkit.get_or_bust( + data_dict, + ['version', 'dataset_id'] + ) + version_list = dataset_version_list(context, data_dict) + for version in version_list: + if version_name in {version['name'], version['id']}: + return version['activity_id'] + + raise toolkit.ObjectNotFound('Version not found in the dataset.') + + +def dataset_has_versions(context, data_dict): + """Check if the dataset has versions. + + :param dataset_id: the id the dataset + :type dataset_id: string + :returns: True if the dataset has at least 1 version + :rtype: boolean + """ + version_list = dataset_version_list(context, data_dict) + if not version_list: + return False + return True diff --git a/ckanext/versions/plugin.py b/ckanext/versions/plugin.py index e858a18..3644aa5 100644 --- a/ckanext/versions/plugin.py +++ b/ckanext/versions/plugin.py @@ -6,7 +6,7 @@ from ckanext.versions import cli, helpers from ckanext.versions.blueprints import blueprint -from ckanext.versions.logic import action, auth +from ckanext.versions.logic import action, auth, dataset_version_action from ckanext.versions.model import tables_exist log = logging.getLogger(__name__) @@ -51,8 +51,13 @@ def get_actions(self): 'resource_version_current': action.resource_version_current, 'resource_version_clear': action.resource_version_clear, 'version_show': action.version_show, + 'version_update': action.version_update, 'version_delete': action.version_delete, 'resource_view_list': action.resource_view_list, + 'dataset_version_create': dataset_version_action.dataset_version_create, + 'dataset_version_restore': dataset_version_action.dataset_version_restore, + 'dataset_version_list': dataset_version_action.dataset_version_list, + 'dataset_version_latest': dataset_version_action.dataset_version_latest } # IAuthFunctions @@ -75,6 +80,7 @@ def get_helpers(self): 'versions_resource_version_from_activity_id': helpers.resource_version_from_activity_id, 'versions_resource_current_version': helpers.resource_current_version, 'versions_download_url': helpers.download_url, + 'dataset_version_for_activity_id': helpers.dataset_version_for_activity_id, } return helper_functions diff --git a/ckanext/versions/tests/__init__.py b/ckanext/versions/tests/__init__.py index 11e7b86..b9f1cb5 100644 --- a/ckanext/versions/tests/__init__.py +++ b/ckanext/versions/tests/__init__.py @@ -1,4 +1,5 @@ from ckan import model +from ckanext.versions.logic.dataset_version_action import dataset_version_create, dataset_version_restore def get_context(user): @@ -6,3 +7,27 @@ def get_context(user): 'model': model, 'user': user if isinstance(user, str) else user['name'] } + + +def assert_version(version, checks): + assert version + for k, v in checks.items(): + assert version[k] == v, "found incorrect %s of version" % k + + +def create_version(dataset_id, user, version_name="Default Name"): + return dataset_version_create( + get_context(user), + { + "dataset_id": dataset_id, + "name": version_name + } + ) + + +def restore_version(dataset_id, version_id, user): + context = get_context(user) + return dataset_version_restore(context, { + 'dataset_id': dataset_id, + 'version_id': version_id + }) diff --git a/ckanext/versions/tests/fixtures.py b/ckanext/versions/tests/fixtures.py index 460aa19..d8019e4 100644 --- a/ckanext/versions/tests/fixtures.py +++ b/ckanext/versions/tests/fixtures.py @@ -1,9 +1,50 @@ import pytest +from ckan.tests import factories from ckanext.versions.model import create_tables, tables_exist +from ckanext.versions.tests import create_version @pytest.fixture def versions_setup(): if not tables_exist(): create_tables() + + +@pytest.fixture() +def org_admin(): + return factories.User(name="admin") + + +@pytest.fixture() +def org_editor(): + return factories.User(name="editor") + + +@pytest.fixture() +def org_member(): + return factories.User(name="member") + + +@pytest.fixture() +def test_organization(org_admin, org_editor, org_member): + return factories.Organization(users=[ + {'name': org_admin['id'], 'capacity': 'admin'}, + {'name': org_editor['id'], 'capacity': 'editor'}, + {'name': org_member['id'], 'capacity': 'member'} + ]) + + +@pytest.fixture() +def test_dataset(test_organization): + return factories.Dataset(owner_org=test_organization['id']) + + +@pytest.fixture() +def test_resource(test_dataset): + return factories.Resource(package_id=test_dataset['id']) + + +@pytest.fixture() +def test_version(test_dataset, org_editor): + return create_version(test_dataset['id'], org_editor, version_name="Version1") diff --git a/ckanext/versions/tests/test_actions.py b/ckanext/versions/tests/test_actions.py index 157b331..36905f2 100644 --- a/ckanext/versions/tests/test_actions.py +++ b/ckanext/versions/tests/test_actions.py @@ -302,6 +302,26 @@ def test_resource_version_current_no_versions(self): assert helpers.call_action('resource_version_current', {}, resource_id=resource['id']) is None +@pytest.mark.usefixtures('clean_db', 'versions_setup') +class TestVersionUpdate(object): + + def test_version_update(self, test_version, org_editor): + context = get_context(org_editor) + updated_version = toolkit.get_action('version_update')( + context, + { + 'version_id': test_version['id'], + 'package_id': test_version['package_id'], + 'name': "updated-name", + 'notes': "updated-notes" + } + ) + + assert test_version['id'] == updated_version['id'] + assert "updated-name" == updated_version['name'] + assert "updated-notes" == updated_version['notes'] + + @pytest.mark.usefixtures('clean_db', 'versions_setup') class TestVersionShow(object): @@ -329,6 +349,34 @@ def test_version_show(self): assert result['notes'] == 'Version notes' assert result['creator_user_id'] == user['id'] + def test_version_show_for_version_name(self): + dataset = factories.Dataset() + resource = factories.Resource( + package_id=dataset['id'], + name='First name' + ) + user = factories.Sysadmin() + context = get_context(user) + + version = resource_version_create( + context, { + 'resource_id': resource['id'], + 'name': '1', + 'notes': 'Version notes' + } + ) + + result = version_show( + context, + {'version_id': version['name'], + 'dataset_id': dataset['id']} + ) + + assert result['id'] == version['id'] + assert result['name'] == '1' + assert result['notes'] == 'Version notes' + assert result['creator_user_id'] == user['id'] + @pytest.mark.usefixtures('clean_db', 'versions_setup') class TestVersionDelete(object): diff --git a/ckanext/versions/tests/test_dataset_version_action.py b/ckanext/versions/tests/test_dataset_version_action.py new file mode 100644 index 0000000..b72666a --- /dev/null +++ b/ckanext/versions/tests/test_dataset_version_action.py @@ -0,0 +1,380 @@ +import pytest + +from ckan import model +from ckan.plugins import toolkit +from ckan.tests import factories +from ckanext.versions.logic.dataset_version_action import ( + dataset_version_create, + dataset_has_versions, + get_activity_id_from_dataset_version_name, + activity_dataset_show, + dataset_version_latest, dataset_version_list +) +from ckanext.versions.model import Version +from ckanext.versions.tests import get_context, assert_version, create_version, restore_version + + +@pytest.mark.usefixtures('clean_db', 'versions_setup') +class TestDatasetVersion(object): + + def test_dataset_version_create_should_create_version(self, org_admin, test_dataset): + version_name = "Test Version 1.0" + version_notes = "Some details about the version" + version = dataset_version_create( + get_context(org_admin), + { + "dataset_id": test_dataset['id'], + "name": version_name, + "notes": version_notes + } + ) + checks = {'package_id': test_dataset['id'], + 'resource_id': None, + 'notes': version_notes, + 'name': version_name, + 'creator_user_id': org_admin['id']} + + assert_version(version, checks) + + def test_dataset_create_version_accepts_dataset_name(self, test_dataset, org_editor): + version = create_version(test_dataset['name'], org_editor) + assert test_dataset['id'] == version['package_id'] + + def test_dataset_create_should_create_version_for_given_activity(self, test_dataset, org_editor): + version_name = "V1.0" + context = get_context(org_editor) + for i in range(2): + toolkit.get_action('package_patch')( + context, + { + "id": test_dataset['id'], + "name": "updated-name%s" % i + } + ) + activities = toolkit.get_action('package_activity_list')( + context, + { + 'id': test_dataset['id'] + } + ) + past_activity_id = activities[-1]['id'] + + version = dataset_version_create( + context, + { + "dataset_id": test_dataset['id'], + "name": version_name, + "activity_id": past_activity_id + } + ) + + checks = {'package_id': test_dataset['id'], + 'resource_id': None, + 'notes': None, + 'name': version_name, + 'activity_id': past_activity_id, + 'creator_user_id': org_editor['id']} + + assert_version(version, checks) + + def test_dataset_create_should_fail_when_incorrect_activity_id(self, test_dataset, org_editor): + context = get_context(org_editor) + with pytest.raises(toolkit.ObjectNotFound, match="Activity not found"): + dataset_version_create( + context, + { + "dataset_id": test_dataset['id'], + "name": "V1.0", + "activity_id": "fake-activity-id" + } + ) + + def test_dataset_version_create_fails_if_version_for_activity_exists(self, test_dataset, org_editor): + create_version(test_dataset['id'], org_editor, version_name="Version1") + with pytest.raises(toolkit.ValidationError, match="Version already exists for this activity"): + create_version(test_dataset['id'], org_editor, version_name="Version2") + + @pytest.mark.parametrize("user_role, can_create_version", [ + ('admin', True), + ('editor', True), + ('member', False), + ]) + def test_dataset_version_create_auth(self, test_organization, test_dataset, user_role, can_create_version): + for user in test_organization['users']: + if user['capacity'] == user_role: + if can_create_version: + create_version(test_dataset['id'], user) + return + else: + with pytest.raises(toolkit.NotAuthorized): + create_version(test_dataset['id'], user) + return + pytest.fail("Couldn't find user with required role %s", user_role) + + def test_dataset_version_create_should_not_create_version_with_same_name(self, test_dataset, org_editor): + version_name = "Not unique name" + create_version(test_dataset['id'], org_editor, version_name=version_name) + with pytest.raises(toolkit.ValidationError): + create_version(test_dataset['id'], org_editor, version_name=version_name) + + def test_dataset_version_create_should_fail_if_dataset_not_exists(self, org_editor): + with pytest.raises(toolkit.ObjectNotFound, match="Dataset not found"): + create_version('fake_dataset_id', org_editor) + + def test_dataset_version_create_returns_valid_activity_id(self, test_organization, org_editor): + old_name = "initial-name" + dataset = factories.Dataset(name=old_name, owner_org=test_organization['id']) + version = create_version(dataset['id'], org_editor) + context = get_context(org_editor) + toolkit.get_action('package_patch')( + context, + { + "id": dataset['id'], + "name": "updated-name" + } + ) + + old_dataset = toolkit.get_action('activity_data_show')( + context, + {'id': version['activity_id']} + )['package'] + + assert old_dataset['name'] == old_name + assert old_dataset['id'] == dataset['id'] + + @pytest.mark.parametrize("object_ref", [ + "id", + "name" + ]) + def test_dataset_version_restore_retrives_dataset(self, test_version, test_dataset, org_editor, object_ref): + context = get_context(org_editor) + dataset_title = test_dataset['title'] + dataset_id = test_dataset['id'] + dataset_ref = test_dataset[object_ref] + version_ref = test_version[object_ref] + toolkit.get_action('package_patch')( + context, + { + "id": dataset_id, + "title": "updated-title" + } + ) + restored_dataset = restore_version(dataset_ref, version_ref, org_editor) + + assert dataset_title == restored_dataset['title'], "restored dataset title does not match" + assert dataset_id == restored_dataset['id'], "restored dataset has different id" + + def test_dataset_version_restore_creates_new_version_after_restore(self, test_version, test_dataset, org_editor): + restore_version(test_dataset['id'], test_version['id'], org_editor) + latest_version = model.Session.query(Version). \ + filter(Version.package_id == test_dataset['id']). \ + order_by(Version.created.desc()).first() + + assert latest_version.id != test_version['id'], "restore action should create a new version" + + def test_dataset_version_restore_fails_if_dataset_not_found(self, test_version, test_dataset, org_editor): + with pytest.raises(toolkit.ObjectNotFound, match="Dataset not found"): + restore_version('fake-dataset-id', test_version['id'], org_editor) + + def test_dataset_version_restore_fails_if_dataset_not_have_version(self, test_version, test_dataset, org_editor): + with pytest.raises(toolkit.ObjectNotFound, match="Version not found"): + restore_version(test_dataset['id'], 'fake-version-id', org_editor) + + @pytest.mark.parametrize("user_role, can_create_version", [ + ('admin', True), + ('editor', True), + ('member', False), + ]) + def test_dataset_version_restore_auth(self, test_version, test_organization, test_dataset, user_role, can_create_version): + for user in test_organization['users']: + if user['capacity'] == user_role: + if can_create_version: + restore_version(test_dataset['id'], test_version['id'], user) + return + else: + with pytest.raises(toolkit.NotAuthorized): + restore_version(test_dataset['id'], test_version['id'], user) + return + pytest.fail("Couldn't find user with required role %s", user_role) + + def test_dataset_version_list_return_all_version(self, test_dataset, org_editor): + context = get_context(org_editor) + version1 = create_version(test_dataset['id'], org_editor, version_name="Version1") + toolkit.get_action('package_patch')( + context, + { + "id": test_dataset['id'], + "name": "updated-name" + } + ) + version2 = create_version(test_dataset['id'], org_editor, version_name="Version2") + + version_list = dataset_version_list( + context, + { + 'dataset_id': test_dataset['id'] + } + ) + version_ids = [v['id'] for v in version_list] + assert 2 == len(version_ids), "only 2 versions created for dataset" + assert version1['id'] in version_ids + assert version2['id'] in version_ids + + def test_dataset_version_list_return_version_in_create_time_desc_order(self, test_dataset, org_editor): + context = get_context(org_editor) + version1 = create_version(test_dataset['id'], org_editor, version_name="Version1") + toolkit.get_action('package_patch')( + context, + { + "id": test_dataset['id'], + "title": "New Title" + } + ) + version2 = create_version(test_dataset['id'], org_editor, version_name="Version2") + + version_list = dataset_version_list( + context, + { + 'dataset_id': test_dataset['id'] + } + ) + assert version2['id'] == version_list[0]['id'], "version2 should be first as newest" + assert version1['id'] == version_list[-1]['id'], "version1 should be last as oldest" + + def test_dataset_version_should_fail_if_dataset_not_exists(self, org_editor): + context = get_context(org_editor) + with pytest.raises(toolkit.ObjectNotFound, match="Dataset not found"): + dataset_version_list( + context, + { + 'dataset_id': 'fake-dataset-id' + } + ) + + def test_dataset_version_should_return_empty_list_if_dataset_no_versions(self, test_dataset, org_editor): + context = get_context(org_editor) + versions_list = dataset_version_list( + context, + { + 'dataset_id': test_dataset['id'] + } + ) + assert [] == versions_list + + def test_dataset_version_latest_show_latest_version(self, test_dataset, org_editor): + context = get_context(org_editor) + version1 = create_version(test_dataset['id'], org_editor, version_name="Version1") + toolkit.get_action('package_patch')( + context, + { + "id": test_dataset['id'], + "name": "updated-name" + } + ) + version2 = create_version(test_dataset['id'], org_editor, version_name="Version2") + + latest_version = dataset_version_latest( + context, + { + 'dataset_id': test_dataset['id'] + } + ) + + assert version1['id'] != latest_version['id'] + assert_version(latest_version, {'id': version2['id'], 'name': version2['name']}) + + def test_dataset_version_latest_raises_when_dataset_not_found(self, org_editor): + context = get_context(org_editor) + with pytest.raises(toolkit.ObjectNotFound, match="Dataset not found"): + dataset_version_latest( + context, + { + 'dataset_id': 'fake-dataset-id' + } + ) + + def test_dataset_version_latest_raises_when_dataset_has_no_versions(self, test_dataset, org_editor): + context = get_context(org_editor) + with pytest.raises(toolkit.ObjectNotFound, match="Versions not found for this dataset"): + dataset_version_latest( + context, + { + 'dataset_id': test_dataset['id'] + } + ) + + def test_activity_dataset_show_returns_correct_dataset(self, test_dataset, org_editor): + version = create_version(test_dataset['id'], org_editor) + context = get_context(org_editor) + updated_name = "updated-name" + new_dataset = toolkit.get_action('package_patch')( + context, + { + "id": test_dataset['id'], + "name": updated_name + } + ) + + old_dataset = activity_dataset_show( + context, + { + 'dataset_id': new_dataset['id'], + 'activity_id': version['activity_id'] + } + ) + + assert test_dataset['name'] == old_dataset['name'] + assert new_dataset['id'] == old_dataset['id'] + + def test_activity_dataset_show_fails_if_no_dataset_in_activity(self, test_dataset, org_editor): + version = create_version(test_dataset['id'], org_editor) + context = get_context(org_editor) + with pytest.raises(toolkit.ObjectNotFound, match='Dataset not found in the activity object.'): + activity_dataset_show( + context, + { + 'dataset_id': 'fake-dataset-id', + 'activity_id': version['activity_id'] + } + ) + + def test_get_activity_id_from_dataset_version_returns_correct(self, test_dataset, org_editor): + version1 = create_version(test_dataset['id'], org_editor, version_name="Version1") + expected_activity_id = version1['activity_id'] + + context = get_context(org_editor) + actual_activity_id = get_activity_id_from_dataset_version_name( + context, + { + 'dataset_id': test_dataset['id'], + 'version': version1['name'] + } + ) + + assert expected_activity_id == actual_activity_id + + def test_get_activity_id_from_dataset_version_raises_not_found(self, test_dataset, org_editor): + context = get_context(org_editor) + with pytest.raises(toolkit.ObjectNotFound, match="Version not found in the dataset"): + get_activity_id_from_dataset_version_name( + context, + { + 'dataset_id': test_dataset['id'], + 'version': 'Fake version name' + } + ) + + def test_new_dataset_has_no_versions(self, test_dataset, org_editor): + context = get_context(org_editor) + assert False is dataset_has_versions( + context, + {'dataset_id': test_dataset['id']} + ) + + def test_dataset_has_versions_after_one_created(self, test_dataset, org_editor): + context = get_context(org_editor) + create_version(test_dataset['id'], org_editor) + assert True is dataset_has_versions( + context, + {'dataset_id': test_dataset['id']} + ) diff --git a/ckanext/versions/tests/test_helpers.py b/ckanext/versions/tests/test_helpers.py index e820f86..1426c5b 100644 --- a/ckanext/versions/tests/test_helpers.py +++ b/ckanext/versions/tests/test_helpers.py @@ -1,6 +1,11 @@ +import mock import pytest +from ckan.plugins import toolkit from ckanext.versions import helpers +from ckanext.versions.helpers import dataset_version_for_activity_id +from ckanext.versions.tests import get_context + @pytest.mark.ckan_config('ckan.site_url', 'http://ckan:5000') def test_version_download_url_without_filename(): @@ -24,3 +29,34 @@ def test_version_download_url_with_external_url(): download_url = helpers.download_url(url, '') assert download_url == url + + +@pytest.mark.usefixtures('clean_db', 'versions_setup', 'with_request_context') +def test_dataset_version_for_activity_id_return_version(org_editor, test_dataset, test_version): + activity_id = test_version['activity_id'] + + with mock.patch('ckanext.versions.helpers.toolkit.g', user=org_editor['name']): + actual_version = dataset_version_for_activity_id(test_dataset['id'], activity_id) + assert test_version['id'] == actual_version['id'] + + +@pytest.mark.usefixtures('clean_db', 'versions_setup', 'with_request_context') +def test_dataset_version_for_activity_returns_none_if_no_version(app, test_dataset, org_editor): + context = get_context(org_editor) + toolkit.get_action('package_patch')( + context, + { + "id": test_dataset['id'], + "name": "updated-name" + } + ) + activities = toolkit.get_action('package_activity_list')( + context, + { + 'id': test_dataset['id'] + } + ) + activity_id = activities[0]['id'] + with mock.patch('ckanext.versions.helpers.toolkit.g', user=org_editor['name']): + version = dataset_version_for_activity_id(test_dataset['id'], activity_id) + assert None is version