x/python-ganttclient
Code Issues Proposed changes

merge trunk

Browse Source
This commit is contained in:
Cory Wright
2010-12-27 11:27:46 -05:00
parent 3950c51fc0 ab4d9a38eb
commit bb3dfcaa5c
34 changed files with 728 additions and 802 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.mailmap
View File

@@ -24,7 +24,6 @@
<todd@ansolabs.com> <todd@rubidine.com>
<vishvananda@gmail.com> <vishvananda@yahoo.com>
<vishvananda@gmail.com> <root@mirror.nasanebula.net>
# These are from people who failed to set a proper committer
. <root@tonbuntu>
. <laner@controller>
. <root@ubuntu>
<vishvananda@gmail.com> <root@ubuntu>
<sleepsonthefloor@gmail.com> <root@tonbuntu>
<rlane@wikimedia.org> <laner@controller>

4
Authors
View File

@@ -6,6 +6,7 @@ Armando Migliaccio <Armando.Migliaccio@eu.citrix.com>
Chris Behrens <cbehrens@codestud.com>
Chmouel Boudjnah <chmouel@chmouel.com>
Cory Wright <corywright@gmail.com>
David Pravec <David.Pravec@danix.org>
Dean Troyer <dtroyer@gmail.com>
Devin Carlen <devin.carlen@gmail.com>
Ed Leafe <ed@leafe.com>
@@ -26,7 +27,9 @@ Michael Gundlach <michael.gundlach@rackspace.com>
Monty Taylor <mordred@inaugust.com>
Paul Voccio <paul@openstack.org>
Rick Clark <rick@openstack.org>
Ryan Lane <rlane@wikimedia.org>
Ryan Lucio <rlucio@internap.com>
Salvatore Orlando <salvatore.orlando@eu.citrix.com>
Sandy Walsh <sandy.walsh@rackspace.com>
Soren Hansen <soren.hansen@rackspace.com>
Thierry Carrez <thierry@openstack.org>
@@ -35,3 +38,4 @@ Trey Morris <trey.morris@rackspace.com>
Vishvananda Ishaya <vishvananda@gmail.com>
Youcef Laribi <Youcef.Laribi@eu.citrix.com>
Zhixue Wu <Zhixue.Wu@citrix.com>

3
bin/nova-combined
View File

