Check SimulationMovement_buildSlapOS with simulator.

master/bt5/slapos_accounting/TestTemplateItem/testSlapOSAccountingSlapOSTriggerBuildAlarm.py

@@ -7,10 +7,40 @@
 
 import transaction
 import functools
+import os
+import tempfile
 from Products.ERP5Type.tests.utils import createZODBPythonScript
 from Products.SlapOS.tests.testSlapOSMixin import \
   testSlapOSMixin
 
+def withAbort(func):
+  @functools.wraps(func)
+  def wrapped(self, *args, **kwargs):
+    try:
+      func(self, *args, **kwargs)
+    finally:
+      transaction.abort()
+  return wrapped
+
+class Simulator:
+  def __init__(self, outfile, method, to_return=None):
+    self.outfile = outfile
+    self.method = method
+    self.to_return = to_return
+
+  def __call__(self, *args, **kwargs):
+    """Simulation Method"""
+    old = open(self.outfile, 'r').read()
+    if old:
+      l = eval(old)
+    else:
+      l = []
+    l.append({'recmethod': self.method,
+      'recargs': args,
+      'reckwargs': kwargs})
+    open(self.outfile, 'w').write(repr(l))
+    return self.to_return
+
 def simulateSimulationMovement_buildSlapOS(func):
   @functools.wraps(func)
   def wrapped(self, *args, **kwargs):
@@ -70,3 +100,85 @@ class TestAlarm(testSlapOSMixin):
     self.assertNotEqual(
         'Not visited by SimulationMovement_buildSlapOS',
         simulation_movement.getTitle())
+
+  @withAbort
+  def test_SimulationMovement_buildSlapOS(self):
+    business_process = self.portal.business_process_module.newContent(
+        portal_type='Business Process')
+    root_business_link = business_process.newContent(
+        portal_type='Business Link')
+    business_link = business_process.newContent(portal_type='Business Link')
+
+    root_applied_rule = self.portal.portal_simulation.newContent(
+        portal_type='Applied Rule')
+    simulation_movement = root_applied_rule.newContent(
+        causality=root_business_link.getRelativeUrl(),
+        portal_type='Simulation Movement')
+
+    applied_rule = simulation_movement.newContent(portal_type='Applied Rule')
+    lower_simulation_movement = applied_rule.newContent(
+        causality=business_link.getRelativeUrl(),
+        portal_type='Simulation Movement')
+
+    build_simulator = tempfile.mkstemp()[1]
+    activate_simulator = tempfile.mkstemp()[1]
+    try:
+      from Products.CMFActivity.ActiveObject import ActiveObject
+      ActiveObject.original_activate = ActiveObject.activate
+      ActiveObject.activate = Simulator(activate_simulator, 'activate',
+          root_applied_rule)
+      from Products.ERP5.Document.BusinessLink import BusinessLink
+      BusinessLink.original_build = BusinessLink.build
+      BusinessLink.build = Simulator(build_simulator, 'build')
+
+      simulation_movement.SimulationMovement_buildSlapOS(tag='root_tag')
+
+      build_value = eval(open(build_simulator).read())
+      activate_value = eval(open(activate_simulator).read())
+
+      self.assertEqual([{
+        'recmethod': 'build',
+        'recargs': (),
+        'reckwargs': {'path': '%s/%%' % root_applied_rule.getPath(),
+        'activate_kw': {'tag': 'root_tag'}}}],
+        build_value
+      )
+      self.assertEqual([{
+        'recmethod': 'activate',
+        'recargs': (),
+        'reckwargs': {'tag': 'build_in_progress_%s_%s' % (
+            root_business_link.getUid(), root_applied_rule.getUid()),
+          'after_tag': 'root_tag', 'activity': 'SQLQueue'}}],
+        activate_value)
+
+      open(build_simulator, 'w').truncate()
+      open(activate_simulator, 'w').truncate()
+
+      lower_simulation_movement.SimulationMovement_buildSlapOS(tag='lower_tag')
+      build_value = eval(open(build_simulator).read())
+      activate_value = eval(open(activate_simulator).read())
+
+      self.assertEqual([{
+        'recmethod': 'build',
+        'recargs': (),
+        'reckwargs': {'path': '%s/%%' % root_applied_rule.getPath(),
+        'activate_kw': {'tag': 'lower_tag'}}}],
+        build_value
+      )
+      self.assertEqual([{
+        'recmethod': 'activate',
+        'recargs': (),
+        'reckwargs': {'tag': 'build_in_progress_%s_%s' % (
+            business_link.getUid(), root_applied_rule.getUid()),
+          'after_tag': 'lower_tag', 'activity': 'SQLQueue'}}],
+        activate_value)
+
+    finally:
+      ActiveObject.activate = ActiveObject.original_activate
+      delattr(ActiveObject, 'original_activate')
+      BusinessLink.build = BusinessLink.original_build
+      delattr(BusinessLink, 'original_build')
+      if os.path.exists(build_simulator):
+        os.unlink(build_simulator)
+      if os.path.exists(activate_simulator):
+        os.unlink(activate_simulator)

master/bt5/slapos_accounting/bt/revision

-36
+37
\ No newline at end of file