From 0bd50ae7fc2f4d03d5e2da547d9de39ddb294653 Mon Sep 17 00:00:00 2001 From: Robert Bartel Date: Thu, 25 Jan 2024 15:19:37 -0500 Subject: [PATCH] Update unit tests for DatasetUser functionality. --- .../core/dmod/test/test_dataset_manager.py | 205 +++++++++++++++++- 1 file changed, 199 insertions(+), 6 deletions(-) diff --git a/python/lib/core/dmod/test/test_dataset_manager.py b/python/lib/core/dmod/test/test_dataset_manager.py index 1d16b056c..3ad77add7 100644 --- a/python/lib/core/dmod/test/test_dataset_manager.py +++ b/python/lib/core/dmod/test/test_dataset_manager.py @@ -1,11 +1,22 @@ import unittest -from uuid import uuid4 -from ..core.dataset import Dataset, DatasetManager, DatasetType, InitialDataAdder +from uuid import uuid4, UUID +from ..core.dataset import Dataset, DatasetManager, DatasetType, DatasetUser, InitialDataAdder from ..core.meta_data import DataCategory, DataDomain, DataFormat, DiscreteRestriction, StandardDatasetIndex from datetime import datetime, timedelta from typing import Any, Dict, List, Optional, Set, Tuple, Union +class MockDatasetUser(DatasetUser): + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._uuid = uuid4() + + @property + def uuid(self) -> UUID: + return self._uuid + + class MockDatasetManager(DatasetManager): ACCESS_LOC = "location_0" @@ -20,7 +31,7 @@ def combine_partials_into_composite(self, dataset_name: str, item_name: str, com def create(self, name: str, category: DataCategory, domain: DataDomain, is_read_only: bool, initial_data: Optional[InitialDataAdder] = None, expires_on: Optional[datetime] = None) -> Dataset: """ - A trivial implementation, intended just to test that the convenience method ::method:`create_temporary` works. + Trivial implementation, intended to do just enough to test other methods provided by abstract superclass. Parameters ---------- @@ -35,8 +46,10 @@ def create(self, name: str, category: DataCategory, domain: DataDomain, is_read_ ------- """ - return Dataset(name=name, category=category, data_domain=domain, is_read_only=is_read_only, expires=expires_on, - access_location=self.ACCESS_LOC) + ds = Dataset(name=name, category=category, data_domain=domain, is_read_only=is_read_only, expires=expires_on, + manager=self, created_on=datetime.now(), access_location=self.ACCESS_LOC) + self._datasets[name] = ds + return ds def delete(self, dataset: Dataset, **kwargs) -> bool: pass @@ -65,6 +78,7 @@ def setUp(self): self._manager = MockDatasetManager() self._ex_datasets: Dict[int, Dataset] = dict() + self._ex_ds_users: Dict[int, DatasetUser] = dict() # Ex dataset 1 ex_idx = 1 @@ -72,6 +86,8 @@ def setUp(self): domain = DataDomain(data_format=DataFormat.NGEN_REALIZATION_CONFIG, discrete_restrictions=[disc_rest]) self._ex_datasets[ex_idx] = Dataset(name="test_ds_1", category=DataCategory.CONFIG, data_domain=domain, access_location=self._manager.ACCESS_LOC, is_read_only=False) + # Ex user 1 + self._ex_ds_users[ex_idx] = MockDatasetUser() def tearDown(self): pass @@ -136,4 +152,181 @@ def test_create_temporary_2_a(self): # Have to manipulate a bit here to make sure the UUIDs match, as create itself doesn't account for them directly dataset_w_create_temp.uuid = dataset_w_create.uuid - self.assertEqual(dataset_w_create, dataset_w_create_temp) \ No newline at end of file + self.assertEqual(dataset_w_create, dataset_w_create_temp) + + def test_link_user_1_a(self): + """ Test that linking a user puts the user in the set of linked users. """ + ex_idx = 1 + base_dataset = self._ex_datasets[ex_idx] + user = self._ex_ds_users[ex_idx] + + dataset = self._manager.create(name=base_dataset.name, category=base_dataset.category, + domain=base_dataset.data_domain, is_read_only=base_dataset.is_read_only) + self.assertFalse(user.uuid in self._manager.get_dataset_user_ids()) + + self._manager.link_user(user=user, dataset=dataset) + self.assertTrue(user.uuid in self._manager.get_dataset_user_ids()) + + def test_link_user_1_b(self): + """ Test that looking up an unlinked a user results in an exception. """ + ex_idx = 1 + base_dataset = self._ex_datasets[ex_idx] + user = self._ex_ds_users[ex_idx] + + self.assertFalse(base_dataset.name in self._manager.datasets) + dataset = self._manager.create(name=base_dataset.name, category=base_dataset.category, + domain=base_dataset.data_domain, is_read_only=base_dataset.is_read_only) + self.assertTrue(base_dataset.name in self._manager.datasets) + + self.assertRaises(ValueError, self._manager.get_dataset_user, user.uuid) + + def test_link_user_1_c(self): + """ Test that linking a user makes it possible to look them up. """ + ex_idx = 1 + base_dataset = self._ex_datasets[ex_idx] + user = self._ex_ds_users[ex_idx] + + dataset = self._manager.create(name=base_dataset.name, category=base_dataset.category, + domain=base_dataset.data_domain, is_read_only=base_dataset.is_read_only) + self.assertFalse(user.uuid in self._manager.get_dataset_user_ids()) + + self._manager.link_user(user=user, dataset=dataset) + lookup_user = self._manager.get_dataset_user(user.uuid) + self.assertEqual(user, lookup_user) + + def test_link_user_1_d(self): + """ Test that linking a user to two datasets associates them with both. """ + ex_idx = 1 + base_dataset = self._ex_datasets[ex_idx] + user = self._ex_ds_users[ex_idx] + + dataset_1 = self._manager.create(name=f"{base_dataset.name}_1", category=base_dataset.category, + domain=base_dataset.data_domain, is_read_only=base_dataset.is_read_only) + dataset_2 = self._manager.create(name=f"{base_dataset.name}_2", category=base_dataset.category, + domain=base_dataset.data_domain, is_read_only=base_dataset.is_read_only) + + self.assertFalse(user.uuid in self._manager.get_dataset_user_ids()) + + self._manager.link_user(user=user, dataset=dataset_1) + self._manager.link_user(user=user, dataset=dataset_2) + + self.assertNotEqual(dataset_1.name, dataset_2.name) + self.assertIn(user.uuid, self._manager.get_user_ids_for_dataset(dataset_1.name)) + self.assertIn(user.uuid, self._manager.get_user_ids_for_dataset(dataset_2.name)) + + def test_get_dataset_user_ids_0_a(self): + """ Test that a manager starts off with no dataset users by default. """ + self.assertEqual(len(self._manager.get_dataset_user_ids()), 0) + + def test_unlink_user_1_a(self): + """ Test that unlinking a user makes it no longer possible to look them up. """ + ex_idx = 1 + base_dataset = self._ex_datasets[ex_idx] + user = self._ex_ds_users[ex_idx] + + dataset = self._manager.create(name=base_dataset.name, category=base_dataset.category, + domain=base_dataset.data_domain, is_read_only=base_dataset.is_read_only) + self.assertFalse(user.uuid in self._manager.get_dataset_user_ids()) + + self._manager.link_user(user=user, dataset=dataset) + self.assertTrue(user.uuid in self._manager.get_dataset_user_ids()) + + self._manager.unlink_user(user=user, dataset=dataset) + self.assertFalse(user.uuid in self._manager.get_dataset_user_ids()) + + def test_unlink_user_1_b(self): + """ Test that unlinking a user takes them out of set for dataset. """ + ex_idx = 1 + base_dataset = self._ex_datasets[ex_idx] + user = self._ex_ds_users[ex_idx] + + dataset = self._manager.create(name=base_dataset.name, category=base_dataset.category, + domain=base_dataset.data_domain, is_read_only=base_dataset.is_read_only) + self.assertFalse(user.uuid in self._manager.get_user_ids_for_dataset(dataset_name=dataset.name)) + + self._manager.link_user(user=user, dataset=dataset) + self.assertTrue(user.uuid in self._manager.get_user_ids_for_dataset(dataset_name=dataset.name)) + + self._manager.unlink_user(user=user, dataset=dataset) + self.assertFalse(user.uuid in self._manager.get_user_ids_for_dataset(dataset_name=dataset.name)) + + def test_unlink_user_1_c(self): + """ Test that unlinking a user takes them out of linked set for that dataset, but not for others. """ + ex_idx = 1 + base_dataset = self._ex_datasets[ex_idx] + user = self._ex_ds_users[ex_idx] + + dataset_1 = self._manager.create(name=f"{base_dataset.name}_1", category=base_dataset.category, + domain=base_dataset.data_domain, is_read_only=base_dataset.is_read_only) + dataset_2 = self._manager.create(name=f"{base_dataset.name}_2", category=base_dataset.category, + domain=base_dataset.data_domain, is_read_only=base_dataset.is_read_only) + + self.assertFalse(user.uuid in self._manager.get_user_ids_for_dataset(dataset_name=dataset_1.name)) + self.assertFalse(user.uuid in self._manager.get_user_ids_for_dataset(dataset_name=dataset_2.name)) + + self._manager.link_user(user=user, dataset=dataset_1) + self._manager.link_user(user=user, dataset=dataset_2) + self.assertTrue(user.uuid in self._manager.get_user_ids_for_dataset(dataset_name=dataset_1.name)) + self.assertTrue(user.uuid in self._manager.get_user_ids_for_dataset(dataset_name=dataset_2.name)) + + self._manager.unlink_user(user=user, dataset=dataset_1) + self.assertFalse(user.uuid in self._manager.get_user_ids_for_dataset(dataset_name=dataset_1.name)) + self.assertTrue(user.uuid in self._manager.get_user_ids_for_dataset(dataset_name=dataset_2.name)) + + def test_unlink_user_1_d(self): + """ Test that unlinked user can still be found if they were linked to others. """ + ex_idx = 1 + base_dataset = self._ex_datasets[ex_idx] + user = self._ex_ds_users[ex_idx] + + dataset_1 = self._manager.create(name=f"{base_dataset.name}_1", category=base_dataset.category, + domain=base_dataset.data_domain, is_read_only=base_dataset.is_read_only) + dataset_2 = self._manager.create(name=f"{base_dataset.name}_2", category=base_dataset.category, + domain=base_dataset.data_domain, is_read_only=base_dataset.is_read_only) + + self.assertFalse(user.uuid in self._manager.get_dataset_user_ids()) + + self._manager.link_user(user=user, dataset=dataset_1) + self._manager.link_user(user=user, dataset=dataset_2) + + self.assertTrue(user.uuid in self._manager.get_dataset_user_ids()) + + self._manager.unlink_user(user=user, dataset=dataset_1) + self.assertTrue(user.uuid in self._manager.get_dataset_user_ids()) + + def test_unlink_user_1_e(self): + """ Test that unlinking returns false if the user was not linked. """ + ex_idx = 1 + base_dataset = self._ex_datasets[ex_idx] + user = self._ex_ds_users[ex_idx] + + dataset_1 = self._manager.create(name=f"{base_dataset.name}_1", category=base_dataset.category, + domain=base_dataset.data_domain, is_read_only=base_dataset.is_read_only) + + self.assertFalse(self._manager.unlink_user(user=user, dataset=dataset_1)) + + def test_unlink_user_1_f(self): + """ Test that unlinking returns true if the user was linked. """ + ex_idx = 1 + base_dataset = self._ex_datasets[ex_idx] + user = self._ex_ds_users[ex_idx] + + dataset_1 = self._manager.create(name=f"{base_dataset.name}_1", category=base_dataset.category, + domain=base_dataset.data_domain, is_read_only=base_dataset.is_read_only) + self._manager.link_user(user=user, dataset=dataset_1) + + self.assertTrue(self._manager.unlink_user(user=user, dataset=dataset_1)) + + def test_get_user_ids_for_dataset_1_a(self): + """ Test that function works after linking. """ + ex_idx = 1 + base_dataset = self._ex_datasets[ex_idx] + user = self._ex_ds_users[ex_idx] + + dataset = self._manager.create(name=base_dataset.name, category=base_dataset.category, + domain=base_dataset.data_domain, is_read_only=base_dataset.is_read_only) + self.assertFalse(user.uuid in self._manager.get_user_ids_for_dataset(dataset_name=dataset.name)) + + self._manager.link_user(user=user, dataset=dataset) + self.assertTrue(user.uuid in self._manager.get_user_ids_for_dataset(dataset_name=dataset.name)) +