@@ -22,6 +22,7 @@
import eventlet
eventlet.monkey_patch()
import gettext
import os
import sys
@@ -33,6 +34,8 @@ possible_topdir = os.path.normpath(os.path.join(os.path.abspath(sys.argv[0]),
if os.path.exists(os.path.join(possible_topdir, 'nova', '__init__.py')):
sys.path.insert(0, possible_topdir)
gettext.install('nova', unicode=1)
from nova import api
from nova import flags
from nova import service

1
bin/nova-dhcpbridge
View File

@@ -110,7 +110,6 @@ def main():
FLAGS.num_networks = 5
path = os.path.abspath(os.path.join(os.path.dirname(__file__),
'..',
'_trial_temp',
'nova.sqlite'))
FLAGS.sql_connection = 'sqlite:///%s' % path
action = argv[1]

3
nova/auth/fakeldap.py
View File

@@ -150,6 +150,9 @@ def _match(key, value, attrs):
"""Match a given key and value against an attribute list."""
if key not in attrs:
return False
# This is a wild card search. Implemented as all or nothing for now.
if value == "*":
return True
if key != "objectclass":
return value in attrs[key]
# it is an objectclass check, so check subclasses

161
nova/auth/ldapdriver.py
View File

@@ -32,11 +32,16 @@ from nova import flags
FLAGS = flags.FLAGS
flags.DEFINE_integer('ldap_schema_version', 2,
'Current version of the LDAP schema')
flags.DEFINE_string('ldap_url', 'ldap://localhost',
'Point this at your ldap server')
flags.DEFINE_string('ldap_password', 'changeme', 'LDAP password')
flags.DEFINE_string('ldap_user_dn', 'cn=Manager,dc=example,dc=com',
'DN of admin user')
flags.DEFINE_string('ldap_user_id_attribute', 'uid', 'Attribute to use as id')
flags.DEFINE_string('ldap_user_name_attribute', 'cn',
'Attribute to use as name')
flags.DEFINE_string('ldap_user_unit', 'Users', 'OID for Users')
flags.DEFINE_string('ldap_user_subtree', 'ou=Users,dc=example,dc=com',
'OU for Users')
@@ -73,10 +78,20 @@ class LdapDriver(object):
Defines enter and exit and therefore supports the with/as syntax.
"""
project_pattern = '(owner=*)'
isadmin_attribute = 'isNovaAdmin'
project_attribute = 'owner'
project_objectclass = 'groupOfNames'
def __init__(self):
"""Imports the LDAP module"""
self.ldap = __import__('ldap')
self.conn = None
if FLAGS.ldap_schema_version == 1:
LdapDriver.project_pattern = '(objectclass=novaProject)'
LdapDriver.isadmin_attribute = 'isAdmin'
LdapDriver.project_attribute = 'projectManager'
LdapDriver.project_objectclass = 'novaProject'
def __enter__(self):
"""Creates the connection to LDAP"""
@@ -104,13 +119,13 @@ class LdapDriver(object):
"""Retrieve project by id"""
dn = 'cn=%s,%s' % (pid,
FLAGS.ldap_project_subtree)
attr = self.__find_object(dn, '(objectclass=novaProject)')
attr = self.__find_object(dn, LdapDriver.project_pattern)
return self.__to_project(attr)
def get_users(self):
"""Retrieve list of users"""
attrs = self.__find_objects(FLAGS.ldap_user_subtree,
'(objectclass=novaUser)')
'(objectclass=novaUser)')
users = []
for attr in attrs:
user = self.__to_user(attr)
@@ -120,7 +135,7 @@ class LdapDriver(object):
def get_projects(self, uid=None):
"""Retrieve list of projects"""
pattern = '(objectclass=novaProject)'
pattern = LdapDriver.project_pattern
if uid:
pattern = "(&%s(member=%s))" % (pattern, self.__uid_to_dn(uid))
attrs = self.__find_objects(FLAGS.ldap_project_subtree,
@@ -139,23 +154,25 @@ class LdapDriver(object):
# Malformed entries are useless, replace attributes found.
attr = []
if 'secretKey' in user.keys():
attr.append((self.ldap.MOD_REPLACE, 'secretKey', \
[secret_key]))
attr.append((self.ldap.MOD_REPLACE, 'secretKey',
[secret_key]))
else:
attr.append((self.ldap.MOD_ADD, 'secretKey', \
[secret_key]))
attr.append((self.ldap.MOD_ADD, 'secretKey',
[secret_key]))
if 'accessKey' in user.keys():
attr.append((self.ldap.MOD_REPLACE, 'accessKey', \
[access_key]))
attr.append((self.ldap.MOD_REPLACE, 'accessKey',
[access_key]))
else:
attr.append((self.ldap.MOD_ADD, 'accessKey', \
[access_key]))
if 'isAdmin' in user.keys():
attr.append((self.ldap.MOD_REPLACE, 'isAdmin', \
[str(is_admin).upper()]))
attr.append((self.ldap.MOD_ADD, 'accessKey',
[access_key]))
if LdapDriver.isadmin_attribute in user.keys():
attr.append((self.ldap.MOD_REPLACE,
LdapDriver.isadmin_attribute,
[str(is_admin).upper()]))
else:
attr.append((self.ldap.MOD_ADD, 'isAdmin', \
[str(is_admin).upper()]))
attr.append((self.ldap.MOD_ADD,
LdapDriver.isadmin_attribute,
[str(is_admin).upper()]))
self.conn.modify_s(self.__uid_to_dn(name), attr)
return self.get_user(name)
else:
@@ -168,12 +185,12 @@ class LdapDriver(object):
'inetOrgPerson',
'novaUser']),
('ou', [FLAGS.ldap_user_unit]),
('uid', [name]),
(FLAGS.ldap_user_id_attribute, [name]),
('sn', [name]),
('cn', [name]),
(FLAGS.ldap_user_name_attribute, [name]),
('secretKey', [secret_key]),
('accessKey', [access_key]),
('isAdmin', [str(is_admin).upper()]),
(LdapDriver.isadmin_attribute, [str(is_admin).upper()]),
]
self.conn.add_s(self.__uid_to_dn(name), attr)
return self.__to_user(dict(attr))
@@ -204,10 +221,10 @@ class LdapDriver(object):
if not manager_dn in members:
members.append(manager_dn)
attr = [
('objectclass', ['novaProject']),
('objectclass', [LdapDriver.project_objectclass]),
('cn', [name]),
('description', [description]),
('projectManager', [manager_dn]),
(LdapDriver.project_attribute, [manager_dn]),
('member', members)]
self.conn.add_s('cn=%s,%s' % (name, FLAGS.ldap_project_subtree), attr)
return self.__to_project(dict(attr))
@@ -223,7 +240,8 @@ class LdapDriver(object):
"manager %s doesn't exist")
% manager_uid)
manager_dn = self.__uid_to_dn(manager_uid)
attr.append((self.ldap.MOD_REPLACE, 'projectManager', manager_dn))
attr.append((self.ldap.MOD_REPLACE, LdapDriver.project_attribute,
manager_dn))
if description:
attr.append((self.ldap.MOD_REPLACE, 'description', description))
self.conn.modify_s('cn=%s,%s' % (project_id,
@@ -283,10 +301,9 @@ class LdapDriver(object):
return roles
else:
project_dn = 'cn=%s,%s' % (project_id, FLAGS.ldap_project_subtree)
roles = self.__find_objects(project_dn,
'(&(&(objectclass=groupOfNames)'
'(!(objectclass=novaProject)))'
'(member=%s))' % self.__uid_to_dn(uid))
query = ('(&(&(objectclass=groupOfNames)(!%s))(member=%s))' %
(LdapDriver.project_pattern, self.__uid_to_dn(uid)))
roles = self.__find_objects(project_dn, query)
return [role['cn'][0] for role in roles]
def delete_user(self, uid):
@@ -300,14 +317,15 @@ class LdapDriver(object):
# Retrieve user by name
user = self.__get_ldap_user(uid)
if 'secretKey' in user.keys():
attr.append((self.ldap.MOD_DELETE, 'secretKey', \
user['secretKey']))
attr.append((self.ldap.MOD_DELETE, 'secretKey',
user['secretKey']))
if 'accessKey' in user.keys():
attr.append((self.ldap.MOD_DELETE, 'accessKey', \
user['accessKey']))
if 'isAdmin' in user.keys():
attr.append((self.ldap.MOD_DELETE, 'isAdmin', \
user['isAdmin']))
attr.append((self.ldap.MOD_DELETE, 'accessKey',
user['accessKey']))
if LdapDriver.isadmin_attribute in user.keys():
attr.append((self.ldap.MOD_DELETE,
LdapDriver.isadmin_attribute,
user[LdapDriver.isadmin_attribute]))
self.conn.modify_s(self.__uid_to_dn(uid), attr)
else:
# Delete entry
@@ -329,7 +347,8 @@ class LdapDriver(object):
if secret_key:
attr.append((self.ldap.MOD_REPLACE, 'secretKey', secret_key))
if admin is not None:
attr.append((self.ldap.MOD_REPLACE, 'isAdmin', str(admin).upper()))
attr.append((self.ldap.MOD_REPLACE, LdapDriver.isadmin_attribute,
str(admin).upper()))
self.conn.modify_s(self.__uid_to_dn(uid), attr)
def __user_exists(self, uid):
@@ -347,7 +366,7 @@ class LdapDriver(object):
def __get_ldap_user(self, uid):
"""Retrieve LDAP user entry by id"""
attr = self.__find_object(self.__uid_to_dn(uid),
'(objectclass=novaUser)')
'(objectclass=novaUser)')
return attr
def __find_object(self, dn, query=None, scope=None):
@@ -383,19 +402,21 @@ class LdapDriver(object):
def __find_role_dns(self, tree):
"""Find dns of role objects in given tree"""
return self.__find_dns(tree,
'(&(objectclass=groupOfNames)(!(objectclass=novaProject)))')
query = ('(&(objectclass=groupOfNames)(!%s))' %
LdapDriver.project_pattern)
return self.__find_dns(tree, query)
def __find_group_dns_with_member(self, tree, uid):
"""Find dns of group objects in a given tree that contain member"""
dns = self.__find_dns(tree,
'(&(objectclass=groupOfNames)(member=%s))' %
self.__uid_to_dn(uid))
query = ('(&(objectclass=groupOfNames)(member=%s))' %
self.__uid_to_dn(uid))
dns = self.__find_dns(tree, query)
return dns
def __group_exists(self, dn):
"""Check if group exists"""
return self.__find_object(dn, '(objectclass=groupOfNames)') is not None
query = '(objectclass=groupOfNames)'
return self.__find_object(dn, query) is not None
@staticmethod
def __role_to_dn(role, project_id=None):
@@ -417,9 +438,9 @@ class LdapDriver(object):
if member_uids is not None:
for member_uid in member_uids:
if not self.__user_exists(member_uid):
raise exception.NotFound(_("Group can't be created "
"because user %s doesn't exist")
% member_uid)
raise exception.NotFound("Group can't be created "
"because user %s doesn't exist" %
member_uid)
members.append(self.__uid_to_dn(member_uid))
dn = self.__uid_to_dn(uid)
if not dn in members:
@@ -434,9 +455,8 @@ class LdapDriver(object):
def __is_in_group(self, uid, group_dn):
"""Check if user is in group"""
if not self.__user_exists(uid):
raise exception.NotFound(_("User %s can't be searched in group "
"because the user doesn't exist")
% uid)
raise exception.NotFound("User %s can't be searched in group "
"because the user doesn't exist" % uid)
if not self.__group_exists(group_dn):
return False
res = self.__find_object(group_dn,
@@ -447,12 +467,11 @@ class LdapDriver(object):
def __add_to_group(self, uid, group_dn):
"""Add user to group"""
if not self.__user_exists(uid):
raise exception.NotFound(_("User %s can't be added to the group "
"because the user doesn't exist")
% uid)
raise exception.NotFound("User %s can't be added to the group "
"because the user doesn't exist" % uid)
if not self.__group_exists(group_dn):
raise exception.NotFound(_("The group at dn %s doesn't exist")
% group_dn)
raise exception.NotFound("The group at dn %s doesn't exist" %
group_dn)
if self.__is_in_group(uid, group_dn):
raise exception.Duplicate(_("User %s is already a member of "
"the group %s") % (uid, group_dn))
@@ -462,18 +481,17 @@ class LdapDriver(object):
def __remove_from_group(self, uid, group_dn):
"""Remove user from group"""
if not self.__group_exists(group_dn):
raise exception.NotFound(_("The group at dn %s doesn't exist")
% group_dn)
raise exception.NotFound("The group at dn %s doesn't exist" %
group_dn)
if not self.__user_exists(uid):
raise exception.NotFound(_("User %s can't be removed from the "
"group because the user doesn't exist")
% uid)
raise exception.NotFound("User %s can't be removed from the "
"group because the user doesn't exist" %
uid)
if not self.__is_in_group(uid, group_dn):
raise exception.NotFound(_("User %s is not a member of the group")
% uid)
raise exception.NotFound("User %s is not a member of the group" %
uid)
# NOTE(vish): remove user from group and any sub_groups
sub_dns = self.__find_group_dns_with_member(
group_dn, uid)
sub_dns = self.__find_group_dns_with_member(group_dn, uid)
for sub_dn in sub_dns:
self.__safe_remove_from_group(uid, sub_dn)
@@ -491,9 +509,8 @@ class LdapDriver(object):
def __remove_from_all(self, uid):
"""Remove user from all roles and projects"""
if not self.__user_exists(uid):
raise exception.NotFound(_("User %s can't be removed from all "
"because the user doesn't exist")
% uid)
raise exception.NotFound("User %s can't be removed from all "
"because the user doesn't exist" % uid)
role_dns = self.__find_group_dns_with_member(
FLAGS.role_project_subtree, uid)
for role_dn in role_dns:
@@ -521,13 +538,13 @@ class LdapDriver(object):
if attr is None:
return None
if ('accessKey' in attr.keys() and 'secretKey' in attr.keys() \
and 'isAdmin' in attr.keys()):
and LdapDriver.isadmin_attribute in attr.keys()):
return {
'id': attr['uid'][0],
'name': attr['cn'][0],
'id': attr[FLAGS.ldap_user_id_attribute][0],
'name': attr[FLAGS.ldap_user_name_attribute][0],
'access': attr['accessKey'][0],
'secret': attr['secretKey'][0],
'admin': (attr['isAdmin'][0] == 'TRUE')}
'admin': (attr[LdapDriver.isadmin_attribute][0] == 'TRUE')}
else:
return None
@@ -539,7 +556,8 @@ class LdapDriver(object):
return {
'id': attr['cn'][0],
'name': attr['cn'][0],
'project_manager_id': self.__dn_to_uid(attr['projectManager'][0]),
'project_manager_id':
self.__dn_to_uid(attr[LdapDriver.project_attribute][0]),
'description': attr.get('description', [None])[0],
'member_ids': [self.__dn_to_uid(x) for x in member_dns]}
@@ -549,9 +567,10 @@ class LdapDriver(object):
return dn.split(',')[0].split('=')[1]
@staticmethod
def __uid_to_dn(dn):
def __uid_to_dn(uid):
"""Convert uid to dn"""
return 'uid=%s,%s' % (dn, FLAGS.ldap_user_subtree)
return (FLAGS.ldap_user_id_attribute + '=%s,%s'
% (uid, FLAGS.ldap_user_subtree))
class FakeLdapDriver(LdapDriver):

46
nova/auth/nova_openldap.schema
View File

@@ -1,7 +1,9 @@
#
# Person object for Nova
# inetorgperson with extra attributes
# Author: Vishvananda Ishaya <vishvananda@yahoo.com>
# Schema version: 2
# Authors: Vishvananda Ishaya <vishvananda@gmail.com>
# Ryan Lane <rlane@wikimedia.org>
#
#
@@ -30,55 +32,19 @@ attributetype (
SINGLE-VALUE
)
attributetype (
novaAttrs:3
NAME 'keyFingerprint'
DESC 'Fingerprint of private key'
EQUALITY caseIgnoreMatch
SUBSTR caseIgnoreSubstringsMatch
SYNTAX 1.3.6.1.4.1.1466.115.121.1.15
SINGLE-VALUE
)
attributetype (
novaAttrs:4
NAME 'isAdmin'
DESC 'Is user an administrator?'
NAME 'isNovaAdmin'
DESC 'Is user an nova administrator?'
EQUALITY booleanMatch
SYNTAX 1.3.6.1.4.1.1466.115.121.1.7
SINGLE-VALUE
)
attributetype (
novaAttrs:5
NAME 'projectManager'
DESC 'Project Managers of a project'
SYNTAX 1.3.6.1.4.1.1466.115.121.1.12
)
objectClass (
novaOCs:1
NAME 'novaUser'
DESC 'access and secret keys'
AUXILIARY
MUST ( uid )
MAY ( accessKey $ secretKey $ isAdmin )
)
objectClass (
novaOCs:2
NAME 'novaKeyPair'
DESC 'Key pair for User'
SUP top
STRUCTURAL
MUST ( cn $ sshPublicKey $ keyFingerprint )
)
objectClass (
novaOCs:3
NAME 'novaProject'
DESC 'Container for project'
SUP groupOfNames
STRUCTURAL
MUST ( cn $ projectManager )
MAY ( accessKey $ secretKey $ isNovaAdmin )
)

13
nova/auth/nova_sun.schema
View File

@@ -1,16 +1,13 @@
#
# Person object for Nova
# inetorgperson with extra attributes
# Author: Vishvananda Ishaya <vishvananda@yahoo.com>
# Modified for strict RFC 4512 compatibility by: Ryan Lane <ryan@ryandlane.com>
# Schema version: 2
# Authors: Vishvananda Ishaya <vishvananda@gmail.com>
# Ryan Lane <rlane@wikimedia.org>
#
# using internet experimental oid arc as per BP64 3.1
dn: cn=schema
attributeTypes: ( 1.3.6.1.3.1.666.666.3.1 NAME 'accessKey' DESC 'Key for accessing data' EQUALITY caseIgnoreMatch SUBSTR caseIgnoreSubstringsMatch SYNTAX 1.3.6.1.4.1.1466.115.121.1.15 SINGLE-VALUE )
attributeTypes: ( 1.3.6.1.3.1.666.666.3.2 NAME 'secretKey' DESC 'Secret key' EQUALITY caseIgnoreMatch SUBSTR caseIgnoreSubstringsMatch SYNTAX 1.3.6.1.4.1.1466.115.121.1.15 SINGLE-VALUE )
attributeTypes: ( 1.3.6.1.3.1.666.666.3.3 NAME 'keyFingerprint' DESC 'Fingerprint of private key' EQUALITY caseIgnoreMatch SUBSTR caseIgnoreSubstringsMatch SYNTAX 1.3.6.1.4.1.1466.115.121.1.15 SINGLE-VALUE)
attributeTypes: ( 1.3.6.1.3.1.666.666.3.4 NAME 'isAdmin' DESC 'Is user an administrator?' EQUALITY booleanMatch SYNTAX 1.3.6.1.4.1.1466.115.121.1.7 SINGLE-VALUE )
attributeTypes: ( 1.3.6.1.3.1.666.666.3.5 NAME 'projectManager' DESC 'Project Managers of a project' SYNTAX 1.3.6.1.4.1.1466.115.121.1.12 )
objectClasses: ( 1.3.6.1.3.1.666.666.4.1 NAME 'novaUser' DESC 'access and secret keys' SUP top AUXILIARY MUST ( uid ) MAY ( accessKey $ secretKey $ isAdmin ) )
objectClasses: ( 1.3.6.1.3.1.666.666.4.2 NAME 'novaKeyPair' DESC 'Key pair for User' SUP top STRUCTURAL MUST ( cn $ sshPublicKey $ keyFingerprint ) )
objectClasses: ( 1.3.6.1.3.1.666.666.4.3 NAME 'novaProject' DESC 'Container for project' SUP groupOfNames STRUCTURAL MUST ( cn $ projectManager ) )
attributeTypes: ( 1.3.6.1.3.1.666.666.3.4 NAME 'isNovaAdmin' DESC 'Is user a nova administrator?' EQUALITY booleanMatch SYNTAX 1.3.6.1.4.1.1466.115.121.1.7 SINGLE-VALUE )
objectClasses: ( 1.3.6.1.3.1.666.666.4.1 NAME 'novaUser' DESC 'access and secret keys' SUP top AUXILIARY MAY ( accessKey $ secretKey $ isNovaAdmin ) )

1
nova/auth/opendj.sh
View File

@@ -32,7 +32,6 @@ abspath=`dirname "$(cd "${0%/*}" 2>/dev/null; echo "$PWD"/"${0##*/}")"`
schemapath='/var/opendj/instance/config/schema'
cp $abspath/openssh-lpk_sun.schema $schemapath/97-openssh-lpk_sun.ldif
cp $abspath/nova_sun.schema $schemapath/98-nova_sun.ldif
chown opendj:opendj $schemapath/97-openssh-lpk_sun.ldif
chown opendj:opendj $schemapath/98-nova_sun.ldif
cat >/etc/ldap/ldap.conf <<LDAP_CONF_EOF

3
nova/auth/slap.sh
View File

@@ -22,7 +22,7 @@ apt-get install -y slapd ldap-utils python-ldap
abspath=`dirname "$(cd "${0%/*}" 2>/dev/null; echo "$PWD"/"${0##*/}")"`
cp $abspath/openssh-lpk_openldap.schema /etc/ldap/schema/openssh-lpk_openldap.schema
cp $abspath/nova_openldap.schema /etc/ldap/schema/nova_openldap.schema
cp $abspath/nova_openldap.schema /etc/ldap/schema/nova.schema
mv /etc/ldap/slapd.conf /etc/ldap/slapd.conf.orig
cat >/etc/ldap/slapd.conf <<SLAPD_CONF_EOF
@@ -33,7 +33,6 @@ cat >/etc/ldap/slapd.conf <<SLAPD_CONF_EOF
include /etc/ldap/schema/core.schema
include /etc/ldap/schema/cosine.schema
include /etc/ldap/schema/inetorgperson.schema
include /etc/ldap/schema/openssh-lpk_openldap.schema
include /etc/ldap/schema/nova.schema
pidfile /var/run/slapd/slapd.pid
argsfile /var/run/slapd/slapd.args

59
nova/fakememcache.py Normal file
View File

@@ -0,0 +1,59 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# 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.
"""Super simple fake memcache client."""
import utils
class Client(object):
"""Replicates a tiny subset of memcached client interface."""
def __init__(self, *args, **kwargs):
"""Ignores the passed in args"""
self.cache = {}
def get(self, key):
"""Retrieves the value for a key or None."""
(timeout, value) = self.cache.get(key, (0, None))
if timeout == 0 or utils.utcnow_ts() < timeout:
return value
return None
def set(self, key, value, time=0, min_compress_len=0):
"""Sets the value for a key."""
timeout = 0
if time != 0:
timeout = utils.utcnow_ts() + time
self.cache[key] = (timeout, value)
return True
def add(self, key, value, time=0, min_compress_len=0):
"""Sets the value for a key if it doesn't exist."""
if not self.get(key) is None:
return False
return self.set(key, value, time, min_compress_len)
def incr(self, key, delta=1):
"""Increments the value for a key."""
value = self.get(key)
if value is None:
return None
new_value = int(value) + delta
self.cache[key] = (self.cache[key][0], str(new_value))
return new_value

122
nova/fakerabbit.py
View File

@@ -25,6 +25,10 @@ from carrot.backends import base
from eventlet import greenthread
EXCHANGES = {}
QUEUES = {}
class Message(base.BaseMessage):
pass
@@ -68,81 +72,63 @@ class Queue(object):
return self._queue.get()
class Backend(object):
""" Singleton backend for testing """
class __impl(base.BaseBackend):
def __init__(self, *args, **kwargs):
#super(__impl, self).__init__(*args, **kwargs)
self._exchanges = {}
self._queues = {}
class Backend(base.BaseBackend):
def queue_declare(self, queue, **kwargs):
global QUEUES
if queue not in QUEUES:
logging.debug(_('Declaring queue %s'), queue)
QUEUES[queue] = Queue(queue)
def _reset_all(self):
self._exchanges = {}
self._queues = {}
def exchange_declare(self, exchange, type, *args, **kwargs):
global EXCHANGES
if exchange not in EXCHANGES:
logging.debug(_('Declaring exchange %s'), exchange)
EXCHANGES[exchange] = Exchange(exchange, type)
def queue_declare(self, queue, **kwargs):
if queue not in self._queues:
logging.debug(_('Declaring queue %s'), queue)
self._queues[queue] = Queue(queue)
def queue_bind(self, queue, exchange, routing_key, **kwargs):
global EXCHANGES
global QUEUES
logging.debug(_('Binding %s to %s with key %s'),
queue, exchange, routing_key)
EXCHANGES[exchange].bind(QUEUES[queue].push, routing_key)
def exchange_declare(self, exchange, type, *args, **kwargs):
if exchange not in self._exchanges:
logging.debug(_('Declaring exchange %s'), exchange)
self._exchanges[exchange] = Exchange(exchange, type)
def declare_consumer(self, queue, callback, *args, **kwargs):
self.current_queue = queue
self.current_callback = callback
def queue_bind(self, queue, exchange, routing_key, **kwargs):
logging.debug(_('Binding %s to %s with key %s'),
queue, exchange, routing_key)
self._exchanges[exchange].bind(self._queues[queue].push,
routing_key)
def consume(self, limit=None):
while True:
item = self.get(self.current_queue)
if item:
self.current_callback(item)
raise StopIteration()
greenthread.sleep(0)
def declare_consumer(self, queue, callback, *args, **kwargs):
self.current_queue = queue
self.current_callback = callback
def get(self, queue, no_ack=False):
global QUEUES
if not queue in QUEUES or not QUEUES[queue].size():
return None
(message_data, content_type, content_encoding) = QUEUES[queue].pop()
message = Message(backend=self, body=message_data,
content_type=content_type,
content_encoding=content_encoding)
message.result = True
logging.debug(_('Getting from %s: %s'), queue, message)
return message
def consume(self, *args, **kwargs):
while True:
item = self.get(self.current_queue)
if item:
self.current_callback(item)
raise StopIteration()
greenthread.sleep(0)
def prepare_message(self, message_data, delivery_mode,
content_type, content_encoding, **kwargs):
"""Prepare message for sending."""
return (message_data, content_type, content_encoding)
def get(self, queue, no_ack=False):
if not queue in self._queues or not self._queues[queue].size():
return None
(message_data, content_type, content_encoding) = \
self._queues[queue].pop()
message = Message(backend=self, body=message_data,
content_type=content_type,
content_encoding=content_encoding)
message.result = True
logging.debug(_('Getting from %s: %s'), queue, message)
return message
def prepare_message(self, message_data, delivery_mode,
content_type, content_encoding, **kwargs):
"""Prepare message for sending."""
return (message_data, content_type, content_encoding)
def publish(self, message, exchange, routing_key, **kwargs):
if exchange in self._exchanges:
self._exchanges[exchange].publish(
message, routing_key=routing_key)
__instance = None
def __init__(self, *args, **kwargs):
if Backend.__instance is None:
Backend.__instance = Backend.__impl(*args, **kwargs)
self.__dict__['_Backend__instance'] = Backend.__instance
def __getattr__(self, attr):
return getattr(self.__instance, attr)
def __setattr__(self, attr, value):
return setattr(self.__instance, attr, value)
def publish(self, message, exchange, routing_key, **kwargs):
global EXCHANGES
if exchange in EXCHANGES:
EXCHANGES[exchange].publish(message, routing_key=routing_key)
def reset_all():
Backend()._reset_all()
global EXCHANGES
global QUEUES
EXCHANGES = {}
QUEUES = {}

2
nova/rpc.py
View File

@@ -245,7 +245,7 @@ def msg_reply(msg_id, reply=None, failure=None):
logging.error(_("Returning exception %s to caller"), message)
logging.error(tb)
failure = (failure[0].__name__, str(failure[1]), tb)
conn = Connection.instance()
conn = Connection.instance(True)
publisher = DirectPublisher(connection=conn, msg_id=msg_id)
try:
publisher.send({'result': reply, 'failure': failure})

54
nova/tests/api_integration.py
View File

@@ -1,54 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# 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.
import boto
from boto.ec2.regioninfo import RegionInfo
import unittest
ACCESS_KEY = 'fake'
SECRET_KEY = 'fake'
CLC_IP = '127.0.0.1'
CLC_PORT = 8773
REGION = 'test'
def get_connection():
return boto.connect_ec2(
aws_access_key_id=ACCESS_KEY,
aws_secret_access_key=SECRET_KEY,
is_secure=False,
region=RegionInfo(None, REGION, CLC_IP),
port=CLC_PORT,
path='/services/Cloud',
debug=99)
class APIIntegrationTests(unittest.TestCase):
def test_001_get_all_images(self):
conn = get_connection()
res = conn.get_all_images()
if __name__ == '__main__':
unittest.main()
#print conn.get_all_key_pairs()
#print conn.create_key_pair
#print conn.create_security_group('name', 'description')

153
nova/tests/quota_unittest.py
View File

@@ -1,153 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# 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.
import logging
from nova import context
from nova import db
from nova import exception
from nova import flags
from nova import quota
from nova import test
from nova import utils
from nova.auth import manager
from nova.api.ec2 import cloud
FLAGS = flags.FLAGS
class QuotaTestCase(test.TestCase):
def setUp(self):
logging.getLogger().setLevel(logging.DEBUG)
super(QuotaTestCase, self).setUp()
self.flags(connection_type='fake',
quota_instances=2,
quota_cores=4,
quota_volumes=2,
quota_gigabytes=20,
quota_floating_ips=1)
self.cloud = cloud.CloudController()
self.manager = manager.AuthManager()
self.user = self.manager.create_user('admin', 'admin', 'admin', True)
self.project = self.manager.create_project('admin', 'admin', 'admin')
self.network = utils.import_object(FLAGS.network_manager)
self.context = context.RequestContext(project=self.project,
user=self.user)
def tearDown(self):
manager.AuthManager().delete_project(self.project)
manager.AuthManager().delete_user(self.user)
super(QuotaTestCase, self).tearDown()
def _create_instance(self, cores=2):
"""Create a test instance"""
inst = {}
inst['image_id'] = 'ami-test'
inst['reservation_id'] = 'r-fakeres'
inst['user_id'] = self.user.id
inst['project_id'] = self.project.id
inst['instance_type'] = 'm1.large'
inst['vcpus'] = cores
inst['mac_address'] = utils.generate_mac()
return db.instance_create(self.context, inst)['id']
def _create_volume(self, size=10):
"""Create a test volume"""
vol = {}
vol['user_id'] = self.user.id
vol['project_id'] = self.project.id
vol['size'] = size
return db.volume_create(self.context, vol)['id']
def test_quota_overrides(self):
"""Make sure overriding a projects quotas works"""
num_instances = quota.allowed_instances(self.context, 100, 'm1.small')
self.assertEqual(num_instances, 2)
db.quota_create(self.context, {'project_id': self.project.id,
'instances': 10})
num_instances = quota.allowed_instances(self.context, 100, 'm1.small')
self.assertEqual(num_instances, 4)
db.quota_update(self.context, self.project.id, {'cores': 100})
num_instances = quota.allowed_instances(self.context, 100, 'm1.small')
self.assertEqual(num_instances, 10)
db.quota_destroy(self.context, self.project.id)
def test_too_many_instances(self):
instance_ids = []
for i in range(FLAGS.quota_instances):
instance_id = self._create_instance()
instance_ids.append(instance_id)
self.assertRaises(quota.QuotaError, self.cloud.run_instances,
self.context,
min_count=1,
max_count=1,
instance_type='m1.small',
image_id='fake')
for instance_id in instance_ids:
db.instance_destroy(self.context, instance_id)
def test_too_many_cores(self):
instance_ids = []
instance_id = self._create_instance(cores=4)
instance_ids.append(instance_id)
self.assertRaises(quota.QuotaError, self.cloud.run_instances,
self.context,
min_count=1,
max_count=1,
instance_type='m1.small',
image_id='fake')
for instance_id in instance_ids:
db.instance_destroy(self.context, instance_id)
def test_too_many_volumes(self):
volume_ids = []
for i in range(FLAGS.quota_volumes):
volume_id = self._create_volume()
volume_ids.append(volume_id)
self.assertRaises(quota.QuotaError, self.cloud.create_volume,
self.context,
size=10)
for volume_id in volume_ids:
db.volume_destroy(self.context, volume_id)
def test_too_many_gigabytes(self):
volume_ids = []
volume_id = self._create_volume(size=20)
volume_ids.append(volume_id)
self.assertRaises(quota.QuotaError,
self.cloud.create_volume,
self.context,
size=10)
for volume_id in volume_ids:
db.volume_destroy(self.context, volume_id)
def test_too_many_addresses(self):
address = '192.168.0.100'
db.floating_ip_create(context.get_admin_context(),
{'address': address, 'host': FLAGS.host})
float_addr = self.network.allocate_floating_ip(self.context,
self.project.id)
# NOTE(vish): This assert never fails. When cloud attempts to
# make an rpc.call, the test just finishes with OK. It
# appears to be something in the magic inline callbacks
# that is breaking.
self.assertRaises(quota.QuotaError, self.cloud.allocate_address,
self.context)
db.floating_ip_destroy(context.get_admin_context(), address)

227
nova/tests/service_unittest.py
View File

@@ -1,227 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# 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.
"""
Unit Tests for remote procedure calls using queue
"""
import mox
from nova import exception
from nova import flags
from nova import rpc
from nova import test
from nova import service
from nova import manager
FLAGS = flags.FLAGS
flags.DEFINE_string("fake_manager", "nova.tests.service_unittest.FakeManager",
"Manager for testing")
class FakeManager(manager.Manager):
"""Fake manager for tests"""
def test_method(self):
return 'manager'
class ExtendedService(service.Service):
def test_method(self):
return 'service'
class ServiceManagerTestCase(test.TestCase):
"""Test cases for Services"""
def test_attribute_error_for_no_manager(self):
serv = service.Service('test',
'test',
'test',
'nova.tests.service_unittest.FakeManager')
self.assertRaises(AttributeError, getattr, serv, 'test_method')
def test_message_gets_to_manager(self):
serv = service.Service('test',
'test',
'test',
'nova.tests.service_unittest.FakeManager')
serv.start()
self.assertEqual(serv.test_method(), 'manager')
def test_override_manager_method(self):
serv = ExtendedService('test',
'test',
'test',
'nova.tests.service_unittest.FakeManager')
serv.start()
self.assertEqual(serv.test_method(), 'service')
class ServiceTestCase(test.TestCase):
"""Test cases for Services"""
def setUp(self):
super(ServiceTestCase, self).setUp()
self.mox.StubOutWithMock(service, 'db')
def test_create(self):
host = 'foo'
binary = 'nova-fake'
topic = 'fake'
# NOTE(vish): Create was moved out of mox replay to make sure that
# the looping calls are created in StartService.
app = service.Service.create(host=host, binary=binary)
self.mox.StubOutWithMock(rpc,
'AdapterConsumer',
use_mock_anything=True)
rpc.AdapterConsumer(connection=mox.IgnoreArg(),
topic=topic,
proxy=mox.IsA(service.Service)).AndReturn(
rpc.AdapterConsumer)
rpc.AdapterConsumer(connection=mox.IgnoreArg(),
topic='%s.%s' % (topic, host),
proxy=mox.IsA(service.Service)).AndReturn(
rpc.AdapterConsumer)
rpc.AdapterConsumer.attach_to_eventlet()
rpc.AdapterConsumer.attach_to_eventlet()
service_create = {'host': host,
'binary': binary,
'topic': topic,
'report_count': 0}
service_ref = {'host': host,
'binary': binary,
'report_count': 0,
'id': 1}
service.db.service_get_by_args(mox.IgnoreArg(),
host,
binary).AndRaise(exception.NotFound())
service.db.service_create(mox.IgnoreArg(),
service_create).AndReturn(service_ref)
self.mox.ReplayAll()
app.start()
app.stop()
self.assert_(app)
# We're testing sort of weird behavior in how report_state decides
# whether it is disconnected, it looks for a variable on itself called
# 'model_disconnected' and report_state doesn't really do much so this
# these are mostly just for coverage
def test_report_state_no_service(self):
host = 'foo'
binary = 'bar'
topic = 'test'
service_create = {'host': host,
'binary': binary,
'topic': topic,
'report_count': 0}
service_ref = {'host': host,
'binary': binary,
'topic': topic,
'report_count': 0,
'id': 1}
service.db.service_get_by_args(mox.IgnoreArg(),
host,
binary).AndRaise(exception.NotFound())
service.db.service_create(mox.IgnoreArg(),
service_create).AndReturn(service_ref)
service.db.service_get(mox.IgnoreArg(),
service_ref['id']).AndReturn(service_ref)
service.db.service_update(mox.IgnoreArg(), service_ref['id'],
mox.ContainsKeyValue('report_count', 1))
self.mox.ReplayAll()
serv = service.Service(host,
binary,
topic,
'nova.tests.service_unittest.FakeManager')
serv.start()
serv.report_state()
def test_report_state_newly_disconnected(self):
host = 'foo'
binary = 'bar'
topic = 'test'
service_create = {'host': host,
'binary': binary,
'topic': topic,
'report_count': 0}
service_ref = {'host': host,
'binary': binary,
'topic': topic,
'report_count': 0,
'id': 1}
service.db.service_get_by_args(mox.IgnoreArg(),
host,
binary).AndRaise(exception.NotFound())
service.db.service_create(mox.IgnoreArg(),
service_create).AndReturn(service_ref)
service.db.service_get(mox.IgnoreArg(),
mox.IgnoreArg()).AndRaise(Exception())
self.mox.ReplayAll()
serv = service.Service(host,
binary,
topic,
'nova.tests.service_unittest.FakeManager')
serv.start()
serv.report_state()
self.assert_(serv.model_disconnected)
def test_report_state_newly_connected(self):
host = 'foo'
binary = 'bar'
topic = 'test'
service_create = {'host': host,
'binary': binary,
'topic': topic,
'report_count': 0}
service_ref = {'host': host,
'binary': binary,
'topic': topic,
'report_count': 0,
'id': 1}
service.db.service_get_by_args(mox.IgnoreArg(),
host,
binary).AndRaise(exception.NotFound())
service.db.service_create(mox.IgnoreArg(),
service_create).AndReturn(service_ref)
service.db.service_get(mox.IgnoreArg(),
service_ref['id']).AndReturn(service_ref)
service.db.service_update(mox.IgnoreArg(), service_ref['id'],
mox.ContainsKeyValue('report_count', 1))
self.mox.ReplayAll()
serv = service.Service(host,
binary,
topic,
'nova.tests.service_unittest.FakeManager')
serv.start()
serv.model_disconnected = True
serv.report_state()
self.assert_(not serv.model_disconnected)

0
nova/tests/access_unittest.py → nova/tests/test_access.py
View File

0
nova/tests/api_unittest.py → nova/tests/test_api.py
View File

0
nova/tests/auth_unittest.py → nova/tests/test_auth.py
View File

30
nova/tests/cloud_unittest.py → nova/tests/test_cloud.py
View File

@@ -22,20 +22,18 @@ import logging
from M2Crypto import BIO
from M2Crypto import RSA
import os
import StringIO
import tempfile
import time
from eventlet import greenthread
from xml.etree import ElementTree
from nova import context
from nova import crypto
from nova import db
from nova import flags
from nova import rpc
from nova import service
from nova import test
from nova import utils
from nova.auth import manager
from nova.compute import power_state
from nova.api.ec2 import cloud
@@ -54,7 +52,8 @@ os.makedirs(IMAGES_PATH)
class CloudTestCase(test.TestCase):
def setUp(self):
super(CloudTestCase, self).setUp()
self.flags(connection_type='fake', images_path=IMAGES_PATH)
self.flags(connection_type='fake',
images_path=IMAGES_PATH)
self.conn = rpc.Connection.instance()
logging.getLogger().setLevel(logging.DEBUG)
@@ -62,27 +61,23 @@ class CloudTestCase(test.TestCase):
# set up our cloud
self.cloud = cloud.CloudController()
# set up a service
self.compute = utils.import_object(FLAGS.compute_manager)
self.compute_consumer = rpc.AdapterConsumer(connection=self.conn,
topic=FLAGS.compute_topic,
proxy=self.compute)
self.compute_consumer.attach_to_eventlet()
self.network = utils.import_object(FLAGS.network_manager)
self.network_consumer = rpc.AdapterConsumer(connection=self.conn,
topic=FLAGS.network_topic,
proxy=self.network)
self.network_consumer.attach_to_eventlet()
# set up services
self.compute = service.Service.create(binary='nova-compute')
self.compute.start()
self.network = service.Service.create(binary='nova-network')
self.network.start()
self.manager = manager.AuthManager()
self.user = self.manager.create_user('admin', 'admin', 'admin', True)
self.project = self.manager.create_project('proj', 'admin', 'proj')
self.context = context.RequestContext(user=self.user,
project=self.project)
project=self.project)
def tearDown(self):
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
self.compute.kill()
self.network.kill()
super(CloudTestCase, self).tearDown()
def _create_key(self, name):
@@ -109,12 +104,13 @@ class CloudTestCase(test.TestCase):
{'address': address,
'host': FLAGS.host})
self.cloud.allocate_address(self.context)
inst = db.instance_create(self.context, {})
inst = db.instance_create(self.context, {'host': FLAGS.host})
fixed = self.network.allocate_fixed_ip(self.context, inst['id'])
ec2_id = cloud.internal_id_to_ec2_id(inst['internal_id'])
self.cloud.associate_address(self.context,
instance_id=ec2_id,
public_ip=address)
greenthread.sleep(0.3)
self.cloud.disassociate_address(self.context,
public_ip=address)
self.cloud.release_address(self.context,

1
nova/tests/compute_unittest.py → nova/tests/test_compute.py
View File

@@ -41,6 +41,7 @@ class ComputeTestCase(test.TestCase):
logging.getLogger().setLevel(logging.DEBUG)
super(ComputeTestCase, self).setUp()
self.flags(connection_type='fake',
stub_network=True,
network_manager='nova.network.manager.FlatManager')
self.compute = utils.import_object(FLAGS.compute_manager)
self.compute_api = compute_api.ComputeAPI()

0
nova/tests/flags_unittest.py → nova/tests/test_flags.py
View File

86
nova/tests/test_middleware.py Normal file
View File

@@ -0,0 +1,86 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# 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.
import datetime
import webob
import webob.dec
import webob.exc
from nova.api import ec2
from nova import flags
from nova import test
from nova import utils
FLAGS = flags.FLAGS
@webob.dec.wsgify
def conditional_forbid(req):
"""Helper wsgi app returns 403 if param 'die' is 1."""
if 'die' in req.params and req.params['die'] == '1':
raise webob.exc.HTTPForbidden()
return 'OK'
class LockoutTestCase(test.TrialTestCase):
"""Test case for the Lockout middleware."""
def setUp(self): # pylint: disable-msg=C0103
super(LockoutTestCase, self).setUp()
utils.set_time_override()
self.lockout = ec2.Lockout(conditional_forbid)
def tearDown(self): # pylint: disable-msg=C0103
utils.clear_time_override()
super(LockoutTestCase, self).tearDown()
def _send_bad_attempts(self, access_key, num_attempts=1):
"""Fail x."""
for i in xrange(num_attempts):
req = webob.Request.blank('/?AWSAccessKeyId=%s&die=1' % access_key)
self.assertEqual(req.get_response(self.lockout).status_int, 403)
def _is_locked_out(self, access_key):
"""Sends a test request to see if key is locked out."""
req = webob.Request.blank('/?AWSAccessKeyId=%s' % access_key)
return (req.get_response(self.lockout).status_int == 403)
def test_lockout(self):
self._send_bad_attempts('test', FLAGS.lockout_attempts)
self.assertTrue(self._is_locked_out('test'))
def test_timeout(self):
self._send_bad_attempts('test', FLAGS.lockout_attempts)
self.assertTrue(self._is_locked_out('test'))
utils.advance_time_seconds(FLAGS.lockout_minutes * 60)
self.assertFalse(self._is_locked_out('test'))
def test_multiple_keys(self):
self._send_bad_attempts('test1', FLAGS.lockout_attempts)
self.assertTrue(self._is_locked_out('test1'))
self.assertFalse(self._is_locked_out('test2'))
utils.advance_time_seconds(FLAGS.lockout_minutes * 60)
self.assertFalse(self._is_locked_out('test1'))
self.assertFalse(self._is_locked_out('test2'))
def test_window_timeout(self):
self._send_bad_attempts('test', FLAGS.lockout_attempts - 1)
self.assertFalse(self._is_locked_out('test'))
utils.advance_time_seconds(FLAGS.lockout_window * 60)
self._send_bad_attempts('test', FLAGS.lockout_attempts - 1)
self.assertFalse(self._is_locked_out('test'))

8
nova/tests/misc_unittest.py → nova/tests/test_misc.py
View File

@@ -22,13 +22,13 @@ from nova.utils import parse_mailmap, str_dict_replace
class ProjectTestCase(test.TestCase):
def test_authors_up_to_date(self):
if os.path.exists('../.bzr'):
if os.path.exists('.bzr'):
contributors = set()
mailmap = parse_mailmap('../.mailmap')
mailmap = parse_mailmap('.mailmap')
import bzrlib.workingtree
tree = bzrlib.workingtree.WorkingTree.open('..')
tree = bzrlib.workingtree.WorkingTree.open('.')
tree.lock_read()
try:
parents = tree.get_parent_ids()
@@ -42,7 +42,7 @@ class ProjectTestCase(test.TestCase):
email = author.split(' ')[-1]
contributors.add(str_dict_replace(email, mailmap))
authors_file = open('../Authors', 'r').read()
authors_file = open('Authors', 'r').read()
missing = set()
for contributor in contributors:

8
nova/tests/network_unittest.py → nova/tests/test_network.py
View File

@@ -26,6 +26,7 @@ from nova import context
from nova import db
from nova import exception
from nova import flags
from nova import service
from nova import test
from nova import utils
from nova.auth import manager
@@ -40,6 +41,7 @@ class NetworkTestCase(test.TestCase):
# NOTE(vish): if you change these flags, make sure to change the
# flags in the corresponding section in nova-dhcpbridge
self.flags(connection_type='fake',
fake_call=True,
fake_network=True,
network_size=16,
num_networks=5)
@@ -56,16 +58,13 @@ class NetworkTestCase(test.TestCase):
# create the necessary network data for the project
user_context = context.RequestContext(project=self.projects[i],
user=self.user)
network_ref = self.network.get_network(user_context)
self.network.set_network_host(context.get_admin_context(),
network_ref['id'])
host = self.network.get_network_host(user_context.elevated())
instance_ref = self._create_instance(0)
self.instance_id = instance_ref['id']
instance_ref = self._create_instance(1)
self.instance2_id = instance_ref['id']
def tearDown(self):
super(NetworkTestCase, self).tearDown()
# TODO(termie): this should really be instantiating clean datastores
# in between runs, one failure kills all the tests
db.instance_destroy(context.get_admin_context(), self.instance_id)
@@ -73,6 +72,7 @@ class NetworkTestCase(test.TestCase):
for project in self.projects:
self.manager.delete_project(project)
self.manager.delete_user(self.user)
super(NetworkTestCase, self).tearDown()
def _create_instance(self, project_num, mac=None):
if not mac:

29
nova/tests/rpc_unittest.py → nova/tests/test_rpc.py
View File

@@ -33,7 +33,7 @@ class RpcTestCase(test.TestCase):
"""Test cases for rpc"""
def setUp(self):
super(RpcTestCase, self).setUp()
self.conn = rpc.Connection.instance()
self.conn = rpc.Connection.instance(True)
self.receiver = TestReceiver()
self.consumer = rpc.AdapterConsumer(connection=self.conn,
topic='test',
@@ -79,6 +79,33 @@ class RpcTestCase(test.TestCase):
except rpc.RemoteError as exc:
self.assertEqual(int(exc.value), value)
def test_nested_calls(self):
"""Test that we can do an rpc.call inside another call"""
class Nested(object):
@staticmethod
def echo(context, queue, value):
"""Calls echo in the passed queue"""
logging.debug("Nested received %s, %s", queue, value)
ret = rpc.call(context,
queue,
{"method": "echo",
"args": {"value": value}})
logging.debug("Nested return %s", ret)
return value
nested = Nested()
conn = rpc.Connection.instance(True)
consumer = rpc.AdapterConsumer(connection=conn,
topic='nested',
proxy=nested)
consumer.attach_to_eventlet()
value = 42
result = rpc.call(self.context,
'nested', {"method": "echo",
"args": {"queue": "test",
"value": value}})
self.assertEqual(value, result)
class TestReceiver(object):
"""Simple Proxy class so the consumer has methods to call

3
nova/tests/scheduler_unittest.py → nova/tests/test_scheduler.py
View File

@@ -48,7 +48,7 @@ class SchedulerTestCase(test.TestCase):
"""Test case for scheduler"""
def setUp(self):
super(SchedulerTestCase, self).setUp()
self.flags(scheduler_driver='nova.tests.scheduler_unittest.TestDriver')
self.flags(scheduler_driver='nova.tests.test_scheduler.TestDriver')
def test_fallback(self):
scheduler = manager.SchedulerManager()
@@ -78,6 +78,7 @@ class SimpleDriverTestCase(test.TestCase):
def setUp(self):
super(SimpleDriverTestCase, self).setUp()
self.flags(connection_type='fake',
stub_network=True,
max_cores=4,
max_gigabytes=4,
network_manager='nova.network.manager.FlatManager',

0
nova/tests/twistd_unittest.py → nova/tests/test_twistd.py
View File

38
nova/tests/virt_unittest.py → nova/tests/test_virt.py
View File

@@ -33,6 +33,7 @@ flags.DECLARE('instances_path', 'nova.compute.manager')
class LibvirtConnTestCase(test.TestCase):
def setUp(self):
super(LibvirtConnTestCase, self).setUp()
self.flags(fake_call=True)
self.manager = manager.AuthManager()
self.user = self.manager.create_user('fake', 'fake', 'fake',
admin=True)
@@ -52,45 +53,43 @@ class LibvirtConnTestCase(test.TestCase):
def test_xml_and_uri_no_ramdisk_no_kernel(self):
instance_data = dict(self.test_instance)
self.do_test_xml_and_uri(instance_data,
expect_kernel=False, expect_ramdisk=False)
self._check_xml_and_uri(instance_data,
expect_kernel=False, expect_ramdisk=False)
def test_xml_and_uri_no_ramdisk(self):
instance_data = dict(self.test_instance)
instance_data['kernel_id'] = 'aki-deadbeef'
self.do_test_xml_and_uri(instance_data,
expect_kernel=True, expect_ramdisk=False)
self._check_xml_and_uri(instance_data,
expect_kernel=True, expect_ramdisk=False)
def test_xml_and_uri_no_kernel(self):
instance_data = dict(self.test_instance)
instance_data['ramdisk_id'] = 'ari-deadbeef'
self.do_test_xml_and_uri(instance_data,
expect_kernel=False, expect_ramdisk=False)
self._check_xml_and_uri(instance_data,
expect_kernel=False, expect_ramdisk=False)
def test_xml_and_uri(self):
instance_data = dict(self.test_instance)
instance_data['ramdisk_id'] = 'ari-deadbeef'
instance_data['kernel_id'] = 'aki-deadbeef'
self.do_test_xml_and_uri(instance_data,
expect_kernel=True, expect_ramdisk=True)
self._check_xml_and_uri(instance_data,
expect_kernel=True, expect_ramdisk=True)
def test_xml_and_uri_rescue(self):
instance_data = dict(self.test_instance)
instance_data['ramdisk_id'] = 'ari-deadbeef'
instance_data['kernel_id'] = 'aki-deadbeef'
self.do_test_xml_and_uri(instance_data,
expect_kernel=True, expect_ramdisk=True,
rescue=True)
self._check_xml_and_uri(instance_data, expect_kernel=True,
expect_ramdisk=True, rescue=True)
def do_test_xml_and_uri(self, instance,
expect_ramdisk, expect_kernel,
rescue=False):
def _check_xml_and_uri(self, instance, expect_ramdisk, expect_kernel,
rescue=False):
user_context = context.RequestContext(project=self.project,
user=self.user)
instance_ref = db.instance_create(user_context, instance)
network_ref = self.network.get_network(user_context)
self.network.set_network_host(context.get_admin_context(),
network_ref['id'])
host = self.network.get_network_host(user_context.elevated())
network_ref = db.project_get_network(context.get_admin_context(),
self.project.id)
fixed_ip = {'address': self.test_ip,
'network_id': network_ref['id']}
@@ -158,7 +157,6 @@ class LibvirtConnTestCase(test.TestCase):
(lambda t: t.find('./devices/serial/source').get(
'path').split('/')[1], 'console.log'),
(lambda t: t.find('./memory').text, '2097152')]
if rescue:
common_checks += [
(lambda t: t.findall('./devices/disk/source')[0].get(
@@ -338,7 +336,7 @@ class NWFilterTestCase(test.TestCase):
self.security_group.id)
instance = db.instance_get(self.context, inst_id)
d = self.fw.setup_nwfilters_for_instance(instance)
self.fw.setup_base_nwfilters()
self.fw.setup_nwfilters_for_instance(instance)
_ensure_all_called()
self.teardown_security_group()
return d

0
nova/tests/volume_unittest.py → nova/tests/test_volume.py
View File

220
nova/tests/test_xenapi.py Normal file
View File

@@ -0,0 +1,220 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright (c) 2010 Citrix Systems, Inc.
#
# 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 suite for XenAPI
"""
import stubout
from nova import db
from nova import context
from nova import flags
from nova import test
from nova import utils
from nova.auth import manager
from nova.compute import instance_types
from nova.compute import power_state
from nova.virt import xenapi_conn
from nova.virt.xenapi import fake
from nova.virt.xenapi import volume_utils
from nova.tests.db import fakes
from nova.tests.xenapi import stubs
FLAGS = flags.FLAGS
class XenAPIVolumeTestCase(test.TestCase):
"""
Unit tests for Volume operations
"""
def setUp(self):
super(XenAPIVolumeTestCase, self).setUp()
self.stubs = stubout.StubOutForTesting()
FLAGS.target_host = '127.0.0.1'
FLAGS.xenapi_connection_url = 'test_url'
FLAGS.xenapi_connection_password = 'test_pass'
fakes.stub_out_db_instance_api(self.stubs)
stubs.stub_out_get_target(self.stubs)
fake.reset()
self.values = {'name': 1, 'id': 1,
'project_id': 'fake',
'user_id': 'fake',
'image_id': 1,
'kernel_id': 2,
'ramdisk_id': 3,
'instance_type': 'm1.large',
'mac_address': 'aa:bb:cc:dd:ee:ff',
}
def _create_volume(self, size='0'):
"""Create a volume object."""
vol = {}
vol['size'] = size
vol['user_id'] = 'fake'
vol['project_id'] = 'fake'
vol['host'] = 'localhost'
vol['availability_zone'] = FLAGS.storage_availability_zone
vol['status'] = "creating"
vol['attach_status'] = "detached"
return db.volume_create(context.get_admin_context(), vol)
def test_create_iscsi_storage(self):
""" This shows how to test helper classes' methods """
stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests)
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
helper = volume_utils.VolumeHelper
helper.XenAPI = session.get_imported_xenapi()
vol = self._create_volume()
info = helper.parse_volume_info(vol['ec2_id'], '/dev/sdc')
label = 'SR-%s' % vol['ec2_id']
description = 'Test-SR'
sr_ref = helper.create_iscsi_storage(session, info, label, description)
srs = fake.get_all('SR')
self.assertEqual(sr_ref, srs[0])
db.volume_destroy(context.get_admin_context(), vol['id'])
def test_parse_volume_info_raise_exception(self):
""" This shows how to test helper classes' methods """
stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests)
session = xenapi_conn.XenAPISession('test_url', 'root', 'test_pass')
helper = volume_utils.VolumeHelper
helper.XenAPI = session.get_imported_xenapi()
vol = self._create_volume()
# oops, wrong mount point!
self.assertRaises(volume_utils.StorageError,
helper.parse_volume_info,
vol['ec2_id'],
'/dev/sd')
db.volume_destroy(context.get_admin_context(), vol['id'])
def test_attach_volume(self):
""" This shows how to test Ops classes' methods """
stubs.stubout_session(self.stubs, stubs.FakeSessionForVolumeTests)
conn = xenapi_conn.get_connection(False)
volume = self._create_volume()
instance = db.instance_create(self.values)
fake.create_vm(instance.name, 'Running')
result = conn.attach_volume(instance.name, volume['ec2_id'],
'/dev/sdc')
def check():
# check that the VM has a VBD attached to it
# Get XenAPI reference for the VM
vms = fake.get_all('VM')
# Get XenAPI record for VBD
vbds = fake.get_all('VBD')
vbd = fake.get_record('VBD', vbds[0])
vm_ref = vbd['VM']
self.assertEqual(vm_ref, vms[0])
check()
def test_attach_volume_raise_exception(self):
""" This shows how to test when exceptions are raised """
stubs.stubout_session(self.stubs,
stubs.FakeSessionForVolumeFailedTests)
conn = xenapi_conn.get_connection(False)
volume = self._create_volume()
instance = db.instance_create(self.values)
fake.create_vm(instance.name, 'Running')
self.assertRaises(Exception,
conn.attach_volume,
instance.name,
volume['ec2_id'],
'/dev/sdc')
def tearDown(self):
super(XenAPIVolumeTestCase, self).tearDown()
self.stubs.UnsetAll()
class XenAPIVMTestCase(test.TestCase):
"""
Unit tests for VM operations
"""
def setUp(self):
super(XenAPIVMTestCase, self).setUp()
self.manager = manager.AuthManager()
self.user = self.manager.create_user('fake', 'fake', 'fake',
admin=True)
self.project = self.manager.create_project('fake', 'fake', 'fake')
self.network = utils.import_object(FLAGS.network_manager)
self.stubs = stubout.StubOutForTesting()
FLAGS.xenapi_connection_url = 'test_url'
FLAGS.xenapi_connection_password = 'test_pass'
fake.reset()
fakes.stub_out_db_instance_api(self.stubs)
fake.create_network('fake', FLAGS.flat_network_bridge)
def test_list_instances_0(self):
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
conn = xenapi_conn.get_connection(False)
instances = conn.list_instances()
self.assertEquals(instances, [])
def test_spawn(self):
stubs.stubout_session(self.stubs, stubs.FakeSessionForVMTests)
values = {'name': 1, 'id': 1,
'project_id': self.project.id,
'user_id': self.user.id,
'image_id': 1,
'kernel_id': 2,
'ramdisk_id': 3,
'instance_type': 'm1.large',
'mac_address': 'aa:bb:cc:dd:ee:ff',
}
conn = xenapi_conn.get_connection(False)
instance = db.instance_create(values)
conn.spawn(instance)
def check():
instances = conn.list_instances()
self.assertEquals(instances, [1])
# Get Nova record for VM
vm_info = conn.get_info(1)
# Get XenAPI record for VM
vms = fake.get_all('VM')
vm = fake.get_record('VM', vms[0])
# Check that m1.large above turned into the right thing.
instance_type = instance_types.INSTANCE_TYPES['m1.large']
mem_kib = long(instance_type['memory_mb']) << 10
mem_bytes = str(mem_kib << 10)
vcpus = instance_type['vcpus']
self.assertEquals(vm_info['max_mem'], mem_kib)
self.assertEquals(vm_info['mem'], mem_kib)
self.assertEquals(vm['memory_static_max'], mem_bytes)
self.assertEquals(vm['memory_dynamic_max'], mem_bytes)
self.assertEquals(vm['memory_dynamic_min'], mem_bytes)
self.assertEquals(vm['VCPUs_max'], str(vcpus))
self.assertEquals(vm['VCPUs_at_startup'], str(vcpus))
# Check that the VM is running according to Nova
self.assertEquals(vm_info['state'], power_state.RUNNING)
# Check that the VM is running according to XenAPI.
self.assertEquals(vm['power_state'], 'Running')
check()
def tearDown(self):
super(XenAPIVMTestCase, self).tearDown()
self.manager.delete_project(self.project)
self.manager.delete_user(self.user)
self.stubs.UnsetAll()

