Spaces:
Sleeping
Sleeping
| import logging | |
| import unittest | |
| import uuid | |
| import os | |
| from storage.domain_dao import InMemDomainDAO | |
| from domain.domain_protocol_test import TimestampTestD | |
| class InMemDomainDAOTest(unittest.TestCase): | |
| def setUp(self): | |
| self.in_mem_dao: InMemDomainDAO = InMemDomainDAO[TimestampTestD]() | |
| self.timestamp_d: TimestampTestD = TimestampTestD(nanos=1) | |
| def test_insert_domain_obj(self): | |
| self.in_mem_dao.insert([self.timestamp_d]) | |
| expected_map = {self.timestamp_d.id: self.timestamp_d} | |
| self.assertEqual(self.in_mem_dao._id_to_domain_obj, expected_map) | |
| def test_insert_domain_obj_raise_on_duplicate_id_in_db(self): | |
| self.in_mem_dao.insert([self.timestamp_d]) | |
| with self.assertRaises(ValueError) as context: | |
| self.in_mem_dao.insert([self.timestamp_d]) | |
| self.assertIn("Duplicate ids exist in DB", str(context.exception)) | |
| def test_insert_domain_obj_raise_on_duplicate_id_arguements(self): | |
| with self.assertRaises(ValueError) as context: | |
| self.in_mem_dao.insert([self.timestamp_d, self.timestamp_d]) | |
| self.assertIn("Duplicate IDs exist within incoming domain_objs", str(context.exception)) | |
| def test_read_by_id(self): | |
| self.in_mem_dao.insert([self.timestamp_d]) | |
| timestamp_d = self.in_mem_dao.read_by_id(self.timestamp_d.id) | |
| self.assertEqual(timestamp_d, self.timestamp_d) | |
| def test_read_by_id_raise_not_found(self): | |
| with self.assertRaises(ValueError): | |
| self.in_mem_dao.read_by_id(self.timestamp_d.id) | |
| def test_read_all(self): | |
| timestamp_d_b = TimestampTestD(2) | |
| self.in_mem_dao.insert([self.timestamp_d, timestamp_d_b]) | |
| expected_timestamps = {self.timestamp_d, timestamp_d_b} | |
| self.assertEqual(expected_timestamps, self.in_mem_dao.read_all()) | |
| def test_load_from_file(self): | |
| file_path = f".bin/{uuid.uuid4()}.jsonl" | |
| with open(file_path, 'w') as f: | |
| f.write(self.timestamp_d.to_json() + '\n') | |
| dao = InMemDomainDAO[TimestampTestD].load_from_file(file_path, TimestampTestD) | |
| os.remove(file_path) | |
| self.assertEqual({self.timestamp_d}, dao.read_all()) | |
| def test_load_from_file_fail_not_found(self): | |
| with self.assertRaises(ValueError): | |
| _ = InMemDomainDAO[TimestampTestD].load_from_file("file_path", TimestampTestD) | |
| def test_save_to_file(self): | |
| file_path = f".bin/{uuid.uuid4()}.jsonl" | |
| self.in_mem_dao.insert([self.timestamp_d]) | |
| self.in_mem_dao.save_to_file(file_path) | |
| created_dao = InMemDomainDAO[TimestampTestD].load_from_file(file_path, TimestampTestD) | |
| os.remove(file_path) | |
| self.assertEqual(self.in_mem_dao.read_all(), created_dao.read_all()) | |
| #TODO: Add test for CacheDomainDAO | |
| if __name__ == '__main__': | |
| logging.basicConfig(level=logging.DEBUG) | |
| unittest.main() | |