test_compound.py

# copyright 2015 LOGILAB S.A. (Paris, FRANCE), all rights reserved.
# contact http://www.logilab.fr -- mailto:contact@logilab.fr
#
# This program is free software: you can redistribute it and/or modify it under
# the terms of the GNU Lesser General Public License as published by the Free
# Software Foundation, either version 2.1 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 Lesser General Public License for more
# details.
#
# You should have received a copy of the GNU Lesser General Public License along
# with this program. If not, see <http://www.gnu.org/licenses/>.
"""cubicweb-compound tests"""

from cubicweb.devtools.testlib import CubicWebTC

from cubes.compound.entities import (CompositeGraph,
                                     copy_entity,
                                     structure_def)
from cubes.compound.views import CloneAction


def sort_keys(dic):
    return dict((k, sorted(v)) for k, v in dic.items())


class CompositeGraphTC(CubicWebTC):

    def test_iter_schema_notinschema(self):
        graph = CompositeGraph(self.schema)
        self.assertEqual(list(graph.iter_schema('Oups')), [])

    def test_iter_schema_singleton(self):
        graph = CompositeGraph(self.schema)
        structure = graph.structure('OnlineAccount')
        self.assertEqual(structure, {})

    def test_iter_schema_agent(self):
        graph = CompositeGraph(self.schema)
        structure = graph.structure('Agent')
        expected = {
            ('account', 'subject'): {
                'OnlineAccount': {}
            },
            ('biography', 'subject'): {
                'Biography': {
                    ('event', 'subject'): {
                        'Event': {},
                        'Anecdote': {},
                    },
                },
            },
            ('narrated_by', 'object'): {
                'Anecdote': {}
            },
        }
        self.assertEqual(structure, expected)

    def test_iter_entities_singleton(self):
        graph = CompositeGraph(self.schema)
        with self.admin_access.repo_cnx() as cnx:
            agent = cnx.create_entity('Agent')
            cnx.commit()
            self.assertEqual(list(graph.iter_entities(agent)), [])

    def test_iter_entities_full(self):
        graph = CompositeGraph(self.schema)
        with self.admin_access.repo_cnx() as cnx:
            agent = cnx.create_entity('Agent')
            account = cnx.create_entity('OnlineAccount', reverse_account=agent)
            bio = cnx.create_entity('Biography', reverse_biography=agent)
            event = cnx.create_entity('Event', reverse_event=bio)
            anecdote1 = cnx.create_entity('Anecdote', reverse_event=bio)
            anecdote2 = cnx.create_entity('Anecdote', reverse_event=bio, narrated_by=agent)
            cnx.commit()
            entities_graph = graph.iter_entities(agent)
            expected = [
                (account, ('account', 'object'), agent),
                (bio, ('biography', 'object'), agent),
                (event, ('event', 'object'), bio),
                (anecdote1, ('event', 'object'), bio),
                (anecdote2, ('event', 'object'), bio),
                (anecdote2, ('narrated_by', 'subject'), agent),
            ]
            self.assertCountEqual(entities_graph, expected)


class ContainerStructureTC(CubicWebTC):

    def test_definition(self):
        container = sort_keys(structure_def(self.schema, 'Agent'))
        expected = {
            'OnlineAccount': [('account', 'object')],
            'Biography': [('biography', 'object')],
            'Event': [('event', 'object')],
            'Anecdote': [('event', 'object'), ('narrated_by', 'subject')],
        }
        self.assertEqual(container, expected)

    def test_adapters(self):
        entity = self.vreg['etypes'].etype_class('Agent')(self)
        self.assertIsNotNone(entity.cw_adapt_to('IContainer'))
        self.assertIsNone(entity.cw_adapt_to('IContained'))
        entity = self.vreg['etypes'].etype_class('OnlineAccount')(self)
        self.assertIsNotNone(entity.cw_adapt_to('IContained'))
        entity = self.vreg['etypes'].etype_class('Group')(self)
        self.assertIsNone(entity.cw_adapt_to('IContained'))


def one(cnx, etype, **kwargs):
    return cnx.find(etype, **kwargs).one()


class ContainerAPITC(CubicWebTC):

    def setup_database(self):
        with self.admin_access.repo_cnx() as cnx:
            agent = cnx.create_entity('Agent')
            cnx.create_entity('OnlineAccount', reverse_account=agent)
            cnx.commit()

    def test_container(self):
        with self.admin_access.repo_cnx() as cnx:
            agent = one(cnx, 'Agent')
            self.assertEqual(agent.cw_adapt_to('IContainer').container, agent)
            account = one(cnx, 'OnlineAccount')
            icontained = account.cw_adapt_to('IContained')
            self.assertEqual(icontained.container, agent)

    def test_parent_relations_direct(self):
        with self.admin_access.repo_cnx() as cnx:
            account = one(cnx, 'OnlineAccount')
            icontained = account.cw_adapt_to('IContained')
            self.assertEqual(icontained.parent_relations,
                             set([('account', 'object')]))
            self.assertEqual(icontained.parent_relation(), ('account', 'object'))

    def test_parent_relations_indirect(self):
        with self.admin_access.repo_cnx() as cnx:
            agent = one(cnx, 'Agent')
            bio = cnx.create_entity('Biography', reverse_biography=agent)
            event = cnx.create_entity('Event', reverse_event=bio)
            cnx.commit()
            icontained = event.cw_adapt_to('IContained')
            self.assertEqual(icontained.parent_relations,
                             set([('event', 'object')]))
            self.assertEqual(icontained.parent_relation(), ('event', 'object'))