20
nova/tests/xenapi/__init__.py Normal file
View File

@@ -0,0 +1,20 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright (c) 2010 Citrix Systems, Inc.
#
# 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:`xenapi` -- Stubs for XenAPI
=================================
"""

103
nova/tests/xenapi/stubs.py Normal file
View File

@@ -0,0 +1,103 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# Copyright (c) 2010 Citrix Systems, Inc.
#
# 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.
"""Stubouts, mocks and fixtures for the test suite"""
from nova.virt import xenapi_conn
from nova.virt.xenapi import fake
from nova.virt.xenapi import volume_utils
def stubout_session(stubs, cls):
"""Stubs out two methods from XenAPISession"""
def fake_import(self):
"""Stubs out get_imported_xenapi of XenAPISession"""
fake_module = 'nova.virt.xenapi.fake'
from_list = ['fake']
return __import__(fake_module, globals(), locals(), from_list, -1)
stubs.Set(xenapi_conn.XenAPISession, '_create_session',
lambda s, url: cls(url))
stubs.Set(xenapi_conn.XenAPISession, 'get_imported_xenapi',
fake_import)
def stub_out_get_target(stubs):
"""Stubs out _get_target in volume_utils"""
def fake_get_target(volume_id):
return (None, None)
stubs.Set(volume_utils, '_get_target', fake_get_target)
class FakeSessionForVMTests(fake.SessionBase):
""" Stubs out a XenAPISession for VM tests """
def __init__(self, uri):
super(FakeSessionForVMTests, self).__init__(uri)
def network_get_all_records_where(self, _1, _2):
return self.xenapi.network.get_all_records()
def host_call_plugin(self, _1, _2, _3, _4, _5):
return ''
def VM_start(self, _1, ref, _2, _3):
vm = fake.get_record('VM', ref)
if vm['power_state'] != 'Halted':
raise fake.Failure(['VM_BAD_POWER_STATE', ref, 'Halted',
vm['power_state']])
vm['power_state'] = 'Running'
vm['is_a_template'] = False
vm['is_control_domain'] = False
class FakeSessionForVolumeTests(fake.SessionBase):
""" Stubs out a XenAPISession for Volume tests """
def __init__(self, uri):
super(FakeSessionForVolumeTests, self).__init__(uri)
def VBD_plug(self, _1, ref):
rec = fake.get_record('VBD', ref)
rec['currently-attached'] = True
def VDI_introduce(self, _1, uuid, _2, _3, _4, _5,
_6, _7, _8, _9, _10, _11):
valid_vdi = False
refs = fake.get_all('VDI')
for ref in refs:
rec = fake.get_record('VDI', ref)
if rec['uuid'] == uuid:
valid_vdi = True
if not valid_vdi:
raise fake.Failure([['INVALID_VDI', 'session', self._session]])
class FakeSessionForVolumeFailedTests(FakeSessionForVolumeTests):
""" Stubs out a XenAPISession for Volume tests: it injects failures """
def __init__(self, uri):
super(FakeSessionForVolumeFailedTests, self).__init__(uri)
def VDI_introduce(self, _1, uuid, _2, _3, _4, _5,
_6, _7, _8, _9, _10, _11):
# This is for testing failure
raise fake.Failure([['INVALID_VDI', 'session', self._session]])
def PBD_unplug(self, _1, ref):
rec = fake.get_record('PBD', ref)
rec['currently-attached'] = False
def SR_forget(self, _1, ref):
pass

125
run_tests.py
View File

@@ -1,125 +0,0 @@
# vim: tabstop=4 shiftwidth=4 softtabstop=4
# 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.
"""
This is our basic test running framework based on Twisted's Trial.
Usage Examples:
# to run all the tests
python run_tests.py
# to run a specific test suite imported here
python run_tests.py NodeConnectionTestCase
# to run a specific test imported here
python run_tests.py NodeConnectionTestCase.test_reboot
# to run some test suites elsewhere
python run_tests.py nova.tests.node_unittest
python run_tests.py nova.tests.node_unittest.NodeConnectionTestCase
Due to our use of multiprocessing it we frequently get some ignorable
'Interrupted system call' exceptions after test completion.
"""
import eventlet
eventlet.monkey_patch()
import __main__
import gettext
import os
import sys
gettext.install('nova', unicode=1)
from twisted.scripts import trial as trial_script
from nova import flags
from nova import twistd
from nova.tests.access_unittest import *
from nova.tests.api_unittest import *
from nova.tests.auth_unittest import *
from nova.tests.cloud_unittest import *
from nova.tests.compute_unittest import *
from nova.tests.flags_unittest import *
from nova.tests.misc_unittest import *
from nova.tests.network_unittest import *
#from nova.tests.objectstore_unittest import *
from nova.tests.quota_unittest import *
from nova.tests.rpc_unittest import *
from nova.tests.scheduler_unittest import *
from nova.tests.service_unittest import *
from nova.tests.twistd_unittest import *
from nova.tests.virt_unittest import *
from nova.tests.volume_unittest import *
FLAGS = flags.FLAGS
flags.DEFINE_bool('flush_db', True,
'Flush the database before running fake tests')
flags.DEFINE_string('tests_stderr', 'run_tests.err.log',
'Path to where to pipe STDERR during test runs.'
' Default = "run_tests.err.log"')
if __name__ == '__main__':
OptionsClass = twistd.WrapTwistedOptions(trial_script.Options)
config = OptionsClass()
argv = config.parseOptions()
FLAGS.verbose = True
# TODO(termie): these should make a call instead of doing work on import
if FLAGS.fake_tests:
from nova.tests.fake_flags import *
else:
from nova.tests.real_flags import *
# Establish redirect for STDERR
sys.stderr.flush()
err = open(FLAGS.tests_stderr, 'w+', 0)
os.dup2(err.fileno(), sys.stderr.fileno())
if len(argv) == 1 and len(config['tests']) == 0:
# If no tests were specified run the ones imported in this file
# NOTE(termie): "tests" is not a flag, just some Trial related stuff
config['tests'].update(['__main__'])
elif len(config['tests']):
# If we specified tests check first whether they are in __main__
for arg in config['tests']:
key = arg.split('.')[0]
if hasattr(__main__, key):
config['tests'].remove(arg)
config['tests'].add('__main__.%s' % arg)
trial_script._initialDebugSetup(config)
trialRunner = trial_script._makeRunner(config)
suite = trial_script._getSuite(config)
if config['until-failure']:
test_result = trialRunner.runUntilFailure(suite)
else:
test_result = trialRunner.run(suite)
if config.tracer:
sys.settrace(None)
results = config.tracer.results()
results.write_results(show_missing=1, summary=False,
coverdir=config.coverdir)
sys.exit(not test_result.wasSuccessful())