openstack/trove
Code Issues Proposed changes

Get rid of nose and Proboscis

Browse Source 
they are not maintained for a long time, and trove already switch
the tests to templest.
This commit removes the usage of them. more details see story.

Story: 2010882
Task: 48606

Depends-On: https://review.opendev.org/c/openstack/trove/+/914228

Change-Id: Ie021e0bd708cf8ed13853dea632312af38190205
This commit is contained in:
wu.chunyang 2023-12-22 17:51:38 +08:00 committed by wu.chunyang
parent 3863b81a18
commit 93f35c7f04
101 changed files with 7 additions and 24209 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
doc/source/contributor/functional_test.rst
View File

@ -12,6 +12,10 @@ However, Trove team is not going to migrate all the existing functional tests to
Since Victoria, the upstream CI jobs keep failing because of the poor performance of the CI devstack host (virtual machine), trove project contributors should guarantee any proposed patch passes both the functional test and trove tempest test by themselves, the code reviewer may ask for the test result.
.. note::
Since Caracal, functional test are removed from Trove project.
Install DevStack
----------------

42
generate_examples.py
View File

@ -1,42 +0,0 @@
#!/usr/bin/env python
# Copyright 2014 OpenStack Foundation
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
#
import argparse
import os
import sys
import run_tests
def import_tests():
from trove.tests.examples import snippets
snippets.monkey_patch_uuid_and_date()
if __name__ == "__main__":
parser = argparse.ArgumentParser(description='Generate Example Snippets')
parser.add_argument('--fix-examples', action='store_true',
help='Fix the examples rather than failing tests.')
args = parser.parse_args()
if args.fix_examples:
os.environ['TESTS_FIX_EXAMPLES'] = 'True'
# Remove the '--fix-examples' argument from sys.argv as it is not a
# valid argument in the run_tests module.
sys.argv.pop(sys.argv.index('--fix-examples'))
run_tests.main(import_tests)

230
integration/README.md
View File

@ -1,230 +0,0 @@
# Trove integration script - trovestack
## Steps to setup environment
Install a fresh Ubuntu 22.04 (jammy) image. We suggest creating a development virtual machine using the image.
1. Login to the machine as root
1. Make sure we have git installed
```
# apt-get update
# apt-get install git-core -y
```
1. Add a user named ubuntu if you do not already have one:
```
# adduser ubuntu
# visudo
```
Add this line to the file below the root user
ubuntu ALL=(ALL:ALL) ALL
Or use this if you dont want to type your password to sudo a command:
ubuntu ALL=(ALL) NOPASSWD: ALL
if /dev/pts/0 does not have read/write for your user
# chmod 666 /dev/pts/0
> Note that this number can change and if you can not connect to the screen session then the /dev/pts/# needs modding like above.
1. Login with ubuntu and download the Trove code.
```shell
# su ubuntu
$ mkdir -p /opt/stack
$ cd /opt/stack
```
> Note that it is important that you clone the repository
here. This is a change from the earlier trove-integration where
you could clone trove-integration anywhere you wanted (like HOME)
and trove would get cloned for you in the right place. Since
trovestack is now in the trove repository, if you wish to test
changes that you have made to trove, it is advisable for you to
have your trove repository in /opt/stack to avoid another trove
repository being cloned for you.
1. Clone this repo and go into the scripts directory
```
$ git clone https://github.com/openstack/trove.git
$ cd trove/integration/scripts/
```
## Running trovestack
Run this to get the command list with a short description of each
$ ./trovestack
### Install Trove
*This brings up trove services and initializes the trove database.*
$ ./trovestack install
### Connecting to the screen session
$ screen -x stack
If that command fails with the error
Cannot open your terminal '/dev/pts/1'
If that command fails with the error chmod the corresponding /dev/pts/#
$ chmod 660 /dev/pts/1
### Navigate the log screens
To produce the list of screens that you can scroll through and select
ctrl+a then "
An example of screen list:
```
..... (full list ommitted)
20 c-vol
21 h-eng
22 h-api
23 h-api-cfn
24 h-api-cw
25 tr-api
26 tr-tmgr
27 tr-cond
```
Alternatively, to go directly to a specific screen window
ctrl+a then '
then enter a number (like 25) or name (like tr-api)
### Detach from the screen session
Allows the services to continue running in the background
ctrl+a then d
### Kick start the build/test-init/build-image commands
*Add mysql as a parameter to set build and add the mysql guest image. This will also populate /etc/trove/test.conf with appropriate values for running the integration tests.*
$ ./trovestack kick-start mysql
### Initialize the test configuration and set up test users (overwrites /etc/trove/test.conf)
$ ./trovestack test-init
### Build guest agent image
The trove guest agent image could be created using `trovestack` script
according to the following command:
```shell
PATH_DEVSTACK_OUTPUT=/opt/stack \
./trovestack build-image \
${datastore_type} \
${guest_os} \
${guest_os_release} \
${dev_mode}
```
- If the script is running as a part of DevStack, the viriable
`PATH_DEVSTACK_OUTPUT` is set automatically.
- if `dev_mode=false`, the trove code for guest agent is injected into the
image at the building time.
- If `dev_mode=true`, no Trove code is injected into the guest image. The guest
agent will download Trove code during the service initialization.
For example, build a Mysql image for Ubuntu jammy operating system:
```shell
$ ./trovestack build-image mysql ubuntu jammy false
```
### Running Integration Tests
Check the values in /etc/trove/test.conf in case it has been re-initialized prior to running the tests. For example, from the previous mysql steps:
"dbaas_datastore": "%datastore_type%",
"dbaas_datastore_version": "%datastore_version%",
should be:
"dbaas_datastore": "mysql",
"dbaas_datastore_version": "5.5",
Once Trove is running on DevStack, you can run the integration tests locally.
$./trovestack int-tests
This will runs all of the blackbox tests by default. Use the `--group` option to run a different group:
$./trovestack int-tests --group=simple_blackbox
You can also specify the `TESTS_USE_INSTANCE_ID` environment variable to have the integration tests use an existing instance for the tests rather than creating a new one.
$./TESTS_DO_NOT_DELETE_INSTANCE=True TESTS_USE_INSTANCE_ID=INSTANCE_UUID ./trovestack int-tests --group=simple_blackbox
## Reset your environment
### Stop all the services running in the screens and refresh the environment
$ killall -9 screen
$ screen -wipe
$ RECLONE=yes ./trovestack install
$ ./trovestack kick-start mysql
or
$ RECLONE=yes ./trovestack install
$ ./trovestack test-init
$ ./trovestack build-image mysql
## Recover after reboot
If the VM was restarted, then the process for bringing up Openstack and Trove is quite simple
$./trovestack start-deps
$./trovestack start
Use screen to ensure all modules have started without error
$screen -r stack
## VMware Fusion 5 speed improvement
Running Ubuntu with KVM or Qemu can be extremely slow without certain optimizations. The following are some VMware settings that can improve performance and may also apply to other virtualization platforms.
1. Shutdown the Ubuntu VM.
2. Go to VM Settings -> Processors & Memory -> Advanced Options.
Check the "Enable hypervisor applications in this virtual machine"
3. Go to VM Settings -> Advanced.
Set the "Troubleshooting" option to "None"
4. After setting these create a snapshot so that in cases where things break down you can revert to a clean snapshot.
5. Boot up the VM and run the `./trovestack install`
6. To verify that KVM is setup properly after the devstack installation you can run these commands.
```
ubuntu@ubuntu:~$ kvm-ok
INFO: /dev/kvm exists
KVM acceleration can be used
```
## VMware Workstation performance improvements
In recent versions of VMWare, you can get much better performance if you enable the right virtualization options. For example, in VMWare Workstation (found in version 10.0.2), click on VM->Settings->Processor.
You should see a box of "Virtualization Engine" options that can be changed only when the VM is shutdown.
Make sure you check "Virtualize Intel VT-x/EPT or AMD-V/RVI" and "Virtualize CPU performance counters". Set the preferred mode to "Automatic".
Then boot the VM and ensure that the proper virtualization is enabled.
```
ubuntu@ubuntu:~$ kvm-ok
INFO: /dev/kvm exists
KVM acceleration can be used
```

44
integration/tests/integration/core.test.conf
View File

@ -1,44 +0,0 @@
{
"report_directory":"rdli-test-report",
"white_box":false,
"test_mgmt":false,
"use_local_ovz":false,
"use_venv":false,
"glance_code_root":"/opt/stack/glance",
"glance_api_conf":"/vagrant/conf/glance-api.conf",
"glance_reg_conf":"/vagrant/conf/glance-reg.conf",
"glance_images_directory": "/glance_images",
"glance_image": "fakey_fakerson.tar.gz",
"instance_flavor_name":"m1.rd-tiny",
"instance_bigger_flavor_name":"m1.rd-smaller",
"nova_code_root":"/opt/stack/nova",
"nova_conf":"/home/vagrant/nova.conf",
"keystone_code_root":"/opt/stack/keystone",
"keystone_conf":"/etc/keystone/keystone.conf",
"trove_code_root":"/opt/stack/trove",
"trove_conf":"/tmp/trove.conf",
"trove_version":"v1.0",
"trove_api_updated":"2012-08-01T00:00:00Z",
"trove_must_have_volume":false,
"trove_can_have_volume":true,
"trove_main_instance_has_volume": true,
"trove_max_accepted_volume_size": 1000,
"trove_max_instances_per_user": 55,
"trove_max_volumes_per_user": 100,
"use_reaper":false,
"root_removed_from_instance_api": true,
"root_timestamp_disabled": false,
"openvz_disabled": false,
"management_api_disabled": true,
"dbaas_image": 1,
"dns_driver":"trove.dns.rsdns.driver.RsDnsDriver",
"dns_instance_entry_factory":"trove.dns.rsdns.driver.RsDnsInstanceEntryFactory",
"databases_page_size": 20,
"instances_page_size": 20,
"users_page_size": 20,
"rabbit_runs_locally":false,
"dns_instance_entry_factory":"trove.dns.rsdns.driver.RsDnsInstanceEntryFactory",
"sentinel": null
}

245
integration/tests/integration/int_tests.py
View File

@ -1,245 +0,0 @@
#!/usr/bin/env python
#
# # Copyright (c) 2011 OpenStack, LLC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Runs the tests.
There are a few initialization issues to deal with.
The first is flags, which must be initialized before any imports. The test
configuration has the same problem (it was based on flags back when the tests
resided outside of the Nova code).
The command line is picked apart so that Nose won't see commands it isn't
compatible with, such as "--flagfile" or "--group".
This script imports all other tests to make them known to Proboscis before
passing control to proboscis.TestProgram which itself calls nose, which then
call unittest.TestProgram and exits.
If "repl" is a command line argument, then the original stdout and stderr is
saved and sys.exit is neutralized so that unittest.TestProgram will not exit
and instead sys.stdout and stderr are restored so that interactive mode can
be used.
"""
import atexit
import gettext
import os
import sys
import proboscis
from nose import config
from nose import core
from tests.colorizer import NovaTestRunner
if os.environ.get("PYDEV_DEBUG", "False") == 'True':
from pydev import pydevd
pydevd.settrace('10.0.2.2', port=7864, stdoutToServer=True,
stderrToServer=True)
def add_support_for_localization():
"""Adds support for localization in the logging.
If ../nova/__init__.py exists, add ../ to Python search path, so that
it will override what happens to be installed in
/usr/(local/)lib/python...
"""
path = os.path.join(os.path.abspath(sys.argv[0]), os.pardir, os.pardir)
possible_topdir = os.path.normpath(path)
if os.path.exists(os.path.join(possible_topdir, 'nova', '__init__.py')):
sys.path.insert(0, possible_topdir)
gettext.install('nova')
MAIN_RUNNER = None
def initialize_rdl_config(config_file):
from trove.common import cfg
from oslo_log import log
from trove.db import get_db_api
conf = cfg.CONF
cfg.parse_args(['int_tests'], default_config_files=[config_file])
log.setup(conf, None)
try:
get_db_api().configure_db(conf)
conf_file = conf.find_file(conf.api_paste_config)
except RuntimeError as error:
import traceback
print(traceback.format_exc())
sys.exit("ERROR: %s" % error)
def _clean_up():
"""Shuts down any services this program has started and shows results."""
from tests.util import report
report.update()
if MAIN_RUNNER is not None:
MAIN_RUNNER.on_exit()
from tests.util.services import get_running_services
for service in get_running_services():
sys.stderr.write("Stopping service ")
for c in service.cmd:
sys.stderr.write(c + " ")
sys.stderr.write("...\n\r")
service.stop()
def import_tests():
# The DNS stuff is problematic. Not loading the other tests allow us to
# run its functional tests only.
ADD_DOMAINS = os.environ.get("ADD_DOMAINS", "False") == 'True'
if not ADD_DOMAINS:
# F401 unused imports needed for tox tests
from trove.tests.api import backups # noqa
from trove.tests.api import configurations # noqa
from trove.tests.api import databases # noqa
from trove.tests.api import datastores # noqa
from trove.tests.api import instances as rd_instances # noqa
from trove.tests.api import instances_actions as acts # noqa
from trove.tests.api import instances_delete # noqa
from trove.tests.api import instances_resize # noqa
from trove.tests.api import limits # noqa
from trove.tests.api.mgmt import datastore_versions # noqa
from trove.tests.api.mgmt import instances_actions as mgmt_acts # noqa
from trove.tests.api import replication # noqa
from trove.tests.api import root # noqa
from trove.tests.api import user_access # noqa
from trove.tests.api import users # noqa
from trove.tests.api import versions # noqa
from trove.tests.db import migrations # noqa
# Groups that exist as core int-tests are registered from the
# trove.tests.int_tests module
from trove.tests import int_tests
def run_main(test_importer):
add_support_for_localization()
# Strip non-nose arguments out before passing this to nosetests
repl = False
nose_args = []
conf_file = "~/test.conf"
show_elapsed = True
groups = []
print("RUNNING TEST ARGS : " + str(sys.argv))
extra_test_conf_lines = []
rdl_config_file = None
nova_flag_file = None
index = 0
while index < len(sys.argv):
arg = sys.argv[index]
if arg[:2] == "-i" or arg == '--repl':
repl = True
elif arg[:7] == "--conf=":
conf_file = os.path.expanduser(arg[7:])
print("Setting TEST_CONF to " + conf_file)
os.environ["TEST_CONF"] = conf_file
elif arg[:8] == "--group=":
groups.append(arg[8:])
elif arg == "--test-config":
if index >= len(sys.argv) - 1:
print('Expected an argument to follow "--test-conf".')
sys.exit()
conf_line = sys.argv[index + 1]
extra_test_conf_lines.append(conf_line)
elif arg[:11] == "--flagfile=":
pass
elif arg[:14] == "--config-file=":
rdl_config_file = arg[14:]
elif arg[:13] == "--nova-flags=":
nova_flag_file = arg[13:]
elif arg.startswith('--hide-elapsed'):
show_elapsed = False
else:
nose_args.append(arg)
index += 1
# Many of the test decorators depend on configuration values, so before
# start importing modules we have to load the test config followed by the
# flag files.
from trove.tests.config import CONFIG
# Find config file.
if not "TEST_CONF" in os.environ:
raise RuntimeError("Please define an environment variable named " +
"TEST_CONF with the location to a conf file.")
file_path = os.path.expanduser(os.environ["TEST_CONF"])
if not os.path.exists(file_path):
raise RuntimeError("Could not find TEST_CONF at " + file_path + ".")
# Load config file and then any lines we read from the arguments.
CONFIG.load_from_file(file_path)
for line in extra_test_conf_lines:
CONFIG.load_from_line(line)
if CONFIG.white_box: # If white-box testing, set up the flags.
# Handle loading up RDL's config file madness.
initialize_rdl_config(rdl_config_file)
# Set up the report, and print out how we're running the tests.
from tests.util import report
from datetime import datetime
report.log("Trove Integration Tests, %s" % datetime.now())
report.log("Invoked via command: " + str(sys.argv))
report.log("Groups = " + str(groups))
report.log("Test conf file = %s" % os.environ["TEST_CONF"])
if CONFIG.white_box:
report.log("")
report.log("Test config file = %s" % rdl_config_file)
report.log("")
report.log("sys.path:")
for path in sys.path:
report.log("\t%s" % path)
# Now that all configurations are loaded its time to import everything
test_importer()
atexit.register(_clean_up)
c = config.Config(stream=sys.stdout,
env=os.environ,
verbosity=3,
plugins=core.DefaultPluginManager())
runner = NovaTestRunner(stream=c.stream,
verbosity=c.verbosity,
config=c,
show_elapsed=show_elapsed,
known_bugs=CONFIG.known_bugs)
MAIN_RUNNER = runner
if repl:
# Turn off the following "feature" of the unittest module in case
# we want to start a REPL.
sys.exit = lambda x: None
proboscis.TestProgram(argv=nose_args, groups=groups, config=c,
testRunner=MAIN_RUNNER).run_and_exit()
sys.stdout = sys.__stdout__
sys.stderr = sys.__stderr__
if __name__ == "__main__":
run_main(import_tests)

95
integration/tests/integration/localhost.test.conf
View File

@ -1,95 +0,0 @@
{
"include-files":["core.test.conf"],
"fake_mode": true,
"dbaas_url":"http://localhost:8779/v1.0",
"version_url":"http://localhost:8779",
"nova_auth_url":"http://localhost:8779/v1.0/auth",
"trove_auth_url":"http://localhost:8779/v1.0/auth",
"trove_client_insecure":false,
"auth_strategy":"fake",
"trove_version":"v1.0",
"trove_api_updated":"2012-08-01T00:00:00Z",
"trove_dns_support":false,
"trove_ip_support":false,
"nova_client": null,
"users": [
{
"auth_user":"admin",
"auth_key":"password",
"tenant":"admin-1000",
"requirements": {
"is_admin":true,
"services": ["trove"]
}
},
{
"auth_user":"jsmith",
"auth_key":"password",
"tenant":"2500",
"requirements": {
"is_admin":false,
"services": ["trove"]
}
},
{
"auth_user":"hub_cap",
"auth_key":"password",
"tenant":"3000",
"requirements": {
"is_admin":false,
"services": ["trove"]
}
}
],
"flavors": [
{
"id": 1,
"name": "m1.tiny",
"ram": 512
},
{
"id": 2,
"name": "m1.small",
"ram": 2048
},
{
"id": 3,
"name": "m1.medium",
"ram": 4096
},
{
"id": 4,
"name": "m1.large",
"ram": 8192
},
{
"id": 5,
"name": "m1.xlarge",
"ram": 16384
},
{
"id": 6,
"name": "tinier",
"ram": 506
},
{
"id": 7,
"name": "m1.rd-tiny",
"ram": 512
},
{
"id": 8,
"name": "m1.rd-smaller",
"ram": 768
}
],
"sentinel": null
}

25
integration/tests/integration/tests/__init__.py
View File

@ -1,25 +0,0 @@
# Copyright (c) 2011 OpenStack, LLC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
:mod:`tests` -- Integration / Functional Tests for Nova
===================================
.. automodule:: tests
:platform: Unix
:synopsis: Tests for Nova.
.. moduleauthor:: Nirmal Ranganathan <nirmal.ranganathan@rackspace.com>
.. moduleauthor:: Tim Simpson <tim.simpson@rackspace.com>
"""

