inventory_list interpolation flow: support at_date & to_date

bt5/tmp_work/TestTemplateItem/portal_components/test.erp5.testInventoryAPI.py

@@ -816,7 +816,7 @@ class TestInventory(InventoryAPITestCase):
                             resource=self.resource.getRelativeUrl(),
                             at_date=date_gmt_1)
 
-  def test_interpolation_method_linear(self):
+  def test_interpolation_method_linear_to_date(self):
     self._makeMovement(
        quantity=10,
        start_date=DateTime("2016/01/01 01:00:00"),
@@ -881,7 +881,72 @@ class TestInventory(InventoryAPITestCase):
       to_date=DateTime("2016/01/01 11:00:00"),
       interpolation_method='linear')
 
-  def test_interpolation_method_XXX_one_for_all(self):
+  def test_interpolation_method_linear_at_date(self):
+    self._makeMovement(
+       quantity=10,
+       start_date=DateTime("2016/01/01 01:00:00"),
+       stop_date=DateTime("2016/01/01 11:00:00"),
+    )
+
+    # With a time frame that does not contain the movement, we have 0%
+    self.assertInventoryEquals(0,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/02/01 00:00:00"),
+      at_date=DateTime("2016/02/02 00:00:00"),
+      interpolation_method='linear')
+
+    # With a time frame that contains the full movement, we have 100% of the quantity
+    self.assertInventoryEquals(10,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 00:00:00"),
+      at_date=DateTime("2016/01/02 00:00:00"),
+      interpolation_method='linear')
+
+    # corner case: exact same time, we also have 100%
+    self.assertInventoryEquals(10,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 01:00:00"),
+      at_date=DateTime("2016/01/01 11:00:00"),
+      interpolation_method='linear')
+
+    # With a time frame containing the 50% of the movement, we have 50% of the quantity
+
+    # time frame start before movement
+    self.assertInventoryEquals(5,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 00:00:00"),
+      at_date=DateTime("2016/01/01 06:00:00"),
+      interpolation_method='linear')
+
+    # time frame start at exact same time as movement
+    self.assertInventoryEquals(5,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 01:00:00"),
+      at_date=DateTime("2016/01/01 06:00:00"),
+      interpolation_method='linear')
+
+    # Time frame is contained inside the movement
+    self.assertInventoryEquals(5,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 02:00:00"),
+      at_date=DateTime("2016/01/01 07:00:00"),
+      interpolation_method='linear')
+
+    # Time frame finishes after movement end
+    self.assertInventoryEquals(5,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 06:00:00"),
+      at_date=DateTime("2016/01/01 12:00:00"),
+      interpolation_method='linear')
+
+    # Time frame finishes at exact same time that movement end
+    self.assertInventoryEquals(5,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 06:00:00"),
+      at_date=DateTime("2016/01/01 11:00:00"),
+      interpolation_method='linear')
+
+  def test_interpolation_method_XXX_one_for_all_to_date(self):
     self._makeMovement(
        quantity=10,
        start_date=DateTime("2016/01/01 01:00:00"),
@@ -947,7 +1012,73 @@ class TestInventory(InventoryAPITestCase):
       to_date=DateTime("2016/01/01 11:00:00"),
       interpolation_method='one_for_all')
 
-  def test_interpolation_method_XXX_all_or_nothing(self):
+  def test_interpolation_method_XXX_one_for_all_at_date(self):
+    self._makeMovement(
+       quantity=10,
+       start_date=DateTime("2016/01/01 01:00:00"),
+       stop_date=DateTime("2016/01/01 11:00:00"),
+    )
+
+    # With a time frame that does not contain the movement, we have 0%
+    self.assertInventoryEquals(0,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/02/01 00:00:00"),
+      at_date=DateTime("2016/02/02 00:00:00"),
+      interpolation_method='one_for_all')
+
+    # With a time frame that contains the full movement, we have 100% of the quantity
+    self.assertInventoryEquals(10,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 00:00:00"),
+      at_date=DateTime("2016/01/02 00:00:00"),
+      interpolation_method='one_for_all')
+
+    # corner case: exact same time, we also have 100%
+    self.assertInventoryEquals(10,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 01:00:00"),
+      at_date=DateTime("2016/01/01 11:00:00"),
+      interpolation_method='one_for_all')
+
+    # With a time frame containing the 50% of the movement, we have 100% of the quantity
+    # this is "one_for_all" XXX naming
+
+    # time frame start before movement
+    self.assertInventoryEquals(10,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 00:00:00"),
+      at_date=DateTime("2016/01/01 06:00:00"),
+      interpolation_method='one_for_all')
+
+    # time frame start at exact same time as movement
+    self.assertInventoryEquals(10,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 01:00:00"),
+      at_date=DateTime("2016/01/01 06:00:00"),
+      interpolation_method='one_for_all')
+
+    # Time frame is contained inside the movement
+    self.assertInventoryEquals(10,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 02:00:00"),
+      at_date=DateTime("2016/01/01 07:00:00"),
+      interpolation_method='one_for_all')
+
+    # Time frame finishes after movement end
+    self.assertInventoryEquals(10,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 06:00:00"),
+      at_date=DateTime("2016/01/01 12:00:00"),
+      interpolation_method='one_for_all')
+
+    # Time frame finishes at exact same time that movement end
+    self.assertInventoryEquals(10,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 06:00:00"),
+      at_date=DateTime("2016/01/01 11:00:00"),
+      interpolation_method='one_for_all')
+
+  def test_interpolation_method_XXX_all_or_nothing_to_date(self):
     self._makeMovement(
        quantity=10,
        start_date=DateTime("2016/01/01 01:00:00"),
@@ -1014,6 +1145,73 @@ class TestInventory(InventoryAPITestCase):
       interpolation_method='all_or_nothing')
 
 
