Commit f2207794 authored by Rafael Monnerat's avatar Rafael Monnerat

slapos_wechat: Add a test for alarm

parent 26c4d8f1
# Copyright (c) 2002-2012 Nexedi SA and Contributors. All Rights Reserved.
from erp5.component.test.SlapOSTestCaseMixin import SlapOSTestCaseMixinWithAbort
from Products.ERP5Type.tests.utils import createZODBPythonScript
from DateTime import DateTime
class TestSlapOSWechatUpdateConfirmedPayment(SlapOSTestCaseMixinWithAbort):
def _simulatePaymentTransaction_startWechatPayment(self):
script_name = 'PaymentTransaction_startWechatPayment'
if script_name in self.portal.portal_skins.custom.objectIds():
raise ValueError('Precondition failed: %s exists in custom' % script_name)
createZODBPythonScript(self.portal.portal_skins.custom,
script_name,
'*args, **kwargs',
'# Script body\n'
"""portal_workflow = context.portal_workflow
portal_workflow.doActionFor(context, action='edit_action', comment='Visited by PaymentTransaction_startWechatPayment') """ )
self.commit()
def _dropPaymentTransaction_startWechatPayment(self):
script_name = 'PaymentTransaction_startWechatPayment'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
self.commit()
def test_alarm_confirmed_draft_wechat(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
payment_mode="wechat",
)
self.portal.portal_workflow._jumpToStateFor(transaction, 'confirmed')
self.tic()
self._simulatePaymentTransaction_startWechatPayment()
try:
self.portal.portal_alarms.slapos_wechat_update_confirmed_payment.activeSense()
self.tic()
finally:
self._dropPaymentTransaction_startWechatPayment()
self.tic()
self.assertEqual(
'Visited by PaymentTransaction_startWechatPayment',
transaction.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_confirmed(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
payment_mode="wechat",
)
self.tic()
self._simulatePaymentTransaction_startWechatPayment()
try:
self.portal.portal_alarms.slapos_wechat_update_confirmed_payment.activeSense()
self.tic()
finally:
self._dropPaymentTransaction_startWechatPayment()
self.tic()
self.assertNotEqual(
'Visited by PaymentTransaction_startWechatPayment',
transaction.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_draft(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
payment_mode="wechat",
)
self.portal.portal_workflow._jumpToStateFor(transaction, 'confirmed')
self.portal.portal_workflow._jumpToStateFor(transaction, 'solved')
self.tic()
self._simulatePaymentTransaction_startWechatPayment()
try:
self.portal.portal_alarms.slapos_wechat_update_confirmed_payment.activeSense()
self.tic()
finally:
self._dropPaymentTransaction_startWechatPayment()
self.tic()
self.assertNotEqual(
'Visited by PaymentTransaction_startWechatPayment',
transaction.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_wechat(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
)
self.portal.portal_workflow._jumpToStateFor(transaction, 'confirmed')
self.tic()
self._simulatePaymentTransaction_startWechatPayment()
try:
self.portal.portal_alarms.slapos_wechat_update_confirmed_payment.activeSense()
self.tic()
finally:
self._dropPaymentTransaction_startWechatPayment()
self.tic()
self.assertNotEqual(
'Visited by PaymentTransaction_startWechatPayment',
transaction.workflow_history['edit_workflow'][-1]['comment'])
def _simulatePaymentTransaction_getTotalPayablePrice(self):
script_name = 'PaymentTransaction_getTotalPayablePrice'
if script_name in self.portal.portal_skins.custom.objectIds():
raise ValueError('Precondition failed: %s exists in custom' % script_name)
createZODBPythonScript(self.portal.portal_skins.custom,
script_name,
'*args, **kwargs',
'# Script body\n'
"""return 1""")
self.commit()
def _simulatePaymentTransaction_getZeroTotalPayablePrice(self):
script_name = 'PaymentTransaction_getTotalPayablePrice'
if script_name in self.portal.portal_skins.custom.objectIds():
raise ValueError('Precondition failed: %s exists in custom' % script_name)
createZODBPythonScript(self.portal.portal_skins.custom,
script_name,
'*args, **kwargs',
'# Script body\n'
"""return 0""")
self.commit()
def _dropPaymentTransaction_getTotalPayablePrice(self):
script_name = 'PaymentTransaction_getTotalPayablePrice'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
self.commit()
def test_not_confirmed_payment(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
payment_mode="wechat",
)
simulation_state = transaction.getSimulationState()
modification_date = transaction.getModificationDate()
self._simulatePaymentTransaction_getTotalPayablePrice()
try:
transaction.PaymentTransaction_startWechatPayment()
finally:
self._dropPaymentTransaction_getTotalPayablePrice()
self.assertEqual(transaction.getSimulationState(), simulation_state)
self.assertEqual(transaction.getModificationDate(), modification_date)
def test_not_wechat_payment(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
)
self.portal.portal_workflow._jumpToStateFor(transaction, 'confirmed')
simulation_state = transaction.getSimulationState()
modification_date = transaction.getModificationDate()
self._simulatePaymentTransaction_getTotalPayablePrice()
try:
transaction.PaymentTransaction_startWechatPayment()
finally:
self._dropPaymentTransaction_getTotalPayablePrice()
self.assertEqual(transaction.getSimulationState(), simulation_state)
self.assertEqual(transaction.getModificationDate(), modification_date)
def test_zero_amount_payment(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
payment_mode="wechat",
)
self.portal.portal_workflow._jumpToStateFor(transaction, 'confirmed')
simulation_state = transaction.getSimulationState()
modification_date = transaction.getModificationDate()
self._simulatePaymentTransaction_getZeroTotalPayablePrice()
try:
transaction.PaymentTransaction_startWechatPayment()
finally:
self._dropPaymentTransaction_getTotalPayablePrice()
self.assertEqual(transaction.getSimulationState(), simulation_state)
self.assertEqual(transaction.getModificationDate(), modification_date)
def test_expected_payment(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
payment_mode="wechat",
)
self.portal.portal_workflow._jumpToStateFor(transaction, 'confirmed')
self._simulatePaymentTransaction_getTotalPayablePrice()
try:
transaction.PaymentTransaction_startWechatPayment()
finally:
self._dropPaymentTransaction_getTotalPayablePrice()
self.assertEqual(transaction.getSimulationState(), 'started')
class TestSlapOSWechatUpdateStartedPayment(SlapOSTestCaseMixinWithAbort):
def test_not_started_payment(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
payment_mode="wechat",
)
simulation_state = transaction.getSimulationState()
modification_date = transaction.getModificationDate()
transaction.PaymentTransaction_updateWechatPaymentStatus()
self.assertEqual(transaction.getSimulationState(), simulation_state)
self.assertEqual(transaction.getModificationDate(), modification_date)
def test_not_wechat_payment(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
)
self.portal.portal_workflow._jumpToStateFor(transaction, 'started')
simulation_state = transaction.getSimulationState()
modification_date = transaction.getModificationDate()
transaction.PaymentTransaction_updateWechatPaymentStatus()
self.assertEqual(transaction.getSimulationState(), simulation_state)
self.assertEqual(transaction.getModificationDate(), modification_date)
def test_not_registered_payment(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
payment_mode="wechat",
)
self.portal.portal_workflow._jumpToStateFor(transaction, 'started')
transaction.PaymentTransaction_updateWechatPaymentStatus()
self.assertEqual(transaction.getSimulationState(), 'started')
def _simulatePaymentTransaction_createPaidWechatEvent(self):
script_name = 'PaymentTransaction_createWechatEvent'
if script_name in self.portal.portal_skins.custom.objectIds():
raise ValueError('Precondition failed: %s exists in custom' % script_name)
createZODBPythonScript(self.portal.portal_skins.custom,
script_name,
'*args, **kwargs',
'# Script body\n'
"""portal_workflow = context.portal_workflow
portal_workflow.doActionFor(context, action='edit_action', comment='Visited by PaymentTransaction_createWechatEvent')
class Foo:
def updateStatus(self):
context.stop()
return Foo()
""" )
self.commit()
def _simulatePaymentTransaction_createNotPaidWechatEvent(self):
script_name = 'PaymentTransaction_createWechatEvent'
if script_name in self.portal.portal_skins.custom.objectIds():
raise ValueError('Precondition failed: %s exists in custom' % script_name)
createZODBPythonScript(self.portal.portal_skins.custom,
script_name,
'*args, **kwargs',
'# Script body\n'
"""portal_workflow = context.portal_workflow
portal_workflow.doActionFor(context, action='edit_action', comment='Visited by PaymentTransaction_createWechatEvent')
class Foo:
def updateStatus(self):
pass
return Foo()
""" )
self.commit()
def _dropPaymentTransaction_createWechatEvent(self):
script_name = 'PaymentTransaction_createWechatEvent'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
self.commit()
def test_paid_payment(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
payment_mode="wechat",
start_date=DateTime(),
)
self.portal.portal_workflow._jumpToStateFor(transaction, 'started')
# Manually generate mapping
transaction.PaymentTransaction_generateWechatId()
self._simulatePaymentTransaction_createPaidWechatEvent()
try:
transaction.PaymentTransaction_updateWechatPaymentStatus()
finally:
self._dropPaymentTransaction_createWechatEvent()
self.assertEqual(
'Visited by PaymentTransaction_createWechatEvent',
transaction.workflow_history['edit_workflow'][-1]['comment'])
self.assertEqual(
"",
transaction.workflow_history['edit_workflow'][-2]['comment'])
self.assertEqual(transaction.getSimulationState(), 'stopped')
def test_not_paid_payment(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
payment_mode="wechat",
start_date=DateTime(),
)
self.portal.portal_workflow._jumpToStateFor(transaction, 'started')
# Manually generate mapping
transaction.PaymentTransaction_generateWechatId()
self._simulatePaymentTransaction_createNotPaidWechatEvent()
try:
transaction.PaymentTransaction_updateWechatPaymentStatus()
finally:
self._dropPaymentTransaction_createWechatEvent()
self.assertEqual(
'Visited by PaymentTransaction_createWechatEvent',
transaction.workflow_history['edit_workflow'][-1]['comment'])
self.assertEqual(transaction.getSimulationState(), 'started')
def _simulatePaymentTransaction_updateWechatPaymentStatus(self):
script_name = 'PaymentTransaction_updateWechatPaymentStatus'
if script_name in self.portal.portal_skins.custom.objectIds():
raise ValueError('Precondition failed: %s exists in custom' % script_name)
createZODBPythonScript(self.portal.portal_skins.custom,
script_name,
'*args, **kwargs',
'# Script body\n'
"""portal_workflow = context.portal_workflow
portal_workflow.doActionFor(context, action='edit_action', comment='Visited by PaymentTransaction_updateWechatPaymentStatus') """ )
self.commit()
def _dropPaymentTransaction_updateWechatPaymentStatus(self):
script_name = 'PaymentTransaction_updateWechatPaymentStatus'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
self.commit()
def test_alarm_started_draft_wechat(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
payment_mode="wechat",
)
self.portal.portal_workflow._jumpToStateFor(transaction, 'started')
self.tic()
self._simulatePaymentTransaction_updateWechatPaymentStatus()
try:
self.portal.portal_alarms.slapos_wechat_update_started_payment.activeSense()
self.tic()
finally:
self._dropPaymentTransaction_updateWechatPaymentStatus()
self.tic()
self.assertEqual(
'Visited by PaymentTransaction_updateWechatPaymentStatus',
transaction.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_started(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
payment_mode="wechat",
)
self.tic()
self._simulatePaymentTransaction_updateWechatPaymentStatus()
try:
self.portal.portal_alarms.slapos_wechat_update_started_payment.activeSense()
self.tic()
finally:
self._dropPaymentTransaction_updateWechatPaymentStatus()
self.tic()
self.assertNotEqual(
'Visited by PaymentTransaction_updateWechatPaymentStatus',
transaction.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_draft(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
payment_mode="wechat",
)
self.portal.portal_workflow._jumpToStateFor(transaction, 'started')
self.portal.portal_workflow._jumpToStateFor(transaction, 'solved')
self.tic()
self._simulatePaymentTransaction_updateWechatPaymentStatus()
try:
self.portal.portal_alarms.slapos_wechat_update_started_payment.activeSense()
self.tic()
finally:
self._dropPaymentTransaction_updateWechatPaymentStatus()
self.tic()
self.assertNotEqual(
'Visited by PaymentTransaction_updateWechatPaymentStatus',
transaction.workflow_history['edit_workflow'][-1]['comment'])
def test_alarm_not_wechat(self):
new_id = self.generateNewId()
transaction = self.portal.accounting_module.newContent(
portal_type='Payment Transaction',
title="Transaction %s" % new_id,
reference="TESTTRANS-%s" % new_id,
)
self.portal.portal_workflow._jumpToStateFor(transaction, 'started')
self.tic()
self._simulatePaymentTransaction_updateWechatPaymentStatus()
try:
self.portal.portal_alarms.slapos_wechat_update_started_payment.activeSense()
self.tic()
finally:
self._dropPaymentTransaction_updateWechatPaymentStatus()
self.tic()
self.assertNotEqual(
'Visited by PaymentTransaction_updateWechatPaymentStatus',
transaction.workflow_history['edit_workflow'][-1]['comment'])
\ No newline at end of file
<?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>_recorded_property_dict</string> </key>
<value>
<persistent> <string encoding="base64">AAAAAAAAAAI=</string> </persistent>
</value>
</item>
<item>
<key> <string>default_reference</string> </key>
<value> <string>testSlapOSWechatAlarm</string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>test.erp5.testSlapOSWechatAlarm</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">AAAAAAAAAAM=</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/>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="3" aka="AAAAAAAAAAM=">
<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">AAAAAAAAAAQ=</string> </persistent>
</value>
</item>
</dictionary>
</value>
</item>
</dictionary>
</pickle>
</record>
<record id="4" aka="AAAAAAAAAAQ=">
<pickle>
<global name="WorkflowHistoryList" module="Products.ERP5Type.patches.WorkflowTool"/>
</pickle>
<pickle>
<tuple>
<none/>
<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>
<none/>
</tuple>
</pickle>
</record>
</ZopeData>
test.erp5.testSlapOSWechatSkins test.erp5.testSlapOSWechatSkins
test.erp5.testSlapOSWechatWorkflow test.erp5.testSlapOSWechatWorkflow
test.erp5.testSlapOSWechatAlarm
\ No newline at end of file
Markdown is supported
0%
or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment