test_update_cache.py

#!/usr/bin/env python
# encoding: utf-8
#
# ** header v3.0
# This file is a part of the CaosDB Project.
#
# Copyright (C) 2020 Henrik tom Wörden
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Affero General Public License as
# published by the Free Software Foundation, either version 3 of the
# License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU Affero General Public License for more details.
#
# You should have received a copy of the GNU Affero General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
#
# ** end header

import os
import unittest
from copy import deepcopy
from tempfile import NamedTemporaryFile

import caosdb as db
from caosadvancedtools.cache import UpdateCache, get_pretty_xml


class CacheTest(unittest.TestCase):
    def get_cont(self, *args):
        ent = db.Record()
        ent2 = db.Record()
        ent2.add_parent(name="Experiment")
        ent2.add_property("lol", ent)
        c = db.Container()
        c.extend([ent, ent2])

        return c

    def setUp(self):
        self.cache = UpdateCache(db_file=NamedTemporaryFile(delete=False).name)
        self.cache.create_cache()
        self.run_id = "235234"

    def test_insert(self):
        c = self.get_cont()
        # we do not want to check database in this unit test and thus replace
        # the lookup
        UpdateCache.get_previous_version = self.get_cont
        c[0].add_property("hallo", "21235")
        c[1].add_property("hallo", "235")
        self.cache.insert(c, self.run_id)

        saved_stuff = self.cache.get_updates(self.run_id)
        assert len(saved_stuff) == 1
        # old version
        assert saved_stuff[0][2] == get_pretty_xml(self.get_cont())
        # new version
        assert saved_stuff[0][3] == get_pretty_xml(c)
        assert len(self.cache.get_updates("sldkfjsldfjsldjf")) == 0

    def tearDown(self):
        os.remove(self.cache.db_file)