Commit ba1edb69 authored by Yusuke Muraoka's avatar Yusuke Muraoka

Add support for nested lines by Order/Delivery Builder.

Move getTotal* and related methods of OrderLine to DeliveryLine,
because nested lines would be used in delivery lines as well as order lines.


git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@25505 20353a03-c40f-0410-a6d1-a30d3c3de9de
parent c11547c2
......@@ -263,16 +263,16 @@ class DeliveryBuilder(OrderBuilder):
simulation_movement_list.append(simulation_movement)
# Collect
root_group = self.collectMovement(simulation_movement_list)
root_group_node = self.collectMovement(simulation_movement_list)
# Build
portal = self.getPortalObject()
delivery_module = getattr(portal, self.getDeliveryModule())
delivery_to_update_list = [delivery]
self._resetUpdated()
delivery_list = self._deliveryGroupProcessing(
delivery_list = self._processDeliveryGroup(
delivery_module,
root_group,
root_group_node,
self.getDeliveryMovementGroupList(),
delivery_to_update_list=delivery_to_update_list,
divergence_list=divergence_to_adopt_list,
......
......@@ -115,8 +115,19 @@ class DeliveryLine(Movement, XMLObject, XMLMatrix, Variated,
return self.getParentValue().isAccountable() and (not self.hasCellContent())
def _getTotalPrice(self, default=0.0, context=None, fast=0):
""" Returns the total price for this line or the cells it contains. """
if not self.hasCellContent(base_id='movement'):
"""
Returns the total price for this line, this line contains, or the cells it contains.
if hasLineContent: return sum of lines total price
if hasCellContent: return sum of cells total price
else: return quantity * price
if fast is argument true, then a SQL method will be used.
"""
if self.hasLineContent():
meta_type = self.meta_type
return sum(l.getTotalPrice(context=context)
for l in self.objectValues() if l.meta_type==meta_type)
elif not self.hasCellContent(base_id='movement'):
return Movement._getTotalPrice(self, default=default, context=context)
elif fast: # Use MySQL
return self.DeliveryLine_zGetTotal()[0].total_price or 0.0
......@@ -129,18 +140,34 @@ class DeliveryLine(Movement, XMLObject, XMLMatrix, Variated,
"""
Returns the quantity if no cell or the total quantity if cells
If fast is equal to 0, we returns the right quantity even
if there is nothing into the catalog or the catalog is not
up to date
if hasLineContent: return sum of lines total quantity
if hasCellContent: return sum of cells total quantity
else: return quantity
if fast argument is true, then a SQL method will be used.
"""
base_id = 'movement'
if not self.hasCellContent(base_id=base_id):
return self.getQuantity()
else:
if self.hasLineContent():
meta_type = self.meta_type
return sum(l.getTotalQuantity() for l in
self.objectValues() if l.meta_type==meta_type)
elif self.hasCellContent(base_id=base_id):
if fast : # Use MySQL
aggregate = self.DeliveryLine_zGetTotal()[0]
return aggregate.total_quantity or 0.0
return sum([cell.getQuantity() for cell in self.getCellValueList()])
else:
return self.getQuantity()
security.declareProtected(Permissions.AccessContentsInformation,
'hasLineContent')
def hasLineContent(self):
"""Return true if the object contains lines.
This method only checks the first sub line because all sub
lines should be same meta type in reality if we have line
inside line.
"""
return len(self) != 0 and self.objectValues()[0].meta_type == self.meta_type
security.declareProtected(Permissions.AccessContentsInformation,
'hasCellContent')
......
......@@ -79,3 +79,7 @@ class MovementGroup(XMLObject):
return sorted([[sorted(x[0], key=lambda x: x.getId()), x[1]] \
for x in self._separate(movement_list)],
key=lambda x: x[0][0].getId())
def isBranch(self):
# self is taken as branch point by the builder if returned value is True.
return False
##############################################################################
#
# 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.
#
##############################################################################
from Products.ERP5.Document.MovementGroup import MovementGroup
class NestedLineMovementGroup(MovementGroup):
"""
This MovementGroup is only used to multiple lines control.
No more effect.
"""
meta_type = 'ERP5 Nested Line Movement Group'
portal_type = 'Nested Line Movement Group'
def _getPropertyDict(self, movement, **kw):
return {}
def test(self, object, property_dict, property_list=None, **kw):
if property_list not in (None, []):
target_property_list = [x for x in self.getTestedPropertyList() \
if x in property_list]
else:
target_property_list = self.getTestedPropertyList()
for prop in target_property_list:
if property_dict[prop] != object.getProperty(prop, None):
return False, property_dict
return True, property_dict
def isBranch(self):
return True
......@@ -118,10 +118,10 @@ class OrderBuilder(XMLObject, Amount, Predicate):
movement_list = [self.restrictedTraverse(relative_url) for relative_url \
in movement_relative_url_list]
# Collect
root_group = self.collectMovement(movement_list)
root_group_node = self.collectMovement(movement_list)
# Build
delivery_list = self.buildDeliveryList(
root_group,
root_group_node,
delivery_relative_url_list=delivery_relative_url_list,
movement_list=movement_list,**kw)
# Call a script after building
......@@ -229,45 +229,48 @@ class OrderBuilder(XMLObject, Amount, Predicate):
movement_group_list = self.getMovementGroupList()
last_line_movement_group = self.getDeliveryMovementGroupList()[-1]
separate_method_name_list = self.getDeliveryCellSeparateOrderList([])
my_root_group = MovementGroupNode(
root_group_node = MovementGroupNode(
separate_method_name_list=separate_method_name_list,
movement_group_list=movement_group_list,
last_line_movement_group=last_line_movement_group)
my_root_group.append(movement_list)
return my_root_group
root_group_node.append(movement_list)
return root_group_node
def _test(self, instance, movement_group_list,
def _test(self, instance, movement_group_node_list,
divergence_list):
result = True
new_property_dict = {}
for movement_group in movement_group_list:
tmp_result, tmp_property_dict = movement_group.test(
for movement_group_node in movement_group_node_list:
tmp_result, tmp_property_dict = movement_group_node.test(
instance, divergence_list)
if not tmp_result:
result = tmp_result
new_property_dict.update(tmp_property_dict)
return result, new_property_dict
def _findUpdatableObject(self, instance_list, movement_group_list,
def _findUpdatableObject(self, instance_list, movement_group_node_list,
divergence_list):
instance = None
property_dict = {}
if not len(instance_list):
for movement_group in movement_group_list:
property_dict.update(movement_group.getGroupEditDict())
for movement_group_node in movement_group_node_list:
property_dict.update(movement_group_node.getGroupEditDict())
else:
# we want to check the original first.
# the original is the delivery of the last (bottom) movement group.
# we want to check the original delivery first.
# so sort instance_list by that current is exists or not.
try:
original = movement_group_list[-1].getMovementList()[0].getDeliveryValue()
current = movement_group_node_list[-1].getMovementList()[0].getDeliveryValue()
portal = self.getPortalObject()
while current != portal:
if current in instance_list:
instance_list.sort(key=lambda x: x != current and 1 or 0)
break
current = current.getParentValue()
except AttributeError:
original = None
if original is not None:
original_id = original.getId()
instance_list.sort(key=lambda x: x.getId() != original_id and 1 or 0)
pass
for instance_to_update in instance_list:
result, property_dict = self._test(
instance_to_update, movement_group_list, divergence_list)
instance_to_update, movement_group_node_list, divergence_list)
if result == True:
instance = instance_to_update
break
......@@ -280,7 +283,7 @@ class OrderBuilder(XMLObject, Amount, Predicate):
buildDeliveryList = UnrestrictedMethod(self._buildDeliveryList)
return buildDeliveryList(*args, **kw)
def _buildDeliveryList(self, movement_group, delivery_relative_url_list=None,
def _buildDeliveryList(self, movement_group_node, delivery_relative_url_list=None,
movement_list=None,**kw):
"""This method is wrapped by UnrestrictedMethod."""
# Parameter initialization
......@@ -304,33 +307,26 @@ class OrderBuilder(XMLObject, Amount, Predicate):
# We do not want to update the same object more than twice in one
# _deliveryGroupProcessing().
self._resetUpdated()
delivery_list = self._deliveryGroupProcessing(
delivery_list = self._processDeliveryGroup(
delivery_module,
movement_group,
movement_group_node,
self.getDeliveryMovementGroupList(),
delivery_to_update_list=delivery_to_update_list,
**kw)
return delivery_list
def _deliveryGroupProcessing(self, *args, **kw):
"""
Build empty delivery from a list of movement
"""
deliveryGroupProcessing = UnrestrictedMethod(self.__deliveryGroupProcessing)
return deliveryGroupProcessing(*args, **kw)
def __deliveryGroupProcessing(self, delivery_module, movement_group,
collect_order_list, movement_group_list=None,
def _processDeliveryGroup(self, delivery_module, movement_group_node,
collect_order_list, movement_group_node_list=None,
delivery_to_update_list=None,
divergence_list=None,
activate_kw=None, force_update=0, **kw):
"""This method is wrapped by UnrestrictedMethod."""
if movement_group_list is None:
movement_group_list = []
if movement_group_node_list is None:
movement_group_node_list = []
if divergence_list is None:
divergence_list = []
# do not use 'append' or '+=' because they are destructive.
movement_group_list = movement_group_list + [movement_group]
movement_group_node_list = movement_group_node_list + [movement_group_node]
# Parameter initialization
if delivery_to_update_list is None:
delivery_to_update_list = []
......@@ -338,12 +334,12 @@ class OrderBuilder(XMLObject, Amount, Predicate):
if len(collect_order_list):
# Get sorted movement for each delivery
for group in movement_group.getGroupList():
new_delivery_list = self._deliveryGroupProcessing(
for grouped_node in movement_group_node.getGroupList():
new_delivery_list = self._processDeliveryGroup(
delivery_module,
group,
grouped_node,
collect_order_list[1:],
movement_group_list=movement_group_list,
movement_group_node_list=movement_group_node_list,
delivery_to_update_list=delivery_to_update_list,
divergence_list=divergence_list,
activate_kw=activate_kw,
......@@ -358,7 +354,7 @@ class OrderBuilder(XMLObject, Amount, Predicate):
if x.getPortalType() == self.getDeliveryPortalType() and \
not self._isUpdated(x, 'delivery')]
delivery, property_dict = self._findUpdatableObject(
delivery_to_update_list, movement_group_list,
delivery_to_update_list, movement_group_node_list,
divergence_list)
# if all deliveries are rejected in case of update, we update the
......@@ -370,7 +366,7 @@ class OrderBuilder(XMLObject, Amount, Predicate):
# Create delivery
try:
old_delivery = self._searchUpByPortalType(
movement_group.getMovementList()[0].getDeliveryValue(),
movement_group_node.getMovementList()[0].getDeliveryValue(),
self.getDeliveryPortalType())
except AttributeError:
old_delivery = None
......@@ -392,7 +388,7 @@ class OrderBuilder(XMLObject, Amount, Predicate):
delivery = delivery_module[cp['new_id']]
# delete non-split movements
keep_id_list = [y.getDeliveryValue().getId() for y in \
movement_group.getMovementList()]
movement_group_node.getMovementList()]
delete_id_list = [x.getId() for x in delivery.contentValues() \
if x.getId() not in keep_id_list]
delivery.deleteContent(delete_id_list)
......@@ -402,10 +398,10 @@ class OrderBuilder(XMLObject, Amount, Predicate):
delivery.edit(**property_dict)
# Then, create delivery line
for group in movement_group.getGroupList():
self._deliveryLineGroupProcessing(
for grouped_node in movement_group_node.getGroupList():
self._processDeliveryLineGroup(
delivery,
group,
grouped_node,
self.getDeliveryLineMovementGroupList()[1:],
divergence_list=divergence_list,
activate_kw=activate_kw,
......@@ -413,28 +409,31 @@ class OrderBuilder(XMLObject, Amount, Predicate):
delivery_list.append(delivery)
return delivery_list
def _deliveryLineGroupProcessing(self, delivery, movement_group,
collect_order_list, movement_group_list=None,
def _processDeliveryLineGroup(self, delivery, movement_group_node,
collect_order_list, movement_group_node_list=None,
divergence_list=None,
activate_kw=None, force_update=0, **kw):
"""
Build delivery line from a list of movement on a delivery
"""
if movement_group_list is None:
movement_group_list = []
if movement_group_node_list is None:
movement_group_node_list = []
if divergence_list is None:
divergence_list = []
# do not use 'append' or '+=' because they are destructive.
movement_group_list = movement_group_list + [movement_group]
movement_group_node_list = movement_group_node_list + [movement_group_node]
if len(collect_order_list):
if len(collect_order_list) and not movement_group_node.getCurrentMovementGroup().isBranch():
# Get sorted movement for each delivery line
for group in movement_group.getGroupList():
self._deliveryLineGroupProcessing(
delivery, group, collect_order_list[1:],
movement_group_list=movement_group_list,
for grouped_node in movement_group_node.getGroupList():
self._processDeliveryLineGroup(
delivery,
grouped_node,
collect_order_list[1:],
movement_group_node_list=movement_group_node_list,
divergence_list=divergence_list,
activate_kw=activate_kw, force_update=force_update)
activate_kw=activate_kw,
force_update=force_update)
else:
# Test if we can update an existing line, or if we need to create a new
# one
......@@ -442,7 +441,7 @@ class OrderBuilder(XMLObject, Amount, Predicate):
portal_type=self.getDeliveryLinePortalType()) if \
not self._isUpdated(x, 'line')]
delivery_line, property_dict = self._findUpdatableObject(
delivery_line_to_update_list, movement_group_list,
delivery_line_to_update_list, movement_group_node_list,
divergence_list)
if delivery_line is not None:
update_existing_line = 1
......@@ -451,7 +450,7 @@ class OrderBuilder(XMLObject, Amount, Predicate):
update_existing_line = 0
try:
old_delivery_line = self._searchUpByPortalType(
movement_group.getMovementList()[0].getDeliveryValue(),
movement_group_node.getMovementList()[0].getDeliveryValue(),
self.getDeliveryLinePortalType())
except AttributeError:
old_delivery_line = None
......@@ -475,7 +474,7 @@ class OrderBuilder(XMLObject, Amount, Predicate):
delivery_line.setVariationCategoryList([])
# delete non-split movements
keep_id_list = [y.getDeliveryValue().getId() for y in \
movement_group.getMovementList()]
movement_group_node.getMovementList()]
delete_id_list = [x.getId() for x in delivery_line.contentValues() \
if x.getId() not in keep_id_list]
delivery_line.deleteContent(delete_id_list)
......@@ -484,34 +483,46 @@ class OrderBuilder(XMLObject, Amount, Predicate):
if property_dict:
delivery_line.edit(**property_dict)
if movement_group_node.getCurrentMovementGroup().isBranch():
for grouped_node in movement_group_node.getGroupList():
self._processDeliveryLineGroup(
delivery_line,
grouped_node,
collect_order_list[1:],
movement_group_node_list=movement_group_node_list,
divergence_list=divergence_list,
activate_kw=activate_kw,
force_update=force_update)
return
# Update variation category list on line
variation_category_dict = dict([(variation_category, True) for
variation_category in
delivery_line.getVariationCategoryList()])
for movement in movement_group.getMovementList():
for movement in movement_group_node.getMovementList():
for category in movement.getVariationCategoryList():
variation_category_dict[category] = True
variation_category_list = sorted(variation_category_dict.keys())
delivery_line.setVariationCategoryList(variation_category_list)
# Then, create delivery movement (delivery cell or complete delivery
# line)
group_list = movement_group.getGroupList()
grouped_node_list = movement_group_node.getGroupList()
# If no group is defined for cell, we need to continue, in order to
# save the quantity value
if len(group_list):
for group in group_list:
self._deliveryCellGroupProcessing(
if len(grouped_node_list):
for grouped_node in grouped_node_list:
self._processDeliveryCellGroup(
delivery_line,
group,
grouped_node,
self.getDeliveryCellMovementGroupList()[1:],
update_existing_line=update_existing_line,
divergence_list=divergence_list,
activate_kw=activate_kw,
force_update=force_update)
else:
self._deliveryCellGroupProcessing(
self._processDeliveryCellGroup(
delivery_line,
movement_group,
movement_group_node,
[],
update_existing_line=update_existing_line,
divergence_list=divergence_list,
......@@ -519,8 +530,8 @@ class OrderBuilder(XMLObject, Amount, Predicate):
force_update=force_update)
def _deliveryCellGroupProcessing(self, delivery_line, movement_group,
collect_order_list, movement_group_list=None,
def _processDeliveryCellGroup(self, delivery_line, movement_group_node,
collect_order_list, movement_group_node_list=None,
update_existing_line=0,
divergence_list=None,
activate_kw=None, force_update=0):
......@@ -528,27 +539,27 @@ class OrderBuilder(XMLObject, Amount, Predicate):
Build delivery cell from a list of movement on a delivery line
or complete delivery line
"""
if movement_group_list is None:
movement_group_list = []
if movement_group_node_list is None:
movement_group_node_list = []
if divergence_list is None:
divergence_list = []
# do not use 'append' or '+=' because they are destructive.
movement_group_list = movement_group_list + [movement_group]
movement_group_node_list = movement_group_node_list + [movement_group_node]
if len(collect_order_list):
# Get sorted movement for each delivery line
for group in movement_group.getGroupList():
self._deliveryCellGroupProcessing(
for grouped_node in movement_group_node.getGroupList():
self._processDeliveryCellGroup(
delivery_line,
group,
grouped_node,
collect_order_list[1:],
movement_group_list=movement_group_list,
movement_group_node_list=movement_group_node_list,
update_existing_line=update_existing_line,
divergence_list=divergence_list,
activate_kw=activate_kw,
force_update=force_update)
else:
movement_list = movement_group.getMovementList()
movement_list = movement_group_node.getMovementList()
if len(movement_list) != 1:
raise CollectError, "DeliveryBuilder: %s unable to distinct those\
movements: %s" % (self.getId(), str(movement_list))
......@@ -574,7 +585,7 @@ class OrderBuilder(XMLObject, Amount, Predicate):
else:
object_to_update_list = []
object_to_update, property_dict = self._findUpdatableObject(
object_to_update_list, movement_group_list,
object_to_update_list, movement_group_node_list,
divergence_list)
if object_to_update is not None:
update_existing_movement = 1
......@@ -586,7 +597,7 @@ class OrderBuilder(XMLObject, Amount, Predicate):
delivery_line.getCellKeyList(base_id=base_id) \
if delivery_line.hasCell(base_id=base_id, *cell_key)]
object_to_update, property_dict = self._findUpdatableObject(
object_to_update_list, movement_group_list,
object_to_update_list, movement_group_node_list,
divergence_list)
if object_to_update is not None:
# We update a existing cell
......@@ -600,7 +611,7 @@ class OrderBuilder(XMLObject, Amount, Predicate):
omit_optional_variation=1)
if not delivery_line.hasCell(base_id=base_id, *cell_key):
try:
old_cell = movement_group.getMovementList()[0].getDeliveryValue()
old_cell = movement_group_node.getMovementList()[0].getDeliveryValue()
except AttributeError:
old_cell = None
if old_cell is None:
......@@ -756,3 +767,8 @@ class OrderBuilder(XMLObject, Amount, Predicate):
def _resetUpdated(self):
tv = getTransactionalVariable(self)
tv['builder_processed_list'] = {}
# for backward compatibilities.
_deliveryGroupProcessing = _processDeliveryGroup
_deliveryLineGroupProcessing = _processDeliveryLineGroup
_deliveryCellGroupProcessing = _processDeliveryCellGroup
......@@ -64,57 +64,6 @@ class OrderLine(DeliveryLine):
# Declarative interfaces
__implements__ = ( Interface.Variated, )
security.declareProtected(Permissions.AccessContentsInformation,
'hasLineContent')
def hasLineContent(self):
"""Return true if the object contains lines.
This method only checks the first sub document because all sub
documents should be Order Line in reality if we have Order Line
inside Order Line.
"""
return len(self) != 0 and self.objectValues()[0].meta_type == self.meta_type
def _getTotalPrice(self, default=0.0, context=None, fast=0):
"""Returns the total price for this order line.
if hasLineContent: return sum of lines total price
if hasCellContent: return sum of cells total price
else: return quantity * price
if fast is argument true, then a SQL method will be used.
"""
if self.hasLineContent():
meta_type = self.meta_type
return sum(l.getTotalPrice(context=context)
for l in self.objectValues() if l.meta_type==meta_type)
return DeliveryLine._getTotalPrice(self,
default=default,
context=context,
fast=fast)
security.declareProtected(Permissions.AccessContentsInformation,
'getTotalQuantity')
def getTotalQuantity(self, fast=0):
"""Returns the total quantity of this order line.
if hasLineContent: return sum of lines total quantity
if hasCellContent: return sum of cells total quantity
else: return quantity
if fast argument is true, then a SQL method will be used.
"""
base_id = 'movement'
if self.hasLineContent():
meta_type = self.meta_type
return sum(l.getTotalQuantity() for l in
self.objectValues() if l.meta_type==meta_type)
elif self.hasCellContent(base_id=base_id):
if fast : # Use MySQL
aggregate = self.DeliveryLine_zGetTotal()[0]
return aggregate.total_quantity or 0.0
return sum([cell.getQuantity() for cell in self.getCellValueList()])
else:
return self.getQuantity()
def applyToOrderLineRelatedMovement(self, portal_type='Simulation Movement',
method_id = 'expand'):
"""
......
......@@ -76,8 +76,7 @@ class MovementGroupNode:
for movement in movement_list[1:]:
# We have a conflict here, because it is forbidden to have
# 2 movements on the same node group
tmp_result = self._separate(movement)
self._movement_list, split_movement = tmp_result
self._movement_list, split_movement = self._separate(movement)
if split_movement is not None:
# We rejected a movement, we need to put it on another line
# Or to create a new one
......@@ -109,6 +108,9 @@ class MovementGroupNode:
del(property_dict[key])
return property_dict
def getCurrentMovementGroup(self):
return self._movement_group
def getMovementList(self):
"""
Return movement list in the current group
......
##############################################################################
# -*- coding: utf8 -*-
#
# 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., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
import unittest
from Products.ERP5Type.UnrestrictedMethod import UnrestrictedMethod
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from Products.ERP5Type.tests.Sequence import SequenceList
from Products.ERP5Type.tests.utils import createZODBPythonScript
from Products.ERP5.tests.testInvoice import TestSaleInvoiceMixin
class TestNestedLineMixin(TestSaleInvoiceMixin):
"""
NestedLineMovementGroup is a mark only for controlling multiple lines in DeliveryBuilder.
We need this feature to make multi-level "Invoice Line"s.
"""
DEFAULT_SEQUENCE = TestSaleInvoiceMixin.PACKING_LIST_DEFAULT_SEQUENCE + \
"""
stepSetReadyPackingList
stepTic
stepUpdateBuilderForMultipleLineList
stepSetPythonScriptForDeliveryBuilder
stepStartPackingList
stepCheckInvoicingRule
stepTic
stepGetRelatedInvoiceFromPackingList
"""
delivery_builder_id = 'sale_invoice_builder'
default_quantity = TestSaleInvoiceMixin.default_quantity
new_order_quantity = TestSaleInvoiceMixin.default_quantity * 3
new_packing_list_quantity = TestSaleInvoiceMixin.default_quantity * 5
new_invoice_quantity = TestSaleInvoiceMixin.default_quantity * 2
def afterSetUp(self):
TestSaleInvoiceMixin.afterSetUp(self)
# Necessary to allow Invoice Line to be included in Invoice Line.
self.allowInvoiceLineContentTypeInInvoiceLine()
def allowInvoiceLineContentTypeInInvoiceLine(self):
return UnrestrictedMethod(self._allowInvoiceLineContentTypeInInvoiceLine)()
def _allowInvoiceLineContentTypeInInvoiceLine(self):
invoice_line_type = self.portal.portal_types['Invoice Line']
if 'Invoice Line' not in invoice_line_type.allowed_content_types:
invoice_line_type.allowed_content_types += ('Invoice Line',)
def stepGetRelatedInvoiceFromPackingList(self, sequence, **kw):
packing_list = sequence.get('packing_list')
related_invoice_list = packing_list \
.getCausalityRelatedValueList(portal_type=self.invoice_portal_type)
invoice = related_invoice_list[0].getObject()
sequence.edit(invoice=invoice)
def stepUpdateBuilderForMultipleLineList(self, **kw):
self.updateBuilderForMultipleLineList()
def updateBuilderForMultipleLineList(self):
return UnrestrictedMethod(self._updateBuilderForMultipleLineList)()
def _updateBuilderForMultipleLineList(self):
delivery_builder = getattr(self.portal.portal_deliveries, self.delivery_builder_id)
delivery_builder.deleteContent(delivery_builder.contentIds())
delivery_builder.newContent(
portal_type='Property Movement Group',
collect_order_group='delivery',
divergence_scope='property',
tested_property_list=('start_date', 'stop_date'),
int_index=1)
delivery_builder.newContent(
portal_type='Category Movement Group',
collect_order_group='delivery',
divergence_scope='category',
tested_property_list=('delivery_mode',
'incoterm',
'source',
'destination',
'source_section',
'destination_section',
'destination_function',
'source_function',
'source_decision',
'destination_decision',
'source_administration',
'destination_administration',
'price_currency'),
int_index=2)
delivery_builder.newContent(
portal_type='Delivery Causality Assignment Movement Group',
collect_order_group='delivery',
int_index=3)
delivery_builder.newContent(
portal_type='Property Movement Group',
collect_order_group='line',
divergence_scope='property',
tested_property_list=('start_date', 'stop_date'),
int_index=1)
# *** test this ***
delivery_builder.newContent(
portal_type='Nested Line Movement Group',
collect_order_group='line',
int_index=2)
delivery_builder.newContent(
portal_type='Category Movement Group',
collect_order_group='line',
divergence_scope='category',
tested_property_list=('resource', 'aggregate', 'base_contribution'),
int_index=3)
delivery_builder.newContent(
portal_type='Base Variant Movement Group',
collect_order_group='line',
int_index=4)
delivery_builder.newContent(
portal_type='Property Movement Group',
collect_order_group='line',
divergence_scope='property',
tested_property_list=('description'),
int_index=5)
delivery_builder.newContent(
portal_type='Variant Movement Group',
collect_order_group='cell',
divergence_scope='category',
int_index=1)
def stepSetExistDeliveriesToSequence(self, sequence=None, **kw):
order = self.portal.sale_order_module.contentValues(portal_type='Sale Order')[0]
packing_list = self.portal.sale_packing_list_module \
.contentValues(portal_type='Sale Packing List')[0]
invoice = self.portal.accounting_module \
.contentValues(portal_type='Sale Invoice Transaction')[0]
sequence.edit(order=order, packing_list=packing_list, invoice=invoice)
def stepUpdateOrder(self, sequence=None, **kw):
movement = sequence.get('order').getMovementList(portal_type='Sale Order Line')[0]
movement.edit(quantity=self.new_order_quantity,
price=self.default_price)
def stepUpdatePackingList(self, sequence=None, **kw):
movement = sequence.get('packing_list') \
.getMovementList(portal_type='Sale Packing List Line')[0]
movement.edit(quantity=self.new_packing_list_quantity)
def stepSetFillContainerLine(self, sequence=None, **kw):
movement = sequence.get('container_line')
movement.edit(quantity=self.new_order_quantity)
def stepUpdateInvoice(self, sequence=None, **kw):
movement = sequence.get('invoice') \
.getMovementList(portal_type='Invoice Line')[0]
movement.edit(quantity=self.new_invoice_quantity)
def stepSetPythonScriptForDeliveryBuilder(self, **kw):
"""
Make a script which returns existing Sale Invoice Transactions,
so that all movements are merged into existing ones.
"""
delivery_select_method_id = 'Test_selectDelivery'
createZODBPythonScript(
self.portal.portal_skins.custom,
delivery_select_method_id,
'movement_list=None',
"""
return context.getPortalObject().portal_catalog(portal_type='Sale Invoice Transaction')
""")
delivery_builder = getattr(self.portal.portal_deliveries, self.delivery_builder_id)
delivery_builder.delivery_select_method_id = delivery_select_method_id
def stepSetSeparateMethodToDeliveryBuilder(self, **kw):
"""
Merge multiple simulation movements into one movement.
"""
delivery_builder = getattr(self.portal.portal_deliveries, self.delivery_builder_id)
delivery_builder.delivery_cell_separate_order = ('calculateAddQuantity',)
def stepAdoptPrevisionPackingListQuantity(self,sequence=None, sequence_list=None):
document = sequence.get('packing_list')
self._solveDivergence(document, 'quantity', 'adopt')
def stepAcceptDecisionPackingListQuantity(self,sequence=None, sequence_list=None):
document = sequence.get('packing_list')
self._solveDivergence(document, 'quantity', 'accept')
def stepAdoptPrevisionInvoiceQuantity(self,sequence=None, sequence_list=None):
document = sequence.get('invoice')
self._solveDivergence(document, 'quantity', 'adopt')
def stepAcceptDecisionInvoiceQuantity(self,sequence=None, sequence_list=None):
document = sequence.get('invoice')
self._solveDivergence(document, 'quantity', 'accept')
class TestNestedLine(TestNestedLineMixin, ERP5TypeTestCase):
quiet = 0
def test_01_IfNested(self, quiet=quiet):
sequence_list = SequenceList()
sequence = sequence_list.addSequenceString(self.DEFAULT_SEQUENCE)
sequence_list.play(self, quiet=quiet)
# order = sequence.get('order')
# packing_list = sequence.get('packing_list')
document = sequence.get('invoice')
self.assertEquals('Sale Invoice Transaction', document.getPortalType())
self.assertEquals(1, len(document))
line = document.objectValues()[0]
self.assertEquals('Invoice Line', line.getPortalType())
self.assertEquals(None, line.getQuantity(None))
self.assertEquals(1, len(line))
line_line = line.objectValues()[0]
self.assertEquals('Invoice Line', line_line.getPortalType())
self.assertEquals(self.default_price * self.default_quantity, document.getTotalPrice())
self.assertEquals(self.default_quantity, document.getTotalQuantity())
self.assertEquals(self.default_price, line_line.getPrice())
self.assertEquals(self.default_quantity, line_line.getQuantity())
def test_02_AdoptingPrevision(self, quiet=quiet):
sequence_list = SequenceList()
sequence = sequence_list.addSequenceString(self.DEFAULT_SEQUENCE + \
"""
stepUpdatePackingList
stepTic
stepAcceptDecisionPackingListQuantity
stepTic
stepCheckInvoiceIsDivergent
stepCheckInvoiceIsDiverged
stepAdoptPrevisionInvoiceQuantity
stepTic
"""
)
sequence_list.play(self, quiet=quiet)
document = sequence.get('invoice')
self.assertEquals('solved', document.getCausalityState())
self.assertEquals(1, len(document))
line = document.objectValues()[0]
self.assertEquals('Invoice Line', line.getPortalType())
self.assertEquals(None, line.getQuantity(None))
self.assertEquals(1, len(line))
line_line = line.objectValues()[0]
self.assertEquals('Invoice Line', line_line.getPortalType())
self.assertEquals(self.default_price * self.new_packing_list_quantity, document.getTotalPrice())
self.assertEquals(self.new_packing_list_quantity, document.getTotalQuantity())
self.assertEquals(self.new_packing_list_quantity, line_line.getQuantity())
def test_03_AcceptingDecision(self, quiet=quiet):
sequence_list = SequenceList()
sequence = sequence_list.addSequenceString(self.DEFAULT_SEQUENCE + \
"""
stepUpdateInvoice
stepTic
stepCheckInvoiceIsDivergent
stepAcceptDecisionInvoiceQuantity
stepTic
stepCheckInvoiceIsNotDivergent
stepCheckPackingListIsDivergent
stepAdoptPrevisionPackingListQuantity
stepTic
"""
)
sequence_list.play(self, quiet=quiet)
document = sequence.get('invoice')
self.assertEquals('solved', document.getCausalityState())
self.assertEquals(1, len(document))
line = document.objectValues()[0]
self.assertEquals('Invoice Line', line.getPortalType())
self.assertEquals(None, line.getQuantity(None))
self.assertEquals(1, len(line))
line_line = line.objectValues()[0]
self.assertEquals('Invoice Line', line_line.getPortalType())
self.assertEquals(self.default_price * self.new_invoice_quantity, document.getTotalPrice())
self.assertEquals(self.new_invoice_quantity, document.getTotalQuantity())
self.assertEquals(self.new_invoice_quantity, line_line.getQuantity())
def test_04_MergingMultipleSaleOrders(self, quiet=quiet):
sequence_list = SequenceList()
sequence = sequence_list.addSequenceString(self.DEFAULT_SEQUENCE + \
"""
stepCreateOrder
stepSetOrderProfile
stepSetOrderPriceCurrency
stepTic
stepCreateOrderLine
stepSetOrderLineResource
stepUpdateOrder
stepOrderOrder
stepTic
stepCheckDeliveryBuilding
stepConfirmOrder
stepTic
stepCheckOrderRule
stepCheckOrderSimulation
stepCheckDeliveryBuilding
stepAddPackingListContainer
stepAddPackingListContainerLine
stepSetFillContainerLine
stepTic
stepSetReadyPackingList
stepTic
stepStartPackingList
stepCheckInvoicingRule
stepTic
stepCheckInvoiceIsDivergent
stepAdoptPrevisionInvoiceQuantity
stepTic
"""
)
sequence_list.play(self, quiet=quiet)
self.assertEquals(1, len(self.portal.accounting_module))
document = self.portal.accounting_module.objectValues()[0]
self.assertEquals('solved', document.getCausalityState())
self.assertEquals(1, len(document))
line = document.objectValues()[0]
self.assertEquals('Invoice Line', line.getPortalType())
self.assertEquals(None, line.getQuantity(None))
self.assertEquals(1, len(line))
line_line = line.objectValues()[0]
self.assertEquals('Invoice Line', line_line.getPortalType())
# The sale invoice summed up from two sale orders.
# The quantity of a sale order is self.default_quantity, and
# that of the other one is self.new_order_quantity.
self.assertEquals(self.default_price * (self.default_quantity + self.new_order_quantity), document.getTotalPrice())
self.assertEquals(self.default_quantity + self.new_order_quantity, document.getTotalQuantity())
self.assertEquals(self.default_quantity + self.new_order_quantity, line_line.getQuantity())
def test_suite():
suite = unittest.TestSuite()
suite.addTest(unittest.makeSuite(TestNestedLine))
return suite
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