testERP5Simulation.py 15.4 KB
Newer Older
1
# -*- coding: utf-8 -*-
2 3 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 30 31 32 33 34
##############################################################################
#
# Copyright (c) 2009 Nexedi SA and Contributors. All Rights Reserved.
#
# 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 adviced 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.
#
##############################################################################
"""
This test is experimental for new simulation implementation.
"""

import unittest
import transaction

35 36
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from Products.ERP5Type.tests.Sequence import SequenceList
37 38 39 40 41 42 43 44 45 46 47
from testPackingList import TestPackingList, TestPackingListMixin

class TestERP5SimulationMixin(TestPackingListMixin):
  def getBusinessTemplateList(self):
    return list(TestPackingListMixin.getBusinessTemplateList(self)) + \
           ['erp5_simulation',]

  def afterSetUp(self, quiet=1, run=1):
    TestPackingListMixin.afterSetUp(self, quiet, run)
    self.validateNewRules()

48 49 50 51 52
  def beforeTearDown(self):
    portal_rules = self.portal.portal_rules
    for rule in portal_rules.objectValues(portal_type='New Order Rule'):
      if rule.getValidationState() == 'validated':
        rule.invalidate()
53

54 55 56 57 58 59 60 61 62
class TestERP5Simulation(TestERP5SimulationMixin, ERP5TypeTestCase):
  run_all_test = 1
  quiet = 0

  def validateNewRules(self):
    # create a New Order Rule document.
    portal_rules = self.portal.portal_rules
    try:
      new_order_rule = filter(
63
        lambda x:x.title == 'New Simple Order Rule',
64 65 66
        portal_rules.objectValues(portal_type='New Order Rule'))[0]
    except IndexError:
      new_order_rule = portal_rules.newContent(
67
        title='New Simple Order Rule',
68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123
        portal_type='New Order Rule',
        reference='default_order_rule',
        version=2,
        )
      # create category divergence testers that is only used for matching
      for i in ('order',):
        new_order_rule.newContent(
          title='%s divergence tester' % i,
          portal_type='Category Membership Divergence Tester',
          tested_property=i,
          divergence_provider=0,
          matching_provider=1)
      # create float divergence testers
      for i in ('converted_quantity',):
        new_order_rule.newContent(
          title='%s divergence tester' % i,
          portal_type='Float Divergence Tester',
          tested_property=i,
          use_delivery_ratio=1,
          quantity_range_min=-1,
          quantity_range_max=2)
    if new_order_rule.getValidationState() != 'validated':
      new_order_rule.validate()

  def _modifyPackingListLineQuantity(self, sequence=None,
      sequence_list=None, delta=0.0):
    """
    Set a increased quantity on packing list lines
    """
    packing_list = sequence.get('packing_list')
    quantity = self.default_quantity + delta
    sequence.edit(line_quantity=quantity)
    for packing_list_line in packing_list.objectValues(
        portal_type=self.packing_list_line_portal_type):
      packing_list_line.edit(quantity=quantity)
    sequence.edit(last_delta=delta)

  def stepIncreasePackingListLineQuantity2(self, sequence=None,
      sequence_list=None, **kw):
    return self._modifyPackingListLineQuantity(sequence, sequence_list, 2.0)

  def stepDecreasePackingListLineQuantity1(self, sequence=None,
      sequence_list=None, **kw):
    return self._modifyPackingListLineQuantity(sequence, sequence_list, -1.0)

  def stepDecreasePackingListLineQuantity10(self, sequence=None,
      sequence_list=None, **kw):
    return self._modifyPackingListLineQuantity(sequence, sequence_list, -10.0)

  def stepSplitAndDeferPackingList(self, sequence=None, sequence_list=None, **kw):
    """
      Do the split and defer action
    """
    packing_list = sequence.get('packing_list')
    solver_tool = self.portal.portal_solvers
    solver_process = solver_tool.newSolverProcess(packing_list)
124
    sequence.edit(solver_process=solver_process)
