Skip to content

erp5
erp5
Commit 662d8798 authored by Yusuke Muraoka's avatar Yusuke Muraoka
Browse files
Options

- changed the TransformationSourcingRule to refer a business process 

  instead of a supply chain
- added a test for TransformationSourcingRule


git-svn-id: https://svn.erp5.org/repos/public/erp5/trunk@27405 20353a03-c40f-0410-a6d1-a30d3c3de9de
parent 73ad6938
Hide whitespace changes
Inline Side-by-side
Showing with 464 additions and 346 deletions
product/ERP5/Document/TransformationRule.py
View file @ 662d8798
......@@ -38,17 +38,12 @@ from Products.ERP5.Document.Rule import Rule
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 @@ class TransformationMovementFactory:
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 @@ class TransformationMovementFactory:
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 @@ class TransformationMovementFactory:
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 @@ class TransformationRule(TransformationRuleMixin, Rule):
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 @@ class TransformationRule(TransformationRuleMixin, Rule):
% (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 @@ class TransformationRule(TransformationRuleMixin, Rule):
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 @@ class TransformationRule(TransformationRuleMixin, Rule):
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 @@ class TransformationRule(TransformationRuleMixin, Rule):
# 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 @@ class TransformationRule(TransformationRuleMixin, Rule):
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 @@ class TransformationRule(TransformationRuleMixin, Rule):
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 @@ which last_phase_path_list is empty.""" % (transformation, business_process)
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)
......
product/ERP5/Document/TransformationSourcingRule.py
View file @ 662d8798
......@@ -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"
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, 'solve')
def solve(self, applied_rule, solution_list):
"""
Solve inconsistency according to a certain number of solutions
templates. This updates the
-> new status -> solved
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.
"""
security.declareProtected(Permissions.ModifyPortalContent, 'diverge')
def diverge(self, applied_rule):
"""
-> new status -> diverged
def getFactory(self):
return SourcingMovementFactory()
This basically sets the rule to "diverged"
and blocks expansion process
"""
# # 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
# """
def isDeliverable(self, m):
resource = m.getResource()
if m.getResource() is None:
return 0
if resource.find('operation/') >= 0:
return 0
else:
return 1
def isOrderable(self, m):
return 0
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()
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"
path = path_list[0]
# 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)
quantity = parent_movement.getNetQuantity() * path.getQuantity()
price = parent_movement.getPrice()
if price is not None:
price *= path.getQuantity()
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,
)
factory.makeMovements(applied_rule)
Rule.expand(self, applied_rule, **kw)
product/ERP5/tests/testMRP.py
View file @ 662d8798
......@@ -37,14 +37,15 @@ from Products.ERP5Type.tests.Sequence import SequenceList
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 @@ class TestMRPMixin(TestBPMMixin):
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)
portal_type=portal_type)
return self._createObject(module, portal_type, **kw)
def createTransformedResource(self, transformation=None):
if transformation is None:
transformation = self.createTransformation()
return transformation.newContent(
portal_type=self.transformed_resource_portal_type)
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 @@ class TestMRPMixin(TestBPMMixin):
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 @@ class TestMRPMixin(TestBPMMixin):
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,20 +206,51 @@ class TestMRPMixin(TestBPMMixin):
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"""
def test_TransformationRule_getHeadProductionPathList(self):
......@@ -181,25 +267,25 @@ class TestMRPImplementation(TestMRPMixin, ERP5TypeTestCase):
set(rule.getHeadProductionPathList(transformation, business_process)))
def test_TransformationRule_expand(self):
transformation = self.createDefaultTransformation()
# mock order
order = self.createDefaultOrder()
order_line = order.objectValues()[0]
"""
Simple case
"""
business_process = self.createSimpleBusinessProcess()
business_process = order.getSpecialiseValue()
# mock order
order = self.portal.production_order_module.newContent(portal_type="Production Order")
order_line = order.newContent(portal_type="Production Order Line")
# paths
path_p2 = '%s/p2' % business_process.getRelativeUrl()
path_p3 = '%s/p3' % business_process.getRelativeUrl()
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?
# 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 @@ class TestMRPImplementation(TestMRPMixin, ERP5TypeTestCase):
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 @@ class TestMRPImplementation(TestMRPMixin, ERP5TypeTestCase):
# 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 @@ class TestMRPImplementation(TestMRPMixin, ERP5TypeTestCase):
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,59 +372,160 @@ class TestMRPImplementation(TestMRPMixin, ERP5TypeTestCase):
# 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()
suite.addTest(unittest.makeSuite(TestMRPImplementation))
......
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
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7