testTradeModelLine.py

# -*- coding: utf-8 -*-
##############################################################################
# Copyright (c) 2009-2010 Nexedi SA and Contributors. All Rights Reserved.
#          Łukasz Nowak <luke@nexedi.com>
#          Fabien Morin <fabien@nexedi.com>
#          Julien Muchembled <jm@nexedi.com>
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly adviced to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################

from UserDict import UserDict
import random
import unittest
import transaction

from Products.ERP5.tests.testBPMCore import TestBPMMixin
from Products.ERP5Type.Base import Base
from Products.ERP5Type.Utils import simple_decorator
from DateTime import DateTime
from Products.ERP5Type.tests.utils import createZODBPythonScript, updateCellList


def save_result_as(name):
  @simple_decorator
  def decorator(function):
    def wrapper(self, *args, **kw):
      result = function(self, *args, **kw)
      self[name] = result
      return result
    return wrapper
  return decorator


class TestTradeModelLineMixin(TestBPMMixin, UserDict):
  """Provides methods to implementations sharing similar logic to Trade Model Lines"""
  # Constants and variables shared by tests
  base_unit_quantity = 0.01
  node_portal_type = 'Organisation'
  order_date = DateTime()
  amount_generator_line_portal_type = 'Trade Model Line'

  def setBaseAmountQuantityMethod(self, base_amount_id, text):
    """Populate TradeModelLine_getBaseAmountQuantityMethod shared script

    This helper method edits the script so that:
    - there's no need to do any cleanup
    - data produced by previous still behaves as expected
    """
    base_amount = self.portal.portal_categories.base_amount
    for name in self.__class__.__name__, self._testMethodName:
      try:
        base_amount = base_amount[name]
      except KeyError:
        base_amount = base_amount.newContent(name)
    try:
      return base_amount[base_amount_id].getRelativeUrl()
    except KeyError:
      base_amount = base_amount.newContent(base_amount_id).getRelativeUrl()
    skin = self.portal.portal_skins.custom
    script_id = self.amount_generator_line_portal_type.replace(' ', '') \
                + '_getBaseAmountQuantityMethod'
    test = "\nif base_application == %r:\n  " % base_amount
    try:
      old_text = '\n' + skin[script_id].body()
    except KeyError:
      old_text = ''
    else:
      skin._delObject(script_id)
    text = test + '\n  '.join(text.splitlines()) + old_text
    createZODBPythonScript(skin, script_id, "base_application", text)
    return base_amount

  def afterSetUp(self):
    UserDict.__init__(self)
    return super(TestTradeModelLineMixin, self).afterSetUp()

  def beforeTearDown(self):
    UserDict.clear(self)
    return super(TestTradeModelLineMixin, self).beforeTearDown()

  def clone(self, document):
    parent = document.getParentValue()
    clone, = parent.manage_pasteObjects(
      parent.manage_copyObjects(ids=document.getId()))
    clone = parent[clone['new_id']]
    try:
      self[clone.getPath()] = self[document.getPath()]
    except KeyError:
      pass
    return clone

  @save_result_as('node')
  def createNode(self, **kw):
    module = self.portal.getDefaultModule(portal_type=self.node_portal_type)
    return module.newContent(portal_type=self.node_portal_type, **kw)

  @save_result_as('resource')
  def createResource(self, portal_type, **kw):
    module = self.portal.getDefaultModule(portal_type=portal_type)
    return module.newContent(portal_type=portal_type, **kw)

  @save_result_as('currency')
  def createCurrency(self):
    return self.createResource('Currency', title='Currency',
                               base_unit_quantity=self.base_unit_quantity)

  @save_result_as('business_process')
  def createBusinessProcess(self, **kw):
    business_process = super(TestTradeModelLineMixin,
        self).createBusinessProcess(**kw)
    if self.business_link_portal_type is not None:
      business_link_list = [
        dict(reference='discount',
             trade_phase='default/discount',
             predecessor='trade_state/invoiced',
             # should successor be trade_state/discounted? There is no
             # such trade_state category
             successor='trade_state/accounted',
        ),
        dict(reference='tax',
             trade_phase='default/tax',
             predecessor='trade_state/invoiced',
             # should successor be trade_state/taxed? There IS such a
             # trade_state category, but the rule that wants to match
             # the Simulation Movement that has this link as causality
             # is default_invoice_transaction_rule, the same as for
             # default/discount, so I'll use the same successor as
             # above. Besides, we'd have to create a new business_link
             # just to get back to accounted, and match it with (or
             # create a new) a portal_rule.
             successor='trade_state/accounted',
        ),
      ]
      for business_link in business_link_list:
        link = self.createBusinessLink(business_process, **business_link)
        self['business_link/' + link.getTradePhaseId()] = link
    return business_process

  @save_result_as('trade_condition')
  def createTradeCondition(self, specialise_value_list,
                           trade_model_line_list=(), **kw):
    module = self.portal.getDefaultModule(
        portal_type=self.trade_condition_portal_type)
    if isinstance(specialise_value_list, Base):
      specialise_value_list = specialise_value_list,
    trade_condition = module.newContent(
        portal_type=self.trade_condition_portal_type,
        title=self.id(),
        specialise_value_list=specialise_value_list,
        **kw)
    for int_index, line_kw in enumerate(trade_model_line_list):
      kw = dict(int_index=int_index)
      kw.update(line_kw)
      self.createTradeModelLine(trade_condition, **kw)
    return trade_condition

  def createTradeModelLine(self, document, **kw):
    line = document.newContent(portal_type='Trade Model Line', **kw)
    reference = line.getReference()
    if reference:
      self['trade_model_line/' + reference] =  line
    return line

  @save_result_as('order')
  def createOrder(self, specialise_value_list, order_line_list=(), **kw):
    module = self.portal.getDefaultModule(portal_type=self.order_portal_type)
    if isinstance(specialise_value_list, Base):
      specialise_value_list = specialise_value_list,
    kw.setdefault('start_date', self.order_date)
    order = module.newContent(
        portal_type=self.order_portal_type,
        title=self.id(),
        specialise_value_list=specialise_value_list,
        **kw)
    for arrow in ('source_value', 'source_section_value',
                  'destination_value', 'destination_section_value'):
      if order.getProperty(arrow) is None:
        order._setProperty(arrow, self.createNode())
    if not order.getPriceCurrency():
      self['price_currency'] = price_currency = self.createCurrency()
      order._setPriceCurrencyValue(price_currency)
    for line_kw in order_line_list:
      order.newContent(portal_type=self.order_line_portal_type, **line_kw)
    return order

  def getAggregatedAmountList(self, amount_generator, *args, **kw):
    return amount_generator.getAggregatedAmountList(*args, **kw)

  def getAggregatedAmountDict(self, amount_generator, partial_check=False,
                              **expected_amount_dict):
    amount_list = self.getAggregatedAmountList(amount_generator)
    amount_dict = {}
    for amount in amount_list:
      reference = amount.getReference()
      try:
        expected_amount = expected_amount_dict.pop(reference)
      except KeyError:
        if not partial_check:
          raise
      else:
        for k, v in expected_amount.iteritems():
          if k == 'causality_value_list':
            self.assertEqual(v, amount.getValueList('causality'))
          else:
            self.assertEqual(v, amount.getProperty(k))
        amount_dict[reference] = amount
    if partial_check:
      for value in expected_amount_dict.itervalues():
        self.assertEqual(None, value)
    else:
      self.assertEqual({}, expected_amount_dict)
    return amount_dict

  def getTradeModelSimulationMovementList(self, delivery_line):
    result_list = []
    for simulation_movement in delivery_line.getDeliveryRelatedValueList(
        portal_type='Simulation Movement'):
      if delivery_line.getPortalType() == self.order_line_portal_type:
        applied_rule, = [x
          for x in simulation_movement.objectValues()
          if x.getSpecialiseReference() == 'default_delivering_rule']
        simulation_movement, = applied_rule.objectValues()
      applied_rule, = [x
        for x in simulation_movement.objectValues()
        if x.getSpecialiseReference() == 'default_invoicing_rule']
      simulation_movement, = applied_rule.objectValues()
      applied_rule, = [x
        for x in simulation_movement.objectValues()
        if x.getSpecialiseReference() == 'default_trade_model_rule']
      result_list.append(applied_rule.objectValues())
    return result_list