+  def test_interpolation_method_XXX_all_or_nothing_at_date(self):
+    self._makeMovement(
+       quantity=10,
+       start_date=DateTime("2016/01/01 01:00:00"),
+       stop_date=DateTime("2016/01/01 11:00:00"),
+    )
+
+    # With a time frame that does not contain the movement, we have 0%
+    self.assertInventoryEquals(0,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/02/01 00:00:00"),
+      at_date=DateTime("2016/02/02 00:00:00"),
+      interpolation_method='all_or_nothing')
+
+    # With a time frame that contains the full movement, we have 100% of the quantity
+    self.assertInventoryEquals(10,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 00:00:00"),
+      at_date=DateTime("2016/01/02 00:00:00"),
+      interpolation_method='all_or_nothing')
+
+    # corner case: exact same time, we also have 0%, because at_date
+    self.assertInventoryEquals(0,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 01:00:00"),
+      at_date=DateTime("2016/01/01 11:00:00"),
+      interpolation_method='all_or_nothing')
+
+    # With a time frame containing the 50% of the movement, we have 0% of the quantity
+    # this is "all or nothing" XXX naming
+
+    # time frame start before movement
+    self.assertInventoryEquals(0,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 00:00:00"),
+      at_date=DateTime("2016/01/01 06:00:00"),
+      interpolation_method='all_or_nothing')
+
+    # time frame start at exact same time as movement
+    self.assertInventoryEquals(0,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 01:00:00"),
+      at_date=DateTime("2016/01/01 06:00:00"),
+      interpolation_method='all_or_nothing')
+
+    # Time frame is contained inside the movement
+    self.assertInventoryEquals(0,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 02:00:00"),
+      at_date=DateTime("2016/01/01 07:00:00"),
+      interpolation_method='all_or_nothing')
+
+    # Time frame finishes after movement end
+    self.assertInventoryEquals(0,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 06:00:00"),
+      at_date=DateTime("2016/01/01 12:00:00"),
+      interpolation_method='all_or_nothing')
+
+    # Time frame finishes at exact same time that movement end
+    self.assertInventoryEquals(0,
+      node_uid=self.node.getUid(),
+      from_date=DateTime("2016/01/01 06:00:00"),
+      at_date=DateTime("2016/01/01 11:00:00"),
+      interpolation_method='all_or_nothing')
+
+
 class TestInventoryList(InventoryAPITestCase):
   """Tests getInventoryList methods.
   """

product/ERP5/Tool/SimulationTool.py

@@ -648,11 +648,11 @@ class SimulationTool(BaseTool):
         if date_dict:
           column_value_dict['date'] = date_dict
         if interpolation_method != 'default':
-          assert from_date and to_date
+          assert from_date and (to_date or at_date)
           # if we consider flow, we also select movement whose mirror date is
           # in the from_date/to_date range and movement whose
           # start_date/stop_date contains the report range.
-          # XXX review this
+          if to_date:
             column_value_dict['date'] = ComplexQuery(
                   Query(date=(from_date, to_date), range='minmax'),
                   Query(mirror_date=(from_date, to_date), range='minmax'),
@@ -666,6 +666,20 @@ class SimulationTool(BaseTool):
                       operator="AND"),
                   operator="OR"
               )
+          else:
+            column_value_dict['date'] = ComplexQuery(
+                  Query(date=(from_date, at_date), range='minngt'),
+                  Query(mirror_date=(from_date, at_date), range='minngt'),
+                  ComplexQuery(
+                      Query(mirror_date=from_date, range='min'),
+                      Query(date=at_date, range='ngt'),
+                      operator="AND"),
+                  ComplexQuery(
+                      Query(date=from_date, range='min'),
+                      Query(mirror_date=at_date, range='ngt'),
+                      operator="AND"),
+                  operator="OR"
+              )
       else:
         column_value_dict['date'] = {'query': [to_date], 'range': 'ngt'}
         column_value_dict['mirror_date'] = {'query': [from_date], 'range': 'nlt'}