125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173
    quantity_solver_decision = filter(
      lambda x:x.getCausalityValue().getTestedProperty()=='converted_quantity',
      solver_process.contentValues())[0]
    # use Quantity Split Solver.
    quantity_solver_decision.setSolverValue(self.portal.portal_types['Quantity Split Solver'])
    # configure for Quantity Split Solver.
    kw = {'delivery_solver':'FIFO',
          'start_date':packing_list.getStartDate() + 10}
    quantity_solver_decision.updateConfiguration(**kw)
    solver_process.buildTargetSolverList()
    solver_process.solve()
    # build split deliveries manually. XXX ad-hoc
    previous_tag = None
    for delivery_builder in packing_list.getBuilderList():
      this_builder_tag = '%s_split_%s' % (packing_list.getPath(),
                                          delivery_builder.getId())
      after_tag = []
      if previous_tag:
        after_tag.append(previous_tag)
      delivery_builder.activate(
        after_method_id=('solve',
                         'immediateReindexObject',
                         'recursiveImmediateReindexObject',), # XXX too brutal.
        after_tag=after_tag,
        ).build(explanation_uid=packing_list.getCausalityValue().getUid())

  def stepCheckPackingListSplitted(self, sequence=None, sequence_list=None, **kw):
    """
      Test if packing list was splitted
    """
    order = sequence.get('order')
    packing_list_list = order.getCausalityRelatedValueList(
                               portal_type=self.packing_list_portal_type)
    self.assertEquals(2,len(packing_list_list))
    packing_list1 = None
    packing_list2 = None
    for packing_list in packing_list_list:
      if packing_list.getUid() == sequence.get('packing_list').getUid():
        packing_list1 = packing_list
      else:
        packing_list2 = packing_list
    sequence.edit(new_packing_list=packing_list2)
    for line in packing_list1.objectValues(
          portal_type= self.packing_list_line_portal_type):
      self.assertEquals(self.default_quantity-10,line.getQuantity())
    for line in packing_list2.objectValues(
          portal_type= self.packing_list_line_portal_type):
      self.assertEquals(10,line.getQuantity())

174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195
  def _checkSolverState(self, sequence=None, sequence_list=None,
                        state='solved'):
    """
      Check if target solvers' state.
    """
    solver_process = sequence.get('solver_process')
    for solver in solver_process.objectValues(
      portal_type=self.portal.getPortalTargetSolverTypeList()):
      self.assertEquals(state, solver.getSolverState())

  def stepCheckSolverIsSolving(self, sequence=None, sequence_list=None, **kw):
    """
      Check if all target solvers have 'solving' state.
    """
    self._checkSolverState(sequence, sequence_list, 'solving')

  def stepCheckSolverIsSolved(self, sequence=None, sequence_list=None, **kw):
    """
      Check if all target solvers have 'solved' state.
    """
    self._checkSolverState(sequence, sequence_list, 'solved')

196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221
  def test_01_splitAndDefer(self, quiet=quiet, run=run_all_test):
    """
      Change the quantity on an delivery line, then
      see if the packing list is divergent and then
      split and defer the packing list
    """
    if not run: return
    sequence_list = SequenceList()

    # Test with a simply order without cell
    sequence_string = self.default_sequence + '\
                      stepIncreasePackingListLineQuantity2 \
                      stepCheckPackingListIsCalculating \
                      stepTic \
                      stepCheckPackingListIsNotDivergent \
                      stepCheckPackingListIsSolved \
                      stepDecreasePackingListLineQuantity1 \
                      stepCheckPackingListIsCalculating \
                      stepTic \
                      stepCheckPackingListIsNotDivergent \
                      stepCheckPackingListIsSolved \
                      stepDecreasePackingListLineQuantity10 \
                      stepCheckPackingListIsCalculating \
                      stepTic \
                      stepCheckPackingListIsDiverged \
                      stepSplitAndDeferPackingList \
222
                      stepCheckSolverIsSolving \
223 224 225
                      stepTic \
                      stepCheckPackingListSplitted \
                      stepCheckPackingListIsSolved \
226
                      stepCheckSolverIsSolved \
227 228 229 230 231 232
                      '
    sequence_list.addSequenceString(sequence_string)

    sequence_list.play(self, quiet=quiet)

class TestERP5SimulationPackingList(TestERP5SimulationMixin, TestPackingList):
233
  def validateNewRules(self):
234 235
    # create a New Order Rule document.
    portal_rules = self.portal.portal_rules
236 237 238 239 240
    try:
      new_order_rule = filter(
        lambda x:x.title == 'New Default Order Rule',
        portal_rules.objectValues(portal_type='New Order Rule'))[0]
    except IndexError:
241
      new_order_rule = portal_rules.newContent(
242
        title='New Default Order Rule',
243 244 245 246 247 248 249 250 251
        portal_type='New Order Rule',
        reference='default_order_rule',
        version=2,
        )
      # create category divergence testers
      for i in ('aggregate',
                'base_application',
                'base_contribution',
                'destination',
252 253 254 255 256 257 258 259 260 261
                'destination_account', # XXX-JPS - Needed ?
                'destination_function', # XXX-JPS - Needed ?
                'destination_project', # XXX-JPS - Needed ?
                'destination_section', 
                'price_currency', # XXX-JPS - Needed ?
                'source', 
                'source_account', # XXX-JPS - Needed ?
                'source_function', # XXX-JPS - Needed ?
                'source_project', # XXX-JPS - Needed ?
                'source_section',): 
262
        new_order_rule.newContent(
263
          title='%s divergence tester' % i,
264
          portal_type='Category Membership Divergence Tester',
265
          tested_property=i)
266
      # create category divergence testers that is also used for matching
267
      for i in ('resource',):
268
        new_order_rule.newContent(
269
          title='%s divergence tester' % i,
270 271 272
          portal_type='Category Membership Divergence Tester',
          tested_property=i,
          matching_provider=1)