class TestTradeModelLine(TestTradeModelLineMixin):

  # Constants and variables shared by tests
  default_discount_ratio = -0.05 # -5%
  default_tax_ratio = 0.196 # 19,6%

  new_discount_ratio = -0.04 # -4%
  new_tax_ratio = 0.22 # 22%

  modified_order_line_price_ratio = 2.0
  modified_packing_list_line_quantity_ratio = 0.4
  modified_invoice_line_quantity_ratio = modified_order_line_quantity_ratio \
      = 2.5

  @save_result_as('product/taxed')
  def createProductTaxed(self):
    return self.createResource('Product',
      title='Product Taxed',
      base_contribution=['base_amount/tax'],
      use='normal')

  @save_result_as('product/discounted')
  def createProductDiscounted(self):
    return self.createResource('Product',
      title='Product Discounted',
      base_contribution=['base_amount/discount'],
      use='normal')

  @save_result_as('product/taxed_discounted')
  def createProductDiscountedTaxed(self):
    return self.createResource('Product',
      title='Product Discounted & Taxed',
      base_contribution=['base_amount/discount', 'base_amount/tax'],
      use='normal')

  @save_result_as('service/tax')
  def createServiceTax(self):
    return self.createResource('Service', title='Tax', use='tax')

  @save_result_as('service/discount')
  def createServiceDiscount(self):
    return self.createResource('Service', title='Discount', use='discount')

  def packPackingList(self, packing_list):
    if packing_list.getContainerState() == 'packed':
      return
    packing_list.manage_delObjects(ids=[q.getId()
      for q in packing_list.objectValues(portal_type='Container')])
    transaction.commit()
    container = packing_list.newContent(portal_type='Container')
    for movement in packing_list.getMovementList():
      container.newContent(portal_type='Container Line',
                           resource=movement.getResource(),
                           quantity=movement.getQuantity())
    transaction.commit()
    self.tic()
    self.assertEqual('packed', packing_list.getContainerState())

  def copyExpectedAmountDict(self, delivery, ratio=1):
    self[delivery.getPath()] = expected_amount_dict = {}
    causality = delivery.getCausalityValue()
    for base_amount, amount_dict in self[causality.getPath()].iteritems():
      expected_amount_dict[base_amount] = new_amount_dict = {}
      for line in delivery.getMovementList():
        line_id = line.getCausalityId()
        if line_id in amount_dict:
          new_amount_dict[line.getId()] = ratio * amount_dict[line_id]

  def acceptDecisionQuantityInvoice(self, invoice):
    solver_process_tool = self.portal.portal_solver_processes
    solver_process = solver_process_tool.newSolverProcess(invoice)
    for quantity_solver_decision in solver_process.contentValues():
      if quantity_solver_decision.getCausalityValue().getTestedProperty() \
        == 'quantity':
        # use Trade Model Solver.
        quantity_solver_decision.setSolverValue(
          self.portal.portal_solvers['Trade Model Solver'])
    solver_process.buildTargetSolverList()
    solver_process.solve()

  def processPackingListBuildInvoice(self, packing_list, build=None):
    self.packPackingList(packing_list)
    transaction.commit()
    self.tic()

    packing_list.start()
    packing_list.stop()
    transaction.commit()
    self.tic()

    invoice, = packing_list.getCausalityRelatedValueList(
      portal_type=self.invoice_portal_type)
    self.assertEqual(5, len(invoice))

    packing_list.deliver()
    transaction.commit()
    self.tic()

    self['invoice'] = invoice
    if build == 'invoice':
      return invoice

    self.checkCausalityState(invoice, 'solved')
    self.checkTradeModelRuleSimulationExpand(packing_list)

    invoice.start()
    transaction.commit()
    self.tic()

    self.checkInvoiceAccountingMovements(invoice)

  ###
  ##  Check methods
  ##

  def checkWithoutBPM(self, order):
    transaction.commit() # clear transactional cache
    order.getSpecialiseValue()._setSpecialise(None)
    self.assertRaises(ValueError, order.expand,
      applied_rule_id=order.getCausalityRelatedId(portal_type='Applied Rule'))
    transaction.abort()

  def checkModelLineOnDelivery(self, delivery):
    for portal_type in (self.business_link_portal_type,
                        self.trade_model_path_portal_type,
                        'Trade Model Line'):
      self.assertRaises(ValueError, delivery.newContent,
                        portal_type=portal_type)

  def checkComposition(self, movement, specialise_value_list, type_count_dict):
    composed = movement.asComposedDocument()
    self.assertFalse(movement in composed._effective_model_list)
    self.assertSameSet(composed.getSpecialiseValueList(),
                       specialise_value_list)
    count = 0
    for portal_type, n in type_count_dict.iteritems():
      count += n
      self.assertEqual(n, len(composed.objectValues(portal_type=portal_type)))
    self.assertTrue(count, len(composed.objectValues()))

  def checkAggregatedAmountList(self, order):
    expected_result_dict = self[order.getPath()]
    def check(movement, movement_id):
      kw = {}
      for reference, result in expected_result_dict.iteritems():
        total_price = result.get(movement_id)
        if total_price:
          model_line = self['trade_model_line/' + reference]
          kw[reference] = dict(total_price=total_price,
            causality_value_list=[model_line],
            base_application_list=model_line.getBaseApplicationList(),
            base_contribution_list=model_line.getBaseContributionList())
      self.getAggregatedAmountDict(movement, **kw)

    check(order, None)
    for line in order.getMovementList():
      check(line, line.getId())

  def checkTradeModelRuleSimulationExpand(self, delivery):
    expected_result_dict = self[delivery.getPath()]
    price_currency = self['price_currency']

    for line in delivery.getMovementList():
      simulation_movement_list, = \
        self.getTradeModelSimulationMovementList(line)
      result_dict = dict((sm.getResourceValue().getUse(), sm)
                         for sm in simulation_movement_list)
      self.assertEqual(len(simulation_movement_list),
                       len(result_dict))
      for use in 'discount', 'tax':
        total_price = expected_result_dict[use].get(line.getId())
        if total_price:
          sm = result_dict.pop(use)
          self.assertEqual(str(sm.getTotalPrice()), str(total_price))
          self.assertEqual(3, len(sm.getCausalityValueList()))
          self.assertEqual(1, len(sm.getCausalityValueList(
            portal_type=self.business_link_portal_type)))
          self.assertEqual(1, len(sm.getCausalityValueList(
            portal_type=self.trade_model_path_portal_type)))
          self.assertEqual(1, len(sm.getCausalityValueList(
            portal_type='Trade Model Line')))
          self.assertEqual(sm.getBaseApplicationList(),
                           ['base_amount/' + use])
          self.assertEqual(sm.getBaseContributionList(),
                           dict(discount=['base_amount/tax'], tax=[])[use])
      self.assertEqual({}, result_dict)

  def checkCausalityState(self, delivery, state):
    self.assertEqual(state, delivery.getCausalityState(),
      delivery.getDivergenceList())

  def checkInvoiceAccountingMovements(self, invoice):
    # Wouldn't it be better to use 'invoice.getAggregatedAmountList()'
    # instead of looking at invoice lines ? We wouldn't have to clear
    # base_contribution_list in test_01a_InvoiceNewTradeConditionOrLineSupport
    line_dict = {}
    for line in invoice.getMovementList():
      if line.getPortalType() == self.invoice_line_portal_type:
        key = line.getResourceValue().getUse()
      else:
        key = ('income_expense', 'payable_receivable', 'vat')[
          ['income', 'expense',
           'liability/payable', 'asset/receivable',
           'liability/payable/collected_vat', 'asset/receivable/refundable_vat',
          ].index(line.getSourceValue().getAccountType()) // 2]
      line_dict.setdefault(key, 0)
      line_dict[key] += line.getTotalPrice()
    self.assertEqual(6, len(line_dict))

    currency_precision = self['price_currency'].getQuantityPrecision()
    rounded_total_price = round(line_dict['normal'], currency_precision)
    rounded_tax_price = round(line_dict['tax'], currency_precision)
    rounded_discount_price = round(line_dict['discount'], currency_precision)
    self.assertEqual(str(abs(line_dict['payable_receivable'])),
        str(rounded_total_price + rounded_tax_price + rounded_discount_price))
    self.assertEqual(str(abs(line_dict['vat'])),
        str(rounded_tax_price))
    self.assertEquals(str(abs(line_dict['income_expense'])),
        str(rounded_total_price + rounded_discount_price))

  ###
  ##  Test cases
  ##

  def test_01_OrderWithSimpleTaxedAndDiscountedLines(self, build=None):
    """Full test case with quite simple linear use case

    Data:
    - 1 SO: 1 taxed, 1 discounted, 1 taxed&discounted
    - 1 TC: tax, discount
    - 1 BP (linked to default BP)

    Checks:
    - composition
    - getAggregatedAmountList
    - expand (before and after modifying quantities on order lines)
    - build of packing list (+ pack) and invoice
    """
    taxed = self.createProductTaxed()
    discounted = self.createProductDiscounted()
    taxed_discounted = self.createProductDiscountedTaxed()
    business_process = self.createBusinessProcess()
    trade_condition = self.createTradeCondition(
      business_process, (
      dict(price=self.default_discount_ratio,
           base_application='base_amount/discount',
           base_contribution='base_amount/tax',
           trade_phase='default/discount',
           resource_value=self.createServiceDiscount(),
           reference='discount',
           int_index=10),
      dict(price=self.default_tax_ratio,
           base_application='base_amount/tax',
           trade_phase='default/tax',
           resource_value=self.createServiceTax(),
           reference='tax',
           int_index=20),
      ))
    order = self.createOrder(trade_condition, (
      dict(price=1, quantity=2, id='taxed',
                                resource_value=taxed),
      dict(price=3, quantity=4, id='discounted',
                                resource_value=discounted),
      dict(price=5, quantity=6, id='taxed_discounted',
                                resource_value=taxed_discounted),
      ))
    discount = {None: (3*4 + 5*6) * self.default_discount_ratio,
                'discounted': (3*4) * self.default_discount_ratio,
                'taxed_discounted': (5*6) * self.default_discount_ratio}
    self[order.getPath()] = dict(
      discount=discount,
      tax={None: (1*2 + 5*6 + discount[None]) * self.default_tax_ratio,
           'taxed': (1*2) * self.default_tax_ratio,
           'discounted': discount['discounted'] * self.default_tax_ratio,
           'taxed_discounted': (5*6 + discount['taxed_discounted'])
                               * self.default_tax_ratio})

    transaction.commit()
    self.tic()

    if not build:
      # Check amount_generator refuses to produce amounts
      # if lines are not ordered correctly.
      self['trade_model_line/tax'].setIntIndex(0)
      self.assertRaises(ValueError, order.getGeneratedAmountList)
      transaction.abort()

      for movement in (order, order['taxed'], order['discounted'],
                       order['taxed_discounted']):
        self.checkComposition(movement, [trade_condition], {
          self.trade_model_path_portal_type: 11,
          self.business_link_portal_type: 7,
          "Trade Model Line": 2})

      self.checkAggregatedAmountList(order)

      order.plan()
      transaction.commit()
      self.tic()

      self.checkTradeModelRuleSimulationExpand(order)

      self.checkWithoutBPM(order)

      order2 = self.clone(order)

      # Multiply prices by 2 and quantities by 2.5
      order['taxed'].edit(price=2, quantity=5)
      order['discounted'].edit(price=6, quantity=10)
      order['taxed_discounted'].edit(price=10, quantity=15)
      transaction.commit()
      self.tic()

      discount = {None: (6*10 + 10*15) * self.default_discount_ratio,
                  'discounted': (6*10) * self.default_discount_ratio,
                  'taxed_discounted': (10*15) * self.default_discount_ratio}
      self[order.getPath()] = dict(
        discount=discount,
        tax={None: (2*5 + 10*15 + discount[None]) * self.default_tax_ratio,
             'taxed': (2*5) * self.default_tax_ratio,
             'discounted': discount['discounted'] * self.default_tax_ratio,
             'taxed_discounted': (10*15 + discount['taxed_discounted'])
                                 * self.default_tax_ratio})

      self.checkTradeModelRuleSimulationExpand(order)

      self.checkAggregatedAmountList(order)
      order = order2

    order.confirm()
    transaction.commit()
    self.tic()

    packing_list, = order.getCausalityRelatedValueList(
      portal_type=self.packing_list_portal_type)
    self.copyExpectedAmountDict(packing_list)

    self['packing_list'] = packing_list
    if build == 'packing_list':
      return packing_list

    return self.processPackingListBuildInvoice(packing_list, build)

  def test_01a_InvoiceNewTradeConditionOrLineSupport(self):
    invoice = self.test_01_OrderWithSimpleTaxedAndDiscountedLines('invoice')

    # on invoice, make specialise point to a new TC and check it diverged
    trade_condition = self['trade_condition']
    new_trade_condition = self.clone(trade_condition)
    line_dict = dict((line.getReference(), line)
                     for line in new_trade_condition.objectValues())
    line_dict['discount'].edit(reference='discount_2',
                               price=self.new_discount_ratio)
    line_dict['tax'].edit(reference='tax_2',
                          price=self.new_tax_ratio)

    self.assertEqual([trade_condition], invoice.getSpecialiseValueList())
    invoice.setSpecialiseValue(new_trade_condition)
    transaction.commit()
    self.tic()
    self.checkCausalityState(invoice, 'diverged')

    # revert to reuse invoice
    invoice.setSpecialiseValue(trade_condition)
    transaction.commit()
    self.tic()
    self.checkCausalityState(invoice, 'solved')

    # check how is supported addition of invoice line to invoice
    for line in self['order'].getMovementList():
      line = invoice.newContent(portal_type=self.invoice_line_portal_type,
                                resource=line.getResource(),
                                quantity=line.getQuantity(),
                                price=line.getPrice())
      # XXX base_contribution_list is automatically copied from the resource
      #     but Invoice Transaction Trade Model Line Builder will not be called
      #     again (or if we call it, lines would be built in a new invoice)
      #     and accounting lines would not match invoice lines.
      #     So we clear the list to make the test simpler.
      #     See also 'checkInvoiceAccountingMovements' method.
      self.assertTrue(line.getBaseContributionList())
      line._setBaseContributionList(())
    transaction.commit()
    self.tic()
    self.checkCausalityState(invoice, 'solved')

    invoice.start()
    transaction.commit()
    self.tic()

    self.checkCausalityState(invoice, 'solved')
    self.checkInvoiceAccountingMovements(invoice)

    invoice.stop()
    invoice.deliver()
    transaction.commit()
    self.tic()

  def test_01b_NewSimulation_InvoiceModifyQuantityAndSolveDivergency(self):
    invoice = self.test_01_OrderWithSimpleTaxedAndDiscountedLines('invoice')

    for line in invoice.getMovementList():
      if line.getResourceValue().getUse() == 'normal':
        line.setQuantity(line.getQuantity() *
          self.modified_invoice_line_quantity_ratio)
    transaction.commit()
    self.tic()
    self.checkCausalityState(invoice, 'diverged')

    self.acceptDecisionQuantityInvoice(invoice)
    transaction.commit()
    self.tic()
    self.checkCausalityState(invoice, 'solved')

  def test_01c_PackingListSplitBuildInvoiceBuild(self):
    packing_list = \
      self.test_01_OrderWithSimpleTaxedAndDiscountedLines('packing_list')

    for line in packing_list.getMovementList():
        line.setQuantity(line.getQuantity() *
          self.modified_packing_list_line_quantity_ratio)
    transaction.commit()
    self.tic()
    self.checkCausalityState(packing_list, 'diverged')

    order = self['order']
    self.checkTradeModelRuleSimulationExpand(order)
    self.copyExpectedAmountDict(packing_list,
        self.modified_packing_list_line_quantity_ratio)

    listbox = [{'listbox_key':line.getRelativeUrl(),
                'choice':'SplitAndDefer'}
               for line in packing_list.getMovementList()
               if line.isDivergent()]
    self.assertEqual(len(order), len(listbox))
    self.portal.portal_workflow.doActionFor(
      packing_list,
      'split_and_defer_action',
      start_date=packing_list.getStartDate() + 15,
      stop_date=packing_list.getStopDate() + 25,
      listbox=listbox)

    transaction.commit()
    self.tic()
    self.checkCausalityState(packing_list, 'solved')
    new_packing_list, = [x for x in order.getCausalityRelatedValueList(
                             portal_type=self.packing_list_portal_type)
                           if x != packing_list]
    self.copyExpectedAmountDict(new_packing_list,
        1 - self.modified_packing_list_line_quantity_ratio)

    invoice_count = len(self.portal
        .accounting_module.objectValues(portal_type=self.invoice_portal_type))
    self.processPackingListBuildInvoice(packing_list)
    # For some time, the following assertion failed. Here was the reason:
    #     With legacy code, only 1 invoice was built after starting the first
    #     packing list. Now, all invoice lines generated by trade model are
    #     built immediately, creating a second invoice before starting the
    #     second packing list, and we end up with 3 invoices. In other words,
    #     the new simulation splits the second invoice, and I am not sure it's
    #     correct.
    #     The difference between old and new simulation is that when the first
    #     invoice is confirmed, the old code does not expand the simulation
    #     tree completely and SaleInvoice_selectTradeModelMovementList can't
    #     find any simulation movements related to the second packing list.
    # This was fixed by removing 'planned' state from
    # {Purchase,Sale}Invoice_selectTradeModelMovementList scripts.
    self.assertEqual(invoice_count + 1, len(self.portal
        .accounting_module.objectValues(portal_type=self.invoice_portal_type)))
    self.processPackingListBuildInvoice(new_packing_list)

  def test_02_OrderWithComplexTaxedAndDiscountedLines(self):
    service_discount = self.createServiceDiscount()
    service_tax = self.createServiceTax()
    business_process = self.createBusinessProcess()
    line_list = [
      dict(price=0.2,
           base_application='base_amount/tax',
           base_contribution='base_amount/total_tax',
           trade_phase='default/tax',
           resource_value=service_tax,
           reference='service_tax',
           int_index=10),
      dict(price=0.32,
           base_application='base_amount/discount',
           base_contribution='base_amount/total_discount',
           trade_phase='default/discount',
           resource_value=service_discount,
           reference='total_dicount_2',
           int_index=10),
      dict(price=0.2,
           base_application='base_amount/tax',
           base_contribution='base_amount/total_tax',
           trade_phase='default/tax',
           resource_value=service_tax,
           reference='service_tax_2',
           int_index=10),
      dict(price=0.12,
           base_application='base_amount/total_tax',
           base_contribution='base_amount/total_discount',
           trade_phase='default/tax',
           resource_value=service_tax,
           reference='tax_3',
           int_index=20),
      dict(price=0.8,
           base_application='base_amount/total_discount',
           trade_phase='default/discount',
           resource_value=service_discount,
           reference='total_discount',
           int_index=30),
      ]
    random.shuffle(line_list)
    trade_condition = self.createTradeCondition(business_process, line_list)

    taxed = self.createProductTaxed()
    discounted = self.createProductDiscounted()
    order = self.createOrder(trade_condition, (
      dict(price=1, quantity=2, id='taxed',
                                resource_value=taxed),
      dict(price=3, quantity=4, id='discounted',
                                resource_value=discounted),
      ))
    discount_price = (3*4) * 0.32
    tax_price = (1*2) * 0.2
    total_tax_price = tax_price * 2 * 0.12
    self[order.getPath()] = dict(
      service_tax={None: tax_price, 'taxed': tax_price},
      total_dicount_2={None: discount_price, 'discounted': discount_price},
      service_tax_2={None: tax_price, 'taxed': tax_price},
      tax_3={None: total_tax_price, 'taxed': total_tax_price},
      total_discount={None: (total_tax_price+discount_price) * 0.8,
                      'taxed': total_tax_price * 0.8,
                      'discounted': discount_price * 0.8})

    transaction.commit()
    self.tic()

    self.checkModelLineOnDelivery(order)

    for movement in order, order['taxed'], order['discounted']:
      self.checkComposition(movement, [trade_condition], {
        self.trade_model_path_portal_type: 11,
        self.business_link_portal_type: 7,
        "Trade Model Line": 5})

    self.checkAggregatedAmountList(order)

  def test_03_VariatedModelLine(self):
    base_amount = self.setBaseAmountQuantityMethod('tax', """\
def getBaseAmountQuantity(delivery_amount, base_application,
                          variation_category_list=(), **kw):
  if variation_category_list:
    quantity = delivery_amount.getGeneratedAmountQuantity(base_application)
    tax_range, = variation_category_list
    if tax_range == 'tax_range/0_200':
      return min(quantity, 200)
    else:
      assert tax_range == 'tax_range/200_inf'
      return max(0, quantity - 200)
  return context.getBaseAmountQuantity(delivery_amount, base_application, **kw)
return getBaseAmountQuantity""")
    business_process = self.createBusinessProcess()
    trade_condition = self.createTradeCondition(business_process, (
      dict(price=0.3,
           base_application=base_amount,
           reference='tax1',
           int_index=10),
      dict(base_application=base_amount,
           base_contribution='base_amount/total_tax',
           reference='tax2',
           int_index=20),
      dict(base_application='base_amount/total_tax',
           base_contribution='base_amount/total',
           reference='tax3',
           int_index=30),
      ))
    def createCells(line, matrix, base_application=(), base_contribution=()):
      range_list = [set() for x in iter(matrix).next()]
      for index in matrix:
        for x, y in zip(range_list, index):
          x.add(y)
      line.setCellRange(*range_list)
      for index, price in matrix.iteritems():
        line.newCell(mapped_value_property='price', price=price,
          base_application_list=[index[i] for i in base_application],
          base_contribution_list=[index[i] for i in base_contribution],
          *index)
    createCells(self['trade_model_line/tax2'], {
      ('tax_range/0_200', 'tax_share/A'): .1,
      ('tax_range/0_200', 'tax_share/B'): .2,
      ('tax_range/200_inf', 'tax_share/A'): .3,
      ('tax_range/200_inf', 'tax_share/B'): .4,
      }, base_application=(0,), base_contribution=(1,))
    createCells(self['trade_model_line/tax3'], {
      ('tax_share/A',): .5,
      ('tax_share/B',): .6,
      }, base_application=(0,))
    from Products.ERP5Type.Document import newTempAmount
    for x in ((100, 30, 10, 20, 5, 12),
              (500, 150, 20, 90, 40, 120, 55, 96)):
      amount = newTempAmount(self.portal, '_',
                            quantity=x[0], price=1,
                            base_contribution=base_amount)
      amount_list = trade_condition.getGeneratedAmountList((amount,))
      self.assertEqual(sorted(x[1:]),
                       sorted(y.getTotalPrice() for y in amount_list))

  def test_tradeModelLineWithFixedPrice(self):
    """
      Check it's possible to have fixed quantity on lines. Sometimes we want
      to say "discount 10 euros" or "pay more 10 euros" instead of saying "10%
      discount from total"
    """
    fixed_quantity = self.setBaseAmountQuantityMethod('fixed_quantity', """\
return lambda *args, **kw: 1""")

    tax = self.createServiceTax()
    trade_condition = self.createTradeCondition((), (
      # create a model line with 100 euros
      dict(reference='A', resource_value=tax, quantity=100, price=1),
      # add a discount of 10 euros
      dict(reference='B', resource_value=tax, quantity=10, price=-1)))
    order = self.createOrder(trade_condition, (
      dict(),
      ))
    self.assertEqual([], self.getAggregatedAmountList(order))
    for line in trade_condition.objectValues():
      line.setBaseApplication(fixed_quantity)
    amount_list = order.getAggregatedAmountList()
    self.assertEqual([-10, 100], sorted(x.getTotalPrice() for x in amount_list))

  def test_BuildTradeModelLineAndAccountingFromOrder(self):
    business_process = self.createBusinessProcess()

    product = self.createProductTaxed()
    tax = self.createServiceTax()
    trade_condition = self.createTradeCondition(
      business_process, (
      dict(reference='VAT',
           price=.15,
           resource_value=tax,
           trade_phase='default/tax',
           base_application='base_amount/tax'),
      ))
    source = self.createNode()
    destination = self.createNode()

    order = self.createOrder(trade_condition, (
      dict(price=100, quantity=10, resource_value=product),
      ),
      source_value=source,
      destination_value=destination,
      source_section_value=source,
      destination_section_value=destination)

    order.plan()
    order.confirm()
    transaction.commit()
    self.tic()

    packing_list = order.getCausalityRelatedValue(
                      portal_type=self.packing_list_portal_type)
    self.assertNotEquals(packing_list, None)
    self.assertEquals(1000, packing_list.getTotalPrice())
    
    packing_list.start()
    packing_list.stop()
    packing_list.deliver()

    transaction.commit()
    self.tic()

    invoice = packing_list.getCausalityRelatedValue(
                      portal_type=self.invoice_portal_type)
    self.assertNotEquals(invoice, None)
    self.assertEquals(2, len(invoice.getMovementList()))
    self.assertEquals(1150, invoice.getTotalPrice())
    self.assertEquals([], invoice.getDivergenceList())
    
    invoice.start()
    transaction.commit()
    self.tic()

    self.assertEquals([], invoice.getDivergenceList())
    accounting_line_list = invoice.getMovementList(
             portal_type=self.invoice_transaction_line_portal_type)
    self.assertEquals(3, len(accounting_line_list))

    receivable_movement_list = [m for m in accounting_line_list if
        m.getSourceValue() == self.receivable_account]
    self.assertEquals(1, len(receivable_movement_list))
    receivable_movement = receivable_movement_list[0]
    self.assertEquals(receivable_movement.getDestinationValue(),
                      self.payable_account)
    self.assertEquals(1150, receivable_movement.getSourceDebit())

    collected_movement_list = [m for m in accounting_line_list if
        m.getSourceValue() == self.collected_tax_account]
    self.assertEquals(1, len(collected_movement_list))
    collected_movement = collected_movement_list[0]
    self.assertEquals(collected_movement.getDestinationValue(),
                      self.refundable_tax_account)
    self.assertEquals(150, collected_movement.getSourceCredit())

    income_movement_list = [m for m in accounting_line_list if
        m.getSourceValue() == self.income_account]
    self.assertEquals(1, len(income_movement_list))
    income_movement = income_movement_list[0]
    self.assertEquals(income_movement.getDestinationValue(),
                      self.expense_account)
    self.assertEquals(1000, income_movement.getSourceCredit())

  def test_BuildTradeModelLineAndAccountingFromInvoice(self):
    business_process = self.createBusinessProcess()

    product = self.createProductTaxed()
    tax = self.createServiceTax()
    currency = self.createResource('Currency', title='EUR')
    trade_condition = self.createTradeCondition(
      business_process, (
      dict(reference='VAT',
           price=.15,
           resource_value=tax,
           trade_phase='default/tax',
           base_application='base_amount/tax'),
      ))
    source = self.createNode()
    destination = self.createNode()

    invoice = self.portal.accounting_module.newContent(
               portal_type=self.invoice_portal_type,
               source_value=source,
               destination_value=destination,
               source_section_value=source,
               destination_section_value=destination,
               specialise_value=trade_condition,
               price_currency_value=currency,
               start_date=self.order_date,
               stop_date=self.order_date,
               created_by_builder=True)
    invoice.newContent(
                portal_type=self.invoice_line_portal_type,
                resource_value=product,
                quantity=10,
                price=100)

    invoice.plan()
    invoice.confirm()
    transaction.commit()
    self.tic()

    self.assertEquals(2, len(invoice.getMovementList()))
    self.assertEquals(1150, invoice.getTotalPrice())
    self.assertEquals([], invoice.getDivergenceList())

    invoice.start()
    transaction.commit()
    self.tic()

    self.assertEquals([], invoice.getDivergenceList())
    accounting_line_list = invoice.getMovementList(
             portal_type=self.invoice_transaction_line_portal_type)
    self.assertEquals(3, len(accounting_line_list))

    receivable_movement_list = [m for m in accounting_line_list if
        m.getSourceValue() == self.receivable_account]
    self.assertEquals(1, len(receivable_movement_list))
    receivable_movement = receivable_movement_list[0]
    self.assertEquals(receivable_movement.getDestinationValue(),
                      self.payable_account)
    self.assertEquals(1150, receivable_movement.getSourceDebit())

    collected_movement_list = [m for m in accounting_line_list if
        m.getSourceValue() == self.collected_tax_account]
    self.assertEquals(1, len(collected_movement_list))
    collected_movement = collected_movement_list[0]
    self.assertEquals(collected_movement.getDestinationValue(),
                      self.refundable_tax_account)
    self.assertEquals(150, collected_movement.getSourceCredit())

    income_movement_list = [m for m in accounting_line_list if
        m.getSourceValue() == self.income_account]
    self.assertEquals(1, len(income_movement_list))
    income_movement = income_movement_list[0]
    self.assertEquals(income_movement.getDestinationValue(),
                      self.expense_account)
    self.assertEquals(1000, income_movement.getSourceCredit())

  def test_tradeModelLineWithTargetLevelSetting(self):
    """
      Test that target level setting can specify a target of trade model line
      and trade model line can works with appropriate context(delivery or
      movement) only.
    """
    bounded_fee = self.setBaseAmountQuantityMethod('bounded_fee', """\
return lambda *args, **kw: min(800,
  context.getBaseAmountQuantity(*args, **kw))""")
    fixed_quantity = self.setBaseAmountQuantityMethod('fixed_quantity', """\
return lambda *args, **kw: 1""")

    tax = self.createServiceTax()
    trade_condition = self.createTradeCondition(self.createBusinessProcess())
    # create a model line and set target level to `delivery`.
    tml = self.createTradeModelLine(trade_condition,
                                    reference='TAX',
                                    resource_value=tax,
                                    base_application='base_amount/tax',
                                    target_delivery=True,
                                    price=0.05)

    # create an order.
    resource_A = self.createResource('Product', title='A')
    resource_B = self.createResource('Product', title='B')
    order = self.createOrder(trade_condition)
    base_contribution_list = 'base_amount/tax', bounded_fee
    kw = {'portal_type': self.order_line_portal_type,
          'base_contribution_list': base_contribution_list}
    order_line_1 = order.newContent(price=1000, quantity=1,
                                    resource_value=resource_A, **kw)
    order_line_2 = order.newContent(price=500, quantity=1,
                                    resource_value=resource_B, **kw)
    amount_list = order.getGeneratedAmountList()
    self.assertEqual([75], [x.getTotalPrice() for x in amount_list])

    # change target level to `movement`.
    tml.setTargetDelivery(False)
    amount_list = order.getGeneratedAmountList()
    self.assertEqual([25, 50], sorted(x.getTotalPrice() for x in amount_list))

    # create other trade model lines.
    # for movement
    extra_fee_a = self.createTradeModelLine(trade_condition,
                                            reference='EXTRA_FEE_A',
                                            resource_value=tax,
                                            base_application=bounded_fee,
                                            price=.2)
    # Extra fee b has a fixed quantity so that this trade model line is applied
    # to all movements by force.
    extra_fee_b = self.createTradeModelLine(trade_condition,
                                            reference='EXTRA_FEE_B',
                                            resource_value=tax,
                                            base_application=fixed_quantity,
                                            price=1)
    # for delivery level
    discount = self.createTradeModelLine(trade_condition,
                                         reference='DISCOUNT_B',
                                         resource_value=tax,
                                         base_application=fixed_quantity,
                                         target_delivery=True,
                                         quantity=10, price=-1)

    transaction.commit() # flush transactional cache

    expected_tax = 1000*0.05, 500*0.05, 500*0.2, 800*0.2, 1, 1, -10
    amount_list = order.getGeneratedAmountList()
    self.assertEqual(sorted(expected_tax),
                     sorted(x.getTotalPrice() for x in amount_list))
    amount_list = order.getAggregatedAmountList()
    expected_tax = 1000*0.05 + 500*0.05, 500*0.2 + 800*0.2, 1 + 1, -10
    self.assertEqual(sorted(expected_tax),
                     sorted(x.getTotalPrice() for x in amount_list))
    # Change target level
    extra_fee_a.setTargetDelivery(True)
    extra_fee_b.setTargetDelivery(True)
    amount_list = order.getAggregatedAmountList()
    expected_tax = 1000*0.05 + 500*0.05, 800*0.2, 1, -10
    self.assertEqual(sorted(expected_tax),
                     sorted(x.getTotalPrice() for x in amount_list))

  def test_tradeModelLineWithRounding(self):
    """
      Test if trade model line works with rounding.
    """
    trade_condition = self.createTradeCondition(self.createBusinessProcess())
    # create a model line and set target level to `delivery`
    tax = self.createTradeModelLine(trade_condition,
                                    reference='TAX',
                                    base_application='base_amount/tax',
                                    base_contribution='base_amount/total_tax',
                                    price=0.05,
                                    target_delivery=True)

    # create a rounding model for tax
    rounding_model = self.portal.portal_roundings.newContent(portal_type='Rounding Model')
    rounding_model.setDecimalRoundingOption('ROUND_DOWN')
    rounding_model.setPrecision(1)
    rounding_model.setRoundedPropertyId('total_price')
    rounding_model._setMembershipCriterionCategoryList(['base_contribution/base_amount/total_tax'])
    rounding_model._setMembershipCriterionBaseCategoryList(['base_contribution'])
    rounding_model.validate()

    # create an order
    resource_A = self.createResource('Product', title='A')
    resource_B = self.createResource('Product', title='B')
    order = self.createOrder(trade_condition)
    order_line_1 = order.newContent(portal_type=self.order_line_portal_type,
                                    price=3333, quantity=1,
                                    resource_value=resource_A,
                                    base_contribution='base_amount/tax')
    order_line_2 = order.newContent(portal_type=self.order_line_portal_type,
                                    price=171, quantity=1,
                                    resource_value=resource_B,
                                    base_contribution='base_amount/tax')

    transaction.commit()
    self.tic()
    # check the result without rounding
    amount, = order.getAggregatedAmountList(rounding=False)
    self.assertEqual((3333+171)*0.05, amount.getTotalPrice()) # 175.2
    # check the result with rounding
    amount, = order.getAggregatedAmountList(rounding=True)
    self.assertEqual(175, amount.getTotalPrice())

    # change tax trade model line to `movement` level
    tax.setTargetDelivery(False)

    def getTotalAmount(amount_list):
      result = 0
      for amount in amount_list:
        if amount.getBaseContribution() in ('base_amount/total',
                                            'base_amount/total_tax'):
          result += amount.getTotalPrice()
      return result

    # check the result without rounding
    amount, = order.getAggregatedAmountList(rounding=False)
    self.assertEqual(3333*0.05+171*0.05, amount.getTotalPrice()) # 175.2
    # check the result with rounding
    amount_list = order.getAggregatedAmountList(rounding=True)
    self.assertEqual(2, len(amount_list)) # XXX 1 or 2 ???
    self.assertEqual(174, getTotalAmount(amount_list))

    # check getAggregatedAmountList result of each movement
    # order line 1
    amount, = order_line_1.getAggregatedAmountList(rounding=False)
    self.assertEqual(3333*0.05, amount.getTotalPrice()) # 166.65
    amount, = order_line_1.getAggregatedAmountList(rounding=True)
    self.assertEqual(166, amount.getTotalPrice())
    # order line 2
    amount, = order_line_2.getAggregatedAmountList(rounding=False)
    self.assertEqual(171*0.05, amount.getTotalPrice()) # 8.55
    amount, = order_line_2.getAggregatedAmountList(rounding=True)
    self.assertEqual(8, amount.getTotalPrice())

    # change rounding model definition
    rounding_model.setDecimalRoundingOption('ROUND_UP')
    rounding_model.setPrecision(1)
    rounding_model.setRoundedPropertyIdList(['total_price', 'quantity'])

    # change quantity
    order_line_1.edit(quantity=3.3333)

    transaction.commit()
    self.tic()

    # check the result without rounding
    amount, = order.getAggregatedAmountList(rounding=False)
    self.assertEqual(3.3333*3333*0.05+171*0.05, amount.getTotalPrice())
    # check the result with rounding
    # both quantity and total price will be rounded so that the expression
    # should be "round_up(round_up(3.3333 * 3333) * 0.05) + round_up(round_up
    # (1* 171) * 0.05)"
    amount_list = order.getAggregatedAmountList(rounding=True)
    self.assertEqual(2, len(amount_list)) # XXX 1 or 2 ???
    self.assertEqual(565, getTotalAmount(amount_list))

    # create a rounding model to round quantity property of order line
    rounding_model_for_quantity = self.portal.portal_roundings.newContent(portal_type='Rounding Model')
    rounding_model_for_quantity.setDecimalRoundingOption('ROUND_DOWN')
    rounding_model_for_quantity.setPrecision(1)
    rounding_model_for_quantity.setRoundedPropertyId('quantity')
    rounding_model_for_quantity._setMembershipCriterionCategoryList(['base_contribution/base_amount/tax'])
    rounding_model_for_quantity._setMembershipCriterionBaseCategoryList(['base_contribution'])
    rounding_model_for_quantity.validate()

    transaction.commit()
    self.tic()

    amount_list = order.getAggregatedAmountList(rounding=True)
    # The expression should be "round_up(round_up(round_down(3.3333) * 3333)
    # * 0.05) + round_up(round_up(round_down(1) * 171) * 0.05)"
    self.assertEqual(2, len(amount_list)) # XXX 1 or 2 ???
    self.assertEqual(509, getTotalAmount(amount_list))

    # create a rounding model to round price property of order line
    rounding_model_for_price = self.portal.portal_roundings.newContent(portal_type='Rounding Model')
    rounding_model_for_price.setDecimalRoundingOption('ROUND_UP')
    rounding_model_for_price.setPrecision(0.1)
    rounding_model_for_price.setRoundedPropertyId('price')
    rounding_model_for_price._setMembershipCriterionCategoryList(['base_contribution/base_amount/tax'])
    rounding_model_for_price._setMembershipCriterionBaseCategoryList(['base_contribution'])
    rounding_model_for_price.validate()

    # change price
    order_line_2.edit(price=171.1234)
    # invalidate rounding model for total price
    rounding_model.invalidate()

    transaction.commit()
    self.tic()

    # check the result without rounding
    amount, = order.getAggregatedAmountList(rounding=False)
    self.assertEqual(3.3333*3333*0.05+171.1234*0.05, amount.getTotalPrice())
    # check the result with rounding
    amount_list = order.getAggregatedAmountList(rounding=True)
    # The expression should be "round_down(3.3333) * round_up(3333) * 0.05 +
    # round_down(1) * round_up(171.1234) * 0.05"
    self.assertEqual(2, len(amount_list)) # XXX 1 or 2 ???
    self.assertEqual(508.51000000000005, getTotalAmount(amount_list))

  def test_tradeModelLineWithEmptyBaseContributionMovement(self):
    """
    Make sure that a movement which does not have any base_contribution values
    does not match to any trade model lines.
    """
    trade_condition = self.createTradeCondition(
      (), (
      dict(price=0.05,
           reference='TAX',
           base_application_list=['base_amount/tax'],
           base_contribution_list=['base_amount/total_tax']),
      ))

    # create an order
    resource_A = self.createResource('Product', title='A')
    order = self.createOrder(
      trade_condition, (
      # create a movement which should be aggregated
      dict(id='1',
           price=100, quantity=1,
           resource_value=resource_A,
           base_contribution_list=['base_amount/tax']),
      # create a movement which base contribution is empty and shoud not be
      # aggregated
      dict(id='2',
           price=31, quantity=1,
           resource_value=resource_A,
           base_contribution_list=[]),
      ))

    transaction.commit()
    self.tic()

    # check the result
    amount, = order.getAggregatedAmountList()
    self.assertEqual(100*0.05, amount.getTotalPrice())


class TestTradeModelLineSale(TestTradeModelLine):
  invoice_portal_type = 'Sale Invoice Transaction'
  invoice_line_portal_type = 'Invoice Line'
  invoice_transaction_line_portal_type = 'Sale Invoice Transaction Line'
  order_portal_type = 'Sale Order'
  order_line_portal_type = 'Sale Order Line'
  packing_list_portal_type = 'Sale Packing List'
  trade_condition_portal_type = 'Sale Trade Condition'


def test_suite():
  suite = unittest.TestSuite()
  suite.addTest(unittest.makeSuite(TestTradeModelLineSale))
  return suite