storage/domain_dao_test.py

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()