273 274 275
      # create category divergence testers that is only used for matching
      for i in ('order',):
        new_order_rule.newContent(
276
          title='%s divergence tester' % i,
277 278 279 280
          portal_type='Category Membership Divergence Tester',
          tested_property=i,
          divergence_provider=0,
          matching_provider=1)
281
      # create variation divergence testers that is also used for matching
282
      for i in ('variation_property_dict',):
283
        # tested_property has no meaning for this tester.
284
        new_order_rule.newContent(
285
          title='%s divergence tester' % i,
286
          portal_type='Variation Divergence Tester',
287 288
          tested_property=i,
          matching_provider=1)
289 290 291 292
      # create datetime divergence testers
      for i in ('start_date',
                'stop_date',):
        new_order_rule.newContent(
293
          title='%s divergence tester' % i,
294 295 296 297 298 299
          portal_type='DateTime Divergence Tester',
          tested_property=i,
          quantity=0)
      # create float divergence testers
      for i in ('quantity',):
        new_order_rule.newContent(
300
          title='%s divergence tester' % i,
301
          portal_type='Float Divergence Tester', # XXX-JPS Quantity Divergence Tester ? (ie. quantity unit)
302 303 304
          tested_property=i,
          use_delivery_ratio=1,
          quantity=0)
305
    if new_order_rule.getValidationState() != 'validated':
306 307
      new_order_rule.validate()

308 309 310 311 312
  def stepAcceptDecisionQuantity(self,sequence=None, sequence_list=None, **kw):
    """
    Solve quantity divergence by using solver tool.
    """
    packing_list = sequence.get('packing_list')
313 314 315 316 317
    solver_tool = self.portal.portal_solvers
    solver_process = solver_tool.newSolverProcess(packing_list)
    quantity_solver_decision = filter(
      lambda x:x.getCausalityValue().getTestedProperty()=='quantity',
      solver_process.contentValues())[0]
318 319 320
    # use Quantity Accept Solver.
    quantity_solver_decision.setSolverValue(self.portal.portal_types['Quantity Accept Solver'])
    solver_process.buildTargetSolverList()
321
    solver_process.solve()
322 323 324
    # XXX-JPS We do not need the divergence message anymore.
    # since the divergence message == the divergence tester itself
    # with its title, description, tested property, etc.
325 326 327 328 329 330

  def stepAcceptDecisionResource(self,sequence=None, sequence_list=None, **kw):
    """
    Solve quantity divergence by using solver tool.
    """
    packing_list = sequence.get('packing_list')
331 332 333 334 335
    solver_tool = self.portal.portal_solvers
    solver_process = solver_tool.newSolverProcess(packing_list)
    resource_solver_decision = filter(
      lambda x:x.getCausalityValue().getTestedProperty()=='resource',
      solver_process.contentValues())[0]
336 337 338
    # use Resource Accept Solver.
    resource_solver_decision.setSolverValue(self.portal.portal_types['Resource Accept Solver'])
    solver_process.buildTargetSolverList()
339 340
    solver_process.solve()

341 342 343 344 345
  def stepSplitAndDeferPackingList(self, sequence=None, sequence_list=None, **kw):
    """
      Do the split and defer action
    """
    packing_list = sequence.get('packing_list')
346 347 348 349 350
    solver_tool = self.portal.portal_solvers
    solver_process = solver_tool.newSolverProcess(packing_list)
    quantity_solver_decision = filter(
      lambda x:x.getCausalityValue().getTestedProperty()=='quantity',
      solver_process.contentValues())[0]
351 352 353
    # use Quantity Split Solver.
    quantity_solver_decision.setSolverValue(self.portal.portal_types['Quantity Split Solver'])
    # configure for Quantity Split Solver.
354 355 356
    kw = {'delivery_solver':'FIFO',
          'start_date':self.datetime + 15,
          'stop_date':self.datetime + 25}
357 358
    quantity_solver_decision.updateConfiguration(**kw)
    solver_process.buildTargetSolverList()
359
    solver_process.solve()
360 361 362 363 364 365 366 367 368 369 370 371 372 373
    # build split deliveries manually. XXX ad-hoc
    previous_tag = None
    for delivery_builder in packing_list.getBuilderList():
      this_builder_tag = '%s_split_%s' % (packing_list.getPath(),
                                          delivery_builder.getId())
      after_tag = []
      if previous_tag:
        after_tag.append(previous_tag)
      delivery_builder.activate(
        after_method_id=('solve',
                         'immediateReindexObject',
                         'recursiveImmediateReindexObject',), # XXX too brutal.
        after_tag=after_tag,
        ).build(explanation_uid=packing_list.getCausalityValue().getUid())
374

375 376 377
def test_suite():
  suite = unittest.TestSuite()
  suite.addTest(unittest.makeSuite(TestERP5Simulation))
378
  suite.addTest(unittest.makeSuite(TestERP5SimulationPackingList))
379
  return suite