WIP: initial scalability test.

scalability_test/__init__.py

+import os.path
+import json
+
+ZOPE_USER_FAMILY = "user"
+ZOPE_ACTIVITIES_FAMILIY = "activities"
+PERSON_KEY = "person_per_hour"
+ORDER_KEY = "sale_order_per_hour"
+
+class Wendelin_scalability():
+
+  def getTestList(self):
+    return ['createDataStream']
+
+  def getTestPath(self):
+    return 'example/'
+
+  def getUsersFilePath(self):
+    return 'example/scalabilityUsers'
+
+  def getUserQuantity(self, test_number):
+    """
+    Return number of users for a given test run.
+    """
+    users_per_zope_process = 8 
+    max_tests = 100
+    return [x * users_per_zope_process for x in range(1, max_tests)][test_number]
+
+  def getTestDuration(self, test_number):
+    return 60*10
+
+  def getTestRepetition(self, test_number):
+    return 3
+
+  def getScalabilityTestUrl(self, instance_information_dict):
+    for frontend in instance_information_dict['frontend-url-list']:
+      if frontend[0] == ZOPE_USER_FAMILY:
+        frontend_address = frontend[1]
+        break
+    return "%s/erp5" % (frontend_address)
+
+  def getScalabilityTestMetricUrl(self, instance_information_dict, **kw):
+    frontend_address = self.getScalabilityTestUrl(instance_information_dict)
+    metrics_url = frontend_address.replace("://",
+                    "://%s:%s@" % (instance_information_dict['user'],
+                                   instance_information_dict['password']))
+    return metrics_url + "/ERP5Site_getScalabilityTestMetric"
+
+  def getBootstrapScalabilityTestUrl(self, instance_information_dict, count=0, **kw):
+    frontend_address = self.getScalabilityTestUrl(instance_information_dict)
+    bootstrap_url = frontend_address.replace("://",
+                      "://%s:%s@" % (instance_information_dict['user'],
+                                     instance_information_dict['password']))
+    bootstrap_url += "/ERP5Site_bootstrapScalabilityTest"
+    bootstrap_url += "?user_quantity=%i" % self.getUserQuantity(count)
+    return bootstrap_url
+
+  def getSiteAvailabilityUrl(self, instance_information_dict, **kw):
+    frontend_address = self.getScalabilityTestUrl(instance_information_dict)
+    site_url = frontend_address.replace("://",
+                    "://%s:%s@" % (instance_information_dict['user'],
+                                   instance_information_dict['password']))
+    return site_url + "/ERP5Site_isReady"
+
+  def getScalabilityTestOutput(self, metric_list):
+    """
+    From the list of metrics taken during a test run, select the best metric
+    for the test output by a specific criteria
+    """
+    if not metric_list: return None
+    output_json = json.loads(metric_list[0])
+    for metric in metric_list:
+      metric_json = json.loads(metric)
+      if metric_json[PERSON_KEY] > output_json[PERSON_KEY]:
+        output_json[PERSON_KEY] = metric_json[PERSON_KEY]
+      if metric_json[ORDER_KEY] > output_json[ORDER_KEY]:
+        output_json[ORDER_KEY] = metric_json[ORDER_KEY]
+    return "Person: %s doc/hour; SaleOrder: %s doc/hour;" % (
+            str(output_json[PERSON_KEY]), str(output_json[ORDER_KEY]))

scalability_test/example/createDataStream.py

+# -*- coding: utf-8 -*-
+
+import datetime
+import random
+import time
+import string
+from utils import *
+
+TMIN_SLEEP = 2
+TMAX_SLEEP = 6
+
+PREFIX_TITLE = ""
+MAX_PRODUCT = 5
+
+
+def createDataStream(result, browser):
+  """
+  Create a Data Stream and upload some data.
+  """
+  # Open ERP5 homepage and log in
+  result('Open', browser.open())
+
+  # Log in unless already logged in by a previous test suite
+  result('Login', browser.mainForm.submitLogin(
+                            sleep=(TMIN_SLEEP, TMAX_SLEEP)))
+
+  # Go to sale Order module
+  result('GotoModule',
+      browser.mainForm.submitSelectModule(value='/data_stream_module',
+                                       sleep=(TMIN_SLEEP, TMAX_SLEEP)))
+
+  # Create a newData Stream (XXX: this will create Data Stream Bucket!) 
+  result('Create',
+    browser.mainForm.submitNew(sleep=(TMIN_SLEEP, TMAX_SLEEP)))
+
+  # Check whether it has been successfully created
+  assert browser.getTransitionMessage() == 'Object created.'
+  my_order_sale_url = browser.url.split("?")[0]
+
+  # Fill the title
+  my_title = PREFIX_TITLE + generateString(6)
+  browser.mainForm.getControl(name='field_my_title').value = my_title
+
+  # Set some random informations
+  my_str = generateString(random.randint(1,100))
+  browser.mainForm.getControl(name='field_my_description').value = my_str
+  
+  # XXX: how to upload data?
+
+  # Submit the changes, record the time elapsed in seconds
+  result('Save',
+    browser.mainForm.submitSave(sleep=(TMIN_SLEEP, TMAX_SLEEP)))
+
+  # Check whether the changes have been successfully updated
+  assert browser.getTransitionMessage() == 'Data updated.'
+  sale_order_url = my_order_sale_url
+
+  # Validate the Data Stream
+  #browser.mainForm.submitSelectWorkflow(value='validate_action')
+  #result('Validate',
+  #    browser.mainForm.submitDialogConfirm(sleep=(TMIN_SLEEP, TMAX_SLEEP)))
+  #assert browser.getTransitionMessage() == 'Status changed.'