class CloneTC(CubicWebTC):

    def setup_database(self):
        with self.admin_access.repo_cnx() as cnx:
            self.create_user(cnx, u'georges')
            cnx.commit()

    def test_copy_entity(self):
        with self.admin_access.repo_cnx() as cnx:
            cnx.create_entity('Agent', name=u'bob')
            cnx.commit()
        with self.new_access(u'georges').repo_cnx() as cnx:
            bob = cnx.find('Agent').one()
            bob2 = copy_entity(bob)
            alice = copy_entity(bob, name=u'alice', knows=bob)
            cnx.commit()
            self.assertEqual(bob2.name, u'bob')
            self.assertEqual(alice.name, u'alice')
            self.assertEqual([x.eid for x in alice.knows], [bob.eid])
            self.assertEqual(alice.created_by[0].login, u'georges')
            self.assertEqual(bob2.created_by[0].login, u'georges')
            self.assertGreater(alice.creation_date, bob.creation_date)
            alice2 = copy_entity(alice)
            cnx.commit()
            alice2.cw_clear_all_caches()
            self.assertEqual(alice2.name, u'alice')
            self.assertEqual([x.eid for x in alice2.knows], [bob.eid])

    def test_clone_property(self):
        """Test .clone property of IClonableAdapter."""
        with self.admin_access.repo_cnx() as cnx:
            bob = cnx.create_entity('Agent', name=u'bob')
            cnx.commit()
            self.assertIsNone(bob.cw_adapt_to('IClonable').clone)
            bobby = cnx.create_entity('Agent', name=u'bobby', clone_of=bob)
            cnx.commit()
            self.assertEqual(bob.cw_adapt_to('IClonable').clone, bobby)
            self.assertIsNone(bobby.cw_adapt_to('IClonable').clone)

    def test_clone_simple(self):
        with self.admin_access.repo_cnx() as cnx:
            bob = cnx.create_entity('Agent', name=u'bob')
            alice_eid = cnx.create_entity(
                'Agent', name=u'alice', knows=bob).eid
            cnx.create_entity('OnlineAccount', reverse_account=bob)
            cnx.commit()
        with self.new_access(u'georges').repo_cnx() as cnx:
            bob = cnx.find('Agent', name=u'bob').one()
            clone = cnx.create_entity('Agent', name=u'bobby')
            bob.cw_adapt_to('IClonable').clone_into(clone)
            clone.cw_clear_all_caches()
            cnx.commit()
            self.assertEqual(clone.name, u'bobby')
            # Non-structural relation "knows" is just a shallow copy.
            self.assertEqual([x.eid for x in clone.knows], [alice_eid])
            # Recursive copy for structural relation "account".
            rset = cnx.execute(
                'Any CA WHERE C account CA, C eid %(clone)s,'
                '             NOT CA identity OA, O account OA,'
                '             O eid %(original)s',
                {'original': bob.eid, 'clone': clone.eid})
            self.assertTrue(rset)

    def test_clone_full(self):
        with self.admin_access.repo_cnx() as cnx:
            agent = cnx.create_entity('Agent', name=u'bob')
            account = cnx.create_entity('OnlineAccount', reverse_account=agent)
            bio = cnx.create_entity('Biography', reverse_biography=agent)
            event = cnx.create_entity('Event', reverse_event=bio)
            anecdote1 = cnx.create_entity('Anecdote', reverse_event=bio)
            anecdote2 = cnx.create_entity('Anecdote', reverse_event=bio, narrated_by=agent)
            cnx.commit()
        with self.new_access(u'georges').repo_cnx() as cnx:
            bob = cnx.find('Agent', name=u'bob').one()
            clone = cnx.create_entity('Agent', name=u'bobby')
            bob.cw_adapt_to('IClonable').clone_into(clone)
            clone.cw_clear_all_caches()
            cnx.commit()
            # Ensure all relation (to new entities) are present on the clone.
            rset = cnx.execute(
                'Any X WHERE X name "bobby", X account AC, X biography B,'
                '            B event E, E is Event, B event A, A narrated_by X')
            self.assertTrue(rset)

    def test_clone_action(self):
        with self.admin_access.web_request() as req:
            entity = req.create_entity('Agent', name=u'bob')
            req.cnx.commit()
            action = self.vreg['actions'].select('copy', req,
                                                 rset=entity.as_rset())
            self.assertIsInstance(action, CloneAction)

    def test_clone_hook(self):
        with self.admin_access.repo_cnx() as cnx:
            agent = cnx.create_entity('Agent', name=u'bob')
            account = cnx.create_entity('OnlineAccount', reverse_account=agent)
            bio = cnx.create_entity('Biography', reverse_biography=agent)
            event = cnx.create_entity('Event', reverse_event=bio)
            anecdote1 = cnx.create_entity('Anecdote', reverse_event=bio)
            anecdote2 = cnx.create_entity('Anecdote', reverse_event=bio, narrated_by=agent)
            cnx.commit()
        with self.new_access(u'georges').repo_cnx() as cnx:
            bob = cnx.find('Agent', name=u'bob').one()
            clone = cnx.create_entity('Agent', name=u'bobby', clone_of=bob)
            cnx.commit()
            clone.cw_clear_all_caches()
            # Ensure all relation (to new entities) are present on the clone.
            rset = cnx.execute(
                'Any X WHERE X name "bobby", X account AC, X biography B,'
                '            B event E, E is Event, B event A, A narrated_by X')
            self.assertTrue(rset)


if __name__ == '__main__':
    from logilab.common.testlib import unittest_main
    unittest_main()