test.erp5.testConflictResolution.py 4.62 KB
Newer Older
1 2
# -*- coding: utf-8 -*-
##############################################################################
3
# Copyright (c) 2010-2011 Nexedi SA and Contributors. All Rights Reserved.
4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29
#          Julien Muchembled <jm@nexedi.com>
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly advised to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# 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 General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA 02110-1301, USA.
#
##############################################################################

import unittest
30
import transaction
31 32
import ZODB
from ZODB.DemoStorage import DemoStorage
33
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
Jérome Perrin's avatar
Jérome Perrin committed
34
from six.moves import range
35

36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57

class TestType(unittest.TestCase):

  def setUp(self):
    self.db = ZODB.DB(DemoStorage())
    self.tm1 = transaction.TransactionManager()
    self.conn1 = self.db.open(transaction_manager=self.tm1)
    self.tm2 = transaction.TransactionManager()
    self.conn2 = self.db.open(transaction_manager=self.tm2)

  def tearDown(self):
    self.db.close()
    del self.tm1, self.conn1, self.tm2, self.conn2, self.db

  def testConflictFreeLog(self):
    from Products.ERP5Type.ConflictFree import ConflictFreeLog
    for t in (1, 404, 4), (500, 407, 3), (1000, 407, 2), (1500, 808, 1):
      self.conn1.root()['x'] = x1 = ConflictFreeLog(bucket_size=t[0])
      self.tm1.commit()
      self.tm2.begin()
      x2 = self.conn2.root()['x']
      x1.append(-1)
Jérome Perrin's avatar
Jérome Perrin committed
58
      x2.extend(range(200))
59 60 61 62
      self.tm1.commit()
      self.tm2.commit()
      self.tm1.begin()
      x1 += 401, 402
Jérome Perrin's avatar
Jérome Perrin committed
63
      x2.extend(range(200, 400))
64 65 66 67 68
      self.tm2.commit()
      x2.append(400)
      self.tm2.commit()
      self.tm1.commit()
      self.tm2.begin()
Jérome Perrin's avatar
Jérome Perrin committed
69
      expected = list(range(-1, 403))
70 71
      self.assertEqual(expected, list(x1))
      self.assertEqual(expected, list(x2))
72
      self.assertEqual(expected[::-1], list(reversed(x1)))
73 74 75 76 77 78 79 80 81 82 83 84 85
      self.assertEqual(len(expected), len(x1))
      self.assertEqual(len(expected), len(x2))
      x1 += x2
      self.assertEqual(t[1], len(x1._log))
      bucket_count = 1
      x = x2._next
      while x not in (x2, None):
        x = x._next
        bucket_count += 1
      self.assertEqual(t[2], bucket_count)


class TestERP5(ERP5TypeTestCase):
86 87

  def getTitle(self):
88
    return "Conflict Resolution: ERP5"
89 90

  def afterSetUp(self):
91
    other_node = self.getOtherZopeNodeList()[0]
92 93
    self.other_node = self.portal.portal_web_services.connect(
      "http://%s%s" % (other_node, self.portal.getPath()),
94
      self.manager_username, self.manager_password, 'xml-rpc')
95 96 97 98 99
    self.login()

  def testZODBCookie(self):
    cookie_name = self._testMethodName
    portal = self.portal
100
    cookie = portal.getCacheCookie(cookie_name) # 0
101
    self.commit()
102 103 104
    portal.newCacheCookie(cookie_name) # 1
    self.other_node.newCacheCookie(cookie_name) # 1
    self.other_node.newCacheCookie(cookie_name) # 2
105
    self.commit()# max(1, 2) + 1
106
    self.assertEqual(cookie + 3, portal.getCacheCookie(cookie_name))
107

108 109
  def testActiveProcess(self):
    active_process = self.portal.portal_activities.newActiveProcess()
110
    self.commit()
111
    remote = self.other_node
112 113
    remote.getId() # force storage sync of remote ZODB connection
                   # (see also Products.ERP5Type.patches.ZODBConnection)
114 115
    for id in active_process.getRelativeUrl().split('/'):
      remote = getattr(remote, id)
Jérome Perrin's avatar
Jérome Perrin committed
116
    for x in range(100):
117 118 119
      active_process.postResult(x)
    remote.testActiveProcess_postResult(100)
    try:
120
      self.commit()
121
    except:
122
      self.abort() # make failure more readable in case of regression
123
      raise
Jérome Perrin's avatar
Jérome Perrin committed
124
    self.assertEqual(sorted(active_process.getResultList()), list(range(101)))
125 126


127 128
def test_suite():
  suite = unittest.TestSuite()
129 130
  suite.addTest(unittest.makeSuite(TestType))
  suite.addTest(unittest.makeSuite(TestERP5))
131
  return suite