scalability_test/example/scalabilityUsers.py

+# Specify user login/password used to run the tests.
+# <password> and <user_quantity> will be automatically replaced by testnode for each configuration
+
+user_tuple = tuple([('scalability_user_%i' % x, "<password>") for x in range(0, <user_quantity>)])
\ No newline at end of file

scalability_test/example/utils.py

+import random
+import string
+
+def selectRandomOption(browser, select_name):
+  """
+  Function to select randomly an option
+
+  @param browser: Browser
+  @type browser: Browser
+  @param select_name: Name of the input
+  @type select_name: string
+  """
+   # Get the option values
+  options = browser.mainForm.getControl(name=select_name).options[1:]
+  if len(options) > 0:
+    # Select randomly one value
+    browser.mainForm.getControl(name=select_name).value = [random.choice(options)]
+
+def generateString(size) :
+  """
+  Function to generate a string randomly (a-z)
+
+  @param size: Size of the string
+  @type size: int
+  """
+  new_string = random.choice(string.ascii_uppercase)
+  new_string = new_string + ''.join(random.choice(string.ascii_lowercase) for x in range(size))
+  return new_string
+
+def fillRelatedObjects(browser, result, name, maximum=1, actionName="", TMIN_SLEEP=0, TMAX_SLEEP=0, col_num=0, text="") :
+  """
+  Function to fill randomly related objetcs
+
+  @param browser: browser
+  @type browser: Browser
+  @param result: result
+  @type result:
+  @param name: Name of the input name attribute
+  @type name: string
+  @param maximum: Max number of related objects
+  @type maximum: int
+  @param actionName: Name of the action (used for backtrace and statistic ?)
+  @type actionName: string
+  @param TMIN_SLEEP: Min time to sleep (in second)
+  @type TMIN_SLEEP: int
+  @param TMAX_SLEEP: Max time to sleep (in second)
+  @type TMAX_SLEEP: int
+  @param col_num: The numero of the column to filter
+  @type col_num: int
+  @param text: Text used to filter
+  @type text: string
+  """
+
+  # Go to the section linked by the input
+  result('GoTo '+actionName+' Relations', browser.mainForm.getControl(
+      name=name).click())
+
+  # Check the status
+  assert (( browser.getTransitionMessage() == 'Please select one (or more) object.' )
+           or ( browser.getTransitionMessage() == 'Please select one object.' ))
+
+  # Check if it is possible to choose many objects
+  if (browser.getTransitionMessage() == 'Please select one object.'):
+    assert ( maximum <= 1 )
+
+  # Filter applied the 'col_num' column with text 'text'
+  if col_num != 0 :
+    browser.mainForm.getListboxControl(line_number=2, column_number=col_num).value = text
+    browser.mainForm.submitDialogUpdate()
+
+  # Get the number of lines
+  page_stop_number = browser.etree.xpath('//span[@class="listbox-current-page-stop-number"]')
+  if len(page_stop_number) > 0:
+    num_line = int(page_stop_number[0].text)
+  else:
+    num_line = 0
+  # Choose randomly one or more objects
+  if ( num_line > 0 ) and ( maximum > 0 ):
+    iteration = random.randint(1, maximum)
+    for i in range(0, iteration):
+      line_number = random.randint(1,num_line) + 2
+      # Check the box corresponding to line_number if not already checked
+      if not browser.mainForm.getListboxControl(line_number=line_number, column_number=1).value == 'checked':
+        browser.mainForm.getListboxControl(line_number=line_number, column_number=1).value = 'checked'
+  result('Submit '+actionName+' Relations',
+      browser.mainForm.submit(name='Base_callDialogMethod:method',
+                                        sleep=(TMIN_SLEEP, TMAX_SLEEP)))
+
+  # Check whether the changes have been successfully updated
+  assert browser.getTransitionMessage() == 'Data updated.'
+
+def fillOneRelatedObjectSimplified(browser, name):
+  """
+  Function to fill randomly related objetcs
+
+  @param browser: browser
+  @type browser: Browser
+  @param name: Name of the input name attribute
+  @type name: string
+  """
+
+  # Go to the section linked by the input
+  browser.mainForm.getControl(name=name).click()
+
+  # Check the status
+  assert ( browser.getTransitionMessage() == 'Please select one object.' )
+
+  # Get the number of line
+  page_stop_number = browser.etree.xpath('//span[@class="listbox-current-page-stop-number"]')
+  if len(page_stop_number) > 0 :
+    num_line = int(page_stop_number[0].text)
+  else :
+    num_line = 0
+
+  # Choose randomly one or more objects
+  if num_line > 0 :
+    line_number = random.randint(1,num_line) + 2
+    # Check the box corresponding to line_number if not already checked
+    browser.mainForm.getListboxControl(line_number=line_number, column_number=1).value = 'checked'
+    browser.mainForm.submit(name='Base_callDialogMethod:method')
+
+    # Check whether the changes have been successfully updated
+    assert browser.getTransitionMessage() == 'Data updated.'
\ No newline at end of file