sobjects/test/unittest_parsers.py
changeset 7399 972ed1843bd8
parent 7351 ed66f236715d
child 7463 a7daba4046db
--- a/sobjects/test/unittest_parsers.py	Thu May 19 10:53:11 2011 +0200
+++ b/sobjects/test/unittest_parsers.py	Thu May 19 10:53:17 2011 +0200
@@ -164,6 +164,45 @@
         stats = dfsource.pull_data(session, force=True, raise_on_error=True)
         self.assertEqual(stats['created'], set())
         self.assertEqual(len(stats['updated']), 2)
+        session.commit()
+
+        # test move to system source
+        self.sexecute('SET X cw_source S WHERE X eid %(x)s, S name "system"', {'x': email.eid})
+        self.commit()
+        rset = self.sexecute('EmailAddress X WHERE X address "syt@logilab.fr"')
+        self.assertEqual(len(rset), 1)
+        e = rset.get_entity(0, 0)
+        self.assertEqual(e.eid, email.eid)
+        self.assertEqual(e.cw_metainformation(), {'source': {'type': u'native', 'uri': u'system'},
+                                                  'type': 'EmailAddress',
+                                                  'extid': None})
+        self.assertEqual(e.cw_source[0].name, 'system')
+        self.assertEqual(e.reverse_use_email[0].login, 'sthenault')
+        self.commit()
+        # test everything is still fine after source synchronization
+        session.set_pool()
+        stats = dfsource.pull_data(session, force=True, raise_on_error=True)
+        rset = self.sexecute('EmailAddress X WHERE X address "syt@logilab.fr"')
+        self.assertEqual(len(rset), 1)
+        e = rset.get_entity(0, 0)
+        self.assertEqual(e.eid, email.eid)
+        self.assertEqual(e.cw_metainformation(), {'source': {'type': u'native', 'uri': u'system'},
+                                                  'type': 'EmailAddress',
+                                                  'extid': None})
+        self.assertEqual(e.cw_source[0].name, 'system')
+        self.assertEqual(e.reverse_use_email[0].login, 'sthenault')
+        session.commit()
+
+        # test delete entity
+        e.cw_delete()
+        self.commit()
+        # test everything is still fine after source synchronization
+        session.set_pool()
+        stats = dfsource.pull_data(session, force=True, raise_on_error=True)
+        rset = self.sexecute('EmailAddress X WHERE X address "syt@logilab.fr"')
+        self.assertEqual(len(rset), 0)
+        rset = self.sexecute('Any X WHERE X use_email E, X login "sthenault"')
+        self.assertEqual(len(rset), 0)
 
 if __name__ == '__main__':
     from logilab.common.testlib import unittest_main