diff --git a/tests_new/integration_tests/modules/catalog/conftest.py b/tests_new/integration_tests/modules/catalog/conftest.py index bcbd2f7bb..fa93399fe 100644 --- a/tests_new/integration_tests/modules/catalog/conftest.py +++ b/tests_new/integration_tests/modules/catalog/conftest.py @@ -7,6 +7,7 @@ create_category, delete_category, list_glossary_associations, + approve_term_association, ) from integration_tests.modules.s3_datasets.queries import update_dataset @@ -17,11 +18,11 @@ @pytest.fixture(scope='function') -def glossary1(client1, group1): +def glossary1(client1, group1, session_id): glos = None try: glos = create_glossary( - client1, name='glossary1', group=group1, read_me='Glossary created for integration testing' + client1, name=f'glossary1-{session_id}', group=group1, read_me='Glossary created for integration testing' ) yield glos finally: @@ -69,65 +70,44 @@ def category_term1(client1, group1, category1): """ -Session glossary elements needed if using associations - WARNING! Associations are applied to the S3_Datasets module Glossaries can only be tested if the S3_datasets module is enabled in the deployment used for testing! """ -@pytest.fixture(scope='session') -def session_glossary1(client1, group1): - glos = None - try: - glos = create_glossary( - client1, name='Sesssion glossary1', group=group1, read_me='Glossary created for integration testing' - ) - yield glos - finally: - if glos: - delete_glossary(client1, node_uri=glos.nodeUri) - - -@pytest.fixture(scope='session') -def session_glossary_term1(client1, group1, session_glossary1): - term = None +@pytest.fixture(scope='function') +def dataset_association1(client1, group1, glossary1, glossary_term1, session_s3_dataset1): + ds_association = None try: - term = create_term( + update_dataset( client1, - name='Session glos_term1', - parent_uri=session_glossary1.nodeUri, - read_me='Term created for integration testing', + datasetUri=session_s3_dataset1.datasetUri, + input={ + 'terms': [glossary_term1.nodeUri], + 'KmsAlias': session_s3_dataset1.KmsAlias, + }, ) - yield term + response = list_glossary_associations(client1, node_uri=glossary1.nodeUri) + ds_association = next( + (assoc for assoc in response.associations.nodes if assoc.targetUri == session_s3_dataset1.datasetUri), None + ) + yield ds_association finally: - if term: - delete_term(client1, node_uri=term.nodeUri) + if ds_association: + update_dataset( + client1, + datasetUri=session_s3_dataset1.datasetUri, + input={ + 'terms': [], + 'KmsAlias': session_s3_dataset1.KmsAlias, + }, + ) -@pytest.fixture(scope='session') -def dataset_association_with_glossary_term1( - client1, group1, session_glossary1, session_glossary_term1, session_s3_dataset1 -): - update_dataset( - client1, - datasetUri=session_s3_dataset1.datasetUri, - input={ - 'terms': [session_glossary_term1.nodeUri], - 'label': session_s3_dataset1.label, - 'description': session_s3_dataset1.description, - 'tags': session_s3_dataset1.tags, - 'stewards': session_s3_dataset1.stewards, - 'topics': session_s3_dataset1.topics, - 'confidentiality': session_s3_dataset1.confidentiality, - 'autoApprovalEnabled': False, - 'enableExpiration': False, - 'KmsAlias': session_s3_dataset1.KmsAlias, - }, - ) - response = list_glossary_associations(client1, node_uri=session_glossary1.nodeUri) - ds_association = next( - (assoc for assoc in response.associations.nodes if assoc.targetUri == session_s3_dataset1.datasetUri), None - ) - yield ds_association +@pytest.fixture(scope='function') +def approved_dataset_association1(client1, glossary1, dataset_association1): + approve_term_association(client1, link_uri=dataset_association1.linkUri) + response = list_glossary_associations(client1, node_uri=glossary1.nodeUri) + association = next((n for n in response.associations.nodes if n.linkUri == dataset_association1.linkUri), None) + yield association diff --git a/tests_new/integration_tests/modules/catalog/test_glossaries.py b/tests_new/integration_tests/modules/catalog/test_glossaries.py index b7636bb74..e5258eb97 100644 --- a/tests_new/integration_tests/modules/catalog/test_glossaries.py +++ b/tests_new/integration_tests/modules/catalog/test_glossaries.py @@ -22,15 +22,15 @@ ) -def test_create_glossary(client1, glossary1): +def test_create_glossary(client1, glossary1, session_id): assert_that(glossary1.nodeUri).is_not_none() - assert_that(glossary1.label).is_equal_to('glossary1') + assert_that(glossary1.label).is_equal_to(f'glossary1-{session_id}') assert_that(glossary1.readme).is_equal_to('Glossary created for integration testing') -def test_get_glossary(client1, glossary1, category1, glossary_term1, category_term1): +def test_get_glossary(client1, glossary1, category1, glossary_term1, category_term1, session_id): response = get_glossary(client1, node_uri=glossary1.nodeUri) - assert_that(response.label).is_equal_to('glossary1') + assert_that(response.label).is_equal_to(f'glossary1-{session_id}') assert_that(response.nodeUri).is_equal_to(glossary1.nodeUri) assert_that(response.stats).contains_entry(categories=1, terms=2) @@ -63,14 +63,12 @@ def test_get_glossary_get_tree(client1, glossary1, category1, glossary_term1, ca assert_that(c_term.path).is_equal_to(f'/{glossary1.nodeUri}/{category1.nodeUri}/{category_term1.nodeUri}') -def test_get_glossary_list_associations( - client1, session_glossary1, session_glossary_term1, dataset_association_with_glossary_term1 -): - response = list_glossary_associations(client1, node_uri=session_glossary1.nodeUri) +def test_get_glossary_list_associations(client1, glossary1, glossary_term1, dataset_association1): + response = list_glossary_associations(client1, node_uri=glossary1.nodeUri) assert_that(response.associations.count).is_equal_to(1) ass = response.associations.nodes[0] assert_that(ass.linkUri).is_not_none() - assert_that(ass.term.nodeUri).is_equal_to(session_glossary_term1.nodeUri) + assert_that(ass.term.nodeUri).is_equal_to(glossary_term1.nodeUri) def test_list_glossaries(client1, glossary1, category1, glossary_term1, category_term1): @@ -100,8 +98,11 @@ def test_update_glossary(client1, group1, glossary1, session_id): def test_delete_glossary(client1, group1): glos = create_glossary(client1, name='glossary1', group=group1, read_me='Glossary created for integration testing') + number_glossaries_before_delete = list_glossaries(client1).count response = delete_glossary(client1, glos.nodeUri) assert_that(response).is_true() + response = list_glossaries(client1) + assert_that(response.count).is_equal_to(number_glossaries_before_delete - 1) def test_delete_glossary_with_categories_and_terms(client1, group1): @@ -135,8 +136,11 @@ def test_delete_category(client1, glossary1): category = create_category( client1, name='glossary1', parent_uri=glossary1.nodeUri, read_me='Category created for integration testing' ) + number_categories_before_delete = get_glossary(client1, node_uri=glossary1.nodeUri).stats.categories response = delete_category(client1, category.nodeUri) assert_that(response).is_true() + response = get_glossary(client1, node_uri=glossary1.nodeUri) + assert_that(response.stats.categories).is_equal_to(number_categories_before_delete - 1) def test_delete_category_with_terms(client1, glossary1): @@ -170,32 +174,41 @@ def test_update_term(client1, glossary_term1, session_id): assert_that(response.readme).is_equal_to(f'UPDATED: {session_id} Glossary term created for integration testing') -def test_delete_term(client1, group1, category1): +def test_delete_term(client1, group1, category1, glossary1): term = create_term( client1, name='toDelete', parent_uri=category1.nodeUri, read_me='Term created for integration testing' ) + number_terms_before_delete = get_glossary(client1, node_uri=glossary1.nodeUri).stats.terms response = delete_term(client1, node_uri=term.nodeUri) assert_that(response).is_true() + response = get_glossary(client1, node_uri=glossary1.nodeUri) + assert_that(response.stats.terms).is_equal_to(number_terms_before_delete - 1) -def test_approve_term_association( - client1, dataset_association_with_glossary_term1, session_glossary1, session_glossary_term1 -): - response = approve_term_association(client1, link_uri=dataset_association_with_glossary_term1.linkUri) - assert_that(response).is_true() - response = list_glossary_associations(client1, node_uri=session_glossary1.nodeUri) - association = next( - (n for n in response.associations.nodes if n.linkUri == dataset_association_with_glossary_term1.linkUri), None - ) - assert_that(association.approvedBySteward).is_equal_to(True) +def test_approve_term_association_unathorized(client2, dataset_association1): + assert_that(approve_term_association).raises(GqlError).when_called_with( + client2, link_uri=dataset_association1.linkUri + ).contains('UnauthorizedOperation', 'ASSOCIATE_GLOSSARY_TERM') + + +def test_approve_term_association(approved_dataset_association1, dataset_association1): + assert_that(approved_dataset_association1.linkUri).is_equal_to(dataset_association1.linkUri) + assert_that(approved_dataset_association1.approvedBySteward).is_equal_to(True) + + +def test_dismiss_term_association_unathorized(client2, approved_dataset_association1): + assert_that(dismiss_term_association).raises(GqlError).when_called_with( + client2, link_uri=approved_dataset_association1.linkUri + ).contains('UnauthorizedOperation', 'ASSOCIATE_GLOSSARY_TERM') -def test_dismiss_term_association(client1, session_glossary1, dataset_association_with_glossary_term1): - response = dismiss_term_association(client1, link_uri=dataset_association_with_glossary_term1.linkUri) +def test_dismiss_term_association(client1, glossary1, approved_dataset_association1): + assert_that(approved_dataset_association1.approvedBySteward).is_equal_to(True) + response = dismiss_term_association(client1, link_uri=approved_dataset_association1.linkUri) assert_that(response).is_true() - response = list_glossary_associations(client1, node_uri=session_glossary1.nodeUri) + response = list_glossary_associations(client1, node_uri=glossary1.nodeUri) association = next( - (n for n in response.associations.nodes if n.linkUri == dataset_association_with_glossary_term1.linkUri), None + (n for n in response.associations.nodes if n.linkUri == approved_dataset_association1.linkUri), None ) assert_that(association.approvedBySteward).is_equal_to(False)