Skip to content

S
slapos.core
Commit a490a99a authored by Łukasz Nowak's avatar Łukasz Nowak
Browse files
Options

Drop VifibMixin

parent 356b842a
Hide whitespace changes
Inline Side-by-side
Showing with 195 additions and 444 deletions
master/product/SlapOS/tests/testSlapOSMixin.py
View file @ a490a99a
......@@ -29,9 +29,14 @@
import random
import transaction
import unittest
import Products.Vifib.tests.VifibMixin
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
import functools
from Products.ERP5Type.tests.utils import DummyMailHost
from Products.ERP5Type.Utils import convertToUpperCase
import os
from AccessControl.SecurityManagement import getSecurityManager, \
setSecurityManager
def withAbort(func):
@functools.wraps(func)
......@@ -42,7 +47,67 @@ def withAbort(func):
transaction.abort()
return wrapped
class testSlapOSMixin(Products.Vifib.tests.VifibMixin.testVifibMixin):
class testSlapOSMixin(ERP5TypeTestCase):
def clearCache(self):
self.portal.portal_caches.clearAllCache()
self.portal.portal_workflow.refreshWorklistCache()
def getDefaultSitePreferenceId(self):
"""Default id, usefull method to override
"""
return "slapos_default_system_preference"
def setUpMemcached(self):
from Products.ERP5Type.tests.ERP5TypeTestCase import\
_getVolatileMemcachedServerDict, _getPersistentMemcachedServerDict
memcached_tool = self.getPortal().portal_memcached
# setup default volatile distributed memcached
connection_dict = _getVolatileMemcachedServerDict()
url_string = '%(hostname)s:%(port)s' % connection_dict
memcached_tool.default_memcached_plugin.setUrlString(url_string)
# setup default persistent distributed memcached
connection_dict = _getPersistentMemcachedServerDict()
url_string = '%(hostname)s:%(port)s' % connection_dict
memcached_tool.persistent_memcached_plugin.setUrlString(url_string)
def createAlarmStep(self):
def makeCallAlarm(alarm):
def callAlarm(*args, **kwargs):
sm = getSecurityManager()
self.login()
try:
alarm.activeSense(params=kwargs)
transaction.commit()
finally:
setSecurityManager(sm)
return callAlarm
for alarm in self.portal.portal_alarms.contentValues():
if alarm.isEnabled():
setattr(self, 'stepCall' + convertToUpperCase(alarm.getId()) \
+ 'Alarm', makeCallAlarm(alarm))
def setupPortalCertificateAuthority(self):
"""Sets up portal_certificate_authority"""
if not self.portal.hasObject('portal_certificate_authority'):
self.portal.manage_addProduct['ERP5'].manage_addTool(
'ERP5 Certificate Authority Tool', None)
self.portal.portal_certificate_authority.certificate_authority_path = \
os.environ['TEST_CA_PATH']
transaction.commit()
# reset test CA to have it always count from 0
open(os.path.join(os.environ['TEST_CA_PATH'], 'serial'), 'w').write('01')
open(os.path.join(os.environ['TEST_CA_PATH'], 'crlnumber'), 'w').write(
'01')
open(os.path.join(os.environ['TEST_CA_PATH'], 'index.txt'), 'w').write('')
def setupPortalAlarms(self):
if not self.portal.portal_alarms.isSubscribed():
self.portal.portal_alarms.subscribe()
self.assertTrue(self.portal.portal_alarms.isSubscribed())
def isLiveTest(self):
return 'ERP5TypeLiveTestCase' in [q.__name__ for q in self.__class__.mro()]
def _setUpDummyMailHost(self):
"""Do not play with NON persistent replacement of MailHost"""
......@@ -59,6 +124,12 @@ class testSlapOSMixin(Products.Vifib.tests.VifibMixin.testVifibMixin):
self.deSetUpPersistentDummyMailHost()
return
def getUserFolder(self):
"""
Return the user folder
"""
return getattr(self.getPortal(), 'acl_users', None)
def afterSetUp(self):
self.login()
self.createAlarmStep()
......
master/product/Vifib/tests/VifibMixin.py deleted 100644 → 0
View file @ 356b842a
# -*- coding: utf-8 -*-
##############################################################################
#
# Copyright (c) 2010 Nexedi SA and Contributors. All Rights Reserved.
# Romain Courteaud <romain@nexedi.com>
#
# WARNING: This program as such is intended to be used by professional
# programmers who take the whole responsibility of assessing all potential
# consequences resulting from its eventual inadequacies and bugs
# End users who are looking for a ready-to-use solution with commercial
# guarantees and support are strongly advised to contract a Free Software
# Service Company
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
#
##############################################################################
import transaction
from Products.ERP5Type.tests.ERP5TypeTestCase import ERP5TypeTestCase
from AccessControl.SecurityManagement import newSecurityManager, \
getSecurityManager, setSecurityManager
from Products.ERP5Type.tests.utils import DummyMailHost
import os
from Products.ERP5Type.Utils import convertToUpperCase
class testVifibMixin(ERP5TypeTestCase):
"""
Mixin class for unit test of Vifib.
"""
run_all_test = 1
def getBusinessTemplateList(self):
"""
Install the business templates.
"""
result = [
'erp5_core_proxy_field_legacy',
'erp5_full_text_myisam_catalog',
'erp5_ingestion_mysql_innodb_catalog',
'erp5_base',
'erp5_jquery',
'erp5_ingestion',
'erp5_web',
'erp5_simulation',
'erp5_pdm',
'erp5_dms',
'erp5_crm',
'erp5_content_translation',
'erp5_trade',
'erp5_accounting',
'erp5_jquery_ui',
'erp5_xhtml_jquery_style',
'erp5_software_pdm',
'erp5_computer_immobilisation',
'erp5_item',
'erp5_invoicing',
'erp5_secure_payment',
'erp5_knowledge_pad',
'erp5_credential',
'erp5_tax_resource',
'erp5_web_download_theme',
'slapos_cloud',
'erp5_open_trade',
'erp5_tiosafe_core',
'erp5_system_event',
'erp5_simplified_invoicing',
'erp5_rss_style',
'erp5_project',
'erp5_payzen_secure_payment',
'erp5_ooo_import',
'erp5_odt_style',
'erp5_ods_style',
'erp5_l10n_fr',
'erp5_km',
'erp5_forge',
'erp5_discount_resource',
'erp5_dhtml_style',
'erp5_credential_oauth2',
'erp5_commerce',
'erp5_bearer_token',
'erp5_administration',
'erp5_accounting_l10n_fr',
'erp5_web_shacache',
'erp5_data_set',
'slapos_accounting',
'vifib_base',
'erp5_workflow',
'erp5_web_shadir',
'vifib_slap',
'slapos_rest_api_tool_portal_type',
'erp5_configurator',
'slapos_cache',
'slapos_slap_tool',
'slapos_category',
'slapos_payzen',
'slapos_pdm',
'erp5_upgrader',
'slapos_rest_api',
'vifib_data',
'slapos_configurator',
'erp5_ui_test_core',
'slapos_erp5',
'slapos_web',
'vifib_upgrader',
'vifib_slapos_accounting',
'vifib_configurator',
'erp5_ui_test',
# All business templates above are the dependency
# tree of vifib_erp5, so if you mofify this list
# make sure you also update dependency of bt.
# You can use portal_templates.resolveBusinessTemplateListDependency(["vifib_erp5"])
# in order to check this list.
'vifib_erp5',
'vifib_test',
'vifib_web_ui_test',
]
return result
def getUserFolder(self):
"""
Return the user folder
"""
return getattr(self.getPortal(), 'acl_users', None)
def loginDefaultUser(self, quiet=0):
"""
Most of the time, we need to login before doing anything
"""
uf = self.getUserFolder()
uf._doAddUser('default_user', 'default_user',
['Assignee', 'Assignor',
'Associate', 'Auditor', 'Author',
], [])
user = uf.getUserById('default_user').__of__(uf)
newSecurityManager(None, user)
def isLiveTest(self):
return 'ERP5TypeLiveTestCase' in [q.__name__ for q in self.__class__.mro()]
def setupPayZenInterface(self):
payzen = self.portal.portal_secure_payments.vifib_payzen
# avoid resetting prepared site
if payzen.getServiceUsername() is None:
payzen.setServiceUsername('12345')
if payzen.getServicePassword() == '':
payzen.setServicePassword('09876')
def setupPortalAlarms(self):
if not self.portal.portal_alarms.isSubscribed():
self.portal.portal_alarms.subscribe()
self.assertTrue(self.portal.portal_alarms.isSubscribed())
def setupPortalCertificateAuthority(self):
"""Sets up portal_certificate_authority"""
if not self.portal.hasObject('portal_certificate_authority'):
self.portal.manage_addProduct['ERP5'].manage_addTool(
'ERP5 Certificate Authority Tool', None)
self.portal.portal_certificate_authority.certificate_authority_path = \
os.environ['TEST_CA_PATH']
transaction.commit()
# reset test CA to have it always count from 0
open(os.path.join(os.environ['TEST_CA_PATH'], 'serial'), 'w').write('01')
open(os.path.join(os.environ['TEST_CA_PATH'], 'crlnumber'), 'w').write(
'01')
open(os.path.join(os.environ['TEST_CA_PATH'], 'index.txt'), 'w').write('')
def createAlarmStep(self):
def makeCallAlarm(alarm):
def callAlarm(*args, **kwargs):
sm = getSecurityManager()
self.login()
try:
alarm.activeSense(params=kwargs)
transaction.commit()
finally:
setSecurityManager(sm)
return callAlarm
for alarm in self.portal.portal_alarms.contentValues():
if alarm.isEnabled():
setattr(self, 'stepCall' + convertToUpperCase(alarm.getId()) \
+ 'Alarm', makeCallAlarm(alarm))
def setUpMemcached(self):
from Products.ERP5Type.tests.ERP5TypeTestCase import\
_getVolatileMemcachedServerDict, _getPersistentMemcachedServerDict
memcached_tool = self.getPortal().portal_memcached
# setup default volatile distributed memcached
connection_dict = _getVolatileMemcachedServerDict()
url_string = '%(hostname)s:%(port)s' % connection_dict
memcached_tool.default_memcached_plugin.setUrlString(url_string)
# setup default persistent distributed memcached
connection_dict = _getPersistentMemcachedServerDict()
url_string = '%(hostname)s:%(port)s' % connection_dict
memcached_tool.persistent_memcached_plugin.setUrlString(url_string)
def afterSetUp(self, quiet=1, run=run_all_test):
"""
Create ERP5 user.
This has to be called only once.
"""
# setup new active process for this test, in order have
# consistency report local for one test
sm = getSecurityManager()
self.login()
try:
self.portal.portal_alarms.vifib_check_consistency.newActiveProcess()
finally:
setSecurityManager(sm)
import random
self.portal.portal_caches.erp5_site_global_id = '%s' % random.random()
self.portal.portal_caches._p_changed = 1
transaction.commit()
self.portal.portal_caches.updateCache()
self.createAlarmStep()
if getattr(self.portal, 'set_up_once_called', 0):
return
else:
self.portal.set_up_once_called = 1
self.bootstrapSite()
self.portal._p_changed = 1
transaction.commit()
self.stabiliseAccounting()
def stabiliseAccounting(self):
self.stepCallSlaposUpdateDeliveryCausalityStateAlarm()
self.tic()
self.stepCallVifibSolveAutomaticallyAlarm()
self.tic()
self.stepCallVifibExpandDeliveryLineAlarm()
self.tic()
self.stepCallSlaposTriggerBuildAlarm()
self.tic()
self.stepCallSlaposUpdateDeliveryCausalityStateAlarm()
self.tic()
self.stepCallVifibSolveAutomaticallyAlarm()
self.tic()
self.stepCallVifibExpandDeliveryLineAlarm()
self.tic()
self.stepCallSlaposTriggerBuildAlarm()
self.tic()
self.stepCallSlaposUpdateDeliveryCausalityStateAlarm()
self.tic()
self.stepCallVifibSolveAutomaticallyAlarm()
self.tic()
self.stepCallStopConfirmedSaleInvoiceTransactionAlarm()
self.tic()
self.stepCallVifibExpandDeliveryLineAlarm()
self.tic()
self.stepCallSlaposTriggerBuildAlarm()
self.tic()
self.stepCallSlaposUpdateDeliveryCausalityStateAlarm()
self.tic()
self.stepCallVifibSolveAutomaticallyAlarm()
self.tic()
def getDefaultSitePreferenceId(self):
"""Default id, usefull method to override
"""
return "slapos_default_system_preference"
def prepareTestUsers(self):
"""
Prepare test users.
"""
isTransitionPossible = self.portal.portal_workflow.isTransitionPossible
for person in self.portal.portal_catalog(
portal_type="Person",
id="test_%",
):
person = person.getObject()
if isTransitionPossible(person, 'validate'):
person.validate()
for assignment in person.contentValues(portal_type='Assignment'):
if isTransitionPossible(assignment, 'open'):
assignment.open()
def markManualCreation(self, document):
self.portal.portal_workflow.doActionFor(document, 'edit_action',
comment='Manually created by test.')
def bootstrapSite(self):
"""
Manager has to create an administrator user first.
"""
if self.isLiveTest():
# nothing to do in Live Test
return
self.setupPortalAlarms()
self.setupPortalCertificateAuthority()
self.setupPayZenInterface()
self.setUpMemcached()
portal = self.getPortal()
if 'MailHost' in portal.objectIds():
portal.manage_delObjects(['MailHost'])
portal._setObject('MailHost', DummyMailHost('MailHost'))
portal.email_from_address = 'romain@nexedi.com'
portal.email_to_address = 'romain@nexedi.com'
self.clearCache()
self.logMessage("Bootstrap Vifib Without Security...")
self.login()
# Invoke Post-configurator script, this invokes all
# alarms related to configuration.
self.portal.BusinessConfiguration_invokePromiseAlarmList()
self.prepareTestUsers()
transaction.commit()
self.tic()
self.logout()
self.loginDefaultUser()
def clearCache(self):
self.portal.portal_caches.clearAllCache()
self.portal.portal_workflow.refreshWorklistCache()
# access related steps
def stepLoginDefaultUser(self, **kw):
self.login('default_user')
def stepLoginTestHrAdmin(self, **kw):
self.login('test_hr_admin')
def stepLoginTestUpdatedVifibUser(self, **kw):
self.login('test_updated_vifib_user')
def stepLoginTestVifibAdmin(self, **kw):
self.login('test_vifib_admin')
def stepLoginTestVifibCustomer(self, **kw):
self.login('test_vifib_customer')
def stepLoginTestVifibCustomerA(self, **kw):
self.login('test_vifib_customer_a')
def stepLoginTestVifibDeveloper(self, **kw):
self.login('test_vifib_developer')
def stepLoginTestVifibMember(self, **kw):
self.login('test_vifib_member')
def stepLoginTestVifibUserAdmin(self, **kw):
self.login('test_vifib_user_admin')
def stepLoginTestVifibUserDeveloper(self, **kw):
self.login('test_vifib_user_developer')
def stepLoginERP5TypeTestCase(self, **kw):
self.login('ERP5TypeTestCase')
def stepLogout(self, **kw):
self.logout()
def checkDivergency(self):
# there shall be no divergency
current_skin = self.app.REQUEST.get('portal_skin', 'View')
try:
# Note: Worklists are cached, so in order to have next correct result
# clear cache
self.clearCache()
self.changeSkin('RSS')
diverged_document_list = self.portal.portal_catalog(
portal_type=self.portal.getPortalDeliveryTypeList(),
causality_state='!= solved'
)
self.assertFalse('to Solve' in self.portal.ERP5Site_viewWorklist(),
'There are unsolved deliveries: %s' % ','.join([
' '.join((q.getTitle(), q.getPath(), q.getCausalityState())) \
for q in diverged_document_list]))
finally:
self.changeSkin(current_skin)
def stepCheckSiteConsistency(self, **kw):
self.portal.portal_alarms.vifib_check_consistency.activeSense()
transaction.commit()
self.tic()
self.assertEqual([], self.portal.portal_alarms.vifib_check_consistency\
.Alarm_getConsistencyCheckReportLineList())
self.assertFalse(self.portal.portal_alarms.vifib_check_consistency.sense())
self.checkDivergency()
def stepCleanTic(self, **kw):
self.tic()
def stepTic(self, **kw):
def activateAlarm():
sm = getSecurityManager()
self.login()
try:
for alarm in self.portal.portal_alarms.contentValues():
if alarm.isEnabled() and (alarm.getId() not in \
('vifib_check_consistency',)):
alarm.activeSense()
finally:
setSecurityManager(sm)
if kw.get('sequence', None) is None:
# in case of using not in sequence commit transaction
transaction.commit()
# trigger activateAlarm before tic
activateAlarm()
transaction.commit()
self.tic()
# retrigger activateAlarm after tic
activateAlarm()
transaction.commit()
# tic after activateAlarm
self.tic()
try:
self.checkDivergency()
except AssertionError:
# try last time to solve deliveries
sm = getSecurityManager()
self.login()
try:
self.portal.portal_alarms.vifib_update_delivery_causality_state\
.activeSense()
transaction.commit()
self.tic()
self.portal.portal_alarms.vifib_solve_automatically.activeSense()
transaction.commit()
self.tic()
finally:
setSecurityManager(sm)
self.checkDivergency()
master/product/Vifib/tests/testVifibSlapWebService.py
View file @ a490a99a
......@@ -109,6 +109,128 @@ class TestVifibSlapWebServiceMixin(testVifibMixin):
assertUserCanAccessDocument =\
AssertPermissionMethod(Permissions.AccessContentsInformation)
def stepTic(self, **kw):
def activateAlarm():
sm = getSecurityManager()
self.login()
try:
for alarm in self.portal.portal_alarms.contentValues():
if alarm.isEnabled() and (alarm.getId() not in \
('vifib_check_consistency',)):
alarm.activeSense()
finally:
setSecurityManager(sm)
if kw.get('sequence', None) is None:
# in case of using not in sequence commit transaction
transaction.commit()
# trigger activateAlarm before tic
activateAlarm()
transaction.commit()
self.tic()
# retrigger activateAlarm after tic
activateAlarm()
transaction.commit()
# tic after activateAlarm
self.tic()
try:
self.checkDivergency()
except AssertionError:
# try last time to solve deliveries
sm = getSecurityManager()
self.login()
try:
self.portal.portal_alarms.vifib_update_delivery_causality_state\
.activeSense()
transaction.commit()
self.tic()
self.portal.portal_alarms.vifib_solve_automatically.activeSense()
transaction.commit()
self.tic()
finally:
setSecurityManager(sm)
self.checkDivergency()
def stepCleanTic(self, **kw):
self.tic()
def stepLogout(self, **kw):
self.logout()
# access related steps
def stepLoginDefaultUser(self, **kw):
self.login('default_user')
def stepLoginTestHrAdmin(self, **kw):
self.login('test_hr_admin')
def stepLoginTestUpdatedVifibUser(self, **kw):
self.login('test_updated_vifib_user')
def stepLoginTestVifibAdmin(self, **kw):
self.login('test_vifib_admin')
def stepLoginTestVifibCustomer(self, **kw):
self.login('test_vifib_customer')
def stepLoginTestVifibCustomerA(self, **kw):
self.login('test_vifib_customer_a')
def stepLoginTestVifibDeveloper(self, **kw):
self.login('test_vifib_developer')
def stepLoginTestVifibMember(self, **kw):
self.login('test_vifib_member')
def stepLoginTestVifibUserAdmin(self, **kw):
self.login('test_vifib_user_admin')
def stepLoginTestVifibUserDeveloper(self, **kw):
self.login('test_vifib_user_developer')
def stepLoginERP5TypeTestCase(self, **kw):
self.login('ERP5TypeTestCase')
def clearCache(self):
self.portal.portal_caches.clearAllCache()
self.portal.portal_workflow.refreshWorklistCache()
def checkDivergency(self):
# there shall be no divergency
current_skin = self.app.REQUEST.get('portal_skin', 'View')
try:
# Note: Worklists are cached, so in order to have next correct result
# clear cache
self.clearCache()
self.changeSkin('RSS')
diverged_document_list = self.portal.portal_catalog(
portal_type=self.portal.getPortalDeliveryTypeList(),
causality_state='!= solved'
)
self.assertFalse('to Solve' in self.portal.ERP5Site_viewWorklist(),
'There are unsolved deliveries: %s' % ','.join([
' '.join((q.getTitle(), q.getPath(), q.getCausalityState())) \
for q in diverged_document_list]))
finally:
self.changeSkin(current_skin)
def stepCheckSiteConsistency(self, **kw):
self.portal.portal_alarms.vifib_check_consistency.activeSense()
transaction.commit()
self.tic()
self.assertEqual([], self.portal.portal_alarms.vifib_check_consistency\
.Alarm_getConsistencyCheckReportLineList())
self.assertFalse(self.portal.portal_alarms.vifib_check_consistency.sense())
self.checkDivergency()
def markManualCreation(self, document):
self.portal.portal_workflow.doActionFor(document, 'edit_action',
comment='Manually created by test.')
def fakeSlapAuth(self):
fakeSlapAuth()
......
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
GitLab Nexedi Edition | About GitLab | About Nexedi | 沪ICP备2021021310号-2 | 沪ICP备2021021310号-7