445
integration/tests/integration/tests/colorizer.py
View File

@ -1,445 +0,0 @@
#!/usr/bin/env python
# Copyright 2010 United States Government as represented by the
# Administrator of the National Aeronautics and Space Administration.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
# Colorizer Code is borrowed from Twisted:
# Copyright (c) 2001-2010 Twisted Matrix Laboratories.
#
# Permission is hereby granted, free of charge, to any person obtaining
# a copy of this software and associated documentation files (the
# "Software"), to deal in the Software without restriction, including
# without limitation the rights to use, copy, modify, merge, publish,
# distribute, sublicense, and/or sell copies of the Software, and to
# permit persons to whom the Software is furnished to do so, subject to
# the following conditions:
#
# The above copyright notice and this permission notice shall be
# included in all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
# EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF
# MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
# NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE
# LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION
# OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION
# WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.
"""Unittest runner for Nova.
To run all tests
python run_tests.py
To run a single test:
python run_tests.py test_compute:ComputeTestCase.test_run_terminate
To run a single test module:
python run_tests.py test_compute
or
python run_tests.py api.test_wsgi
"""
import gettext
import heapq
import logging
import os
import unittest
import sys
import time
gettext.install('nova')
from nose import config
from nose import core
from nose import result
from proboscis import case
from proboscis import SkipTest
class _AnsiColorizer(object):
"""
A colorizer is an object that loosely wraps around a stream, allowing
callers to write text to the stream in a particular color.
Colorizer classes must implement C{supported()} and C{write(text, color)}.
"""
_colors = dict(black=30, red=31, green=32, yellow=33,
blue=34, magenta=35, cyan=36, white=37)
def __init__(self, stream):
self.stream = stream
def supported(cls, stream=sys.stdout):
"""
A class method that returns True if the current platform supports
coloring terminal output using this method. Returns False otherwise.
"""
if not stream.isatty():
return False # auto color only on TTYs
try:
import curses
except ImportError:
return False
else:
try:
try:
return curses.tigetnum("colors") > 2
except curses.error:
curses.setupterm()
return curses.tigetnum("colors") > 2
except:
raise
# guess false in case of error
return False
supported = classmethod(supported)
def write(self, text, color):
"""
Write the given text to the stream in the given color.
@param text: Text to be written to the stream.
@param color: A string label for a color. e.g. 'red', 'white'.
"""
color = self._colors[color]
self.stream.write('\x1b[%s;1m%s\x1b[0m' % (color, text))
class _Win32Colorizer(object):
"""
See _AnsiColorizer docstring.
"""
def __init__(self, stream):
from win32console import GetStdHandle, STD_OUT_HANDLE, \
FOREGROUND_RED, FOREGROUND_BLUE, FOREGROUND_GREEN, \
FOREGROUND_INTENSITY
red, green, blue, bold = (FOREGROUND_RED, FOREGROUND_GREEN,
FOREGROUND_BLUE, FOREGROUND_INTENSITY)
self.stream = stream
self.screenBuffer = GetStdHandle(STD_OUT_HANDLE)
self._colors = {
'normal': red | green | blue,
'red': red | bold,
'green': green | bold,
'blue': blue | bold,
'yellow': red | green | bold,
'magenta': red | blue | bold,
'cyan': green | blue | bold,
'white': red | green | blue | bold
}
def supported(cls, stream=sys.stdout):
try:
import win32console
screenBuffer = win32console.GetStdHandle(
win32console.STD_OUT_HANDLE)
except ImportError:
return False
import pywintypes
try:
screenBuffer.SetConsoleTextAttribute(
win32console.FOREGROUND_RED |
win32console.FOREGROUND_GREEN |
win32console.FOREGROUND_BLUE)
except pywintypes.error:
return False
else:
return True
supported = classmethod(supported)
def write(self, text, color):
color = self._colors[color]
self.screenBuffer.SetConsoleTextAttribute(color)
self.stream.write(text)
self.screenBuffer.SetConsoleTextAttribute(self._colors['normal'])
class _NullColorizer(object):
"""
See _AnsiColorizer docstring.
"""
def __init__(self, stream):
self.stream = stream
def supported(cls, stream=sys.stdout):
return True
supported = classmethod(supported)
def write(self, text, color):
self.stream.write(text)
def get_elapsed_time_color(elapsed_time):
if elapsed_time > 1.0:
return 'yellow'
elif elapsed_time > 0.25:
return 'cyan'
else:
return 'green'
class NovaTestResult(case.TestResult):
def __init__(self, *args, **kw):
self.show_elapsed = kw.pop('show_elapsed')
self.known_bugs = kw.pop('known_bugs', {})
super(NovaTestResult, self).__init__(*args, **kw)
self.num_slow_tests = 5
self.slow_tests = [] # this is a fixed-sized heap
self._last_case = None
self.colorizer = None
# NOTE(vish): reset stdout for the terminal check
stdout = sys.stdout
sys.stdout = sys.__stdout__
for colorizer in [_Win32Colorizer, _AnsiColorizer, _NullColorizer]:
if colorizer.supported():
self.colorizer = colorizer(self.stream)
break
sys.stdout = stdout
# NOTE(lorinh): Initialize start_time in case a sqlalchemy-migrate
# error results in it failing to be initialized later. Otherwise,
# _handleElapsedTime will fail, causing the wrong error message to
# be outputted.
self.start_time = time.time()
def _intercept_known_bugs(self, test, err):
name = str(test)
excuse = self.known_bugs.get(name, None)
if excuse:
tracker_id, error_string = excuse
if error_string in str(err[1]):
skip = SkipTest("KNOWN BUG: %s\n%s"
% (tracker_id, str(err[1])))
self.onError(test)
super(NovaTestResult, self).addSkip(test, skip)
else:
result = (RuntimeError, RuntimeError(
'Test "%s" contains known bug %s.\n'
'Expected the following error string:\n%s\n'
'What was seen was the following:\n%s\n'
'If the bug is no longer happening, please change '
'the test config.'
% (name, tracker_id, error_string, str(err))), None)
self.onError(test)
super(NovaTestResult, self).addError(test, result)
return True
return False
def getDescription(self, test):
return str(test)
def _handleElapsedTime(self, test):
self.elapsed_time = time.time() - self.start_time
item = (self.elapsed_time, test)
# Record only the n-slowest tests using heap
if len(self.slow_tests) >= self.num_slow_tests:
heapq.heappushpop(self.slow_tests, item)
else:
heapq.heappush(self.slow_tests, item)
def _writeElapsedTime(self, test):
color = get_elapsed_time_color(self.elapsed_time)
self.colorizer.write(" %.2f" % self.elapsed_time, color)
def _writeResult(self, test, long_result, color, short_result, success):
if self.showAll:
self.colorizer.write(long_result, color)
if self.show_elapsed and success:
self._writeElapsedTime(test)
self.stream.writeln()
elif self.dots:
self.stream.write(short_result)
self.stream.flush()
# NOTE(vish): copied from unittest with edit to add color
def addSuccess(self, test):
if self._intercept_known_bugs(test, None):
return
unittest.TestResult.addSuccess(self, test)
self._handleElapsedTime(test)
self._writeResult(test, 'OK', 'green', '.', True)
# NOTE(vish): copied from unittest with edit to add color
def addFailure(self, test, err):
if self._intercept_known_bugs(test, err):
return
self.onError(test)
unittest.TestResult.addFailure(self, test, err)
self._handleElapsedTime(test)
self._writeResult(test, 'FAIL', 'red', 'F', False)
# NOTE(vish): copied from nose with edit to add color
def addError(self, test, err):
"""Overrides normal addError to add support for
errorClasses. If the exception is a registered class, the
error will be added to the list for that class, not errors.
"""
if self._intercept_known_bugs(test, err):
return
self.onError(test)
self._handleElapsedTime(test)
stream = getattr(self, 'stream', None)
ec, ev, tb = err
try:
exc_info = self._exc_info_to_string(err, test)
except TypeError:
# 2.3 compat
exc_info = self._exc_info_to_string(err)
for cls, (storage, label, isfail) in self.errorClasses.items():
if result.isclass(ec) and issubclass(ec, cls):
if isfail:
test.passed = False
storage.append((test, exc_info))
# Might get patched into a streamless result
if stream is not None:
if self.showAll:
message = [label]
detail = result._exception_detail(err[1])
if detail:
message.append(detail)
stream.writeln(": ".join(message))
elif self.dots:
stream.write(label[:1])
return
self.errors.append((test, exc_info))
test.passed = False
if stream is not None:
self._writeResult(test, 'ERROR', 'red', 'E', False)
@staticmethod
def get_doc(cls_or_func):
"""Grabs the doc abbreviated doc string."""
try:
return cls_or_func.__doc__.split("\n")[0].strip()
except (AttributeError, IndexError):
return None
def startTest(self, test):
unittest.TestResult.startTest(self, test)
self.start_time = time.time()
test_name = None
try:
entry = test.test.__proboscis_case__.entry
if entry.method:
current_class = entry.method.im_class
test_name = self.get_doc(entry.home) or entry.home.__name__
else:
current_class = entry.home
except AttributeError:
current_class = test.test.__class__
if self.showAll:
if current_class.__name__ != self._last_case:
self.stream.writeln(current_class.__name__)
self._last_case = current_class.__name__
try:
doc = self.get_doc(current_class)
except (AttributeError, IndexError):
doc = None
if doc:
self.stream.writeln(' ' + doc)
if not test_name:
if hasattr(test.test, 'shortDescription'):
test_name = test.test.shortDescription()
if not test_name:
test_name = test.test._testMethodName
self.stream.write('\t%s' % str(test_name).ljust(60))
self.stream.flush()
class NovaTestRunner(core.TextTestRunner):
def __init__(self, *args, **kwargs):
self.show_elapsed = kwargs.pop('show_elapsed')
self.known_bugs = kwargs.pop('known_bugs', {})
self.__result = None
self.__finished = False
self.__start_time = None
super(NovaTestRunner, self).__init__(*args, **kwargs)
def _makeResult(self):
self.__result = NovaTestResult(
self.stream,
self.descriptions,
self.verbosity,
self.config,
show_elapsed=self.show_elapsed,
known_bugs=self.known_bugs)
self.__start_time = time.time()
return self.__result
def _writeSlowTests(self, result_):
# Pare out 'fast' tests
slow_tests = [item for item in result_.slow_tests
if get_elapsed_time_color(item[0]) != 'green']
if slow_tests:
slow_total_time = sum(item[0] for item in slow_tests)
self.stream.writeln("Slowest %i tests took %.2f secs:"
% (len(slow_tests), slow_total_time))
for elapsed_time, test in sorted(slow_tests, reverse=True):
time_str = "%.2f" % elapsed_time
self.stream.writeln(" %s %s" % (time_str.ljust(10), test))
def on_exit(self):
if self.__result is None:
print("Exiting before tests even started.")
else:
if not self.__finished:
msg = "Tests aborted, trying to print available results..."
print(msg)
stop_time = time.time()
self.__result.printErrors()
self.__result.printSummary(self.__start_time, stop_time)
self.config.plugins.finalize(self.__result)
if self.show_elapsed:
self._writeSlowTests(self.__result)
def run(self, test):
result_ = super(NovaTestRunner, self).run(test)
if self.show_elapsed:
self._writeSlowTests(result_)
self.__finished = True
return result_
if __name__ == '__main__':
logging.setup()
# If any argument looks like a test name but doesn't have "nova.tests" in
# front of it, automatically add that so we don't have to type as much
show_elapsed = True
argv = []
test_fixture = os.getenv("UNITTEST_FIXTURE", "trove")
for x in sys.argv:
if x.startswith('test_'):
argv.append('%s.tests.%s' % (test_fixture, x))
elif x.startswith('--hide-elapsed'):
show_elapsed = False
else:
argv.append(x)
testdir = os.path.abspath(os.path.join(test_fixture, "tests"))
c = config.Config(stream=sys.stdout,
env=os.environ,
verbosity=3,
workingDir=testdir,
plugins=core.DefaultPluginManager())
runner = NovaTestRunner(stream=c.stream,
verbosity=c.verbosity,
config=c,
show_elapsed=show_elapsed)
sys.exit(not core.run(config=c, testRunner=runner, argv=argv))

