Commit 8150caa2 authored by Sebastien Robin's avatar Sebastien Robin

simulation: add tests for Split Move Solver and for FIFO solver

parent 38f33857
No related merge requests found
......@@ -71,17 +71,23 @@ class QuantitySplitMoveSolver(QuantitySplitSolver):
delivery = diverged_simulation_movement.getDeliveryValue().getParentValue()
assert delivery.isDelivery()
trade_phase = self.getDeliveryValue().getTradePhase()
assert trade_phase is not None, "Unable to solve, no trade phase is defined for %s" % diverged_simulation_movement.getRelativeUrl()
business_link_list = diverged_simulation_movement.asComposedDocument().getBusinessLinkValueList(trade_phase=trade_phase)
assert len(business_link_list) == 1, "Expected to find only one business link for trade_phase %s, but found %r" % (trade_phase,
assert trade_phase is not None, "Unable to solve, no trade phase is defined for %s" % \
diverged_simulation_movement.getRelativeUrl()
business_link_list = diverged_simulation_movement.asComposedDocument(
).getBusinessLinkValueList(trade_phase=trade_phase)
assert len(business_link_list) == 1, \
"Expected to find only one business link for trade_phase %s, but found %r" % (trade_phase,
[x.getRelativeUrl() for x in business_link_list])
business_link, = business_link_list
delivery_builder_list = business_link.getDeliveryBuilderValueList()
assert len(delivery_builder_list) == 1, "Expected to find only one builder on business link, but found %r" % (
delivery_builder_list = [x for x in business_link.getDeliveryBuilderValueList() \
if x.getDeliveryPortalType() == delivery.getPortalType()]
assert len(delivery_builder_list) == 1, \
"Expected to find only one builder on business link, but found %r" % (
business_link.getRelativeUrl(), [x.getRelativeUrl() for x in delivery_builder_list])
delivery_builder, = delivery_builder_list
# Update simulation movements to make sure they match the new delivery
delivery_level_movement_group = [x for x in delivery_builder.objectValues() if x.getCollectOrderGroup() == "delivery"]
delivery_level_movement_group = [x for x in delivery_builder.objectValues() \
if x.getCollectOrderGroup() == "delivery"]
delivery_to_move = portal.unrestrictedTraverse(configuration_dict['delivery_url'])
assert delivery_to_move.getDivergenceList() == []
for movement_group in delivery_level_movement_group:
......
......@@ -115,8 +115,10 @@ class TestOrderMixin(SubcontentReindexingWrapper):
#
# 1. All calculations are done relative to the same time
# 2. We don't get random failures when tests run close to midnight
self.datetime = DateTime() - 2
self.pinDateTime(self.datetime)
self.loginByUserName('test_user')
self.begin = self.datetime
def beforeTearDown(self):
self.unpinDateTime()
......
......@@ -322,18 +322,7 @@ class TestPackingListMixin(TestOrderMixin):
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():
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())
self.callPackingListBuilderList(packing_list)
def stepSplitAndDeferDoNothingPackingList(self, sequence=None, sequence_list=None, **kw):
"""
......@@ -389,12 +378,10 @@ class TestPackingListMixin(TestOrderMixin):
packing_list1 = packing_list
else:
packing_list2 = packing_list
for line in packing_list1.objectValues(
portal_type= self.packing_list_line_portal_type):
self.assertEqual(self.default_quantity-2,line.getQuantity())
for line in packing_list2.objectValues(
portal_type= self.packing_list_line_portal_type):
self.assertEqual(2,line.getQuantity())
line1, = packing_list1.objectValues(portal_type=self.packing_list_line_portal_type)
self.assertEqual(self.default_quantity-2,line1.getQuantity())
line2, = packing_list2.objectValues(portal_type=self.packing_list_line_portal_type)
self.assertEqual(2,line2.getQuantity())
def stepCheckPackingListNotSplitted(self, sequence=None, sequence_list=None, **kw):
"""
......@@ -1013,6 +1000,20 @@ class TestPackingListMixin(TestOrderMixin):
portal_type=self.packing_list_line_portal_type)
self._checkRecordedProperty(movement_list, 'resource', False)
def callPackingListBuilderList(self, packing_list):
# build split deliveries manually. XXX ad-hoc
previous_tag = None
for delivery_builder in packing_list.getBuilderList():
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())
class TestPackingList(TestPackingListMixin, ERP5TypeTestCase) :
run_all_test = 1
......@@ -1484,6 +1485,122 @@ class TestPackingList(TestPackingListMixin, ERP5TypeTestCase) :
sequence_list.play(self, quiet=quiet)
def stepSplitAndMovePackingList(self, sequence=None, sequence_list=None, **kw):
"""
Do the split and move to another delivery action
"""
packing_list = sequence.get('packing_list')
new_packing_list = sequence.get('new_packing_list')
solver_process_tool = self.portal.portal_solver_processes
solver_process = solver_process_tool.newSolverProcess(packing_list)
quantity_solver_decision, = [x for x in solver_process.contentValues()
if 'quantity' in x.getCausalityValue().getTestedPropertyList()]
# use Quantity Split Solver.
quantity_solver_decision.setSolverValue(self.portal.portal_solvers['Quantity Split Move Solver'])
# configure for Quantity Split Solver.
kw = {'delivery_solver':'FIFO Delivery Solver',
'delivery_url': new_packing_list.getRelativeUrl()}
quantity_solver_decision.updateConfiguration(**kw)
solver_process.buildTargetSolverList()
solver_process.solve()
def stepCheckSeveralDivergenceAction(self, sequence=None, sequence_list=None, **kw):
"""
Do the split and move to another delivery action
"""
packing_list1 = sequence.get('packing_list')
line1, = packing_list1.objectValues(portal_type=self.packing_list_line_portal_type)
packing_list2 = sequence.get('new_packing_list')
# Make sure we can split and defer the new packing list, this would make
# use of FIFO delivery solver in the case we have several lines
line2, = packing_list2.objectValues(portal_type=self.packing_list_line_portal_type)
line2.setQuantity(1.5)
self.tic()
self.assertEqual('diverged', packing_list2.getCausalityState())
solver_process, = [x for x in packing_list2.getSolverValueList() if \
x.getValidationState() == "draft"]
quantity_solver_decision, = [x for x in solver_process.contentValues()
if 'quantity' in x.getCausalityValue().getTestedPropertyList()]
# use Quantity Split Solver.
quantity_solver_decision.setSolverValue(self.portal.portal_solvers['Quantity Split Solver'])
# configure for Quantity Split Solver.
kw = {'delivery_solver':'FIFO Delivery Solver',
'start_date':self.datetime + 35,
'stop_date':self.datetime + 45}
quantity_solver_decision.updateConfiguration(**kw)
solver_process.buildTargetSolverList()
solver_process.solve()
self.callPackingListBuilderList(packing_list2)
self.tic()
packing_list1, packing_list2, packing_list3 = self.getCreatedTypeList(
self.packing_list_portal_type)
line3, = packing_list3.objectValues(portal_type=self.packing_list_line_portal_type)
self.assertEqual(0.5, line3.getQuantity())
self.assertEqual('solved', packing_list1.getCausalityState())
self.assertEqual('solved', packing_list2.getCausalityState())
self.assertEqual('solved', packing_list3.getCausalityState())
# And now make sure we could accept a new quantity in case we have several
# simulation movements
line2.setQuantity(1.2)
self.tic()
self.assertEqual('diverged', packing_list2.getCausalityState())
solver_process, = [x for x in packing_list2.getSolverValueList() if \
x.getValidationState() == "draft"]
quantity_solver_decision, = [x for x in solver_process.contentValues()
if 'quantity' in x.getCausalityValue().getTestedPropertyList()]
# use Quantity Split Solver.
quantity_solver_decision.setSolverValue(self.portal.portal_solvers['Accept Solver'])
kw = {'tested_property_list':['quantity']}
quantity_solver_decision.updateConfiguration(**kw)
solver_process.buildTargetSolverList()
solver_process.solve()
self.callPackingListBuilderList(packing_list2)
self.tic()
packing_list1, packing_list2, packing_list3 = self.getCreatedTypeList(
self.packing_list_portal_type)
self.assertEqual('solved', packing_list1.getCausalityState())
self.assertEqual('solved', packing_list2.getCausalityState())
self.assertEqual('solved', packing_list3.getCausalityState())
self.assertEqual(self.default_quantity-2, line1.getQuantity())
self.assertEqual(1.2, line2.getQuantity())
self.assertEqual(0.5, line3.getQuantity())
def test_11b_PackingListDecreaseTwoTimesQuantityAndMoveToDelivery(self,
quiet=quiet, run=run_all_test):
"""
Change the quantity on an delivery line, then
split and defer the packing list. Then decrease again the quantity,
and use solver "split and move" to move the quantity to the second packing
list. The second packing list would be solved by the "split and move"
solver
"""
if not run: return
sequence_list = SequenceList()
sequence_string = self.default_sequence + """
DecreasePackingListLineQuantity
CheckPackingListIsCalculating
Tic
CheckPackingListIsDiverged
SplitAndDeferPackingList
Tic
CheckPackingListIsSolved
CheckPackingListSplitted
DecreasePackingListLineQuantity
CheckPackingListIsCalculating
Tic
CheckPackingListIsDiverged
SplitAndMovePackingList
Tic
CheckNewPackingListIsSolved
CheckPackingListSplittedTwoTimes
CheckSeveralDivergenceAction
"""
sequence_list.addSequenceString(sequence_string)
sequence_list.play(self, quiet=quiet)
def test_SplitAndDeferDoNothing(self, quiet=quiet, run=run_all_test):
"""
Use split & defer to solve a divergence, but choose do nothing for all
......
......@@ -96,6 +96,7 @@ from Products.ERP5Type.tests.ProcessingNodeTestCase import \
onsetup(patchActivityTool)()
from AccessControl.SecurityManagement import newSecurityManager, noSecurityManager
from Products.ZSQLCatalog.SQLCatalog import Query
from Acquisition import aq_base
......@@ -412,6 +413,21 @@ class ERP5TypeTestCaseMixin(ProcessingNodeTestCase, PortalTestCase):
return getattr(self.portal, 'currency_module',
getattr(self.portal, 'currency', None))
def getCreatedTypeList(self, portal_type, from_date=None):
"""
Convenient method to retrieve new documents created in the testn in
particular documents that are created indirectly, like trough activities.
"begin" attribute of test class instance could be initialized in an
afterSetup.
"""
if from_date is None:
from_date = getattr(self, 'begin')
type_list = [x.getObject() for x in self.portal.portal_catalog(
portal_type=portal_type,
query=Query(creation_date=from_date, range="min"))]
type_list.sort(key=lambda x: x.getCreationDate())
return type_list
def _addPropertySheet(self, portal_type_name,
property_sheet_name='TestPropertySheet',
deprecated=None):
......
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment