Source

roktar / tests / test_data_handling.py

import shutil
import unittest
from uuid import UUID
from datetime import datetime
from cPickle import loads

from roktar import Roktar
from roktar.configuration import Configuration
from roktar.configuration_backends.json_backend import JSONConfiguration
from roktar.backends.sqlite_backend import SQLiteBackend


class DataHandlingTests(unittest.TestCase):
    def setUp(self):
        configuration = Configuration(JSONConfiguration, "test_roktar")
        self.roktar = Roktar(configuration, SQLiteBackend)

    def tearDown(self):
        self.roktar.close()
        shutil.rmtree("test_roktar")

    def test_create_entity(self):
        entity_id = self.roktar.create_entity(dict(hello="world"))
        entity = self.roktar.get_entity(entity_id)
        self.assertEqual(
            entity, dict(hello="world"), "Entity has an unexpected value"
        )

        changes = list(self.roktar.get_changes(entity_id))
        self.assertEqual(len(changes), 1, "Entity didn't remember it's changes")

        change = changes[0]
        change_uuid = UUID(change["change_uuid"])
        change_time = datetime.fromtimestamp(change["change_time"])
        self.assertEqual(change_uuid.version, 4)
        self.assertEqual(change["parent_uuid"], None)
        self.assertLess(
            (datetime.now() - change_time).seconds, 10,
            "Bad timestamp"
        )
        self.assertEqual(change["entity_id"], entity_id)
        self.assertEqual(change["field_name"], "hello")
        self.assertEqual(loads(str(change["old_value"])), None)
        self.assertEqual(loads(str(change["new_value"])), "world")
        
        all_entities = list(self.roktar.get_all_entities())
        self.assertEqual(len(all_entities), 1)
        e_id, e = all_entities[0]
        self.assertEqual(e_id, entity_id)
        self.assertEqual(e, entity)