Skip to content
Projects
Groups
Snippets
Help
Loading...
Help
Support
Keyboard shortcuts
?
Submit feedback
Contribute to GitLab
Sign in / Register
Toggle navigation
erp5
Project overview
Project overview
Details
Activity
Releases
Repository
Repository
Files
Commits
Branches
Tags
Contributors
Graph
Compare
Issues
0
Issues
0
List
Boards
Labels
Milestones
Merge Requests
0
Merge Requests
0
CI / CD
CI / CD
Pipelines
Jobs
Schedules
Analytics
Analytics
CI / CD
Repository
Value Stream
Wiki
Wiki
Snippets
Snippets
Members
Members
Collapse sidebar
Close sidebar
Activity
Graph
Create a new issue
Jobs
Commits
Issue Boards
Open sidebar
Thomas Gambier
erp5
Commits
a16b5c77
Commit
a16b5c77
authored
Sep 03, 2020
by
Arnaud Fontaine
Browse files
Options
Browse Files
Download
Email Patches
Plain Diff
ZODB Components: erp5_simulation_performance_test: Migrate Unit Test.
parent
ba0d6c7b
Changes
5
Show whitespace changes
Inline
Side-by-side
Showing
5 changed files
with
945 additions
and
832 deletions
+945
-832
bt5/erp5_simulation_performance_test/TestTemplateItem/portal_components/test.erp5.testSimulationElementPerformance.py
..._components/test.erp5.testSimulationElementPerformance.py
+832
-0
bt5/erp5_simulation_performance_test/TestTemplateItem/portal_components/test.erp5.testSimulationElementPerformance.xml
...components/test.erp5.testSimulationElementPerformance.xml
+104
-0
bt5/erp5_simulation_performance_test/bt/template_test_id_list
...erp5_simulation_performance_test/bt/template_test_id_list
+1
-0
bt5/erp5_simulation_performance_test/bt/test_dependency_list
bt5/erp5_simulation_performance_test/bt/test_dependency_list
+8
-0
product/ERP5/tests/testSimulationElementPerformance.py
product/ERP5/tests/testSimulationElementPerformance.py
+0
-832
No files found.
bt5/erp5_simulation_performance_test/TestTemplateItem/portal_components/test.erp5.testSimulationElementPerformance.py
0 → 100644
View file @
a16b5c77
##############################################################################
#
# Copyright (c) 2010 Nexedi SA and Contributors. All Rights Reserved.
# Yoshinori Okuji <yo@nexedi.com>
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsability 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
# garantees 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.
#
##############################################################################
"""Performance tests for simulation.
The purpose of these tests is to measure the performance of the following
basic operations in simulation:
- the creation of new Applied Rules and the expansion of simulation
- the expansion of simulation from existing Applied Rules
- the expansion of simulation which results in convergence
- the expansion of simulation which results in divergence
- the adoption of previsions by a Target Solver
- the acceptance of decisions by a Target Solver
- the creation of new Applied Rules from partially simulated documents
- the creation of new Deliveries by Delivery Builders
- the addition of new Movements into existing Deliveries by Delivery Builders
For the testing environment, these document types are used for Deliveries:
- Sale Order
- Sale Packing List
- Sale Invoice
Since Sale Invoices consist of Movements generated from Sale Packing Lists,
and Movements generated by Trade Model Lines, the simulation trees consist
of at most 4 levels in this order:
- Order Simulation Rule
- Packing List Simulation Rule
- Invoice Simulation Rule
- Trade Model Simulation Rule
In order to simulate various patterns of building, the following rules
are defined:
- Sale Order is completed at "ordered"
- Sale Packing List is completed at "stopped"
- Sale Packing Lists are grouped by destination and by Sale Order
- Sale Invoices are not
So, the relationship between Sale Order and Sale Packing List is 1:N,
and that between Sale Packing List and Sale Invoice is M:N.
"""
import
unittest
from
time
import
time
import
gc
import
subprocess
from
DateTime
import
DateTime
from
Products.ERP5Type.tests.ERP5TypeTestCase
import
ERP5TypeTestCase
from
zLOG
import
LOG
from
Products.ERP5Type.tests.utils
import
LogInterceptor
from
Products.ERP5Type.tests.Sequence
import
SequenceList
# Define variable to chek if performance are good or not
# XXX These variable are specific to the testing environment
# (pystone results: min: < 131578.9 - mean: ~ 139768.5 - max: > 147058.8)
# Historical values are here to remember original values on this
# specific testing environment. We must always try to stay below max
# historical values.
# Only for information
HISTORICAL_EXPECTED_MIN_MAX_TIME
=
{
'Creation Of New Applied Rules'
:
(
34.8
,
36.8
),
'Expansion of Existing Applied Rules'
:
(
16.9
,
19.5
),
'Creation of New Sale Packing Lists'
:
(
37.4
,
38.9
),
'Expansion of Converged Changes'
:
(
22.0
,
24.8
),
'Expansion of Diverged Changes'
:
(
24.6
,
26.9
),
'Adoption of Previsions'
:
(
17.9
,
18.7
),
'Acceptance of Decisions'
:
(
14.5
,
15.2
),
'Creation of New Applied Rules from Partially Simulated Deliveries'
:
(
51.2
,
52.8
),
'Creation Of New Sale Invoices'
:
(
90.0
,
98.5
),
'Addition of New Invoices Lines'
:
(
145.1
,
157.6
),
}
EXPECTED_MIN_MAX_TIME
=
{
'Creation Of New Applied Rules'
:
(
7.38
,
11.38
),
# (30, 35)
'Expansion of Existing Applied Rules'
:
(
9.98
,
15.98
),
# (41, 47)
'Creation of New Sale Packing Lists'
:
(
6.12
,
10.12
),
# (27, 32)
'Expansion of Converged Changes'
:
(
11.12
,
17.12
),
# (48, 54)
'Expansion of Diverged Changes'
:
(
18.58
,
24.58
),
# (53, 59)
'Adoption of Previsions'
:
(
26.78
,
34.78
),
# (105, 114)
'Acceptance of Decisions'
:
(
27.3
,
35.3
),
# (104, 113)
'Creation of New Applied Rules from Partially Simulated Deliveries'
:
(
9.05
,
15.05
),
# (38, 44)
'Creation Of New Sale Invoices'
:
(
11.05
,
17.05
),
# (48, 54)
'Addition of New Invoices Lines'
:
(
29.11
,
37.11
)
# (116, 125)
}
class
TestSimulationPerformance
(
ERP5TypeTestCase
,
LogInterceptor
):
def
getTitle
(
self
):
return
"Simulation Performance"
def
getBusinessTemplateList
(
self
):
"""
Return the list of business templates.
"""
return
(
'erp5_core_proxy_field_legacy'
,
'erp5_base'
,
'erp5_pdm'
,
'erp5_simulation'
,
'erp5_trade'
,
'erp5_accounting'
,
'erp5_invoicing'
,
'erp5_simplified_invoicing'
,
'erp5_simulation_performance_test'
)
def
afterSetUp
(
self
):
"""
Executed before each test_*.
"""
self
.
login
()
portal
=
self
.
getPortal
()
# Validate some documents.
for
rule
in
portal
.
portal_rules
.
objectValues
():
if
rule
.
getValidationState
()
!=
'validated'
\
and
rule
.
getId
().
startswith
(
'test_'
):
rule
.
validate
()
for
trade_condition
in
portal
.
sale_trade_condition_module
.
objectValues
():
if
trade_condition
.
getValidationState
()
!=
'validated'
:
trade_condition
.
validate
()
business_process
=
portal
.
business_process_module
.
test_bp
if
business_process
.
getValidationState
()
!=
'validated'
:
business_process
.
validate
()
subprocess
.
call
(
'sync'
)
#self.prof = hotshot.Profile('hoge.prof')
def
beforeTearDown
(
self
):
#self.prof.close()
self
.
abort
()
portal
=
self
.
getPortal
()
for
module_id
in
(
'sale_order_module'
,
'sale_packing_list_module'
,
'accounting_module'
,
'portal_simulation'
):
module
=
portal
[
module_id
]
module
.
manage_delObjects
(
list
(
module
.
objectIds
()))
self
.
tic
()
def
_getMinMaxTime
(
self
,
target
):
return
EXPECTED_MIN_MAX_TIME
[
target
]
def
_checkSimulation
(
self
,
number_of_sale_orders
=
None
,
number_of_sale_order_lines
=
None
,
number_of_additional_sale_packing_list_lines
=
None
,
measurable
=
True
):
sequence_string
=
"""
Initialize
Tic
UseClientAsDestinationDecision
UseClientAsDestinationAdministration
UseResourceAsCommodityProduct
CreateSaleOrders
Tic
PlanSaleOrders
MeasureCreationOfNewAppliedRules
Tic
CheckSaleOrderSimulation
UseRecipient1AsDestinationDecision
ModifySaleOrders
MeasureExpansionOfExistingAppliedRules
Tic
CheckSaleOrderSimulation
OrderSaleOrders
MeasureCreationOfNewSalePackingLists
Tic
CheckSaleOrderSimulation
CheckBuiltSalePackingLists
UseRecipient2AsDestinationDecision
ModifySaleOrders
MeasureExpansionOfConvergedChanges
Tic
CheckSaleOrderSimulation
CheckConvergedSalePackingLists
UseRecipient1AsDestinationAdministration
ModifySaleOrders
MeasureExpansionOfDivergedChanges
Tic
CheckSaleOrderSimulation
CheckDivergedSalePackingLists
SetSolverTargetToDestinationAdministration
AdoptPrevisionToSalePackingLists
MeasureAdoptionOfPrevisions
Tic
CheckConvergedSalePackingLists
UseRecipient2AsDestinationAdministration
ModifySaleOrders
Tic
CheckSaleOrderSimulation
CheckDivergedSalePackingLists
SetSolverTargetToDestinationAdministration
AcceptDecisionOfSalePackingLists
MeasureAcceptanceOfDecisions
Tic
UseRecipient1AsDestinationAdministration
CheckConvergedSalePackingLists
ConfirmSaleOrders
Tic
UseResourceAsCommodityProduct
AddNewSalePackingLinesIntoSalePackingLists
MeasureCreationOfNewAppliedRulesFromPartiallySimulatedDeliveries
Tic
CheckSalePackingListSimulation
StopSalePackingLists
MeasureCreationOfNewSaleInvoices
Tic
CheckBuiltSaleInvoices
UseResourceAsLuxuryProduct
AddNewSalePackingLinesIntoSalePackingLists
DeliverSalePackingLists
MeasureAdditionOfNewInvoiceLinesIntoSaleInvoices
Tic
CheckAddedSaleInvoices
TestResult
"""
sequence
=
SequenceList
()
self
.
assertTrue
(
isinstance
(
number_of_sale_orders
,
int
))
self
.
assertTrue
(
number_of_sale_orders
>
0
)
self
.
assertTrue
(
isinstance
(
number_of_sale_order_lines
,
int
))
self
.
assertTrue
(
number_of_sale_order_lines
>
0
)
self
.
assertTrue
(
isinstance
(
number_of_additional_sale_packing_list_lines
,
int
))
self
.
assertTrue
(
number_of_additional_sale_packing_list_lines
>
0
)
self
.
assertTrue
(
measurable
in
(
True
,
False
))
self
.
_init_dict
=
dict
(
number_of_sale_orders
=
number_of_sale_orders
,
number_of_sale_order_lines
=
number_of_sale_order_lines
,
number_of_additional_sale_packing_list_lines
=
number_of_additional_sale_packing_list_lines
,
measurable
=
measurable
)
sequence
.
addSequenceString
(
sequence_string
)
sequence
.
play
(
self
)
def
stepInitialize
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
edit
(
**
self
.
_init_dict
)
def
stepTic
(
self
,
sequence
,
sequence_list
=
None
,
**
kw
):
target
=
sequence
.
get
(
'measure_target'
)
measurable
=
target
and
sequence
.
get
(
'measurable'
)
# Reduce noise effect.
subprocess
.
call
(
'sync'
)
gc
.
collect
()
if
measurable
:
before_time
=
time
()
#self.prof.start()
self
.
tic
()
#self.prof.stop()
if
measurable
:
after_time
=
time
()
amount_of_time
=
after_time
-
before_time
min_time
,
max_time
=
self
.
_getMinMaxTime
(
target
)
print
"
\
n
%s took %.4f (%.4f < %.4f < %.4f)"
\
%
(
target
,
amount_of_time
,
min_time
,
amount_of_time
,
max_time
)
# Reset the target to make sure that the same target is not
# measured again.
sequence
.
edit
(
measure_target
=
None
)
# Set the result.
result
=
sequence
.
setdefault
(
'result'
,
[])
result
.
append
((
target
,
min_time
,
amount_of_time
,
max_time
))
def
stepMeasureCreationOfNewAppliedRules
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Creation Of New Applied Rules'
)
def
stepMeasureExpansionOfExistingAppliedRules
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Expansion of Existing Applied Rules'
)
def
stepMeasureCreationOfNewSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Creation of New Sale Packing Lists'
)
def
stepMeasureExpansionOfConvergedChanges
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Expansion of Converged Changes'
)
def
stepMeasureExpansionOfDivergedChanges
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Expansion of Diverged Changes'
)
def
stepMeasureAdoptionOfPrevisions
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Adoption of Previsions'
)
def
stepMeasureAcceptanceOfDecisions
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Acceptance of Decisions'
)
def
stepMeasureCreationOfNewAppliedRulesFromPartiallySimulatedDeliveries
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Creation of New Applied Rules from Partially Simulated Deliveries'
)
def
stepMeasureCreationOfNewSaleInvoices
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Creation Of New Sale Invoices'
)
def
stepMeasureAdditionOfNewInvoiceLinesIntoSaleInvoices
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Addition of New Invoices Lines'
)
def
stepUseClientAsDestinationDecision
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'destination_decision'
,
'organisation_module/client'
)
def
stepUseRecipient1AsDestinationDecision
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'destination_decision'
,
'organisation_module/recipient1'
)
def
stepUseRecipient2AsDestinationDecision
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'destination_decision'
,
'organisation_module/recipient2'
)
def
stepUseClientAsDestinationAdministration
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'destination_administration'
,
'organisation_module/client'
)
def
stepUseRecipient1AsDestinationAdministration
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'destination_administration'
,
'organisation_module/recipient1'
)
def
stepUseRecipient2AsDestinationAdministration
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'destination_administration'
,
'organisation_module/recipient2'
)
def
stepUseResourceAsCommodityProduct
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'resource'
,
'product_module/commodity'
)
def
stepUseResourceAsLuxuryProduct
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'resource'
,
'product_module/luxury'
)
def
stepCreateSaleOrders
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_order_module
base_date
=
DateTime
(
'2010-08-08'
)
number_of_sale_orders
=
sequence
.
get
(
'number_of_sale_orders'
)
number_of_sale_order_lines
=
sequence
.
get
(
'number_of_sale_order_lines'
)
destination_decision
=
sequence
.
get
(
'destination_decision'
)
destination_administration
=
sequence
.
get
(
'destination_administration'
)
resource
=
sequence
.
get
(
'resource'
)
for
i
in
xrange
(
number_of_sale_orders
):
start_date
=
base_date
+
i
stop_date
=
base_date
+
i
+
1
order
=
module
.
newContent
(
portal_type
=
'Sale Order'
,
specialise
=
'sale_trade_condition_module/test_stc'
,
destination_decision
=
destination_decision
,
destination_administration
=
destination_administration
,
start_date
=
start_date
,
stop_date
=
stop_date
)
# Set the rest through the trade condition.
order
.
SaleOrder_applySaleTradeCondition
()
for
_
in
xrange
(
number_of_sale_order_lines
):
order
.
newContent
(
portal_type
=
'Sale Order Line'
,
resource
=
resource
,
quantity
=
1.0
)
def
stepModifySaleOrders
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_order_module
destination_decision
=
sequence
.
get
(
'destination_decision'
)
destination_administration
=
sequence
.
get
(
'destination_administration'
)
for
order
in
module
.
contentValues
(
portal_type
=
'Sale Order'
):
order
.
edit
(
destination_decision
=
destination_decision
,
destination_administration
=
destination_administration
)
self
.
assertEqual
(
order
.
getDestinationDecision
(),
destination_decision
)
self
.
assertEqual
(
order
.
getDestinationAdministration
(),
destination_administration
)
def
stepPlanSaleOrders
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_order_module
for
order
in
module
.
contentValues
(
portal_type
=
'Sale Order'
):
self
.
assertEqual
(
order
.
getSimulationState
(),
'draft'
)
order
.
plan
()
self
.
assertEqual
(
order
.
getSimulationState
(),
'planned'
)
def
stepOrderSaleOrders
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_order_module
for
order
in
module
.
contentValues
(
portal_type
=
'Sale Order'
):
self
.
assertEqual
(
order
.
getSimulationState
(),
'planned'
)
order
.
order
()
self
.
assertEqual
(
order
.
getSimulationState
(),
'ordered'
)
def
stepConfirmSaleOrders
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_order_module
for
order
in
module
.
contentValues
(
portal_type
=
'Sale Order'
):
self
.
assertEqual
(
order
.
getSimulationState
(),
'ordered'
)
order
.
confirm
()
self
.
assertEqual
(
order
.
getSimulationState
(),
'confirmed'
)
def
stepCheckSaleOrderSimulation
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_order_module
number_of_sale_orders
=
sequence
.
get
(
'number_of_sale_orders'
)
number_of_sale_order_lines
=
sequence
.
get
(
'number_of_sale_order_lines'
)
destination_decision
=
sequence
.
get
(
'destination_decision'
)
destination_administration
=
sequence
.
get
(
'destination_administration'
)
resource
=
sequence
.
get
(
'resource'
)
self
.
assertEqual
(
module
.
objectCount
(),
number_of_sale_orders
)
for
order
in
module
.
contentValues
(
portal_type
=
'Sale Order'
):
causality_list
=
order
.
getCausalityRelatedValueList
(
portal_type
=
'Applied Rule'
)
self
.
assertEqual
(
len
(
causality_list
),
1
)
applied_rule
=
causality_list
[
0
]
self
.
assertEqual
(
applied_rule
.
getPortalType
(),
'Applied Rule'
)
rule
=
applied_rule
.
getSpecialiseValue
()
self
.
assertNotEquals
(
rule
,
None
)
self
.
assertEqual
(
rule
.
getReference
(),
'test_order_root_simulation_rule'
)
self
.
assertEqual
(
applied_rule
.
objectCount
(),
number_of_sale_order_lines
)
for
simulation_movement
in
applied_rule
.
objectValues
():
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
self
.
assertEqual
(
simulation_movement
.
getQuantity
(),
1.0
)
self
.
assertEqual
(
simulation_movement
.
getResource
(),
resource
)
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),
{
'business_process_module/test_bp/order_path'
,
'business_process_module/test_bp/order'
})
self
.
assertEqual
(
simulation_movement
.
getDestinationDecision
(),
destination_decision
)
self
.
assertEqual
(
simulation_movement
.
getDestinationAdministration
(),
destination_administration
)
def
stepCheckBuiltSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
number_of_sale_orders
=
sequence
.
get
(
'number_of_sale_orders'
)
number_of_sale_order_lines
=
sequence
.
get
(
'number_of_sale_order_lines'
)
destination_decision
=
sequence
.
get
(
'destination_decision'
)
destination_administration
=
sequence
.
get
(
'destination_administration'
)
resource
=
sequence
.
get
(
'resource'
)
self
.
assertEqual
(
module
.
objectCount
(),
number_of_sale_orders
)
for
packing_list
in
module
.
contentValues
():
self
.
assertEqual
(
packing_list
.
getSimulationState
(),
'confirmed'
)
self
.
assertEqual
(
packing_list
.
getCausalityState
(),
'solved'
)
self
.
assertEqual
(
packing_list
.
getDestinationDecision
(),
destination_decision
)
self
.
assertEqual
(
packing_list
.
getDestinationAdministration
(),
destination_administration
)
self
.
assertEqual
(
packing_list
.
objectCount
(),
1
)
for
line
in
packing_list
.
objectValues
():
self
.
assertEqual
(
line
.
getResource
(),
resource
)
self
.
assertAlmostEquals
(
line
.
getQuantity
(),
1.0
*
number_of_sale_order_lines
)
simulation_movement_list
=
line
.
getDeliveryRelatedValueList
()
self
.
assertEqual
(
len
(
simulation_movement_list
),
number_of_sale_order_lines
)
for
simulation_movement
in
simulation_movement_list
:
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),{
'business_process_module/test_bp/deliver_path'
,
'business_process_module/test_bp/deliver'
})
def
stepCheckBuiltSaleInvoices
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
accounting_module
self
.
assertEqual
(
module
.
objectCount
(),
1
)
number_of_sale_orders
=
sequence
.
get
(
'number_of_sale_orders'
)
number_of_sale_order_lines
=
sequence
.
get
(
'number_of_sale_order_lines'
)
number_of_additional_sale_packing_list_lines
\
=
sequence
.
get
(
'number_of_additional_sale_packing_list_lines'
)
for
invoice
in
module
.
contentValues
():
self
.
assertEqual
(
invoice
.
getSimulationState
(),
'confirmed'
)
self
.
assertEqual
(
invoice
.
getCausalityState
(),
'solved'
)
self
.
assertEqual
(
invoice
.
objectCount
(),
2
)
invoice_line_list
=
invoice
.
objectValues
(
sort_on
=
'resource'
)
commodity_invoice_line
=
invoice_line_list
[
0
]
self
.
assertEqual
(
commodity_invoice_line
.
getResource
(),
'product_module/commodity'
)
self
.
assertAlmostEquals
(
commodity_invoice_line
.
getQuantity
(),
1.0
*
(
number_of_sale_order_lines
\
+
number_of_additional_sale_packing_list_lines
)
\
*
number_of_sale_orders
)
simulation_movement_list
\
=
commodity_invoice_line
.
getDeliveryRelatedValueList
()
self
.
assertEqual
(
len
(
simulation_movement_list
),
(
number_of_sale_order_lines
\
+
number_of_additional_sale_packing_list_lines
)
\
*
number_of_sale_orders
)
for
simulation_movement
in
simulation_movement_list
:
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),
{
'business_process_module/test_bp/invoice_path'
,
'business_process_module/test_bp/invoice'
})
tax_invoice_line
=
invoice_line_list
[
1
]
self
.
assertEqual
(
tax_invoice_line
.
getResource
(),
'service_module/vat_low'
)
self
.
assertAlmostEquals
(
tax_invoice_line
.
getQuantity
(),
commodity_invoice_line
.
getPrice
()
\
*
(
number_of_sale_order_lines
\
+
number_of_additional_sale_packing_list_lines
)
\
*
number_of_sale_orders
)
simulation_movement_list
\
=
tax_invoice_line
.
getDeliveryRelatedValueList
()
self
.
assertEqual
(
len
(
simulation_movement_list
),
(
number_of_sale_order_lines
\
+
number_of_additional_sale_packing_list_lines
)
\
*
number_of_sale_orders
)
for
simulation_movement
in
simulation_movement_list
:
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),
{
'business_process_module/test_bp/invoice_path'
,
'business_process_module/test_bp/invoice'
,
'sale_trade_condition_module/test_stc/vat_low'
})
def
stepCheckAddedSaleInvoices
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
accounting_module
self
.
assertEqual
(
module
.
objectCount
(),
1
)
number_of_sale_orders
=
sequence
.
get
(
'number_of_sale_orders'
)
number_of_sale_order_lines
=
sequence
.
get
(
'number_of_sale_order_lines'
)
number_of_additional_sale_packing_list_lines
\
=
sequence
.
get
(
'number_of_additional_sale_packing_list_lines'
)
for
invoice
in
module
.
contentValues
():
self
.
assertEqual
(
invoice
.
getSimulationState
(),
'confirmed'
)
self
.
assertEqual
(
invoice
.
getCausalityState
(),
'solved'
)
self
.
assertEqual
(
invoice
.
objectCount
(),
4
)
invoice_line_list
=
invoice
.
objectValues
(
sort_on
=
'resource'
)
commodity_invoice_line
=
invoice_line_list
[
0
]
self
.
assertEqual
(
commodity_invoice_line
.
getResource
(),
'product_module/commodity'
)
self
.
assertAlmostEquals
(
commodity_invoice_line
.
getQuantity
(),
1.0
*
(
number_of_sale_order_lines
\
+
number_of_additional_sale_packing_list_lines
)
\
*
number_of_sale_orders
)
simulation_movement_list
\
=
commodity_invoice_line
.
getDeliveryRelatedValueList
()
self
.
assertEqual
(
len
(
simulation_movement_list
),
(
number_of_sale_order_lines
\
+
number_of_additional_sale_packing_list_lines
)
\
*
number_of_sale_orders
)
for
simulation_movement
in
simulation_movement_list
:
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),
{
'business_process_module/test_bp/invoice_path'
,
'business_process_module/test_bp/invoice'
})
luxury_invoice_line
=
invoice_line_list
[
1
]
self
.
assertEqual
(
luxury_invoice_line
.
getResource
(),
'product_module/luxury'
)
self
.
assertAlmostEquals
(
luxury_invoice_line
.
getQuantity
(),
1.0
*
number_of_additional_sale_packing_list_lines
\
*
number_of_sale_orders
)
simulation_movement_list
\
=
luxury_invoice_line
.
getDeliveryRelatedValueList
()
self
.
assertEqual
(
len
(
simulation_movement_list
),
number_of_additional_sale_packing_list_lines
\
*
number_of_sale_orders
)
for
simulation_movement
in
simulation_movement_list
:
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),
{
'business_process_module/test_bp/invoice_path'
,
'business_process_module/test_bp/invoice'
})
high_tax_invoice_line
=
invoice_line_list
[
2
]
self
.
assertEqual
(
high_tax_invoice_line
.
getResource
(),
'service_module/vat_high'
)
self
.
assertAlmostEquals
(
high_tax_invoice_line
.
getQuantity
(),
luxury_invoice_line
.
getPrice
()
\
*
number_of_additional_sale_packing_list_lines
\
*
number_of_sale_orders
)
simulation_movement_list
\
=
high_tax_invoice_line
.
getDeliveryRelatedValueList
()
self
.
assertEqual
(
len
(
simulation_movement_list
),
number_of_additional_sale_packing_list_lines
\
*
number_of_sale_orders
)
for
simulation_movement
in
simulation_movement_list
:
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),
{
'business_process_module/test_bp/invoice_path'
,
'business_process_module/test_bp/invoice'
,
'sale_trade_condition_module/test_stc/vat_high'
})
low_tax_invoice_line
=
invoice_line_list
[
3
]
self
.
assertEqual
(
low_tax_invoice_line
.
getResource
(),
'service_module/vat_low'
)
self
.
assertAlmostEquals
(
low_tax_invoice_line
.
getQuantity
(),
commodity_invoice_line
.
getPrice
()
\
*
(
number_of_sale_order_lines
\
+
number_of_additional_sale_packing_list_lines
)
\
*
number_of_sale_orders
)
simulation_movement_list
\
=
low_tax_invoice_line
.
getDeliveryRelatedValueList
()
self
.
assertEqual
(
len
(
simulation_movement_list
),
(
number_of_sale_order_lines
\
+
number_of_additional_sale_packing_list_lines
)
\
*
number_of_sale_orders
)
for
simulation_movement
in
simulation_movement_list
:
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),
{
'business_process_module/test_bp/invoice_path'
,
'business_process_module/test_bp/invoice'
,
'sale_trade_condition_module/test_stc/vat_low'
})
def
stepCheckConvergedSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
for
packing_list
in
module
.
contentValues
():
self
.
assertEqual
(
packing_list
.
getCausalityState
(),
'solved'
)
def
stepCheckDivergedSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
for
packing_list
in
module
.
contentValues
():
self
.
assertEqual
(
packing_list
.
getCausalityState
(),
'diverged'
)
def
stepCheckSalePackingListSimulation
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
number_of_additional_sale_packing_list_lines
\
=
sequence
.
get
(
'number_of_additional_sale_packing_list_lines'
)
destination_decision
=
sequence
.
get
(
'destination_decision'
)
destination_administration
=
sequence
.
get
(
'destination_administration'
)
for
packing_list
in
module
.
contentValues
(
portal_type
=
'Sale Packing List'
):
self
.
assertEqual
(
packing_list
.
getCausalityState
(),
'solved'
)
causality_list
=
packing_list
.
getCausalityRelatedValueList
(
portal_type
=
'Applied Rule'
)
self
.
assertEqual
(
len
(
causality_list
),
1
)
applied_rule
=
causality_list
[
0
]
self
.
assertEqual
(
applied_rule
.
getPortalType
(),
'Applied Rule'
)
rule
=
applied_rule
.
getSpecialiseValue
()
self
.
assertNotEquals
(
rule
,
None
)
self
.
assertEqual
(
rule
.
getReference
(),
'test_delivery_root_simulation_rule'
)
self
.
assertEqual
(
applied_rule
.
objectCount
(),
number_of_additional_sale_packing_list_lines
)
for
simulation_movement
in
applied_rule
.
objectValues
():
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
delivery_list
=
simulation_movement
.
getDeliveryValueList
()
self
.
assertEqual
(
len
(
delivery_list
),
1
)
delivery
=
delivery_list
[
0
]
self
.
assertEqual
(
delivery
.
getPortalType
(),
'Sale Packing List Line'
)
self
.
assertEqual
(
simulation_movement
.
getQuantity
(),
delivery
.
getQuantity
())
self
.
assertEqual
(
simulation_movement
.
getResource
(),
delivery
.
getResource
())
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),
{
'business_process_module/test_bp/deliver_path'
,
'business_process_module/test_bp/deliver'
})
self
.
assertEqual
(
simulation_movement
.
getDestinationDecision
(),
destination_decision
)
self
.
assertEqual
(
simulation_movement
.
getDestinationAdministration
(),
destination_administration
)
def
stepSetSolverTargetToDestinationAdministration
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'solver_target'
,
'destination_administration'
)
def
stepAdoptPrevisionToSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
solver_target
=
sequence
.
get
(
'solver_target'
)
self
.
assertTrue
(
isinstance
(
solver_target
,
str
))
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
for
packing_list
in
module
.
contentValues
(
portal_type
=
'Sale Packing List'
):
divergence
=
packing_list
.
getDivergenceList
()[
0
]
self
.
assertEqual
(
divergence
.
tested_property
,
solver_target
)
packing_list
.
activate
().
SalePackingList_solveForTesting
(
solver_target
,
divergence
.
prevision_value
)
# XXX: Because divergence is ignored for 'destination_decision',
# its value is actually undefined after solving.
# Old solver updated the value on SPL at the same time as it
# adopted prevision for another property, but this is not the case
# anymore. So fix SPL for the rest of the sequence:
packing_list
.
setDestinationDecisionList
(
packing_list
.
getCausalityValue
().
getDestinationDecisionList
())
# Make sure that the same taget is not used again.
sequence
.
edit
(
solver_target
=
None
)
def
stepAcceptDecisionOfSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
solver_target
=
sequence
.
get
(
'solver_target'
)
self
.
assertTrue
(
isinstance
(
solver_target
,
str
))
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
for
packing_list
in
module
.
contentValues
(
portal_type
=
'Sale Packing List'
):
divergence
=
packing_list
.
getDivergenceList
()[
0
]
self
.
assertEqual
(
divergence
.
tested_property
,
solver_target
)
packing_list
.
activate
().
SalePackingList_solveForTesting
(
solver_target
,
divergence
.
decision_value
)
# Make sure that the same taget is not used again.
sequence
.
edit
(
solver_target
=
None
)
def
stepAddNewSalePackingLinesIntoSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
number
=
sequence
.
get
(
'number_of_additional_sale_packing_list_lines'
)
resource
=
sequence
.
get
(
'resource'
)
for
packing_list
in
module
.
contentValues
(
portal_type
=
'Sale Packing List'
):
for
_
in
xrange
(
number
):
packing_list
.
newContent
(
portal_type
=
'Sale Packing List Line'
,
resource
=
resource
,
quantity
=
1.0
)
def
stepStopSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
for
packing_list
in
module
.
contentValues
(
portal_type
=
'Sale Packing List'
):
self
.
assertEqual
(
packing_list
.
getSimulationState
(),
'confirmed'
)
packing_list
.
stop
()
self
.
assertEqual
(
packing_list
.
getSimulationState
(),
'stopped'
)
def
stepDeliverSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
for
packing_list
in
module
.
contentValues
(
portal_type
=
'Sale Packing List'
):
self
.
assertEqual
(
packing_list
.
getSimulationState
(),
'stopped'
)
packing_list
.
deliver
()
self
.
assertEqual
(
packing_list
.
getSimulationState
(),
'delivered'
)
def
stepTestResult
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
measurable
=
sequence
.
get
(
'measurable'
)
if
measurable
:
result
=
sequence
.
get
(
'result'
)
if
result
:
print
''
failure_list
=
[]
for
target
,
min_time
,
real_time
,
max_time
in
result
:
condition
=
(
min_time
<
real_time
<
max_time
)
print
'%s%s: %.4f < %.4f < %.4f'
\
%
(
condition
and
' '
or
'!'
,
target
,
min_time
,
real_time
,
max_time
)
if
not
condition
:
failure_list
.
append
(
target
)
self
.
assertTrue
(
not
failure_list
,
'these tests failed: %s'
%
(
', '
.
join
(
failure_list
)))
def
test_01_CheckLogicAndMakeCacheHot
(
self
):
"""Check the correctness of the logic as well as making the cache hot.
"""
message
=
'Check Logic'
LOG
(
'Testing... '
,
0
,
message
)
#self.getPortal().portal_activities.manage_enableActivityTracking()
self
.
_checkSimulation
(
number_of_sale_orders
=
1
,
number_of_sale_order_lines
=
1
,
number_of_additional_sale_packing_list_lines
=
1
,
measurable
=
False
)
#self.getPortal().portal_activities.manage_disableActivityTracking()
def
test_02_CheckPerformance
(
self
):
"""Check the performance.
"""
message
=
'Check Performance'
LOG
(
'Testing... '
,
0
,
message
)
self
.
_checkSimulation
(
number_of_sale_orders
=
5
,
number_of_sale_order_lines
=
5
,
number_of_additional_sale_packing_list_lines
=
5
,
measurable
=
True
)
def
test_suite
():
suite
=
unittest
.
TestSuite
()
suite
.
addTest
(
unittest
.
makeSuite
(
TestSimulationPerformance
))
return
suite
bt5/erp5_simulation_performance_test/TestTemplateItem/portal_components/test.erp5.testSimulationElementPerformance.xml
0 → 100644
View file @
a16b5c77
<?xml version="1.0"?>
<ZopeData>
<record
id=
"1"
aka=
"AAAAAAAAAAE="
>
<pickle>
<global
name=
"Test Component"
module=
"erp5.portal_type"
/>
</pickle>
<pickle>
<dictionary>
<item>
<key>
<string>
default_reference
</string>
</key>
<value>
<string>
testSimulationElementPerformance
</string>
</value>
</item>
<item>
<key>
<string>
default_source_reference
</string>
</key>
<value>
<string>
Products.ERP5.tests.testSimulationElementPerformance
</string>
</value>
</item>
<item>
<key>
<string>
id
</string>
</key>
<value>
<string>
test.erp5.testSimulationElementPerformance
</string>
</value>
</item>
<item>
<key>
<string>
portal_type
</string>
</key>
<value>
<string>
Test Component
</string>
</value>
</item>
<item>
<key>
<string>
sid
</string>
</key>
<value>
<none/>
</value>
</item>
<item>
<key>
<string>
text_content_error_message
</string>
</key>
<value>
<tuple/>
</value>
</item>
<item>
<key>
<string>
text_content_warning_message
</string>
</key>
<value>
<tuple/>
</value>
</item>
<item>
<key>
<string>
version
</string>
</key>
<value>
<string>
erp5
</string>
</value>
</item>
<item>
<key>
<string>
workflow_history
</string>
</key>
<value>
<persistent>
<string
encoding=
"base64"
>
AAAAAAAAAAI=
</string>
</persistent>
</value>
</item>
</dictionary>
</pickle>
</record>
<record
id=
"2"
aka=
"AAAAAAAAAAI="
>
<pickle>
<global
name=
"PersistentMapping"
module=
"Persistence.mapping"
/>
</pickle>
<pickle>
<dictionary>
<item>
<key>
<string>
data
</string>
</key>
<value>
<dictionary>
<item>
<key>
<string>
component_validation_workflow
</string>
</key>
<value>
<persistent>
<string
encoding=
"base64"
>
AAAAAAAAAAM=
</string>
</persistent>
</value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle>
</record>
<record
id=
"3"
aka=
"AAAAAAAAAAM="
>
<pickle>
<global
name=
"WorkflowHistoryList"
module=
"Products.ERP5Type.Workflow"
/>
</pickle>
<pickle>
<dictionary>
<item>
<key>
<string>
_log
</string>
</key>
<value>
<list>
<dictionary>
<item>
<key>
<string>
action
</string>
</key>
<value>
<string>
validate
</string>
</value>
</item>
<item>
<key>
<string>
validation_state
</string>
</key>
<value>
<string>
validated
</string>
</value>
</item>
</dictionary>
</list>
</value>
</item>
</dictionary>
</pickle>
</record>
</ZopeData>
bt5/erp5_simulation_performance_test/bt/template_test_id_list
0 → 100644
View file @
a16b5c77
test.erp5.testSimulationElementPerformance
\ No newline at end of file
bt5/erp5_simulation_performance_test/bt/test_dependency_list
0 → 100644
View file @
a16b5c77
erp5_full_text_mroonga_catalog
erp5_core_proxy_field_legacy
erp5_base
erp5_pdm
erp5_simulation
erp5_trade
erp5_accounting
erp5_invoicing
\ No newline at end of file
product/ERP5/tests/testSimulationElementPerformance.py
deleted
100644 → 0
View file @
ba0d6c7b
##############################################################################
#
# Copyright (c) 2010 Nexedi SA and Contributors. All Rights Reserved.
# Yoshinori Okuji <yo@nexedi.com>
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsability 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
# garantees 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.
#
##############################################################################
"""Performance tests for simulation.
The purpose of these tests is to measure the performance of the following
basic operations in simulation:
- the creation of new Applied Rules and the expansion of simulation
- the expansion of simulation from existing Applied Rules
- the expansion of simulation which results in convergence
- the expansion of simulation which results in divergence
- the adoption of previsions by a Target Solver
- the acceptance of decisions by a Target Solver
- the creation of new Applied Rules from partially simulated documents
- the creation of new Deliveries by Delivery Builders
- the addition of new Movements into existing Deliveries by Delivery Builders
For the testing environment, these document types are used for Deliveries:
- Sale Order
- Sale Packing List
- Sale Invoice
Since Sale Invoices consist of Movements generated from Sale Packing Lists,
and Movements generated by Trade Model Lines, the simulation trees consist
of at most 4 levels in this order:
- Order Simulation Rule
- Packing List Simulation Rule
- Invoice Simulation Rule
- Trade Model Simulation Rule
In order to simulate various patterns of building, the following rules
are defined:
- Sale Order is completed at "ordered"
- Sale Packing List is completed at "stopped"
- Sale Packing Lists are grouped by destination and by Sale Order
- Sale Invoices are not
So, the relationship between Sale Order and Sale Packing List is 1:N,
and that between Sale Packing List and Sale Invoice is M:N.
"""
import
unittest
from
time
import
time
import
gc
import
subprocess
from
DateTime
import
DateTime
from
Products.ERP5Type.tests.ERP5TypeTestCase
import
ERP5TypeTestCase
from
zLOG
import
LOG
from
Products.ERP5Type.tests.utils
import
LogInterceptor
from
Products.ERP5Type.tests.Sequence
import
SequenceList
# Define variable to chek if performance are good or not
# XXX These variable are specific to the testing environment
# (pystone results: min: < 131578.9 - mean: ~ 139768.5 - max: > 147058.8)
# Historical values are here to remember original values on this
# specific testing environment. We must always try to stay below max
# historical values.
# Only for information
HISTORICAL_EXPECTED_MIN_MAX_TIME
=
{
'Creation Of New Applied Rules'
:
(
34.8
,
36.8
),
'Expansion of Existing Applied Rules'
:
(
16.9
,
19.5
),
'Creation of New Sale Packing Lists'
:
(
37.4
,
38.9
),
'Expansion of Converged Changes'
:
(
22.0
,
24.8
),
'Expansion of Diverged Changes'
:
(
24.6
,
26.9
),
'Adoption of Previsions'
:
(
17.9
,
18.7
),
'Acceptance of Decisions'
:
(
14.5
,
15.2
),
'Creation of New Applied Rules from Partially Simulated Deliveries'
:
(
51.2
,
52.8
),
'Creation Of New Sale Invoices'
:
(
90.0
,
98.5
),
'Addition of New Invoices Lines'
:
(
145.1
,
157.6
),
}
EXPECTED_MIN_MAX_TIME
=
{
'Creation Of New Applied Rules'
:
(
7.38
,
11.38
),
# (30, 35)
'Expansion of Existing Applied Rules'
:
(
9.98
,
15.98
),
# (41, 47)
'Creation of New Sale Packing Lists'
:
(
6.12
,
10.12
),
# (27, 32)
'Expansion of Converged Changes'
:
(
11.12
,
17.12
),
# (48, 54)
'Expansion of Diverged Changes'
:
(
18.58
,
24.58
),
# (53, 59)
'Adoption of Previsions'
:
(
26.78
,
34.78
),
# (105, 114)
'Acceptance of Decisions'
:
(
27.3
,
35.3
),
# (104, 113)
'Creation of New Applied Rules from Partially Simulated Deliveries'
:
(
9.05
,
15.05
),
# (38, 44)
'Creation Of New Sale Invoices'
:
(
11.05
,
17.05
),
# (48, 54)
'Addition of New Invoices Lines'
:
(
29.11
,
37.11
)
# (116, 125)
}
class
TestSimulationPerformance
(
ERP5TypeTestCase
,
LogInterceptor
):
def
getTitle
(
self
):
return
"Simulation Performance"
def
getBusinessTemplateList
(
self
):
"""
Return the list of business templates.
"""
return
(
'erp5_core_proxy_field_legacy'
,
'erp5_base'
,
'erp5_pdm'
,
'erp5_simulation'
,
'erp5_trade'
,
'erp5_accounting'
,
'erp5_invoicing'
,
'erp5_simplified_invoicing'
,
'erp5_simulation_performance_test'
)
def
afterSetUp
(
self
):
"""
Executed before each test_*.
"""
self
.
login
()
portal
=
self
.
getPortal
()
# Validate some documents.
for
rule
in
portal
.
portal_rules
.
objectValues
():
if
rule
.
getValidationState
()
!=
'validated'
\
and
rule
.
getId
().
startswith
(
'test_'
):
rule
.
validate
()
for
trade_condition
in
portal
.
sale_trade_condition_module
.
objectValues
():
if
trade_condition
.
getValidationState
()
!=
'validated'
:
trade_condition
.
validate
()
business_process
=
portal
.
business_process_module
.
test_bp
if
business_process
.
getValidationState
()
!=
'validated'
:
business_process
.
validate
()
subprocess
.
call
(
'sync'
)
#self.prof = hotshot.Profile('hoge.prof')
def
beforeTearDown
(
self
):
#self.prof.close()
self
.
abort
()
portal
=
self
.
getPortal
()
for
module_id
in
(
'sale_order_module'
,
'sale_packing_list_module'
,
'accounting_module'
,
'portal_simulation'
):
module
=
portal
[
module_id
]
module
.
manage_delObjects
(
list
(
module
.
objectIds
()))
self
.
tic
()
def
_getMinMaxTime
(
self
,
target
):
return
EXPECTED_MIN_MAX_TIME
[
target
]
def
_checkSimulation
(
self
,
number_of_sale_orders
=
None
,
number_of_sale_order_lines
=
None
,
number_of_additional_sale_packing_list_lines
=
None
,
measurable
=
True
):
sequence_string
=
"""
Initialize
Tic
UseClientAsDestinationDecision
UseClientAsDestinationAdministration
UseResourceAsCommodityProduct
CreateSaleOrders
Tic
PlanSaleOrders
MeasureCreationOfNewAppliedRules
Tic
CheckSaleOrderSimulation
UseRecipient1AsDestinationDecision
ModifySaleOrders
MeasureExpansionOfExistingAppliedRules
Tic
CheckSaleOrderSimulation
OrderSaleOrders
MeasureCreationOfNewSalePackingLists
Tic
CheckSaleOrderSimulation
CheckBuiltSalePackingLists
UseRecipient2AsDestinationDecision
ModifySaleOrders
MeasureExpansionOfConvergedChanges
Tic
CheckSaleOrderSimulation
CheckConvergedSalePackingLists
UseRecipient1AsDestinationAdministration
ModifySaleOrders
MeasureExpansionOfDivergedChanges
Tic
CheckSaleOrderSimulation
CheckDivergedSalePackingLists
SetSolverTargetToDestinationAdministration
AdoptPrevisionToSalePackingLists
MeasureAdoptionOfPrevisions
Tic
CheckConvergedSalePackingLists
UseRecipient2AsDestinationAdministration
ModifySaleOrders
Tic
CheckSaleOrderSimulation
CheckDivergedSalePackingLists
SetSolverTargetToDestinationAdministration
AcceptDecisionOfSalePackingLists
MeasureAcceptanceOfDecisions
Tic
UseRecipient1AsDestinationAdministration
CheckConvergedSalePackingLists
ConfirmSaleOrders
Tic
UseResourceAsCommodityProduct
AddNewSalePackingLinesIntoSalePackingLists
MeasureCreationOfNewAppliedRulesFromPartiallySimulatedDeliveries
Tic
CheckSalePackingListSimulation
StopSalePackingLists
MeasureCreationOfNewSaleInvoices
Tic
CheckBuiltSaleInvoices
UseResourceAsLuxuryProduct
AddNewSalePackingLinesIntoSalePackingLists
DeliverSalePackingLists
MeasureAdditionOfNewInvoiceLinesIntoSaleInvoices
Tic
CheckAddedSaleInvoices
TestResult
"""
sequence
=
SequenceList
()
self
.
assertTrue
(
isinstance
(
number_of_sale_orders
,
int
))
self
.
assertTrue
(
number_of_sale_orders
>
0
)
self
.
assertTrue
(
isinstance
(
number_of_sale_order_lines
,
int
))
self
.
assertTrue
(
number_of_sale_order_lines
>
0
)
self
.
assertTrue
(
isinstance
(
number_of_additional_sale_packing_list_lines
,
int
))
self
.
assertTrue
(
number_of_additional_sale_packing_list_lines
>
0
)
self
.
assertTrue
(
measurable
in
(
True
,
False
))
self
.
_init_dict
=
dict
(
number_of_sale_orders
=
number_of_sale_orders
,
number_of_sale_order_lines
=
number_of_sale_order_lines
,
number_of_additional_sale_packing_list_lines
=
number_of_additional_sale_packing_list_lines
,
measurable
=
measurable
)
sequence
.
addSequenceString
(
sequence_string
)
sequence
.
play
(
self
)
def
stepInitialize
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
edit
(
**
self
.
_init_dict
)
def
stepTic
(
self
,
sequence
,
sequence_list
=
None
,
**
kw
):
target
=
sequence
.
get
(
'measure_target'
)
measurable
=
target
and
sequence
.
get
(
'measurable'
)
# Reduce noise effect.
subprocess
.
call
(
'sync'
)
gc
.
collect
()
if
measurable
:
before_time
=
time
()
#self.prof.start()
self
.
tic
()
#self.prof.stop()
if
measurable
:
after_time
=
time
()
amount_of_time
=
after_time
-
before_time
min_time
,
max_time
=
self
.
_getMinMaxTime
(
target
)
print
"
\
n
%s took %.4f (%.4f < %.4f < %.4f)"
\
%
(
target
,
amount_of_time
,
min_time
,
amount_of_time
,
max_time
)
# Reset the target to make sure that the same target is not
# measured again.
sequence
.
edit
(
measure_target
=
None
)
# Set the result.
result
=
sequence
.
setdefault
(
'result'
,
[])
result
.
append
((
target
,
min_time
,
amount_of_time
,
max_time
))
def
stepMeasureCreationOfNewAppliedRules
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Creation Of New Applied Rules'
)
def
stepMeasureExpansionOfExistingAppliedRules
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Expansion of Existing Applied Rules'
)
def
stepMeasureCreationOfNewSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Creation of New Sale Packing Lists'
)
def
stepMeasureExpansionOfConvergedChanges
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Expansion of Converged Changes'
)
def
stepMeasureExpansionOfDivergedChanges
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Expansion of Diverged Changes'
)
def
stepMeasureAdoptionOfPrevisions
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Adoption of Previsions'
)
def
stepMeasureAcceptanceOfDecisions
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Acceptance of Decisions'
)
def
stepMeasureCreationOfNewAppliedRulesFromPartiallySimulatedDeliveries
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Creation of New Applied Rules from Partially Simulated Deliveries'
)
def
stepMeasureCreationOfNewSaleInvoices
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Creation Of New Sale Invoices'
)
def
stepMeasureAdditionOfNewInvoiceLinesIntoSaleInvoices
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'measure_target'
,
'Addition of New Invoices Lines'
)
def
stepUseClientAsDestinationDecision
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'destination_decision'
,
'organisation_module/client'
)
def
stepUseRecipient1AsDestinationDecision
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'destination_decision'
,
'organisation_module/recipient1'
)
def
stepUseRecipient2AsDestinationDecision
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'destination_decision'
,
'organisation_module/recipient2'
)
def
stepUseClientAsDestinationAdministration
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'destination_administration'
,
'organisation_module/client'
)
def
stepUseRecipient1AsDestinationAdministration
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'destination_administration'
,
'organisation_module/recipient1'
)
def
stepUseRecipient2AsDestinationAdministration
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'destination_administration'
,
'organisation_module/recipient2'
)
def
stepUseResourceAsCommodityProduct
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'resource'
,
'product_module/commodity'
)
def
stepUseResourceAsLuxuryProduct
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'resource'
,
'product_module/luxury'
)
def
stepCreateSaleOrders
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_order_module
base_date
=
DateTime
(
'2010-08-08'
)
number_of_sale_orders
=
sequence
.
get
(
'number_of_sale_orders'
)
number_of_sale_order_lines
=
sequence
.
get
(
'number_of_sale_order_lines'
)
destination_decision
=
sequence
.
get
(
'destination_decision'
)
destination_administration
=
sequence
.
get
(
'destination_administration'
)
resource
=
sequence
.
get
(
'resource'
)
for
i
in
xrange
(
number_of_sale_orders
):
start_date
=
base_date
+
i
stop_date
=
base_date
+
i
+
1
order
=
module
.
newContent
(
portal_type
=
'Sale Order'
,
specialise
=
'sale_trade_condition_module/test_stc'
,
destination_decision
=
destination_decision
,
destination_administration
=
destination_administration
,
start_date
=
start_date
,
stop_date
=
stop_date
)
# Set the rest through the trade condition.
order
.
SaleOrder_applySaleTradeCondition
()
for
j
in
xrange
(
number_of_sale_order_lines
):
order
.
newContent
(
portal_type
=
'Sale Order Line'
,
resource
=
resource
,
quantity
=
1.0
)
def
stepModifySaleOrders
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_order_module
destination_decision
=
sequence
.
get
(
'destination_decision'
)
destination_administration
=
sequence
.
get
(
'destination_administration'
)
for
order
in
module
.
contentValues
(
portal_type
=
'Sale Order'
):
order
.
edit
(
destination_decision
=
destination_decision
,
destination_administration
=
destination_administration
)
self
.
assertEqual
(
order
.
getDestinationDecision
(),
destination_decision
)
self
.
assertEqual
(
order
.
getDestinationAdministration
(),
destination_administration
)
def
stepPlanSaleOrders
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_order_module
for
order
in
module
.
contentValues
(
portal_type
=
'Sale Order'
):
self
.
assertEqual
(
order
.
getSimulationState
(),
'draft'
)
order
.
plan
()
self
.
assertEqual
(
order
.
getSimulationState
(),
'planned'
)
def
stepOrderSaleOrders
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_order_module
for
order
in
module
.
contentValues
(
portal_type
=
'Sale Order'
):
self
.
assertEqual
(
order
.
getSimulationState
(),
'planned'
)
order
.
order
()
self
.
assertEqual
(
order
.
getSimulationState
(),
'ordered'
)
def
stepConfirmSaleOrders
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_order_module
for
order
in
module
.
contentValues
(
portal_type
=
'Sale Order'
):
self
.
assertEqual
(
order
.
getSimulationState
(),
'ordered'
)
order
.
confirm
()
self
.
assertEqual
(
order
.
getSimulationState
(),
'confirmed'
)
def
stepCheckSaleOrderSimulation
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_order_module
number_of_sale_orders
=
sequence
.
get
(
'number_of_sale_orders'
)
number_of_sale_order_lines
=
sequence
.
get
(
'number_of_sale_order_lines'
)
destination_decision
=
sequence
.
get
(
'destination_decision'
)
destination_administration
=
sequence
.
get
(
'destination_administration'
)
resource
=
sequence
.
get
(
'resource'
)
self
.
assertEqual
(
module
.
objectCount
(),
number_of_sale_orders
)
for
order
in
module
.
contentValues
(
portal_type
=
'Sale Order'
):
causality_list
=
order
.
getCausalityRelatedValueList
(
portal_type
=
'Applied Rule'
)
self
.
assertEqual
(
len
(
causality_list
),
1
)
applied_rule
=
causality_list
[
0
]
self
.
assertEqual
(
applied_rule
.
getPortalType
(),
'Applied Rule'
)
rule
=
applied_rule
.
getSpecialiseValue
()
self
.
assertNotEquals
(
rule
,
None
)
self
.
assertEqual
(
rule
.
getReference
(),
'test_order_root_simulation_rule'
)
self
.
assertEqual
(
applied_rule
.
objectCount
(),
number_of_sale_order_lines
)
for
simulation_movement
in
applied_rule
.
objectValues
():
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
self
.
assertEqual
(
simulation_movement
.
getQuantity
(),
1.0
)
self
.
assertEqual
(
simulation_movement
.
getResource
(),
resource
)
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),
{
'business_process_module/test_bp/order_path'
,
'business_process_module/test_bp/order'
})
self
.
assertEqual
(
simulation_movement
.
getDestinationDecision
(),
destination_decision
)
self
.
assertEqual
(
simulation_movement
.
getDestinationAdministration
(),
destination_administration
)
def
stepCheckBuiltSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
number_of_sale_orders
=
sequence
.
get
(
'number_of_sale_orders'
)
number_of_sale_order_lines
=
sequence
.
get
(
'number_of_sale_order_lines'
)
destination_decision
=
sequence
.
get
(
'destination_decision'
)
destination_administration
=
sequence
.
get
(
'destination_administration'
)
resource
=
sequence
.
get
(
'resource'
)
self
.
assertEqual
(
module
.
objectCount
(),
number_of_sale_orders
)
for
packing_list
in
module
.
contentValues
():
self
.
assertEqual
(
packing_list
.
getSimulationState
(),
'confirmed'
)
self
.
assertEqual
(
packing_list
.
getCausalityState
(),
'solved'
)
self
.
assertEqual
(
packing_list
.
getDestinationDecision
(),
destination_decision
)
self
.
assertEqual
(
packing_list
.
getDestinationAdministration
(),
destination_administration
)
self
.
assertEqual
(
packing_list
.
objectCount
(),
1
)
for
line
in
packing_list
.
objectValues
():
self
.
assertEqual
(
line
.
getResource
(),
resource
)
self
.
assertAlmostEquals
(
line
.
getQuantity
(),
1.0
*
number_of_sale_order_lines
)
simulation_movement_list
=
line
.
getDeliveryRelatedValueList
()
self
.
assertEqual
(
len
(
simulation_movement_list
),
number_of_sale_order_lines
)
for
simulation_movement
in
simulation_movement_list
:
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),{
'business_process_module/test_bp/deliver_path'
,
'business_process_module/test_bp/deliver'
})
def
stepCheckBuiltSaleInvoices
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
accounting_module
self
.
assertEqual
(
module
.
objectCount
(),
1
)
number_of_sale_orders
=
sequence
.
get
(
'number_of_sale_orders'
)
number_of_sale_order_lines
=
sequence
.
get
(
'number_of_sale_order_lines'
)
number_of_additional_sale_packing_list_lines
\
=
sequence
.
get
(
'number_of_additional_sale_packing_list_lines'
)
for
invoice
in
module
.
contentValues
():
self
.
assertEqual
(
invoice
.
getSimulationState
(),
'confirmed'
)
self
.
assertEqual
(
invoice
.
getCausalityState
(),
'solved'
)
self
.
assertEqual
(
invoice
.
objectCount
(),
2
)
invoice_line_list
=
invoice
.
objectValues
(
sort_on
=
'resource'
)
commodity_invoice_line
=
invoice_line_list
[
0
]
self
.
assertEqual
(
commodity_invoice_line
.
getResource
(),
'product_module/commodity'
)
self
.
assertAlmostEquals
(
commodity_invoice_line
.
getQuantity
(),
1.0
*
(
number_of_sale_order_lines
\
+
number_of_additional_sale_packing_list_lines
)
\
*
number_of_sale_orders
)
simulation_movement_list
\
=
commodity_invoice_line
.
getDeliveryRelatedValueList
()
self
.
assertEqual
(
len
(
simulation_movement_list
),
(
number_of_sale_order_lines
\
+
number_of_additional_sale_packing_list_lines
)
\
*
number_of_sale_orders
)
for
simulation_movement
in
simulation_movement_list
:
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),
{
'business_process_module/test_bp/invoice_path'
,
'business_process_module/test_bp/invoice'
})
tax_invoice_line
=
invoice_line_list
[
1
]
self
.
assertEqual
(
tax_invoice_line
.
getResource
(),
'service_module/vat_low'
)
self
.
assertAlmostEquals
(
tax_invoice_line
.
getQuantity
(),
commodity_invoice_line
.
getPrice
()
\
*
(
number_of_sale_order_lines
\
+
number_of_additional_sale_packing_list_lines
)
\
*
number_of_sale_orders
)
simulation_movement_list
\
=
tax_invoice_line
.
getDeliveryRelatedValueList
()
self
.
assertEqual
(
len
(
simulation_movement_list
),
(
number_of_sale_order_lines
\
+
number_of_additional_sale_packing_list_lines
)
\
*
number_of_sale_orders
)
for
simulation_movement
in
simulation_movement_list
:
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),
{
'business_process_module/test_bp/invoice_path'
,
'business_process_module/test_bp/invoice'
,
'sale_trade_condition_module/test_stc/vat_low'
})
def
stepCheckAddedSaleInvoices
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
accounting_module
self
.
assertEqual
(
module
.
objectCount
(),
1
)
number_of_sale_orders
=
sequence
.
get
(
'number_of_sale_orders'
)
number_of_sale_order_lines
=
sequence
.
get
(
'number_of_sale_order_lines'
)
number_of_additional_sale_packing_list_lines
\
=
sequence
.
get
(
'number_of_additional_sale_packing_list_lines'
)
for
invoice
in
module
.
contentValues
():
self
.
assertEqual
(
invoice
.
getSimulationState
(),
'confirmed'
)
self
.
assertEqual
(
invoice
.
getCausalityState
(),
'solved'
)
self
.
assertEqual
(
invoice
.
objectCount
(),
4
)
invoice_line_list
=
invoice
.
objectValues
(
sort_on
=
'resource'
)
commodity_invoice_line
=
invoice_line_list
[
0
]
self
.
assertEqual
(
commodity_invoice_line
.
getResource
(),
'product_module/commodity'
)
self
.
assertAlmostEquals
(
commodity_invoice_line
.
getQuantity
(),
1.0
*
(
number_of_sale_order_lines
\
+
number_of_additional_sale_packing_list_lines
)
\
*
number_of_sale_orders
)
simulation_movement_list
\
=
commodity_invoice_line
.
getDeliveryRelatedValueList
()
self
.
assertEqual
(
len
(
simulation_movement_list
),
(
number_of_sale_order_lines
\
+
number_of_additional_sale_packing_list_lines
)
\
*
number_of_sale_orders
)
for
simulation_movement
in
simulation_movement_list
:
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),
{
'business_process_module/test_bp/invoice_path'
,
'business_process_module/test_bp/invoice'
})
luxury_invoice_line
=
invoice_line_list
[
1
]
self
.
assertEqual
(
luxury_invoice_line
.
getResource
(),
'product_module/luxury'
)
self
.
assertAlmostEquals
(
luxury_invoice_line
.
getQuantity
(),
1.0
*
number_of_additional_sale_packing_list_lines
\
*
number_of_sale_orders
)
simulation_movement_list
\
=
luxury_invoice_line
.
getDeliveryRelatedValueList
()
self
.
assertEqual
(
len
(
simulation_movement_list
),
number_of_additional_sale_packing_list_lines
\
*
number_of_sale_orders
)
for
simulation_movement
in
simulation_movement_list
:
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),
{
'business_process_module/test_bp/invoice_path'
,
'business_process_module/test_bp/invoice'
})
high_tax_invoice_line
=
invoice_line_list
[
2
]
self
.
assertEqual
(
high_tax_invoice_line
.
getResource
(),
'service_module/vat_high'
)
self
.
assertAlmostEquals
(
high_tax_invoice_line
.
getQuantity
(),
luxury_invoice_line
.
getPrice
()
\
*
number_of_additional_sale_packing_list_lines
\
*
number_of_sale_orders
)
simulation_movement_list
\
=
high_tax_invoice_line
.
getDeliveryRelatedValueList
()
self
.
assertEqual
(
len
(
simulation_movement_list
),
number_of_additional_sale_packing_list_lines
\
*
number_of_sale_orders
)
for
simulation_movement
in
simulation_movement_list
:
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),
{
'business_process_module/test_bp/invoice_path'
,
'business_process_module/test_bp/invoice'
,
'sale_trade_condition_module/test_stc/vat_high'
})
low_tax_invoice_line
=
invoice_line_list
[
3
]
self
.
assertEqual
(
low_tax_invoice_line
.
getResource
(),
'service_module/vat_low'
)
self
.
assertAlmostEquals
(
low_tax_invoice_line
.
getQuantity
(),
commodity_invoice_line
.
getPrice
()
\
*
(
number_of_sale_order_lines
\
+
number_of_additional_sale_packing_list_lines
)
\
*
number_of_sale_orders
)
simulation_movement_list
\
=
low_tax_invoice_line
.
getDeliveryRelatedValueList
()
self
.
assertEqual
(
len
(
simulation_movement_list
),
(
number_of_sale_order_lines
\
+
number_of_additional_sale_packing_list_lines
)
\
*
number_of_sale_orders
)
for
simulation_movement
in
simulation_movement_list
:
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),
{
'business_process_module/test_bp/invoice_path'
,
'business_process_module/test_bp/invoice'
,
'sale_trade_condition_module/test_stc/vat_low'
})
def
stepCheckConvergedSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
for
packing_list
in
module
.
contentValues
():
self
.
assertEqual
(
packing_list
.
getCausalityState
(),
'solved'
)
def
stepCheckDivergedSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
for
packing_list
in
module
.
contentValues
():
self
.
assertEqual
(
packing_list
.
getCausalityState
(),
'diverged'
)
def
stepCheckSalePackingListSimulation
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
number_of_additional_sale_packing_list_lines
\
=
sequence
.
get
(
'number_of_additional_sale_packing_list_lines'
)
destination_decision
=
sequence
.
get
(
'destination_decision'
)
destination_administration
=
sequence
.
get
(
'destination_administration'
)
for
packing_list
in
module
.
contentValues
(
portal_type
=
'Sale Packing List'
):
self
.
assertEqual
(
packing_list
.
getCausalityState
(),
'solved'
)
causality_list
=
packing_list
.
getCausalityRelatedValueList
(
portal_type
=
'Applied Rule'
)
self
.
assertEqual
(
len
(
causality_list
),
1
)
applied_rule
=
causality_list
[
0
]
self
.
assertEqual
(
applied_rule
.
getPortalType
(),
'Applied Rule'
)
rule
=
applied_rule
.
getSpecialiseValue
()
self
.
assertNotEquals
(
rule
,
None
)
self
.
assertEqual
(
rule
.
getReference
(),
'test_delivery_root_simulation_rule'
)
self
.
assertEqual
(
applied_rule
.
objectCount
(),
number_of_additional_sale_packing_list_lines
)
for
simulation_movement
in
applied_rule
.
objectValues
():
self
.
assertEqual
(
simulation_movement
.
getPortalType
(),
'Simulation Movement'
)
delivery_list
=
simulation_movement
.
getDeliveryValueList
()
self
.
assertEqual
(
len
(
delivery_list
),
1
)
delivery
=
delivery_list
[
0
]
self
.
assertEqual
(
delivery
.
getPortalType
(),
'Sale Packing List Line'
)
self
.
assertEqual
(
simulation_movement
.
getQuantity
(),
delivery
.
getQuantity
())
self
.
assertEqual
(
simulation_movement
.
getResource
(),
delivery
.
getResource
())
self
.
assertEqual
(
set
(
simulation_movement
.
getCausalityList
()),
{
'business_process_module/test_bp/deliver_path'
,
'business_process_module/test_bp/deliver'
})
self
.
assertEqual
(
simulation_movement
.
getDestinationDecision
(),
destination_decision
)
self
.
assertEqual
(
simulation_movement
.
getDestinationAdministration
(),
destination_administration
)
def
stepSetSolverTargetToDestinationAdministration
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
sequence
.
set
(
'solver_target'
,
'destination_administration'
)
def
stepAdoptPrevisionToSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
solver_target
=
sequence
.
get
(
'solver_target'
)
self
.
assertTrue
(
isinstance
(
solver_target
,
str
))
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
for
packing_list
in
module
.
contentValues
(
portal_type
=
'Sale Packing List'
):
divergence
=
packing_list
.
getDivergenceList
()[
0
]
self
.
assertEqual
(
divergence
.
tested_property
,
solver_target
)
packing_list
.
activate
().
SalePackingList_solveForTesting
(
solver_target
,
divergence
.
prevision_value
)
# XXX: Because divergence is ignored for 'destination_decision',
# its value is actually undefined after solving.
# Old solver updated the value on SPL at the same time as it
# adopted prevision for another property, but this is not the case
# anymore. So fix SPL for the rest of the sequence:
packing_list
.
setDestinationDecisionList
(
packing_list
.
getCausalityValue
().
getDestinationDecisionList
())
# Make sure that the same taget is not used again.
sequence
.
edit
(
solver_target
=
None
)
def
stepAcceptDecisionOfSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
solver_target
=
sequence
.
get
(
'solver_target'
)
self
.
assertTrue
(
isinstance
(
solver_target
,
str
))
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
for
packing_list
in
module
.
contentValues
(
portal_type
=
'Sale Packing List'
):
divergence
=
packing_list
.
getDivergenceList
()[
0
]
self
.
assertEqual
(
divergence
.
tested_property
,
solver_target
)
packing_list
.
activate
().
SalePackingList_solveForTesting
(
solver_target
,
divergence
.
decision_value
)
# Make sure that the same taget is not used again.
sequence
.
edit
(
solver_target
=
None
)
def
stepAddNewSalePackingLinesIntoSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
number
=
sequence
.
get
(
'number_of_additional_sale_packing_list_lines'
)
resource
=
sequence
.
get
(
'resource'
)
for
packing_list
in
module
.
contentValues
(
portal_type
=
'Sale Packing List'
):
for
i
in
xrange
(
number
):
packing_list
.
newContent
(
portal_type
=
'Sale Packing List Line'
,
resource
=
resource
,
quantity
=
1.0
)
def
stepStopSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
for
packing_list
in
module
.
contentValues
(
portal_type
=
'Sale Packing List'
):
self
.
assertEqual
(
packing_list
.
getSimulationState
(),
'confirmed'
)
packing_list
.
stop
()
self
.
assertEqual
(
packing_list
.
getSimulationState
(),
'stopped'
)
def
stepDeliverSalePackingLists
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
portal
=
self
.
getPortal
()
module
=
portal
.
sale_packing_list_module
for
packing_list
in
module
.
contentValues
(
portal_type
=
'Sale Packing List'
):
self
.
assertEqual
(
packing_list
.
getSimulationState
(),
'stopped'
)
packing_list
.
deliver
()
self
.
assertEqual
(
packing_list
.
getSimulationState
(),
'delivered'
)
def
stepTestResult
(
self
,
sequence
=
None
,
sequence_list
=
None
,
**
kw
):
measurable
=
sequence
.
get
(
'measurable'
)
if
measurable
:
result
=
sequence
.
get
(
'result'
)
if
result
:
print
''
failure_list
=
[]
for
target
,
min_time
,
real_time
,
max_time
in
result
:
condition
=
(
min_time
<
real_time
<
max_time
)
print
'%s%s: %.4f < %.4f < %.4f'
\
%
(
condition
and
' '
or
'!'
,
target
,
min_time
,
real_time
,
max_time
)
if
not
condition
:
failure_list
.
append
(
target
)
self
.
assertTrue
(
not
failure_list
,
'these tests failed: %s'
%
(
', '
.
join
(
failure_list
)))
def
test_01_CheckLogicAndMakeCacheHot
(
self
):
"""Check the correctness of the logic as well as making the cache hot.
"""
message
=
'Check Logic'
LOG
(
'Testing... '
,
0
,
message
)
#self.getPortal().portal_activities.manage_enableActivityTracking()
self
.
_checkSimulation
(
number_of_sale_orders
=
1
,
number_of_sale_order_lines
=
1
,
number_of_additional_sale_packing_list_lines
=
1
,
measurable
=
False
)
#self.getPortal().portal_activities.manage_disableActivityTracking()
def
test_02_CheckPerformance
(
self
):
"""Check the performance.
"""
message
=
'Check Performance'
LOG
(
'Testing... '
,
0
,
message
)
self
.
_checkSimulation
(
number_of_sale_orders
=
5
,
number_of_sale_order_lines
=
5
,
number_of_additional_sale_packing_list_lines
=
5
,
measurable
=
True
)
def
test_suite
():
suite
=
unittest
.
TestSuite
()
suite
.
addTest
(
unittest
.
makeSuite
(
TestSimulationPerformance
))
return
suite
Write
Preview
Markdown
is supported
0%
Try again
or
attach a new file
Attach a file
Cancel
You are about to add
0
people
to the discussion. Proceed with caution.
Finish editing this message first!
Cancel
Please
register
or
sign in
to comment