63
integration/tests/integration/tests/initialize.py
View File

@ -1,63 +0,0 @@
# Copyright (c) 2011 OpenStack, LLC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import os
from proboscis import test
from proboscis.asserts import fail
from tests.util.services import Service
from trove.tests.config import CONFIG
def dbaas_url():
return str(CONFIG.values.get("dbaas_url"))
def nova_url():
return str(CONFIG.values.get("nova_client")['url'])
class Daemon(object):
"""Starts a daemon."""
def __init__(self, alternate_path=None, conf_file_name=None,
extra_cmds=None, service_path_root=None, service_path=None):
# The path to the daemon bin if the other one doesn't work.
self.alternate_path = alternate_path
self.extra_cmds = extra_cmds or []
# The name of a test config value which points to a conf file.
self.conf_file_name = conf_file_name
# The name of a test config value, which is inserted into the service_path.
self.service_path_root = service_path_root
# The first path to the daemon bin we try.
self.service_path = service_path or "%s"
def run(self):
# Print out everything to make it
print("Looking for config value %s..." % self.service_path_root)
print(CONFIG.values[self.service_path_root])
path = self.service_path % CONFIG.values[self.service_path_root]
print("Path = %s" % path)
if not os.path.exists(path):
path = self.alternate_path
if path is None:
fail("Could not find path to %s" % self.service_path_root)
conf_path = str(CONFIG.values[self.conf_file_name])
cmds = CONFIG.python_cmd_list() + [path] + self.extra_cmds + \
[conf_path]
print("Running cmds: %s" % cmds)
self.service = Service(cmds)
if not self.service.is_service_alive():
self.service.start()

