[Erp5-report] r27405 - in /erp5/trunk/products/ERP5: Document/ tests/
nobody at svn.erp5.org
nobody at svn.erp5.org
Mon Jun 8 12:02:15 CEST 2009
Author: yusuke
Date: Mon Jun 8 12:02:09 2009
New Revision: 27405
URL: http://svn.erp5.org?rev=27405&view=rev
Log:
- changed the TransformationSourcingRule to refer a business process
instead of a supply chain
- added a test for TransformationSourcingRule
Modified:
erp5/trunk/products/ERP5/Document/TransformationRule.py
erp5/trunk/products/ERP5/Document/TransformationSourcingRule.py
erp5/trunk/products/ERP5/tests/testMRP.py
Modified: erp5/trunk/products/ERP5/Document/TransformationRule.py
URL: http://svn.erp5.org/erp5/trunk/products/ERP5/Document/TransformationRule.py?rev=27405&r1=27404&r2=27405&view=diff
==============================================================================
--- erp5/trunk/products/ERP5/Document/TransformationRule.py [utf8] (original)
+++ erp5/trunk/products/ERP5/Document/TransformationRule.py [utf8] Mon Jun 8 12:02:09 2009
@@ -38,17 +38,12 @@
from Products.ERP5.Document.SimulationMovement import SimulationMovement
from Products.ERP5Type.Errors import TransformationRuleError
-class TransformationMovementFactory:
- def __init__(self):
- self.default = None # base information to use for making movements
- self.produced_list = list()
- self.consumed_list = list()
-
- def requestProduced(self, **produced):
- self.produced_list.append(produced)
-
- def requestConsumed(self, **consumed):
- self.consumed_list.append(consumed)
+class MovementFactory:
+ def getRequestList(self):
+ """
+ return the list of a request which to be used to apply movements
+ """
+ raise NotImplementedError, 'Must be implemented'
def _getCausalityList(self, causality=None, causality_value=None,
causality_list=None, causality_value_list=None,
@@ -63,17 +58,6 @@
return [causality_value.getRelativeUrl()
for causality_value in causality_value_list]
- def getRequestList(self):
- _list = []
- for (request_list, sign) in ((self.produced_list, -1),
- (self.consumed_list, 1)):
- for request in request_list:
- d = self.default.copy()
- d.update(request)
- d['quantity'] *= sign
- _list.append(d)
- return _list
-
def makeMovements(self, applied_rule):
"""
make movements under the applied_rule by requests
@@ -84,10 +68,6 @@
key = tuple(sorted(movement.getCausalityList()))
movement_dict[key] = movement
- """
- produced quantity should be represented by minus quantity on movement.
- because plus quantity is consumed.
- """
for request in self.getRequestList():
# get movement by causality
key = tuple(sorted(self._getCausalityList(**request)))
@@ -121,6 +101,37 @@
diff_movement.edit(**request)
+class TransformationMovementFactory(MovementFactory):
+ def __init__(self):
+ self.product = None # base information to use for making movements
+ self.produced_list = list()
+ self.consumed_list = list()
+
+ def requestProduced(self, **produced):
+ self.produced_list.append(produced)
+
+ def requestConsumed(self, **consumed):
+ self.consumed_list.append(consumed)
+
+ def getRequestList(self):
+ """
+ return the list of a request which to be used to apply movements
+ """
+ _list = []
+ """
+ produced quantity should be represented by minus quantity on movement.
+ because plus quantity is consumed.
+ """
+ for (request_list, sign) in ((self.produced_list, -1),
+ (self.consumed_list, 1)):
+ for request in request_list:
+ d = self.product.copy()
+ d.update(request)
+ d['quantity'] *= sign
+ _list.append(d)
+ return _list
+
+
class TransformationRuleMixin(Base):
security = ClassSecurityInfo()
@@ -266,7 +277,7 @@
head_production_path_list = self.getHeadProductionPathList(transformation,
business_process)
factory = self.getFactory()
- factory.default = dict(
+ factory.product = dict(
resource=transformation.getResource(),
quantity=parent_movement.getNetQuantity(),
quantity_unit=parent_movement.getQuantityUnit(),
@@ -299,6 +310,21 @@
% (phase, business_process)
for path in phase_path_list:
+ # source, source_section
+ source_section = path.getSourceSection() # only support a static access
+ source_method_id = path.getSourceMethodId()
+ if source_method_id is None:
+ source = path.getSource()
+ else:
+ source = getattr(path, source_method_id)()
+ # destination, destination_section
+ destination_section = path.getDestinationSection() # only support a static access
+ destination_method_id = path.getDestinationMethodId()
+ if destination_method_id is None:
+ destination = path.getDestination()
+ else:
+ destination = getattr(path, destination_method_id)()
+
start_date = path.getExpectedStartDate(explanation)
stop_date = path.getExpectedStopDate(explanation)
predecessor_remaining_phase_list = path.getPredecessorValue()\
@@ -307,7 +333,6 @@
successor_remaining_phase_list = path.getSuccessorValue()\
.getRemainingTradePhaseList(explanation,
trade_phase_list=trade_phase_list)
- destination = path.getDestination()
# checking which is not last path of transformation
if len(successor_remaining_phase_list) != 0:
@@ -317,9 +342,10 @@
start_date=start_date,
stop_date=stop_date,
# when last path of transformation, path.getQuantity() will be return 1.
- quantity=factory.default['quantity'] * path.getQuantity(),
+ quantity=factory.product['quantity'] * path.getQuantity(),
+ source_section=source_section,
+ destination_section=destination_section,
destination=destination,
- #destination_section=???,
trade_phase_value_list=successor_remaining_phase_list)
else:
# for making movement of last product of the transformation
@@ -333,12 +359,19 @@
# trade phase of product is must be empty []
if last_prop_dict.get('trade_phase_value_list', None) is None:
last_prop_dict['trade_phase_value_list'] = successor_remaining_phase_list
+ if last_prop_dict.get('source_section', None) is None:
+ last_prop_dict['source_section'] = source_section
+ # for the source, it is not need, because the produced.
+ if last_prop_dict.get('destination_section', None) is None:
+ last_prop_dict['destination_section'] = destination_section
if last_prop_dict.get('destination', None) is None:
last_prop_dict['destination'] = destination
if last_prop_dict['start_date'] != start_date or\
last_prop_dict['stop_date'] != stop_date or\
last_prop_dict['trade_phase_value_list'] != successor_remaining_phase_list or\
+ last_prop_dict['source_section'] != source_section or\
+ last_prop_dict['destination_section'] != destination_section or\
last_prop_dict['destination'] != destination:
raise TransformationRuleError,\
"""Returned property is different on Transformation %r and Business Process %r"""\
@@ -350,9 +383,10 @@
causality_value=path,
start_date=start_date,
stop_date=stop_date,
- quantity=factory.default['quantity'] * path.getQuantity(),
- source=path.getSource(),
- #source_section=???,
+ quantity=factory.product['quantity'] * path.getQuantity(),
+ source_section=source_section,
+ destination_section=destination_section,
+ source=source,
trade_phase_value_list=predecessor_remaining_phase_list)
# consumed movement
@@ -362,11 +396,12 @@
start_date=start_date,
stop_date=stop_date,
resource=amount.getResource(),
- quantity=factory.default['quantity'] * amount.getQuantity()\
+ quantity=factory.product['quantity'] * amount.getQuantity()\
/ amount.getEfficiency() * path.getQuantity(),
quantity_unit=amount.getQuantityUnit(),
- source=path.getSource(),
- #source_section=???,
+ source_section=source_section,
+ destination_section=destination_section,
+ source=source,
trade_phase=path.getTradePhase())
"""
@@ -398,8 +433,7 @@
factory.requestProduced(
causality_value_list=last_phase_path_list,
# when last path of transformation, path.getQuantity() will be return 1.
- quantity=factory.default['quantity'] * path.getQuantity(),
- #destination_section=???,
+ quantity=factory.product['quantity'] * path.getQuantity(),
**last_prop_dict)
factory.makeMovements(applied_rule)
Modified: erp5/trunk/products/ERP5/Document/TransformationSourcingRule.py
URL: http://svn.erp5.org/erp5/trunk/products/ERP5/Document/TransformationSourcingRule.py?rev=27405&r1=27404&r2=27405&view=diff
==============================================================================
--- erp5/trunk/products/ERP5/Document/TransformationSourcingRule.py [utf8] (original)
+++ erp5/trunk/products/ERP5/Document/TransformationSourcingRule.py [utf8] Mon Jun 8 12:02:09 2009
@@ -27,240 +27,114 @@
#
##############################################################################
-import ExtensionClass
-
from AccessControl import ClassSecurityInfo
from Acquisition import aq_base, aq_parent, aq_inner, aq_acquire
from Products.CMFCore.utils import getToolByName
from Products.ERP5Type import Permissions, PropertySheet, Constraint, interfaces
from Products.ERP5.Document.Rule import Rule
+from Products.ERP5.Document.TransformationRule import MovementFactory, TransformationRuleMixin
from zLOG import LOG
-class ProductionOrderError(Exception): pass
class TransformationSourcingRuleError(Exception): pass
-class TransformationSourcingRuleMixin(ExtensionClass.Base):
+class SourcingMovementFactory(MovementFactory):
+ def __init__(self):
+ self.request_list = list()
+
+ def requestSourcing(self, **sourcing):
+ self.request_list.append(sourcing)
+
+ def getRequestList(self):
+ return self.request_list
+
+class TransformationSourcingRule(TransformationRuleMixin, Rule):
"""
- Mixin class used by TransformationSourcingRule and TransformationRule
+ Transformation Sourcing Rule object make sure
+ items required in a Transformation are sourced
"""
+ # CMF Type Definition
+ meta_type = 'ERP5 Transformation Sourcing Rule'
+ portal_type = 'Transformation Sourcing Rule'
# Declarative security
security = ClassSecurityInfo()
-
- security.declareProtected(Permissions.View,
- 'getSupplyChain')
- def getSupplyChain(self, applied_rule):
- """
- Get the SupplyChain.
- """
- # Get the SupplyChain to use
- supply_chain_portal_type = "Supply Chain"
- order = applied_rule.getRootAppliedRule().getCausalityValue()
- supply_chain = order.getSpecialiseValue(
- portal_type=supply_chain_portal_type)
- if supply_chain is None:
- raise ProductionOrderError,\
- "No SupplyChain defined on %s" % str(order)
- else:
- return supply_chain
-
- def getCurrentSupplyLink(self, movement):
- """
- Get the current SupplyLink
- """
- # Get the current supply link
- supply_link_portal_type = "Supply Link"
- current_supply_link = movement.getCausalityValue(
- portal_type=supply_link_portal_type)
- return current_supply_link
-
- security.declareProtected(Permissions.ModifyPortalContent,
- '_buildMovementList')
- def _buildMovementList(self, applied_rule, movement_dict,activate_kw=None,**kw):
- """
- For each movement in the dictionnary, test if the movement already
- exists.
- If not, create it.
- Then, update the movement attributes.
- """
- for movement_id in movement_dict.keys():
- movement = applied_rule.get(movement_id)
- # Create the movement if it does not exist
- if movement is None:
- movement = applied_rule.newContent(
- portal_type=self.simulation_movement_portal_type,
- id=movement_id,
- activate_kw=activate_kw
- )
- # We shouldn't modify frozen movements
- elif movement.isFrozen():
- # FIXME: this is not perfect, instead of just skipping this one, we
- # should generate a compensation movement
- continue
- # Update movement properties
- movement.edit(activate_kw=activate_kw, **(movement_dict[movement_id]))
-
- security.declareProtected(Permissions.View, 'getTransformation')
- def getTransformation(self, movement):
- """
- Get transformation related to used by the applied rule.
- """
- production_order_movement = movement.getRootSimulationMovement().\
- getOrderValue()
- # XXX Acquisition can be use instead
- parent_uid = production_order_movement.getParentUid()
- explanation_uid = production_order_movement.getExplanationUid()
- if parent_uid == explanation_uid:
- production_order_line = production_order_movement
- else:
- production_order_line = production_order_movement.getParentValue()
- script = production_order_line._getTypeBasedMethod('_getTransformation')
- if script is not None:
- transformation = script()
- else:
- line_transformation = production_order_line.objectValues(
- portal_type=self.getPortalTransformationTypeList())
- if len(line_transformation)==1:
- transformation = line_transformation[0]
- else:
- transformation = production_order_line.getSpecialiseValue(
- portal_type=self.getPortalTransformationTypeList())
- return transformation
-
-class TransformationSourcingRule(TransformationSourcingRuleMixin, Rule):
- """
- Transformation Sourcing Rule object make sure
- items required in a Transformation are sourced
- """
- # CMF Type Definition
- meta_type = 'ERP5 Transformation Sourcing Rule'
- portal_type = 'Transformation Sourcing Rule'
- # Declarative security
- security = ClassSecurityInfo()
- security.declareObjectProtected(Permissions.AccessContentsInformation)
- __implements__ = ( interfaces.IPredicate,
- interfaces.IRule )
- # Default Properties
- property_sheets = ( PropertySheet.Base
+ security.declareObjectProtected(Permissions.AccessContentsInformation)
+ __implements__ = ( interfaces.IPredicate,
+ interfaces.IRule )
+ # Default Properties
+ property_sheets = ( PropertySheet.Base
, PropertySheet.XMLObject
, PropertySheet.CategoryCore
, PropertySheet.DublinCore
, PropertySheet.Task
)
- # Class variable
- simulation_movement_portal_type = "Simulation Movement"
+ def getFactory(self):
+ return SourcingMovementFactory()
- security.declareProtected(Permissions.ModifyPortalContent, 'expand')
- def expand(self, applied_rule, activate_kw=None,**kw):
- """
- Expands the current movement downward.
- -> new status -> expanded
- An applied rule can be expanded only if its parent movement
- is expanded.
- """
- parent_movement = applied_rule.getParentValue()
- # Calculate the previous supply link
- supply_chain = self.getSupplyChain(parent_movement.getParentValue())
- parent_supply_link = self.getCurrentSupplyLink(parent_movement)
- previous_supply_link_list = supply_chain.\
- getPreviousPackingListSupplyLinkList(
- parent_supply_link,
- movement=parent_movement)
- if len(previous_supply_link_list) == 0:
- raise TransformationSourcingRuleError,\
- "Expand must not be called on %r" %\
- applied_rule.getRelativeUrl()
- else:
- movement_dict = {}
- for previous_supply_link in previous_supply_link_list:
- # Calculate the source
- source_value = None
- source_node = previous_supply_link.getSourceValue()
- if source_node is not None:
- source_value = source_node.getDestinationValue()
- source_section_value = previous_supply_link.getSourceSectionValue()
- # Generate the dict
- stop_date = parent_movement.getStartDate()
- movement_dict.update({
- "ts": {
- 'source_value': source_value,
- 'source_section_value': source_section_value,
- 'destination_value': parent_movement.getSourceValue(),
- 'destination_section_value': \
- parent_movement.getSourceSectionValue(),
- 'resource_value': parent_movement.getResourceValue(),
- 'variation_category_list': parent_movement.\
- getVariationCategoryList(),
- "variation_property_dict": \
- parent_movement.getVariationPropertyDict(),
- 'quantity': parent_movement.getNetQuantity(), # getNetQuantity to support efficency from transformation
- 'price': parent_movement.getPrice(),
- 'quantity_unit': parent_movement.getQuantityUnit(),
- 'start_date': previous_supply_link.calculateStartDate(stop_date),
- 'stop_date': stop_date,
- 'deliverable': 1,
- # Save the value of the current supply link
- 'causality_value': previous_supply_link,
- }
- })
- # Build the movement
- self._buildMovementList(applied_rule, movement_dict,
- activate_kw=activate_kw)
- # Create one submovement which sources the transformation
- Rule.expand(self, applied_rule, activate_kw=activate_kw, **kw)
+ security.declareProtected(Permissions.ModifyPortalContent, 'expand')
+ def expand(self, applied_rule, **kw):
+ """
+ Expands the current movement downward.
+ -> new status -> expanded
+ An applied rule can be expanded only if its parent movement
+ is expanded.
+ """
+ parent_movement = applied_rule.getParentValue()
+ explanation = self.getExplanation(movement=parent_movement)
+ state = parent_movement.getCausalityValue().getPredecessorValue()
+ path_list = state.getSuccessorRelatedValueList()
- security.declareProtected(Permissions.ModifyPortalContent, 'solve')
- def solve(self, applied_rule, solution_list):
- """
- Solve inconsistency according to a certain number of solutions
- templates. This updates the
+ if len(path_list) == 0:
+ raise TransformationSourcingRuleError,\
+ "Not found deliverable business path"
+ if len(path_list) > 1:
+ raise TransformationSourcingRuleError,\
+ "Found 2 or more deliverable business path"
- -> new status -> solved
+ path = path_list[0]
- This applies a solution to an applied rule. Once
- the solution is applied, the parent movement is checked.
- If it does not diverge, the rule is reexpanded. If not,
- diverge is called on the parent movement.
- """
+ # source, source_section
+ source_section = path.getSourceSection() # only support a static access
+ source_method_id = path.getSourceMethodId()
+ if source_method_id is None:
+ source = path.getSource()
+ else:
+ source = getattr(path, source_method_id)()
+ # destination, destination_section
+ destination_section = path.getDestinationSection() # only support a static access
+ destination_method_id = path.getDestinationMethodId()
+ if destination_method_id is None:
+ destination = path.getDestination()
+ else:
+ destination = getattr(path, destination_method_id)()
- security.declareProtected(Permissions.ModifyPortalContent, 'diverge')
- def diverge(self, applied_rule):
- """
- -> new status -> diverged
+ start_date = path.getExpectedStartDate(explanation)
+ stop_date = path.getExpectedStopDate(explanation)
- This basically sets the rule to "diverged"
- and blocks expansion process
- """
+ quantity = parent_movement.getNetQuantity() * path.getQuantity()
+ price = parent_movement.getPrice()
+ if price is not None:
+ price *= path.getQuantity()
-# # Solvers
-# security.declareProtected(Permissions.View, 'isDivergent')
-# def isDivergent(self, applied_rule):
-# """
-# Returns 1 if divergent rule
-# """
-#
-# security.declareProtected(Permissions.View, 'getDivergenceList')
-# def getDivergenceList(self, applied_rule):
-# """
-# Returns a list Divergence descriptors
-# """
-#
-# security.declareProtected(Permissions.View, 'getSolverList')
-# def getSolverList(self, applied_rule):
-# """
-# Returns a list Divergence solvers
-# """
+ factory = self.getFactory()
+ factory.requestSourcing(
+ causality_value=path,
+ source=source,
+ source_section=source_section,
+ destination=destination,
+ destination_section=destination_section,
+ resource=parent_movement.getResource(),
+ variation_category_list=parent_movement.getVariationCategoryList(),
+ variation_property_dict=parent_movement.getVariationPropertyDict(),
+ quantity=quantity,
+ price=price,
+ quantity_unit=parent_movement.getQuantityUnit(),
+ start_date=start_date,
+ stop_date=stop_date,
+ deliverable=1,
+ )
- def isDeliverable(self, m):
- resource = m.getResource()
- if m.getResource() is None:
- return 0
- if resource.find('operation/') >= 0:
- return 0
- else:
- return 1
+ factory.makeMovements(applied_rule)
+ Rule.expand(self, applied_rule, **kw)
- def isOrderable(self, m):
- return 0
-
Modified: erp5/trunk/products/ERP5/tests/testMRP.py
URL: http://svn.erp5.org/erp5/trunk/products/ERP5/tests/testMRP.py?rev=27405&r1=27404&r2=27405&view=diff
==============================================================================
--- erp5/trunk/products/ERP5/tests/testMRP.py [utf8] (original)
+++ erp5/trunk/products/ERP5/tests/testMRP.py [utf8] Mon Jun 8 12:02:09 2009
@@ -37,14 +37,15 @@
from Products.CMFCore.utils import getToolByName
from Products.ERP5Type.tests.utils import reindex
-from Products.ERP5.Document.TransformationRule import TransformationRule
-
from Products.ERP5.tests.testBPMCore import TestBPMMixin
class TestMRPMixin(TestBPMMixin):
transformation_portal_type = 'Transformation'
transformed_resource_portal_type = 'Transformation Transformed Resource'
product_portal_type = 'Product'
+ organisation_portal_type = 'Organisation'
+ order_portal_type = 'Production Order'
+ order_line_portal_type = 'Production Order Line'
def setUpOnce(self):
self.portal = self.getPortalObject()
@@ -57,17 +58,38 @@
for rule in rule_tool.contentValues(
portal_type=rule_tool.getPortalRuleTypeList()):
rule.invalidate()
-
- def createTransformation(self):
+
+ def _createDocument(self, portal_type, **kw):
module = self.portal.getDefaultModule(
- portal_type=self.transformation_portal_type)
- return module.newContent(portal_type=self.transformation_portal_type)
-
- def createTransformedResource(self, transformation=None):
- if transformation is None:
- transformation = self.createTransformation()
- return transformation.newContent(
- portal_type=self.transformed_resource_portal_type)
+ portal_type=portal_type)
+ return self._createObject(module, portal_type, **kw)
+
+ def _createObject(self, parent, portal_type, id=None, **kw):
+ o = None
+ if id is not None:
+ o = parent.get(str(id), None)
+ if o is None:
+ o = parent.newContent(portal_type=portal_type)
+ o.edit(**kw)
+ return o
+
+ def createTransformation(self, **kw):
+ return self._createDocument(self.transformation_portal_type, **kw)
+
+ def createProduct(self, **kw):
+ return self._createDocument(self.product_portal_type, **kw)
+
+ def createOrganisation(self, **kw):
+ return self._createDocument(self.organisation_portal_type, **kw)
+
+ def createOrder(self, **kw):
+ return self._createDocument(self.order_portal_type, **kw)
+
+ def createOrderLine(self, order, **kw):
+ return self._createObject(order, self.order_line_portal_type, **kw)
+
+ def createTransformedResource(self, transformation, **kw):
+ return self._createObject(transformation, self.transformed_resource_portal_type, **kw)
@reindex
def createCategories(self):
@@ -78,39 +100,56 @@
self.createCategoriesInCategory(category_tool.trade_phase.mrp,
['p' + str(i) for i in range(5)]) # phase0 ~ 4
- def createProduct(self):
- module = self.portal.getDefaultModule(
- portal_type=self.product_portal_type)
- return module.newContent(portal_type=self.product_portal_type)
-
+ @reindex
+ def createDefaultOrder(self, transformation=None, business_process=None):
+ if transformation is None:
+ transformation = self.createDefaultTransformation()
+ if business_process is None:
+ business_process = self.createSimpleBusinessProcess()
+
+ base_date = DateTime()
+
+ order = self.createOrder(specialise_value=business_process,
+ start_date=base_date,
+ stop_date=base_date+3)
+ order_line = self.createOrderLine(order,
+ quantity=10,
+ resource=transformation.getResource(),
+ specialise_value=transformation)
+ # XXX in some case, specialise_value is not related to order_line by edit,
+ # but by setSpecialise() is ok, Why?
+ order_line.setSpecialiseValue(transformation)
+ return order
+
@reindex
def createDefaultTransformation(self):
- resource1 = self.createProduct()
- resource2 = self.createProduct()
- resource3 = self.createProduct()
- resource4 = self.createProduct()
- resource5 = self.createProduct()
- transformation = self.createTransformation()
- amount1 = self.createTransformedResource(transformation=transformation)
- amount2 = self.createTransformedResource(transformation=transformation)
- amount3 = self.createTransformedResource(transformation=transformation)
- amount4 = self.createTransformedResource(transformation=transformation)
-
- resource1.edit(title='product', quantity_unit_list=['weight/kg'])
- resource2.edit(title='triangle', quantity_unit_list=['weight/kg'])
- resource3.edit(title='box', quantity_unit_list=['weight/kg'])
- resource4.edit(title='circle', quantity_unit_list=['weight/kg'])
- resource5.edit(title='banana', quantity_unit_list=['weight/kg'])
-
- transformation.edit(resource_value=resource1)
- amount1.edit(resource_value=resource2, quantity=3,
- quantity_unit_list=['weight/kg'], trade_phase='mrp/p2')
- amount2.edit(resource_value=resource3, quantity=1,
- quantity_unit_list=['weight/kg'], trade_phase='mrp/p2')
- amount3.edit(resource_value=resource4, quantity=4,
- quantity_unit_list=['weight/kg'], trade_phase='mrp/p3')
- amount4.edit(resource_value=resource5, quantity=1,
- quantity_unit_list=['weight/kg'], trade_phase='mrp/p3')
+ resource1 = self.createProduct(id='1', quantity_unit_list=['weight/kg'])
+ resource2 = self.createProduct(id='2', quantity_unit_list=['weight/kg'])
+ resource3 = self.createProduct(id='3', quantity_unit_list=['weight/kg'])
+ resource4 = self.createProduct(id='4', quantity_unit_list=['weight/kg'])
+ resource5 = self.createProduct(id='5', quantity_unit_list=['weight/kg'])
+
+ transformation = self.createTransformation(resource_value=resource5)
+ self.createTransformedResource(transformation=transformation,
+ resource_value=resource1,
+ quantity=3,
+ quantity_unit_list=['weight/kg'],
+ trade_phase='mrp/p2')
+ self.createTransformedResource(transformation=transformation,
+ resource_value=resource2,
+ quantity=1,
+ quantity_unit_list=['weight/kg'],
+ trade_phase='mrp/p2')
+ self.createTransformedResource(transformation=transformation,
+ resource_value=resource3,
+ quantity=4,
+ quantity_unit_list=['weight/kg'],
+ trade_phase='mrp/p3')
+ self.createTransformedResource(transformation=transformation,
+ resource_value=resource4,
+ quantity=1,
+ quantity_unit_list=['weight/kg'],
+ trade_phase='mrp/p3')
return transformation
@reindex
@@ -125,18 +164,34 @@
business_state_partial = self.createBusinessState(business_process)
business_state_done = self.createBusinessState(business_process)
+ # organisations
+ source_section = self.createOrganisation(title='source_section')
+ source = self.createOrganisation(title='source')
+ destination_section = self.createOrganisation(title='destination_section')
+ destination = self.createOrganisation(title='destination')
+
business_process.edit(referential_date='stop_date')
business_path_p2.edit(id='p2',
predecessor_value=business_state_ready,
successor_value=business_state_partial,
quantity=1,
- trade_phase=['mrp/p2'])
+ trade_phase=['mrp/p2'],
+ source_section_value=source_section,
+ source_value=source,
+ destination_section_value=destination_section,
+ destination_value=destination,
+ )
business_path_p3.edit(id='p3',
predecessor_value=business_state_partial,
successor_value=business_state_done,
quantity=1,
deliverable=1, # root explanation
- trade_phase=['mrp/p3'])
+ trade_phase=['mrp/p3'],
+ source_section_value=source_section,
+ source_value=source,
+ destination_section_value=destination_section,
+ destination_value=destination,
+ )
return business_process
@reindex
@@ -151,19 +206,50 @@
business_state_ready = self.createBusinessState(business_process)
business_state_partial = self.createBusinessState(business_process)
+ # organisations
+ source_section = self.createOrganisation(title='source_section')
+ source = self.createOrganisation(title='source')
+ destination_section = self.createOrganisation(title='destination_section')
+ destination = self.createOrganisation(title='destination')
+
business_process.edit(referential_date='stop_date')
business_path_p2.edit(id='p2',
predecessor_value=business_state_ready,
successor_value=business_state_partial,
quantity=1,
- trade_phase=['mrp/p2'])
+ trade_phase=['mrp/p2'],
+ source_section_value=source_section,
+ source_value=source,
+ destination_section_value=destination_section,
+ destination_value=destination,
+ )
business_path_p3.edit(id='p3',
predecessor_value=business_state_ready,
successor_value=business_state_partial,
quantity=1,
deliverable=1, # root explanation
- trade_phase=['mrp/p3'])
+ trade_phase=['mrp/p3'],
+ source_section_value=source_section,
+ source_value=source,
+ destination_section_value=destination_section,
+ destination_value=destination,
+ )
return business_process
+
+ @reindex
+ def beforeTearDown(self):
+ super(TestMRPMixin, self).beforeTearDown()
+ transaction.abort()
+ for module in (
+ self.portal.organisation_module,
+ self.portal.production_order_module,
+ self.portal.transformation_module,
+ self.portal.business_process_module,
+ # don't remove document because reuse it for testing of id
+ # self.portal.product_module,
+ self.portal.portal_simulation,):
+ module.manage_delObjects(list(module.objectIds()))
+ transaction.commit()
class TestMRPImplementation(TestMRPMixin, ERP5TypeTestCase):
"""the test for implementation"""
@@ -181,25 +267,25 @@
set(rule.getHeadProductionPathList(transformation, business_process)))
def test_TransformationRule_expand(self):
- transformation = self.createDefaultTransformation()
-
- """
- Simple case
- """
- business_process = self.createSimpleBusinessProcess()
-
# mock order
- order = self.portal.production_order_module.newContent(portal_type="Production Order")
- order_line = order.newContent(portal_type="Production Order Line")
-
- base_date = DateTime()
- order.edit(specialise_value=business_process,
- start_date=base_date,
- stop_date=base_date+3,
- source_section_value=order,
- source_value=order)
- order_line.edit(quantity=10)
- order_line.setSpecialiseValue(transformation) # XXX Why can not define by edit?
+ order = self.createDefaultOrder()
+ order_line = order.objectValues()[0]
+
+ business_process = order.getSpecialiseValue()
+
+ # paths
+ path_p2 = '%s/p2' % business_process.getRelativeUrl()
+ path_p3 = '%s/p3' % business_process.getRelativeUrl()
+
+ # organisations
+ path = business_process.objectValues(
+ portal_type=self.portal.getPortalBusinessPathTypeList())[0]
+ source_section = path.getSourceSection()
+ source = path.getSource()
+ destination_section = path.getDestinationSection()
+ destination = path.getDestination()
+ consumed_organisations = (source_section, source, destination_section, None)
+ produced_organisations = (source_section, None, destination_section, destination)
# don't need another rules, just need TransformationRule for test
self.invalidateRules()
@@ -213,7 +299,7 @@
applied_rule.edit(causality_value=order)
movement.edit(order_value=order_line,
quantity=order_line.getQuantity(),
- resource=transformation.getResource())
+ resource=order_line.getResource())
# test mock
applied_rule = movement.newContent(potal_type='Applied Rule')
@@ -222,28 +308,51 @@
# assertion
expected_value_set = set([
- (('business_process_module/1/p2',), 'product_module/1', 'mrp/p3', -10),
- (('business_process_module/1/p2',), 'product_module/2', 'mrp/p2', 30),
- (('business_process_module/1/p2',), 'product_module/3', 'mrp/p2', 10),
- (('business_process_module/1/p3',), 'product_module/1', 'mrp/p3', 10),
- (('business_process_module/1/p3',), 'product_module/4', 'mrp/p3', 40),
- (('business_process_module/1/p3',), 'product_module/5', 'mrp/p3', 10),
- (('business_process_module/1/p3',), 'product_module/1', None, -10)])
+ ((path_p2,), 'product_module/5', produced_organisations, 'mrp/p3', -10),
+ ((path_p2,), 'product_module/1', consumed_organisations, 'mrp/p2', 30),
+ ((path_p2,), 'product_module/2', consumed_organisations, 'mrp/p2', 10),
+ ((path_p3,), 'product_module/5', consumed_organisations, 'mrp/p3', 10),
+ ((path_p3,), 'product_module/3', consumed_organisations, 'mrp/p3', 40),
+ ((path_p3,), 'product_module/4', consumed_organisations, 'mrp/p3', 10),
+ ((path_p3,), 'product_module/5', produced_organisations, None, -10)])
movement_list = applied_rule.objectValues()
self.assertEquals(len(expected_value_set), len(movement_list))
movement_value_set = set([])
for movement in movement_list:
movement_value_set |= set([(tuple(movement.getCausalityList()),
movement.getResource(),
+ (movement.getSourceSection(),
+ movement.getSource(),
+ movement.getDestinationSection(),
+ movement.getDestination(),), # organisations
movement.getTradePhase(),
movement.getQuantity())])
self.assertEquals(expected_value_set, movement_value_set)
- """
- Concurrent case
- """
+ def test_TransformationRule_expand_concurrent(self):
business_process = self.createConcurrentBusinessProcess()
- order.edit(specialise_value=business_process)
+
+ # mock order
+ order = self.createDefaultOrder(business_process=business_process)
+ order_line = order.objectValues()[0]
+
+ # phases
+ phase_p2 = '%s/p2' % business_process.getRelativeUrl()
+ phase_p3 = '%s/p3' % business_process.getRelativeUrl()
+
+ # organisations
+ path = business_process.objectValues(
+ portal_type=self.portal.getPortalBusinessPathTypeList())[0]
+ source_section = path.getSourceSection()
+ source = path.getSource()
+ destination_section = path.getDestinationSection()
+ destination = path.getDestination()
+ organisations = (source_section, source, destination_section, destination)
+ consumed_organisations = (source_section, source, destination_section, None)
+ produced_organisations = (source_section, None, destination_section, destination)
+
+ # don't need another rules, just need TransformationRule for test
+ self.invalidateRules()
self.stepTic()
@@ -254,7 +363,7 @@
applied_rule.edit(causality_value=order)
movement.edit(order_value=order_line,
quantity=order_line.getQuantity(),
- resource=transformation.getResource())
+ resource=order_line.getResource())
# test mock
applied_rule = movement.newContent(potal_type='Applied Rule')
@@ -263,58 +372,159 @@
# assertion
expected_value_set = set([
- (('business_process_module/2/p2',), 'product_module/2', 'mrp/p2', 30),
- (('business_process_module/2/p2',), 'product_module/3', 'mrp/p2', 10),
- (('business_process_module/2/p3',), 'product_module/4', 'mrp/p3', 40),
- (('business_process_module/2/p3',), 'product_module/5', 'mrp/p3', 10),
- (('business_process_module/2/p2', 'business_process_module/2/p3'), 'product_module/1', None, -10)])
+ ((phase_p2,), 'product_module/1', consumed_organisations, 'mrp/p2', 30),
+ ((phase_p2,), 'product_module/2', consumed_organisations, 'mrp/p2', 10),
+ ((phase_p3,), 'product_module/3', consumed_organisations, 'mrp/p3', 40),
+ ((phase_p3,), 'product_module/4', consumed_organisations, 'mrp/p3', 10),
+ ((phase_p2, phase_p3), 'product_module/5', produced_organisations, None, -10)])
movement_list = applied_rule.objectValues()
self.assertEquals(len(expected_value_set), len(movement_list))
movement_value_set = set([])
for movement in movement_list:
movement_value_set |= set([(tuple(movement.getCausalityList()),
movement.getResource(),
+ (movement.getSourceSection(),
+ movement.getSource(),
+ movement.getDestinationSection(),
+ movement.getDestination(),), # organisations
movement.getTradePhase(),
movement.getQuantity())])
self.assertEquals(expected_value_set, movement_value_set)
+ def test_TransformationRule_expand_reexpand(self):
"""
test case of difference when any movement are frozen
by using above result
"""
- # update relation
+ self.test_TransformationRule_expand_concurrent()
+
self.stepTic()
- for movement in movement_list:
+ applied_rule = self.portal.portal_simulation.objectValues()[0]
+
+ business_process = applied_rule.getCausalityValue().getSpecialiseValue()
+
+ # phases
+ phase_p2 = '%s/p2' % business_process.getRelativeUrl()
+ phase_p3 = '%s/p3' % business_process.getRelativeUrl()
+
+ # organisations
+ path = business_process.objectValues(
+ portal_type=self.portal.getPortalBusinessPathTypeList())[0]
+ source_section = path.getSourceSection()
+ source = path.getSource()
+ destination_section = path.getDestinationSection()
+ destination = path.getDestination()
+ consumed_organisations = (source_section, source, destination_section, None)
+ produced_organisations = (source_section, None, destination_section, destination)
+
+ movement = applied_rule.objectValues()[0]
+ applied_rule = movement.objectValues()[0]
+
+ # these movements are made by transformation
+ for movement in applied_rule.objectValues():
movement.edit(quantity=1)
- # XXX change state isFrozen of movement to 1,
- # but I think this way might be wrong.
+ # set the state value of isFrozen to 1,
movement._baseSetFrozen(1)
# re-expand
+ rule = self.portal.portal_rules.default_transformation_rule
rule.expand(applied_rule)
# assertion
expected_value_set = set([
- (('business_process_module/2/p2',), 'product_module/2', 'mrp/p2', 1), # Frozen
- (('business_process_module/2/p2',), 'product_module/2', 'mrp/p2', 29),
- (('business_process_module/2/p2',), 'product_module/3', 'mrp/p2', 1), # Frozen
- (('business_process_module/2/p2',), 'product_module/3', 'mrp/p2', 9),
- (('business_process_module/2/p3',), 'product_module/4', 'mrp/p3', 1), # Frozen
- (('business_process_module/2/p3',), 'product_module/4', 'mrp/p3', 39),
- (('business_process_module/2/p3',), 'product_module/5', 'mrp/p3', 1), # Frozen
- (('business_process_module/2/p3',), 'product_module/5', 'mrp/p3', 9),
- (('business_process_module/2/p2', 'business_process_module/2/p3'), 'product_module/1', None, 1), # Frozen
- (('business_process_module/2/p2', 'business_process_module/2/p3'), 'product_module/1', None, -11)])
+ ((phase_p2,), 'product_module/1', consumed_organisations, 'mrp/p2', 1), # Frozen
+ ((phase_p2,), 'product_module/1', consumed_organisations, 'mrp/p2', 29),
+ ((phase_p2,), 'product_module/2', consumed_organisations, 'mrp/p2', 1), # Frozen
+ ((phase_p2,), 'product_module/2', consumed_organisations, 'mrp/p2', 9),
+ ((phase_p3,), 'product_module/3', consumed_organisations, 'mrp/p3', 1), # Frozen
+ ((phase_p3,), 'product_module/3', consumed_organisations, 'mrp/p3', 39),
+ ((phase_p3,), 'product_module/4', consumed_organisations, 'mrp/p3', 1), # Frozen
+ ((phase_p3,), 'product_module/4', consumed_organisations, 'mrp/p3', 9),
+ ((phase_p2, phase_p3), 'product_module/5', produced_organisations, None, 1), # Frozen
+ ((phase_p2, phase_p3), 'product_module/5', produced_organisations, None, -11)])
movement_list = applied_rule.objectValues()
self.assertEquals(len(expected_value_set), len(movement_list))
movement_value_set = set([])
for movement in movement_list:
movement_value_set |= set([(tuple(movement.getCausalityList()),
movement.getResource(),
+ (movement.getSourceSection(),
+ movement.getSource(),
+ movement.getDestinationSection(),
+ movement.getDestination(),), # organisations
movement.getTradePhase(),
movement.getQuantity())])
self.assertEquals(expected_value_set, movement_value_set)
+
+ def test_TransformationSourcingRule_expand(self):
+ # mock order
+ order = self.createDefaultOrder()
+ order_line = order.objectValues()[0]
+
+ # don't need another rules, just need TransformationSourcingRule for test
+ self.invalidateRules()
+
+ self.stepTic()
+
+ business_process = order.getSpecialiseValue()
+
+ # get last path of a business process
+ # in simple business path, the last is between "partial_produced" and "done"
+ causality_path = None
+ for state in business_process.objectValues(
+ portal_type=self.portal.getPortalBusinessStateTypeList()):
+ if len(state.getRemainingTradePhaseList(self.portal)) == 0:
+ causality_path = state.getSuccessorRelatedValue()
+
+ # phases
+ phase_p2 = '%s/p2' % business_process.getRelativeUrl()
+
+ # organisations
+ source_section = causality_path.getSourceSection()
+ source = causality_path.getSource()
+ destination_section = causality_path.getDestinationSection()
+ destination = causality_path.getDestination()
+ organisations = (source_section, source, destination_section, destination)
+
+ # sourcing resource
+ sourcing_resource = order_line.getResource()
+
+ # alter simulations of the order
+ # root
+ applied_rule = self.portal.portal_simulation.newContent(portal_type='Applied Rule')
+ movement = applied_rule.newContent(portal_type='Simulation Movement')
+ applied_rule.edit(causality_value=order)
+ movement.edit(order_value=order_line,
+ causality_value=causality_path,
+ quantity=order_line.getQuantity(),
+ resource=sourcing_resource,
+ )
+
+ self.stepTic()
+
+ # test mock
+ applied_rule = movement.newContent(potal_type='Applied Rule')
+
+ rule = self.portal.portal_rules.default_transformation_sourcing_rule
+ rule.expand(applied_rule)
+
+ # assertion
+ expected_value_set = set([
+ ((phase_p2,), sourcing_resource, organisations, 10)])
+ movement_list = applied_rule.objectValues()
+ self.assertEquals(len(expected_value_set), len(movement_list))
+ movement_value_set = set([])
+ for movement in movement_list:
+ movement_value_set |= set([(tuple(movement.getCausalityList()),
+ movement.getResource(),
+ (movement.getSourceSection(),
+ movement.getSource(),
+ movement.getDestinationSection(),
+ movement.getDestination(),), # organisations
+ movement.getQuantity())])
+ self.assertEquals(expected_value_set, movement_value_set)
+
def test_suite():
suite = unittest.TestSuite()
More information about the Erp5-report
mailing list