Commit 18c60ac7 authored by Rafael Monnerat's avatar Rafael Monnerat

slapos_subscription_request: Partially refactor and extend scenario tests

Keep source[_section] None on the New template tests

   This feature (with values set) is covered on Dual organisation settings.
   Unify code into mixin for Chinese scenarios
   Update Payzen for the new API
   Remove duplication and fix up assert
   Implement Test for Parallel subscriptions

      Test 2 organisations rather them a single one with parallel Subscriptions
      Move more code into the Mixin, for reuse it.
parent e015be48
Pipeline #12994 failed with stage
in 0 seconds
...@@ -12,6 +12,7 @@ ...@@ -12,6 +12,7 @@
<item>aggregate</item> <item>aggregate</item>
<item>causality</item> <item>causality</item>
<item>destination_section</item> <item>destination_section</item>
<item>source</item>
<item>source_section</item> <item>source_section</item>
<item>specialise</item> <item>specialise</item>
</portal_type> </portal_type>
......
...@@ -21,206 +21,23 @@ ...@@ -21,206 +21,23 @@
from erp5.component.test.testSlapOSSubscriptionScenario import TestSlapOSSubscriptionScenarioMixin from erp5.component.test.testSlapOSSubscriptionScenario import TestSlapOSSubscriptionScenarioMixin
from erp5.component.test.SlapOSTestCaseMixin import changeSkin from erp5.component.test.SlapOSTestCaseMixin import changeSkin
from Products.ERP5Type.tests.utils import createZODBPythonScript
class TestSlapOSSubscriptionChineseScenarioMixin(TestSlapOSSubscriptionScenarioMixin): class TestSlapOSSubscriptionChineseScenarioMixin(TestSlapOSSubscriptionScenarioMixin):
def afterSetUp(self): def afterSetUp(self):
TestSlapOSSubscriptionScenarioMixin.afterSetUp(self) TestSlapOSSubscriptionScenarioMixin.afterSetUp(self)
self.expected_individual_price_without_tax = 1573.3333333333335
self.expected_individual_price_with_tax = 1888.00
self.expected_reservation_fee = 188.00
self.expected_reservation_fee_without_tax = 188
self.expected_reservation_quantity_tax = 0
self.expected_reservation_tax = 0
self.expected_price_currency = "currency_module/CNY" self.expected_price_currency = "currency_module/CNY"
self.normal_user = None self.normal_user = None
self.expected_notification_language = "zh" self.expected_notification_language = "zh"
self.login() self.login()
self.createNotificationMessage("subscription_request-confirmation-with-password", language="zh",
text_content='CHINESE! ${name} ${login_name} ${login_password}')
self.createNotificationMessage("subscription_request-confirmation-without-password", language="zh",
text_content='CHINESE! ${name} ${login_name}')
self.createNotificationMessage("subscription_request-instance-is-ready", language="zh",
text_content='CHINESE! ${name} ${subscription_title} ${hosting_subscription_relative_url}')
self.createNotificationMessage("subscription_request-payment-is-ready", language="zh",
text_content='CHINESE! ${name} ${subscription_title} ${payment_relative_relative_url}')
def _simulatePaymentTransaction_getVADSUrlDict(self):
script_name = 'PaymentTransaction_getVADSUrlDict'
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'
"""payment_transaction_url = context.getRelativeUrl()
return dict(vads_url_already_registered="%s/already_registered" % (payment_transaction_url),
vads_url_cancel="%s/cancel" % (payment_transaction_url),
vads_url_error="%s/error" % (payment_transaction_url),
vads_url_referral="%s/referral" % (payment_transaction_url),
vads_url_refused="%s/refused" % (payment_transaction_url),
vads_url_success="%s/success" % (payment_transaction_url),
vads_url_return="%s/return" % (payment_transaction_url),
)""")
def _dropPaymentTransaction_getVADSUrlDict(self):
script_name = 'PaymentTransaction_getVADSUrlDict'
if script_name in self.portal.portal_skins.custom.objectIds():
self.portal.portal_skins.custom.manage_delObjects(script_name)
@changeSkin('Hal') @changeSkin('Hal')
def _requestSubscription(self, **kw): def _requestSubscription(self, **kw):
if self.cloud_invitation_token is not None: return self._requestSubscriptionViaChineseWebsite(**kw)
kw["token"] = self.cloud_invitation_token.getId()
if 'target_language' not in kw:
kw["target_language"] = "zh"
kw["subscription_reference"] = self.subscription_condition.getReference().replace("_zh", "")
original_mode = self.portal.portal_secure_payments.slapos_wechat_test.getWechatMode()
self._simulatePaymentTransaction_getVADSUrlDict()
try:
self.portal.portal_secure_payments.slapos_wechat_test.setWechatMode("UNITTEST")
self.logout()
self.changeSkin('Hal')
return self.web_site.hateoas.SubscriptionRequestModule_requestSubscription(**kw)
finally:
self._dropPaymentTransaction_getVADSUrlDict()
self.portal.portal_secure_payments.slapos_wechat_test.setWechatMode(original_mode)
def createSubscriptionCondition(self, slave=False): def createSubscriptionCondition(self, slave=False):
self.subscription_condition = self.portal.subscription_condition_module.newContent( self.subscription_condition = self.createChineseSubscriptionCondition(
portal_type="Subscription Condition", slave=slave)
title="TestSubscriptionChineseScenario",
short_tile="Test Your Chinese Scenario",
description="This is a Chinese test",
url_string=self.generateNewSoftwareReleaseUrl(),
root_slave=slave,
price=1888.00,
price_currency="currency_module/CNY",
default_source_reference="default",
reference="rapidvm%s_zh" % 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={},
source=self.expected_source,
source_section=self.expected_source_section
)
self.subscription_condition.validate()
self.subscription_condition.updateLocalRolesOnSecurityGroups()
self.tic()
def _payPayment(self, subscription_request):
quantity = subscription_request.getQuantity()
# Check Payment
payment = self._getRelatedPaymentValue(subscription_request)
self.assertEqual(self.expected_price_currency, payment.getPriceCurrency())
self.assertEqual(-self.expected_reservation_fee*quantity,
payment.PaymentTransaction_getTotalPayablePrice())
self.assertEqual(payment.getSimulationState(), "started")
# Pay 188 CNY per VM
data_kw = {
'result_code': 'SUCCESS',
'trade_state': 'SUCCESS',
'total_fee': self.expected_reservation_fee*100*quantity,
'fee_type': 'CNY',
}
# Wechat_processUpdate will mark payment as payed by stopping it.
payment.PaymentTransaction_createWechatEvent().WechatEvent_processUpdate(data_kw)
return payment
def checkAndPayFirstMonth(self, subscription_request):
self.login()
original_mode = self.portal.portal_secure_payments.slapos_wechat_test.getWechatMode()
self._simulatePaymentTransaction_getVADSUrlDict()
try:
person = subscription_request.getDestinationSectionValue()
quantity = subscription_request.getQuantity()
self.portal.portal_secure_payments.slapos_wechat_test.setWechatMode("UNITTEST")
self.login(person.getUserId())
self.useWechatManually(self.web_site, person.getUserId(), is_email_expected=False)
payment = self.portal.portal_catalog.getResultValue(
portal_type="Payment Transaction",
simulation_state="started")
self.assertEqual(payment.getSourceSection(), self.expected_source_section)
self.assertEqual(payment.getSourcePayment(), "%s/bank_account" % self.expected_source_section)
authAmount = (int(self.expected_individual_price_with_tax*100)*1-int(self.expected_reservation_fee*100))*quantity
self.assertEqual(int(payment.PaymentTransaction_getTotalPayablePrice()*100),
-authAmount)
self.assertEqual(payment.getPriceCurrency(), self.expected_price_currency)
self.logout()
self.login()
data_kw = {
'result_code': 'SUCCESS',
'trade_state': 'SUCCESS',
'total_fee': authAmount,
'fee_type': 'CNY',
}
# Wechat_processUpdate will mark payment as payed by stopping it.
payment.PaymentTransaction_createWechatEvent().WechatEvent_processUpdate(data_kw)
finally:
self._dropPaymentTransaction_getVADSUrlDict()
self.portal.portal_secure_payments.slapos_wechat_test.setWechatMode(original_mode)
def checkAndPaySecondMonth(self, subscription_request):
self.login()
original_mode = self.portal.portal_secure_payments.slapos_wechat_test.getWechatMode()
self._simulatePaymentTransaction_getVADSUrlDict()
try:
person = subscription_request.getDestinationSectionValue()
quantity = subscription_request.getQuantity()
self.portal.portal_secure_payments.slapos_wechat_test.setWechatMode("UNITTEST")
self.login(person.getUserId())
self.useWechatManually(self.web_site, person.getUserId())
payment = self.portal.portal_catalog.getResultValue(
portal_type="Payment Transaction",
simulation_state="started")
authAmount = int(self.expected_individual_price_with_tax*100)*quantity
self.assertEqual(int(payment.PaymentTransaction_getTotalPayablePrice()*100),
-authAmount)
self.assertEqual(payment.getPriceCurrency(), self.expected_price_currency)
self.logout()
self.login()
data_kw = {
'result_code': 'SUCCESS',
'trade_state': 'SUCCESS',
'total_fee': authAmount,
'fee_type': 'CNY',
}
# Wechat_processUpdate will mark payment as payed by stopping it.
payment.PaymentTransaction_createWechatEvent().WechatEvent_processUpdate(data_kw)
finally:
self._dropPaymentTransaction_getVADSUrlDict()
self.portal.portal_secure_payments.slapos_wechat_test.setWechatMode(original_mode)
class TestSlapOSSubscriptionChineseScenario(TestSlapOSSubscriptionChineseScenarioMixin): class TestSlapOSSubscriptionChineseScenario(TestSlapOSSubscriptionChineseScenarioMixin):
......
...@@ -29,13 +29,22 @@ class testSlapOSSubscriptionDualOrganisationScenario(TestSlapOSSubscriptionScena ...@@ -29,13 +29,22 @@ class testSlapOSSubscriptionDualOrganisationScenario(TestSlapOSSubscriptionScena
self.expected_source = fr_organisation.getRelativeUrl() self.expected_source = fr_organisation.getRelativeUrl()
self.expected_source_section = fr_organisation.getRelativeUrl() self.expected_source_section = fr_organisation.getRelativeUrl()
self.expected_zh_reservation_fee = 188.00
self.subscription_condition.edit( self.subscription_condition.edit(
source=self.expected_source, source=self.expected_source,
source_section=self.expected_source_section source_section=self.expected_source_section
) )
self.expected_case2_source = zh_organisation.getRelativeUrl() self.subscription_condition_zh = self.createChineseSubscriptionCondition()
self.expected_case2_source_section = zh_organisation.getRelativeUrl() self.expected_zh_source = zh_organisation.getRelativeUrl()
self.expected_zh_source_section = zh_organisation.getRelativeUrl()
self.subscription_condition_zh.edit(
source=self.expected_zh_source,
source_section=self.expected_zh_source_section
)
self.portal.portal_caches.clearAllCache() self.portal.portal_caches.clearAllCache()
self.tic() self.tic()
...@@ -64,3 +73,194 @@ class testSlapOSSubscriptionDualOrganisationScenario(TestSlapOSSubscriptionScena ...@@ -64,3 +73,194 @@ class testSlapOSSubscriptionDualOrganisationScenario(TestSlapOSSubscriptionScena
# Messages are in chinese, when subscribed via chinese website. Even if the english language is # Messages are in chinese, when subscribed via chinese website. Even if the english language is
# english # english
self._test_subscription_scenario_with_existing_user(amount=1, language="en") self._test_subscription_scenario_with_existing_user(amount=1, language="en")
def requestAndCheckDualHostingSubscription(self, amount, name,
default_email_text, language_list=['en', 'en']):
self.logout()
request_kw = dict(
subscription_reference=self.subscription_condition.getReference(),
amount=amount,
name=name,
default_email_text=default_email_text,
confirmation_required=False,
REQUEST=self.portal.REQUEST)
all_subscription_requested_list = []
for language in language_list:
if language == "zh":
self._requestSubscriptionViaChineseWebsite(**request_kw)
subscription_condition = self.subscription_condition_zh
expected_price_currency = "currency_module/CNY"
expected_source_section = self.expected_zh_source_section
else:
self._requestSubscription(**request_kw)
subscription_condition = self.subscription_condition
expected_price_currency = "currency_module/EUR"
expected_source_section = self.expected_source_section
self.login()
# I'm not sure if this is realistic
self.tic()
subscription_request_list = self.getSubscriptionRequestList(
default_email_text, subscription_condition)
for subscription_request in subscription_request_list:
self.assertEqual(language,
subscription_request.getLanguage())
self.assertEqual(expected_price_currency,
subscription_request.getPriceCurrency())
self.assertEqual(expected_source_section,
subscription_request.getSourceSection())
self.checkDraftSubscriptionRequest(subscription_request,
default_email_text,
subscription_request.getSpecialiseValue(),
amount=amount)
self.tic()
all_subscription_requested_list.extend(subscription_request_list)
self.checkAndPaySubscriptionPayment(all_subscription_requested_list)
self.tic()
# Call alarm to check payment and invoice and move foward to planned.
self.stepCallSlaposSubscriptionRequestProcessDraftAlarm()
self.tic()
for subscription_request in all_subscription_requested_list:
self.checkPlannedSubscriptionRequest(subscription_request,
default_email_text,
subscription_request.getSpecialiseValue())
# 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()
for subscription_request in all_subscription_requested_list:
self.checkOrderedSubscriptionRequest(subscription_request,
default_email_text,
subscription_request.getSpecialiseValue())
# Call alarm to make the request of the instance?
self.stepCallSlaposSubscriptionRequestProcessOrderedAlarm()
self.tic()
# The alarms might be called multiple times for move each step
self.stepCallSlaposSubscriptionRequestProcessOrderedAlarm()
self.tic()
for subscription_request in all_subscription_requested_list:
sale_packing_list_list = self.portal.portal_catalog(
causality_uid = subscription_request.getUid(),
title="Reservation Deduction",
portal_type="Sale Packing List"
)
self.assertEqual(len(sale_packing_list_list), 1)
sale_packing_list = sale_packing_list_list[0]
self.assertEqual(sale_packing_list.getPriceCurrency(),
subscription_request.getPriceCurrency())
self.assertEqual(sale_packing_list.getSpecialise(),
"sale_trade_condition_module/slapos_reservation_refund_trade_condition")
if subscription_request.getPriceCurrency() == "currency_module/CNY":
expected_reservation_fee = self.expected_zh_reservation_fee
else:
expected_reservation_fee = self.expected_reservation_fee
self.assertEqual(round(sale_packing_list.getTotalPrice(), 2),
-round(expected_reservation_fee*amount, 2))
return all_subscription_requested_list
def test_subscription_scenario_with_dual_organisation(self):
amount = 1
language = "en"
# Call as anonymous... check response?
default_email_text = "abc%s@nexedi.com" % self.new_id
name="ABC %s" % self.new_id
self.login()
self.createNormalUser(default_email_text, name, language)
self.tic()
self.subscription_server = self.createPublicServerForAdminUser()
self.login()
# Extra software from zh version
subscription_server_software = self.subscription_condition_zh.getUrlString()
self.supplySoftware(self.subscription_server, subscription_server_software)
self.tic()
self.logout()
subscription_request_list = self.requestAndCheckDualHostingSubscription(
amount, name, default_email_text, language_list=["en", "zh"])
self._checkSubscriptionDeploymentAndSimulation(
subscription_request_list, default_email_text,
self.subscription_server)
if not self.expected_free_reservation:
for subscription_request in subscription_request_list:
if subscription_request.getPriceCurrency() == "currency_module/CNY":
self.checkAndPayFirstMonthViaWechat(subscription_request)
else:
self.checkAndPayFirstMonth(subscription_request)
self.tic()
self.stepCallSlaposSubscriptionRequestProcessConfirmedAlarm()
self.tic()
for subscription_request in subscription_request_list:
self.assertTrue(
subscription_request.SubscriptionRequest_testPaymentBalance())
self.assertEqual('start_requested',
subscription_request.getAggregateValue().getSlapState())
# On the second loop that email is send and state is moved to started
self.stepCallSlaposSubscriptionRequestProcessConfirmedAlarm()
self.tic()
for subscription_request in subscription_request_list:
self.assertTrue(
subscription_request.SubscriptionRequest_testPaymentBalance())
self.assertEqual('start_requested',
subscription_request.getAggregateValue().getSlapState())
self.checkStartedSubscriptionRequest(subscription_request,
default_email_text,
subscription_request.getSpecialiseValue())
for subscription_request in subscription_request_list:
self.assertEqual(self.normal_user,
subscription_request.getDestinationSectionValue())
if self.skip_destroy_and_check:
return
for subscription_request in subscription_request_list:
self.assertEqual('start_requested',
subscription_request.getAggregateValue().getSlapState())
# Destroy all instances and process
hosting_subscription = subscription_request.getAggregateValue()
hosting_subscription.HostingSubscription_requestPerson('destroyed')
self.tic()
self.stepCallSlaposSubscriptionRequestProcessStartedAlarm()
self.tic()
self.checkStoppedSubscriptionRequest(subscription_request,
default_email_text,
subscription_request.getSpecialiseValue())
return default_email_text, name
\ No newline at end of file
...@@ -51,7 +51,9 @@ ...@@ -51,7 +51,9 @@
<item> <item>
<key> <string>text_content_warning_message</string> </key> <key> <string>text_content_warning_message</string> </key>
<value> <value>
<tuple/> <tuple>
<string>W: 78, 2: Dangerous default value [] as argument (dangerous-default-value)</string>
</tuple>
</value> </value>
</item> </item>
<item> <item>
......
...@@ -51,6 +51,10 @@ class testSlapOSSubscriptionCloudInvitationTokenScenario(TestSlapOSSubscriptionC ...@@ -51,6 +51,10 @@ class testSlapOSSubscriptionCloudInvitationTokenScenario(TestSlapOSSubscriptionC
self.expected_reservation_quantity_tax = 0.0 self.expected_reservation_quantity_tax = 0.0
self.expected_reservation_tax = 0.0 self.expected_reservation_tax = 0.0
self.expected_free_reservation = 1 self.expected_free_reservation = 1
self.expected_zh_reservation_fee = 0.0
self.expected_zh_reservation_quantity_tax = 0.0
self.expected_zh_reservation_tax = 0.0
self.expected_zh_free_reservation = 1
self.cloud_invitation_token = self.makeCloudInvitationToken( self.cloud_invitation_token = self.makeCloudInvitationToken(
max_invoice_delay=99, max_invoice_delay=99,
...@@ -74,6 +78,11 @@ class testSlapOSSubscriptionCloudInvitationTokenScenario(TestSlapOSSubscriptionC ...@@ -74,6 +78,11 @@ class testSlapOSSubscriptionCloudInvitationTokenScenario(TestSlapOSSubscriptionC
self.expected_reservation_quantity_tax = 0.0 self.expected_reservation_quantity_tax = 0.0
self.expected_reservation_tax = 0.0 self.expected_reservation_tax = 0.0
self.expected_free_reservation = 1 self.expected_free_reservation = 1
self.expected_zh_reservation_fee = 0.0
self.expected_zh_reservation_quantity_tax = 0.0
self.expected_zh_reservation_tax = 0.0
self.expected_zh_free_reservation = 1
self._test_two_subscription_scenario(amount=1, self._test_two_subscription_scenario(amount=1,
create_invitation=True, create_invitation=True,
max_invoice_delay=99, max_invoice_delay=99,
......
...@@ -29,6 +29,14 @@ class testSlapOSSubscriptionNewTemplateChineseScenario(TestSlapOSSubscriptionChi ...@@ -29,6 +29,14 @@ class testSlapOSSubscriptionNewTemplateChineseScenario(TestSlapOSSubscriptionChi
self.expected_source = organisation.getRelativeUrl() self.expected_source = organisation.getRelativeUrl()
self.expected_source_section = organisation.getRelativeUrl() self.expected_source_section = organisation.getRelativeUrl()
# Set those values (source and source section) are only meaninfull if
# the templates on preferences differ from Chinese and European organisations.
self.subscription_condition.edit(
source=None,
source_section=None
)
self.portal.portal_caches.clearAllCache() self.portal.portal_caches.clearAllCache()
self.tic() self.tic()
......
...@@ -29,6 +29,14 @@ class testSlapOSSubscriptionNewTemplateScenario(TestSlapOSSubscriptionScenarioMi ...@@ -29,6 +29,14 @@ class testSlapOSSubscriptionNewTemplateScenario(TestSlapOSSubscriptionScenarioMi
self.expected_source = organisation.getRelativeUrl() self.expected_source = organisation.getRelativeUrl()
self.expected_source_section = organisation.getRelativeUrl() self.expected_source_section = organisation.getRelativeUrl()
# Set those values (source and source section) are only meaninfull if
# the templates on preferences differ from Chinese and European organisations.
self.subscription_condition.edit(
source=None,
source_section=None
)
self.portal.portal_caches.clearAllCache() self.portal.portal_caches.clearAllCache()
self.tic() self.tic()
......
...@@ -2,12 +2,12 @@ Subscription Condition Module | business_application ...@@ -2,12 +2,12 @@ Subscription Condition Module | business_application
Subscription Condition | aggregate Subscription Condition | aggregate
Subscription Condition | follow_up Subscription Condition | follow_up
Subscription Condition | source Subscription Condition | source
Subscription Condition | source
Subscription Condition | source_section Subscription Condition | source_section
Subscription Request Module | business_application Subscription Request Module | business_application
Subscription Request | aggregate Subscription Request | aggregate
Subscription Request | causality Subscription Request | causality
Subscription Request | destination_section Subscription Request | destination_section
Subscription Request | source
Subscription Request | source_section Subscription Request | source_section
Subscription Request | specialise Subscription Request | specialise
Trial Condition Module | business_application Trial Condition Module | business_application
......
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