0
integration/tests/integration/tests/util/__init__.py
View File

76
integration/tests/integration/tests/util/report.py
View File

@ -1,76 +0,0 @@
"""Creates a report for the test.
"""
import os
import shutil
from os import path
from trove.tests.config import CONFIG
USE_LOCAL_OVZ = CONFIG.use_local_ovz
class Reporter(object):
"""Saves the logs from a test run."""
def __init__(self, root_path):
self.root_path = root_path
if not path.exists(self.root_path):
os.mkdir(self.root_path)
for file in os.listdir(self.root_path):
if file.endswith(".log"):
os.remove(path.join(self.root_path, file))
def _find_all_instance_ids(self):
instances = []
if USE_LOCAL_OVZ:
for dir in os.listdir("/var/lib/vz/private"):
instances.append(dir)
return instances
def log(self, msg):
with open("%s/report.log" % self.root_path, 'a') as file:
file.write(str(msg) + "\n")
def _save_syslog(self):
try:
shutil.copyfile("/var/log/syslog", "host-syslog.log")
except (shutil.Error, IOError) as err:
self.log("ERROR logging syslog : %s" % (err))
def _update_instance(self, id):
root = "%s/%s" % (self.root_path, id)
def save_file(path, short_name):
if USE_LOCAL_OVZ:
try:
shutil.copyfile("/var/lib/vz/private/%s/%s" % (id, path),
"%s-%s.log" % (root, short_name))
except (shutil.Error, IOError) as err:
self.log("ERROR logging %s for instance id %s! : %s"
% (path, id, err))
else:
#TODO: Can we somehow capture these (maybe SSH to the VM)?
pass
save_file("/var/log/firstboot", "firstboot")
save_file("/var/log/syslog", "syslog")
save_file("/var/log/nova/guest.log", "nova-guest")
def _update_instances(self):
for id in self._find_all_instance_ids():
self._update_instance(id)
def update(self):
self._update_instances()
self._save_syslog()
REPORTER = Reporter(CONFIG.report_directory)
def log(msg):
REPORTER.log(msg)
def update():
REPORTER.update()

