Commit 71c9e9ea authored by Rafael Monnerat's avatar Rafael Monnerat

slapos_subscription_request: Test Subscription Scenarios with Chinese currency

parent 5e5d4803
# -*- coding: utf-8 -*-
##############################################################################
#
# Copyright (c) 2012 Nexedi SA and Contributors. All Rights Reserved.
#
##############################################################################
from erp5.component.test.testSlapOSSubscriptionScenario import TestSlapOSSubscriptionScenarioMixin
class TestSlapOSSubscriptionChineseScenario(TestSlapOSSubscriptionScenarioMixin):
def afterSetUp(self):
TestSlapOSSubscriptionScenarioMixin.afterSetUp(self)
self.expected_individual_price_without_tax = 1573.3333333333335
self.expected_individual_price_with_tax = 1888.00
def createSubscriptionCondition(self, slave=False):
self.subscription_condition = self.portal.subscription_condition_module.newContent(
portal_type="Subscription Condition",
title="TestSubscriptionChineseScenario",
short_tile="Test Your Chinese Scenario",
description="This is a Chinese test",
url_string=self.generateNewSoftwareReleaseUrl(),
root_slave=slave,
price=1888.00,
resource="currency_module/RMB",
default_source_reference="default",
reference="rapidvm%s" % self.new_id,
# Aggregate and Follow up to web pages for product description and
# Terms of service
sla_xml='<?xml version="1.0" encoding="utf-8"?>\n<instance>\n</instance>',
text_content='<?xml version="1.0" encoding="utf-8"?>\n<instance>\n</instance>',
user_input={}
)
self.subscription_condition.validate()
self.subscription_condition.updateLocalRolesOnSecurityGroups()
self.tic()
def test_subscription_scenario(self):
self._test_subscription_scenario(amount=1)
def test_subscription_with_3_vms_scenario(self):
self._test_subscription_scenario(amount=3)
def test_two_subscription_scenario(self):
self._test_two_subscription_scenario(amount=1)
<?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>testSlapOSSubscriptionChineseScenario</string> </value>
</item>
<item>
<key> <string>description</string> </key>
<value>
<none/>
</value>
</item>
<item>
<key> <string>id</string> </key>
<value> <string>test.erp5.testSlapOSSubscriptionChineseScenario</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>
......@@ -9,9 +9,12 @@ from erp5.component.test.SlapOSTestCaseDefaultScenarioMixin import DefaultScenar
from erp5.component.test.SlapOSTestCaseMixin import changeSkin
from DateTime import DateTime
class TestSlapOSTrialScenario(DefaultScenarioMixin):
class TestSlapOSSubscriptionScenarioMixin(DefaultScenarioMixin):
def afterSetUp(self):
self.expected_individual_price_without_tax = 162.50
self.expected_individual_price_with_tax = 195.00
self.login()
self.portal.portal_alarms.slapos_subscription_request_process_draft.setEnabled(True)
self.portal.portal_alarms.slapos_subscription_request_process_ordered.setEnabled(True)
......@@ -36,6 +39,13 @@ class TestSlapOSTrialScenario(DefaultScenarioMixin):
self.cleanUpSubscriptionRequest()
self.tic()
self.login()
self.createSubscriptionCondition()
# some preparation
self.logout()
self.web_site = self.portal.web_site_module.hostingjs
def cleanUpSubscriptionRequest(self):
for subscription_request in self.portal.portal_catalog(
portal_type="Subscription Request",
......@@ -182,10 +192,11 @@ class TestSlapOSTrialScenario(DefaultScenarioMixin):
# 195 is the month payment
# 195*3 is the 3 months to pay upfront to use.
# 25 is the reservation fee deduction.
authAmount = (int(self.expected_individual_price_with_tax*100)*3-2500)*quantity
data_kw = {
'errorCode': '0',
'transactionStatus': '6',
'authAmount': (19500*3-2500)*quantity,
'authAmount': authAmount,
'authDevise': '978',
}
payment.PaymentTransaction_createPayzenEvent().PayzenEvent_processUpdate(data_kw, True)
......@@ -306,8 +317,10 @@ class TestSlapOSTrialScenario(DefaultScenarioMixin):
quantity = subscription_request.getQuantity()
# The values are without tax
self.assertEqual(sale_packing_list_line.getQuantity(), 3*quantity)
self.assertEqual(sale_packing_list_line.getPrice(), 162.50)
self.assertEqual(sale_packing_list_line.getTotalPrice(), 3*162.50*quantity)
self.assertEqual(sale_packing_list_line.getPrice(),
self.expected_individual_price_without_tax)
self.assertEqual(round(sale_packing_list_line.getTotalPrice()),
3*self.expected_individual_price_without_tax*quantity)
self.assertEqual(sale_packing_list.getCausality(),
subscription_request.getRelativeUrl())
......@@ -325,7 +338,7 @@ class TestSlapOSTrialScenario(DefaultScenarioMixin):
subscription_request.getRelativeUrl())
@changeSkin('Hal')
def _requestSubscription(self, **kw):
return self.web_site.hateoas.SubscriptionRequestModule_requestSubscription(**kw)
return self.web_site.hateoas.SubscriptionRequestModule_requestSubscritption(**kw)
def getAggregatedSalePackingList(self, subscription_request, specialise):
person_uid = subscription_request.getDestinationSectionValue().getUid()
......@@ -351,17 +364,7 @@ class TestSlapOSTrialScenario(DefaultScenarioMixin):
destination_decision_uid=person_uid,
specialise_uid=specialise_uid)
def test_subscription_scenario(self, amount=1):
""" The admin creates an computer, user can request instances on it"""
self.login()
self.createSubscriptionCondition()
# some preparation
self.logout()
self.web_site = self.portal.web_site_module.hostingjs
# Call here, We should create the instance in advance...
def createPublicServerForAdminUser(self):
# hooray, now it is time to create computers
self.login(self.admin_user.getUserId())
......@@ -383,10 +386,12 @@ class TestSlapOSTrialScenario(DefaultScenarioMixin):
self.tic()
self.logout()
return subscription_server
# Call as anonymous... check response?
default_email_text = "abc%s@nexedi.com" % self.new_id
name="ABC %s" % self.new_id
def requestAndCheckHostingSubscription(self, amount, name,
default_email_text):
self.logout()
self._requestSubscription(
subscription_reference=self.subscription_condition.getReference(),
amount=amount,
......@@ -426,6 +431,13 @@ class TestSlapOSTrialScenario(DefaultScenarioMixin):
self.stepCallSlaposSubscriptionRequestProcessOrderedAlarm()
self.tic()
return subscription_request
def checkSubscriptionDeploymentAndSimulation(self, default_email_text, subscription_server):
subscription_request_list = self.getSubscriptionRequestList(
default_email_text, self.subscription_condition)
for subscription_request in subscription_request_list:
# Check if instance was requested
self.checkRelatedInstance(subscription_request)
......@@ -436,6 +448,7 @@ class TestSlapOSTrialScenario(DefaultScenarioMixin):
self.simulateSlapgridCP(subscription_server)
self.tic()
for subscription_request in subscription_request_list:
# Re-check, as instance shouldn't be allocated until
# the confirmation of the new Payment.
self.checkRelatedInstance(subscription_request)
......@@ -469,14 +482,16 @@ class TestSlapOSTrialScenario(DefaultScenarioMixin):
specialise_subscription_uid = preference_tool.getPreferredAggregatedSubscriptionSaleTradeCondition()
specialise_uid = preference_tool.getPreferredAggregatedSaleTradeCondition()
for subscription_request in subscription_request_list:
sale_packing_list_list = self.getAggregatedSalePackingList(
subscription_request, specialise_subscription_uid)
self.assertEqual(1, len(sale_packing_list_list))
self.checkAggregatedSalePackingList(subscription_request, sale_packing_list_list[0])
self.assertEqual(3, len(self.getSubscriptionSalePackingList(
subscription_request)))
expected_sale_packing_list_amount = len(subscription_request_list) * 3
self.assertEqual(expected_sale_packing_list_amount,
len(self.getSubscriptionSalePackingList(subscription_request)))
self.assertEqual(0, len(self.getAggregatedSalePackingList(
subscription_request, specialise_uid)))
......@@ -486,6 +501,7 @@ class TestSlapOSTrialScenario(DefaultScenarioMixin):
accounting_date=DateTime('2222/01/01'))
self.tic()
for subscription_request in subscription_request_list:
self.assertEqual(1, len(self.getAggregatedSalePackingList(
subscription_request, specialise_subscription_uid)))
......@@ -493,6 +509,7 @@ class TestSlapOSTrialScenario(DefaultScenarioMixin):
self.stepCallSlaposStartConfirmedAggregatedSubscriptionSalePackingListAlarm()
self.tic()
for subscription_request in subscription_request_list:
self.assertEqual(0, len(self.getAggregatedSalePackingList(
subscription_request, specialise_uid)))
......@@ -555,6 +572,7 @@ class TestSlapOSTrialScenario(DefaultScenarioMixin):
self.stepCallSlaposCrmCreateRegularisationRequestAlarm()
self.tic()
for subscription_request in subscription_request_list:
self.checkAndPayFirstTreeMonth(subscription_request)
self.tic()
......@@ -563,6 +581,7 @@ class TestSlapOSTrialScenario(DefaultScenarioMixin):
self.stepCallSlaposSubscriptionRequestProcessOrderedAlarm()
self.tic()
for subscription_request in subscription_request_list:
# Check if instance was already allocated, in this case, it shouldn't
# until the allocation alarm kicks in.
self.checkRelatedInstance(subscription_request)
......@@ -575,85 +594,36 @@ class TestSlapOSTrialScenario(DefaultScenarioMixin):
self.simulateSlapgridCP(subscription_server)
self.tic()
for subscription_request in subscription_request_list:
# Re-check, now it should be allocated.
self.checkAllocationOnRelatedInstance(subscription_request)
def test_subscription_with_3_vms_scenario(self):
self.test_subscription_scenario(amount=3)
def test_two_subscription_scenario(self, amount=1):
def _test_subscription_scenario(self, amount=1):
""" The admin creates an computer, user can request instances on it"""
self.login()
self.createSubscriptionCondition()
# some preparation
self.logout()
self.web_site = self.portal.web_site_module.hostingjs
# Call here, We should create the instance in advance...
# hooray, now it is time to create computers
self.login(self.admin_user.getUserId())
# Create a Public Server for admin user, and allow
subscription_server_title = 'Trial Public Server for Admin User %s' % self.new_id
subscription_server_id = self.requestComputer(subscription_server_title)
subscription_server = self.portal.portal_catalog.getResultValue(
portal_type='Computer', reference=subscription_server_id)
self.setAccessToMemcached(subscription_server)
self.assertNotEqual(None, subscription_server)
self.setServerOpenSubscription(subscription_server)
# and install some software on them
subscription_server_software = self.subscription_condition.getUrlString()
self.supplySoftware(subscription_server, subscription_server_software)
# format the computers
self.formatComputer(subscription_server)
self.tic()
self.logout()
subscription_server = self.createPublicServerForAdminUser()
# Call as anonymous... check response?
default_email_text = "abc%s@nexedi.com" % self.new_id
name="ABC %s" % self.new_id
self._requestSubscription(
subscription_reference=self.subscription_condition.getReference(),
amount=amount,
name=name,
default_email_text=default_email_text,
REQUEST=self.portal.REQUEST)
self.login()
# I'm not sure if this is realistic
self.tic()
subscription_request = self.getSubscriptionRequest(
default_email_text, self.subscription_condition)
self.checkDraftSubscriptionRequest(subscription_request,
default_email_text, self.subscription_condition)
# Check Payment and pay it.
self.checkAndPaySubscriptionPayment(subscription_request)
self.tic()
self.requestAndCheckHostingSubscription(amount, name, default_email_text)
# Call alarm to check payment and invoice and move foward to planned.
self.stepCallSlaposSubscriptionRequestProcessDraftAlarm()
self.tic()
self.checkSubscriptionDeploymentAndSimulation(
default_email_text, subscription_server)
self.checkPlannedSubscriptionRequest(subscription_request,
default_email_text, self.subscription_condition)
def _test_two_subscription_scenario(self, amount=1):
""" The admin creates an computer, user can request instances on it"""
# Call alarm to mark subscription request as ordered, bootstrap the user
# and check if email is sent, once done move to ordered state.
self.stepCallSlaposSubscriptionRequestProcessPlannedAlarm()
self.tic()
subscription_server = self.createPublicServerForAdminUser()
self.checkOrderedSubscriptionRequest(subscription_request,
default_email_text, self.subscription_condition)
# Call as anonymous... check response?
default_email_text = "abc%s@nexedi.com" % self.new_id
name="ABC %s" % self.new_id
self.requestAndCheckHostingSubscription(amount, name, default_email_text)
self.logout()
# Request a second one
# Request a second one, without require confirmation and verifing the second subscription request
self._requestSubscription(
subscription_reference=self.subscription_condition.getReference(),
amount=amount,
......@@ -694,165 +664,16 @@ class TestSlapOSTrialScenario(DefaultScenarioMixin):
self.stepCallSlaposSubscriptionRequestProcessOrderedAlarm()
self.tic()
subscription_request_list = self.getSubscriptionRequestList(
default_email_text, self.subscription_condition)
for subscription_request in subscription_request_list:
# Check if instance was requested
self.checkRelatedInstance(subscription_request)
self.stepCallSlaposAllocateInstanceAlarm()
self.tic()
# now instantiate it on computer and set some nice connection dict
self.simulateSlapgridCP(subscription_server)
self.tic()
for subscription_request in subscription_request_list:
# Re-check, as instance shouldn't be allocated until
# the confirmation of the new Payment.
self.checkRelatedInstance(subscription_request)
# check the Open Sale Order coverage
self.stepCallSlaposRequestUpdateHostingSubscriptionOpenSaleOrderAlarm()
self.tic()
# generate simulation for open order
self.stepCallUpdateOpenOrderSimulationAlarm()
self.tic()
# build subscription packing list
self.stepCallSlaposTriggerBuildAlarm()
self.tic()
# stabilise build deliveries and expand them
self.stepCallSlaposManageBuildingCalculatingDeliveryAlarm()
self.tic()
# build aggregated packing list
self.stepCallSlaposTriggerAggregatedDeliveryOrderBuilderAlarm()
self.tic()
# stabilise aggregated deliveries and expand them
self.stepCallSlaposManageBuildingCalculatingDeliveryAlarm()
self.tic()
# check if Packing list is generated with the right trade condition
preference_tool = self.portal.portal_preferences
specialise_subscription_uid = preference_tool.getPreferredAggregatedSubscriptionSaleTradeCondition()
specialise_uid = preference_tool.getPreferredAggregatedSaleTradeCondition()
for subscription_request in subscription_request_list:
sale_packing_list_list = self.getAggregatedSalePackingList(
subscription_request, specialise_subscription_uid)
self.assertEqual(1, len(sale_packing_list_list))
self.checkAggregatedSalePackingList(subscription_request, sale_packing_list_list[0])
self.assertEqual(6, len(self.getSubscriptionSalePackingList(
subscription_request)))
self.assertEqual(0, len(self.getAggregatedSalePackingList(
subscription_request, specialise_uid)))
# Call this alarm shouldn't affect the delivery
self.stepCallSlaposStartConfirmedAggregatedSalePackingListAlarm(
accounting_date=DateTime('2222/01/01'))
self.tic()
for subscription_request in subscription_request_list:
self.assertEqual(1, len(self.getAggregatedSalePackingList(
subscription_request, specialise_subscription_uid)))
# Call this alarm shouldn't affect the delivery
self.stepCallSlaposStartConfirmedAggregatedSubscriptionSalePackingListAlarm()
self.tic()
for subscription_request in subscription_request_list:
self.assertEqual(0, len(self.getAggregatedSalePackingList(
subscription_request, specialise_uid)))
# stabilise aggregated deliveries and expand them
self.stepCallSlaposManageBuildingCalculatingDeliveryAlarm()
self.tic()
# deliver aggregated deliveries
self.stepCallSlaposDeliverStartedAggregatedSalePackingListAlarm()
self.tic()
# stabilise aggregated deliveries and expand them
self.stepCallSlaposManageBuildingCalculatingDeliveryAlarm()
self.tic()
# build aggregated invoices
self.stepCallSlaposTriggerBuildAlarm()
self.tic()
# stabilise aggregated invoices and expand them
self.stepCallSlaposManageBuildingCalculatingDeliveryAlarm()
self.tic()
# update invoices with their tax & discount
self.stepCallSlaposTriggerBuildAlarm()
self.tic()
self.stepCallSlaposManageBuildingCalculatingDeliveryAlarm()
self.tic()
# update invoices with their tax & discount transaction lines
self.stepCallSlaposTriggerBuildAlarm()
self.tic()
self.stepCallSlaposManageBuildingCalculatingDeliveryAlarm()
self.tic()
# stop the invoices and solve them again
self.stepCallSlaposStopConfirmedAggregatedSaleInvoiceTransactionAlarm()
self.tic()
self.stepCallSlaposManageBuildingCalculatingDeliveryAlarm()
self.tic()
builder = self.portal.portal_orders.slapos_payment_transaction_builder
for _ in range(500):
# build the aggregated payment
self.stepCallSlaposTriggerPaymentTransactionOrderBuilderAlarm()
self.tic()
# If there is something unbuild recall alarm.
if len(builder.OrderBuilder_generateUnrelatedInvoiceList()):
break
# start the payzen payment
self.stepCallSlaposPayzenUpdateConfirmedPaymentAlarm()
self.tic()
# stabilise the payment deliveries and expand them
self.stepCallSlaposManageBuildingCalculatingDeliveryAlarm()
self.tic()
# trigger the CRM interaction
self.stepCallSlaposCrmCreateRegularisationRequestAlarm()
self.tic()
for subscription_request in subscription_request_list:
self.checkAndPayFirstTreeMonth(subscription_request)
self.tic()
# After the payment re-call the Alarm in order to confirm the subscription
# Request.
self.stepCallSlaposSubscriptionRequestProcessOrderedAlarm()
self.tic()
for subscription_request in subscription_request_list:
# Check if instance was already allocated, in this case, it shouldn't
# until the allocation alarm kicks in.
self.checkRelatedInstance(subscription_request)
self.checkSubscriptionDeploymentAndSimulation(
default_email_text, subscription_server)
# Now really allocate the instance
self.stepCallSlaposAllocateInstanceAlarm()
self.tic()
class TestSlapOSSubscriptionScenario(TestSlapOSSubscriptionScenarioMixin):
# now instantiate it on computer and set some nice connection dict
self.simulateSlapgridCP(subscription_server)
self.tic()
def test_subscription_scenario(self):
self._test_subscription_scenario(amount=1)
for subscription_request in subscription_request_list:
# Re-check, now it should be allocated.
self.checkAllocationOnRelatedInstance(subscription_request)
def test_subscription_with_3_vms_scenario(self):
self._test_subscription_scenario(amount=3)
def test_two_subscription_scenario(self):
self._test_two_subscription_scenario(amount=1)
\ No newline at end of file
......@@ -117,6 +117,7 @@
</item>
</dictionary>
</list>
<none/>
</tuple>
</pickle>
</record>
......
......@@ -4,3 +4,4 @@ test.erp5.testSlapOSTrialScenario
test.erp5.testSlapOSSubscriptionAlarm
test.erp5.testSlapOSSubscriptionSkins
test.erp5.testSlapOSSubscriptionScenario
test.erp5.testSlapOSSubscriptionChineseScenario
\ 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