110
integration/tests/integration/tests/util/rpc.py
View File

@ -1,110 +0,0 @@
# Copyright (c) 2012 OpenStack, LLC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""
Test utility for RPC checks.
Functionality to check for rabbit here depends on having rabbit running on
the same machine as the tests, so that the rabbitmqctl commands will function.
The functionality is turned on or off by the test config "rabbit_runs_locally".
"""
import re
from trove.tests.config import CONFIG
from services import start_proc
if CONFIG.values.get('rabbit_runs_locally', False) == True:
DIRECT_ACCESS = True
class Rabbit(object):
def declare_queue(self, topic):
"""Call this to declare a queue from Python."""
#from trove.rpc.impl_kombu import Connection
from trove.openstack.common.rpc import create_connection
with create_connection() as conn:
consumer = conn.declare_topic_consumer(topic=topic)
def get_queue_items(self, queue_name):
"""Determines if the queue exists and if so the message count.
If the queue exists the return value is an integer, otherwise
its None.
Be careful because queue_name is used in a regex and can't have
any unescaped characters.
"""
proc = start_proc(["/usr/bin/sudo", "rabbitmqctl", "list_queues"],
shell=False)
for line in iter(proc.stdout.readline, ""):
print("LIST QUEUES:" + line)
m = re.search(r"%s\s+([0-9]+)" % queue_name, line)
if m:
return int(m.group(1))
return None
@property
def is_alive(self):
"""Calls list_queues, should fail."""
try:
stdout, stderr = self.run(0, "rabbitmqctl", "list_queues")
for lines in stdout, stderr:
for line in lines:
if "no_exists" in line:
return False
return True
except Exception:
return False
def reset(self):
out, err = self.run(0, "rabbitmqctl", "reset")
print(out)
print(err)
def run(self, check_exit_code, *cmd):
cmds = ["/usr/bin/sudo"] + list(cmd)
proc = start_proc(cmds)
lines = proc.stdout.readlines()
err_lines = proc.stderr.readlines()
return lines, err_lines
def start(self):
print("Calling rabbitmqctl start_app")
out = self.run(0, "rabbitmqctl", "start_app")
print(out)
out, err = self.run(0, "rabbitmqctl", "change_password", "guest",
CONFIG.values['rabbit_password'])
print(out)
print(err)
def stop(self):
print("Calling rabbitmqctl stop_app")
out = self.run(0, "rabbitmqctl", "stop_app")
print(out)
else:
DIRECT_ACCESS = False
class Rabbit(object):
def __init__(self):
raise RuntimeError("rabbit_runs_locally is set to False in the "
"test config, so this test cannot be run.")

280
integration/tests/integration/tests/util/services.py
View File

@ -1,280 +0,0 @@
# Copyright (c) 2011 OpenStack, LLC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
"""Functions to initiate and shut down services needed by the tests."""
import os
import re
import subprocess
import time
from collections import namedtuple
from httplib2 import Http
from nose.plugins.skip import SkipTest
from proboscis import decorators
def _is_web_service_alive(url):
"""Does a HTTP GET request to see if the web service is up."""
client = Http()
try:
resp = client.request(url, 'GET')
return resp != None
except Exception:
return False
_running_services = []
def get_running_services():
""" Returns the list of services which this program has started."""
return _running_services
def start_proc(cmd, shell=False):
"""Given a command, starts and returns a process."""
env = os.environ.copy()
proc = subprocess.Popen(
cmd,
shell=shell,
stdin=subprocess.PIPE,
bufsize=0,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
env=env
)
return proc
MemoryInfo = namedtuple("MemoryInfo", ['mapped', 'writeable', 'shared'])
class Service(object):
"""Starts and stops a service under test.
The methods to start and stop the service will not actually do anything
if they detect the service is already running on this machine. This is
because it may be useful for developers to start the services themselves
some other way.
"""
# TODO(tim.simpson): Hard to follow, consider renaming certain attributes.
def __init__(self, cmd):
"""Defines a service to run."""
if not isinstance(cmd, list):
raise TypeError()
self.cmd = cmd
self.do_not_manage_proc = False
self.proc = None
def __del__(self):
if self.is_running:
self.stop()
def ensure_started(self):
"""Starts the service if it is not running."""
if not self.is_running:
self.start()
def find_proc_id(self):
"""Finds and returns the process id."""
if not self.cmd:
return False
# The cmd[1] signifies the executable python script. It gets invoked