openstack/panko
Code Issues Proposed changes

Bump hacking to 0.7.0

Browse Source 
Fixes H602 errors

Change-Id: I071507efd9eb40240fa630ebfded8b83e54f3ce5
This commit is contained in:
Zhongyue Luo 2013-08-16 09:47:10 +08:00
parent ac000155a0
commit 6bee3e5f65
25 changed files with 334 additions and 354 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
test-requirements.txt
View File

@ -1,7 +1,7 @@
pep8==1.4.5
pyflakes==0.7.2
flake8==2.0
hacking>=0.5.6,<0.7
hacking>=0.5.6,<0.8
coverage>=3.6
discover
httplib2

48
tests/api/v1/test_list_events_scenarios.py
View File

@ -86,93 +86,93 @@ class TestListEvents(tests_api.TestBase,
def test_empty_project(self):
data = self.get('/projects/no-such-project/meters/instance')
self.assertEquals({'events': []}, data)
self.assertEqual({'events': []}, data)
def test_by_project(self):
data = self.get('/projects/project1/meters/instance')
self.assertEquals(2, len(data['events']))
self.assertEqual(2, len(data['events']))
def test_by_project_non_admin(self):
data = self.get('/projects/project1/meters/instance',
headers={"X-Roles": "Member",
"X-Project-Id": "project1"})
self.assertEquals(2, len(data['events']))
self.assertEqual(2, len(data['events']))
def test_by_project_wrong_tenant(self):
resp = self.get('/projects/project1/meters/instance',
headers={"X-Roles": "Member",
"X-Project-Id": "this-is-my-project"})
self.assertEquals(404, resp.status_code)
self.assertEqual(404, resp.status_code)
def test_by_project_with_timestamps(self):
data = self.get('/projects/project1/meters/instance',
start_timestamp=datetime.datetime(2012, 7, 2, 10, 42))
self.assertEquals(0, len(data['events']))
self.assertEqual(0, len(data['events']))
def test_empty_resource(self):
data = self.get('/resources/no-such-resource/meters/instance')
self.assertEquals({'events': []}, data)
self.assertEqual({'events': []}, data)
def test_by_resource(self):
data = self.get('/resources/resource-id/meters/instance')
self.assertEquals(2, len(data['events']))
self.assertEqual(2, len(data['events']))
def test_by_resource_non_admin(self):
data = self.get('/resources/resource-id-alternate/meters/instance',
headers={"X-Roles": "Member",
"X-Project-Id": "project2"})
self.assertEquals(1, len(data['events']))
self.assertEqual(1, len(data['events']))
def test_by_resource_some_tenant(self):
data = self.get('/resources/resource-id/meters/instance',
headers={"X-Roles": "Member",
"X-Project-Id": "project2"})
self.assertEquals(0, len(data['events']))
self.assertEqual(0, len(data['events']))
def test_empty_source(self):
data = self.get('/sources/no-such-source/meters/instance')
self.assertEquals({'events': []}, data)
self.assertEqual({'events': []}, data)
def test_by_source(self):
data = self.get('/sources/source1/meters/instance')
self.assertEquals(3, len(data['events']))
self.assertEqual(3, len(data['events']))
def test_by_source_non_admin(self):
data = self.get('/sources/source1/meters/instance',
headers={"X-Roles": "Member",
"X-Project-Id": "project2"})
self.assertEquals(1, len(data['events']))
self.assertEqual(1, len(data['events']))
def test_by_source_with_timestamps(self):
data = self.get('/sources/source1/meters/instance',
end_timestamp=datetime.datetime(2012, 7, 2, 10, 42))
self.assertEquals(2, len(data['events']))
self.assertEqual(2, len(data['events']))
def test_empty_user(self):
data = self.get('/users/no-such-user/meters/instance')
self.assertEquals({'events': []}, data)
self.assertEqual({'events': []}, data)
def test_by_user(self):
data = self.get('/users/user-id/meters/instance')
self.assertEquals(2, len(data['events']))
self.assertEqual(2, len(data['events']))
def test_by_user_non_admin(self):
data = self.get('/users/user-id/meters/instance',
headers={"X-Roles": "Member",
"X-Project-Id": "project1"})
self.assertEquals(2, len(data['events']))
self.assertEqual(2, len(data['events']))
def test_by_user_wrong_tenant(self):
data = self.get('/users/user-id/meters/instance',
headers={"X-Roles": "Member",
"X-Project-Id": "project2"})
self.assertEquals(0, len(data['events']))
self.assertEqual(0, len(data['events']))
def test_by_user_with_timestamps(self):
data = self.get('/users/user-id/meters/instance',
start_timestamp=datetime.datetime(2012, 7, 2, 10, 41),
end_timestamp=datetime.datetime(2012, 7, 2, 10, 42))
self.assertEquals(1, len(data['events']))
self.assertEqual(1, len(data['events']))
def test_template_list_event(self):
rv = self.get('/resources/resource-id/meters/instance',
@ -187,35 +187,35 @@ class TestListEventsMetaquery(TestListEvents,
def test_metaquery1(self):
q = '/sources/source1/meters/instance'
data = self.get('%s?metadata.tag=self.counter2' % q)
self.assertEquals(1, len(data['events']))
self.assertEqual(1, len(data['events']))
def test_metaquery1_wrong_tenant(self):
q = '/sources/source1/meters/instance'
data = self.get('%s?metadata.tag=self.counter2' % q,
headers={"X-Roles": "Member",
"X-Project-Id": "project1"})
self.assertEquals(0, len(data['events']))
self.assertEqual(0, len(data['events']))
def test_metaquery2(self):
q = '/sources/source1/meters/instance'
data = self.get('%s?metadata.tag=self.counter' % q)
self.assertEquals(2, len(data['events']))
self.assertEqual(2, len(data['events']))
def test_metaquery2_non_admin(self):
q = '/sources/source1/meters/instance'
data = self.get('%s?metadata.tag=self.counter' % q,
headers={"X-Roles": "Member",
"X-Project-Id": "project1"})
self.assertEquals(2, len(data['events']))
self.assertEqual(2, len(data['events']))
def test_metaquery3(self):
q = '/sources/source1/meters/instance'
data = self.get('%s?metadata.display_name=test-server' % q)
self.assertEquals(3, len(data['events']))
self.assertEqual(3, len(data['events']))
def test_metaquery3_with_project(self):
q = '/sources/source1/meters/instance'
data = self.get('%s?metadata.display_name=test-server' % q,
headers={"X-Roles": "Member",
"X-Project-Id": "project2"})
self.assertEquals(1, len(data['events']))
self.assertEqual(1, len(data['events']))

76
tests/api/v1/test_list_meters_scenarios.py
View File

@ -41,7 +41,7 @@ class TestListEmptyMeters(tests_api.TestBase,
def test_empty(self):
data = self.get('/meters')
self.assertEquals({'meters': []}, data)
self.assertEqual({'meters': []}, data)
class TestListMeters(tests_api.TestBase,
@ -118,94 +118,90 @@ class TestListMeters(tests_api.TestBase,
def test_list_meters(self):
data = self.get('/meters')
self.assertEquals(4, len(data['meters']))
self.assertEquals(set(r['resource_id'] for r in data['meters']),
set(['resource-id',
'resource-id2',
'resource-id3',
'resource-id4']))
self.assertEquals(set(r['name'] for r in data['meters']),
set(['meter.test',
'meter.mine']))
self.assertEqual(4, len(data['meters']))
self.assertEqual(set(r['resource_id'] for r in data['meters']),
set(['resource-id',
'resource-id2',
'resource-id3',
'resource-id4']))
self.assertEqual(set(r['name'] for r in data['meters']),
set(['meter.test', 'meter.mine']))
def test_list_meters_non_admin(self):
data = self.get('/meters',
headers={"X-Roles": "Member",
"X-Project-Id": "project-id"})
self.assertEquals(2, len(data['meters']))
self.assertEquals(set(r['resource_id'] for r in data['meters']),
set(['resource-id',
'resource-id2']))
self.assertEquals(set(r['name'] for r in data['meters']),
set(['meter.test',
'meter.mine']))
self.assertEqual(2, len(data['meters']))
self.assertEqual(set(r['resource_id'] for r in data['meters']),
set(['resource-id', 'resource-id2']))
self.assertEqual(set(r['name'] for r in data['meters']),
set(['meter.test', 'meter.mine']))
def test_with_resource(self):
data = self.get('/resources/resource-id/meters')
ids = set(r['name'] for r in data['meters'])
self.assertEquals(set(['meter.test']), ids)
self.assertEqual(set(['meter.test']), ids)
def test_with_source(self):
data = self.get('/sources/test_list_resources/meters')
ids = set(r['resource_id'] for r in data['meters'])
self.assertEquals(set(['resource-id',
'resource-id2',
'resource-id3',
'resource-id4']), ids)
self.assertEqual(set(['resource-id',
'resource-id2',
'resource-id3',
'resource-id4']), ids)
def test_with_source_non_admin(self):
data = self.get('/sources/test_list_resources/meters',
headers={"X-Roles": "Member",
"X-Project-Id": "project-id2"})
ids = set(r['resource_id'] for r in data['meters'])
self.assertEquals(set(['resource-id3',
'resource-id4']), ids)
self.assertEqual(set(['resource-id3', 'resource-id4']), ids)
def test_with_source_non_existent(self):
data = self.get('/sources/test_list_resources_dont_exist/meters')
self.assertEquals(data['meters'], [])
self.assertEqual(data['meters'], [])
def test_with_user(self):
data = self.get('/users/user-id/meters')
nids = set(r['name'] for r in data['meters'])
self.assertEquals(set(['meter.mine', 'meter.test']), nids)
self.assertEqual(set(['meter.mine', 'meter.test']), nids)
rids = set(r['resource_id'] for r in data['meters'])
self.assertEquals(set(['resource-id', 'resource-id2']), rids)
self.assertEqual(set(['resource-id', 'resource-id2']), rids)
def test_with_user_non_admin(self):
data = self.get('/users/user-id/meters',
headers={"X-Roles": "Member",
"X-Project-Id": "project-id"})
nids = set(r['name'] for r in data['meters'])
self.assertEquals(set(['meter.mine', 'meter.test']), nids)
self.assertEqual(set(['meter.mine', 'meter.test']), nids)
rids = set(r['resource_id'] for r in data['meters'])
self.assertEquals(set(['resource-id', 'resource-id2']), rids)
self.assertEqual(set(['resource-id', 'resource-id2']), rids)
def test_with_user_wrong_tenant(self):
data = self.get('/users/user-id/meters',
headers={"X-Roles": "Member",
"X-Project-Id": "project666"})
self.assertEquals(data['meters'], [])
self.assertEqual(data['meters'], [])
def test_with_user_non_existent(self):
data = self.get('/users/user-id-foobar123/meters')
self.assertEquals(data['meters'], [])
self.assertEqual(data['meters'], [])
def test_with_project(self):
data = self.get('/projects/project-id2/meters')
ids = set(r['resource_id'] for r in data['meters'])
self.assertEquals(set(['resource-id3', 'resource-id4']), ids)
self.assertEqual(set(['resource-id3', 'resource-id4']), ids)
def test_with_project_non_admin(self):
data = self.get('/projects/project-id2/meters',
headers={"X-Roles": "Member",
"X-Project-Id": "project-id2"})
ids = set(r['resource_id'] for r in data['meters'])
self.assertEquals(set(['resource-id3', 'resource-id4']), ids)
self.assertEqual(set(['resource-id3', 'resource-id4']), ids)
def test_with_project_wrong_tenant(self):
data = self.get('/projects/project-id2/meters',
@ -215,7 +211,7 @@ class TestListMeters(tests_api.TestBase,
def test_with_project_non_existent(self):
data = self.get('/projects/jd-was-here/meters')
self.assertEquals(data['meters'], [])
self.assertEqual(data['meters'], [])
class TestListMetersMetaquery(TestListMeters,
@ -223,32 +219,32 @@ class TestListMetersMetaquery(TestListMeters,
def test_metaquery1(self):
data = self.get('/meters?metadata.tag=self.counter')
self.assertEquals(1, len(data['meters']))
self.assertEqual(1, len(data['meters']))
def test_metaquery1_non_admin(self):
data = self.get('/meters?metadata.tag=self.counter',
headers={"X-Roles": "Member",
"X-Project-Id": "project-id"})
self.assertEquals(1, len(data['meters']))
self.assertEqual(1, len(data['meters']))
def test_metaquery1_wrong_tenant(self):
data = self.get('/meters?metadata.tag=self.counter',
headers={"X-Roles": "Member",
"X-Project-Id": "project-666"})
self.assertEquals(0, len(data['meters']))
self.assertEqual(0, len(data['meters']))
def test_metaquery2(self):
data = self.get('/meters?metadata.tag=four.counter')
self.assertEquals(1, len(data['meters']))
self.assertEqual(1, len(data['meters']))
def test_metaquery2_non_admin(self):
data = self.get('/meters?metadata.tag=four.counter',
headers={"X-Roles": "Member",
"X-Project-Id": "project-id2"})
self.assertEquals(1, len(data['meters']))
self.assertEqual(1, len(data['meters']))
def test_metaquery2_non_admin_wrong_project(self):
data = self.get('/meters?metadata.tag=four.counter',
headers={"X-Roles": "Member",
"X-Project-Id": "project-666"})
self.assertEquals(0, len(data['meters']))
self.assertEqual(0, len(data['meters']))

10
tests/api/v1/test_list_projects_scenarios.py
View File

@ -41,7 +41,7 @@ class TestListEmptyProjects(tests_api.TestBase,
def test_empty(self):
data = self.get('/projects')
self.assertEquals({'projects': []}, data)
self.assertEqual({'projects': []}, data)
class TestListProjects(tests_api.TestBase,
@ -89,20 +89,20 @@ class TestListProjects(tests_api.TestBase,
def test_projects(self):
data = self.get('/projects')
self.assertEquals(['project-id', 'project-id2'], data['projects'])
self.assertEqual(['project-id', 'project-id2'], data['projects'])
def test_projects_non_admin(self):
data = self.get('/projects',
headers={"X-Roles": "Member",
"X-Project-Id": "project-id"})
self.assertEquals(['project-id'], data['projects'])
self.assertEqual(['project-id'], data['projects'])
def test_with_source(self):
data = self.get('/sources/test_list_users/projects')
self.assertEquals(['project-id2'], data['projects'])
self.assertEqual(['project-id2'], data['projects'])
def test_with_source_non_admin(self):
data = self.get('/sources/test_list_users/projects',
headers={"X-Roles": "Member",
"X-Project-Id": "project-id2"})
self.assertEquals(['project-id2'], data['projects'])
self.assertEqual(['project-id2'], data['projects'])

74
tests/api/v1/test_list_resources_scenarios.py
View File

@ -41,7 +41,7 @@ class TestListEmptyResources(tests_api.TestBase,
def test_empty(self):
data = self.get('/resources')
self.assertEquals({'resources': []}, data)
self.assertEqual({'resources': []}, data)
class TestListResourcesBase(tests_api.TestBase,
@ -113,20 +113,19 @@ class TestListResources(TestListResourcesBase):
def test_list_resources(self):
data = self.get('/resources')
self.assertEquals(3, len(data['resources']))
self.assertEquals(set(r['resource_id'] for r in data['resources']),
set(['resource-id',
'resource-id-alternate',
'resource-id2']))
self.assertEqual(3, len(data['resources']))
self.assertEqual(set(r['resource_id'] for r in data['resources']),
set(['resource-id',
'resource-id-alternate',
'resource-id2']))
def test_list_resources_non_admin(self):
data = self.get('/resources',
headers={"X-Roles": "Member",
"X-Project-Id": "project-id"})
self.assertEquals(2, len(data['resources']))
self.assertEquals(set(r['resource_id'] for r in data['resources']),
set(['resource-id',
'resource-id-alternate']))
self.assertEqual(2, len(data['resources']))
self.assertEqual(set(r['resource_id'] for r in data['resources']),
set(['resource-id', 'resource-id-alternate']))
def test_list_resources_with_timestamps(self):
data = self.get('/resources',
@ -134,9 +133,8 @@ class TestListResources(TestListResourcesBase):
2012, 7, 2, 10, 41).isoformat(),
end_timestamp=datetime.datetime(
2012, 7, 2, 10, 43).isoformat())
self.assertEquals(set(r['resource_id'] for r in data['resources']),
set(['resource-id-alternate',
'resource-id2']))
self.assertEqual(set(r['resource_id'] for r in data['resources']),
set(['resource-id-alternate', 'resource-id2']))
def test_list_resources_with_timestamps_non_admin(self):
data = self.get('/resources',
@ -146,23 +144,22 @@ class TestListResources(TestListResourcesBase):
2012, 7, 2, 10, 43).isoformat(),
headers={"X-Roles": "Member",
"X-Project-Id": "project-id"})
self.assertEquals(set(r['resource_id'] for r in data['resources']),
set(['resource-id-alternate']))
self.assertEqual(set(r['resource_id'] for r in data['resources']),
set(['resource-id-alternate']))
def test_with_source(self):
data = self.get('/sources/test_list_resources/resources')
ids = set(r['resource_id'] for r in data['resources'])
self.assertEquals(set(['resource-id',
'resource-id2',
'resource-id-alternate']), ids)
self.assertEqual(set(['resource-id',
'resource-id2',
'resource-id-alternate']), ids)
def test_with_source_non_admin(self):
data = self.get('/sources/test_list_resources/resources',
headers={"X-Roles": "Member",
"X-Project-Id": "project-id"})
ids = set(r['resource_id'] for r in data['resources'])
self.assertEquals(set(['resource-id',
'resource-id-alternate']), ids)
self.assertEqual(set(['resource-id', 'resource-id-alternate']), ids)
def test_with_source_with_timestamps(self):
data = self.get('/sources/test_list_resources/resources',
@ -171,8 +168,7 @@ class TestListResources(TestListResourcesBase):
end_timestamp=datetime.datetime(
2012, 7, 2, 10, 43).isoformat())
ids = set(r['resource_id'] for r in data['resources'])
self.assertEquals(set(['resource-id2',
'resource-id-alternate']), ids)
self.assertEqual(set(['resource-id2', 'resource-id-alternate']), ids)
def test_with_source_with_timestamps_non_admin(self):
data = self.get('/sources/test_list_resources/resources',
@ -183,30 +179,30 @@ class TestListResources(TestListResourcesBase):
end_timestamp=datetime.datetime(
2012, 7, 2, 10, 43).isoformat())
ids = set(r['resource_id'] for r in data['resources'])
self.assertEquals(set(['resource-id-alternate']), ids)
self.assertEqual(set(['resource-id-alternate']), ids)
def test_with_source_non_existent(self):
data = self.get('/sources/test_list_resources_dont_exist/resources')
self.assertEquals(data['resources'], [])
self.assertEqual(data['resources'], [])
def test_with_user(self):
data = self.get('/users/user-id/resources')
ids = set(r['resource_id'] for r in data['resources'])
self.assertEquals(set(['resource-id', 'resource-id-alternate']), ids)
self.assertEqual(set(['resource-id', 'resource-id-alternate']), ids)
def test_with_user_non_admin(self):
data = self.get('/users/user-id/resources',
headers={"X-Roles": "Member",
"X-Project-Id": "project-id"})
ids = set(r['resource_id'] for r in data['resources'])
self.assertEquals(set(['resource-id', 'resource-id-alternate']), ids)
self.assertEqual(set(['resource-id', 'resource-id-alternate']), ids)
def test_with_user_wrong_tenant(self):
data = self.get('/users/user-id/resources',
headers={"X-Roles": "Member",
"X-Project-Id": "project-jd"})
ids = set(r['resource_id'] for r in data['resources'])
self.assertEquals(set(), ids)
self.assertEqual(set(), ids)
def test_with_user_with_timestamps(self):
data = self.get('/users/user-id/resources',
@ -215,7 +211,7 @@ class TestListResources(TestListResourcesBase):
end_timestamp=datetime.datetime(
2012, 7, 2, 10, 42).isoformat())
ids = set(r['resource_id'] for r in data['resources'])
self.assertEquals(set(), ids)
self.assertEqual(set(), ids)
def test_with_user_with_timestamps_non_admin(self):
data = self.get('/users/user-id/resources',
@ -226,23 +222,23 @@ class TestListResources(TestListResourcesBase):
end_timestamp=datetime.datetime(
2012, 7, 2, 10, 42).isoformat())
ids = set(r['resource_id'] for r in data['resources'])
self.assertEquals(set(), ids)
self.assertEqual(set(), ids)
def test_with_user_non_existent(self):
data = self.get('/users/user-id-foobar123/resources')
self.assertEquals(data['resources'], [])
self.assertEqual(data['resources'], [])
def test_with_project(self):
data = self.get('/projects/project-id/resources')
ids = set(r['resource_id'] for r in data['resources'])
self.assertEquals(set(['resource-id', 'resource-id-alternate']), ids)
self.assertEqual(set(['resource-id', 'resource-id-alternate']), ids)
def test_with_project_non_admin(self):
data = self.get('/projects/project-id/resources',
headers={"X-Roles": "Member",
"X-Project-Id": "project-id"})
ids = set(r['resource_id'] for r in data['resources'])
self.assertEquals(set(['resource-id', 'resource-id-alternate']), ids)
self.assertEqual(set(['resource-id', 'resource-id-alternate']), ids)
def test_with_project_with_timestamp(self):
data = self.get('/projects/project-id/resources',
@ -251,7 +247,7 @@ class TestListResources(TestListResourcesBase):
end_timestamp=datetime.datetime(
2012, 7, 2, 10, 41).isoformat())
ids = set(r['resource_id'] for r in data['resources'])
self.assertEquals(set(['resource-id']), ids)
self.assertEqual(set(['resource-id']), ids)
def test_with_project_with_timestamp_non_admin(self):
data = self.get('/projects/project-id/resources',
@ -262,11 +258,11 @@ class TestListResources(TestListResourcesBase):
end_timestamp=datetime.datetime(
2012, 7, 2, 10, 41).isoformat())
ids = set(r['resource_id'] for r in data['resources'])
self.assertEquals(set(['resource-id']), ids)
self.assertEqual(set(['resource-id']), ids)
def test_with_project_non_existent(self):
data = self.get('/projects/jd-was-here/resources')
self.assertEquals(data['resources'], [])
self.assertEqual(data['resources'], [])
class TestListResourcesMetaquery(TestListResourcesBase,
@ -275,23 +271,23 @@ class TestListResourcesMetaquery(TestListResourcesBase,
def test_metaquery1(self):
q = '/sources/test_list_resources/resources'
data = self.get('%s?metadata.display_name=test-server' % q)
self.assertEquals(3, len(data['resources']))
self.assertEqual(3, len(data['resources']))
def test_metaquery1_non_admin(self):
q = '/sources/test_list_resources/resources'
data = self.get('%s?metadata.display_name=test-server' % q,
headers={"X-Roles": "Member",
"X-Project-Id": "project-id"})
self.assertEquals(2, len(data['resources']))
self.assertEqual(2, len(data['resources']))
def test_metaquery2(self):
q = '/sources/test_list_resources/resources'
data = self.get('%s?metadata.tag=self.counter4' % q)
self.assertEquals(1, len(data['resources']))
self.assertEqual(1, len(data['resources']))
def test_metaquery2_non_admin(self):
q = '/sources/test_list_resources/resources'
data = self.get('%s?metadata.tag=self.counter4' % q,
headers={"X-Roles": "Member",
"X-Project-Id": "project-id"})
self.assertEquals(1, len(data['resources']))
self.assertEqual(1, len(data['resources']))

10
tests/api/v1/test_list_users_scenarios.py
View File

@ -41,7 +41,7 @@ class TestListEmptyUsers(tests_api.TestBase,
def test_empty(self):
data = self.get('/users')
self.assertEquals({'users': []}, data)
self.assertEqual({'users': []}, data)
class TestListUsers(tests_api.TestBase,
@ -92,22 +92,22 @@ class TestListUsers(tests_api.TestBase,
def test_users(self):
data = self.get('/users')
self.assertEquals(['user-id', 'user-id2'], data['users'])
self.assertEqual(['user-id', 'user-id2'], data['users'])
def test_users_non_admin(self):
data = self.get('/users',
headers={"X-Roles": "Member",
"X-User-Id": "user-id",
"X-Project-Id": "project-id"})
self.assertEquals(['user-id'], data['users'])
self.assertEqual(['user-id'], data['users'])
def test_with_source(self):
data = self.get('/sources/test_list_users/users')
self.assertEquals(['user-id'], data['users'])
self.assertEqual(['user-id'], data['users'])
def test_with_source_non_admin(self):
data = self.get('/sources/test_list_users/users',
headers={"X-Roles": "Member",
"X-User-Id": "user-id",
"X-Project-Id": "project-id"})
self.assertEquals(['user-id'], data['users'])
self.assertEqual(['user-id'], data['users'])

6
tests/api/v2/test_acl_scenarios.py
View File

@ -164,7 +164,7 @@ class TestAPIACL(FunctionalTest,
"bc23a9d531064583ace8f67dad60f6bb",
})
ids = set(r['resource_id'] for r in data)
self.assertEquals(set(['resource-good', 'resource-56']), ids)
self.assertEqual(set(['resource-good', 'resource-56']), ids)
def test_with_non_admin_missing_project_query(self):
data = self.get_json('/meters',
@ -172,7 +172,7 @@ class TestAPIACL(FunctionalTest,
"X-Auth-Token": VALID_TOKEN2,
"X-Project-Id": "project-good"})
ids = set(r['resource_id'] for r in data)
self.assertEquals(set(['resource-good', 'resource-56']), ids)
self.assertEqual(set(['resource-good', 'resource-56']), ids)
def test_with_non_admin(self):
data = self.get_json('/meters',
@ -183,7 +183,7 @@ class TestAPIACL(FunctionalTest,
'value': 'project-good',
}])
ids = set(r['resource_id'] for r in data)
self.assertEquals(set(['resource-good', 'resource-56']), ids)
self.assertEqual(set(['resource-good', 'resource-56']), ids)
def test_non_admin_wrong_project(self):
data = self.get_json('/meters',

47
tests/api/v2/test_alarm_scenarios.py
View File

@ -38,7 +38,7 @@ class TestListEmptyAlarms(FunctionalTest,
def test_empty(self):
data = self.get_json('/alarms')
self.assertEquals([], data)
self.assertEqual([], data)
class TestAlarms(FunctionalTest,
@ -72,14 +72,11 @@ class TestAlarms(FunctionalTest,
def test_list_alarms(self):
data = self.get_json('/alarms')
self.assertEquals(3, len(data))
self.assertEquals(set(r['name'] for r in data),
set(['name1',
'name2',
'name3']))
self.assertEquals(set(r['counter_name'] for r in data),
set(['meter.test',
'meter.mine']))
self.assertEqual(3, len(data))
self.assertEqual(set(r['name'] for r in data),
set(['name1', 'name2', 'name3']))
self.assertEqual(set(r['counter_name'] for r in data),
set(['meter.test', 'meter.mine']))
def test_get_alarm(self):
alarms = self.get_json('/alarms',
@ -88,14 +85,14 @@ class TestAlarms(FunctionalTest,
}])
for a in alarms:
print('%s: %s' % (a['name'], a['alarm_id']))
self.assertEquals(alarms[0]['name'], 'name1')
self.assertEquals(alarms[0]['counter_name'], 'meter.test')
self.assertEqual(alarms[0]['name'], 'name1')
self.assertEqual(alarms[0]['counter_name'], 'meter.test')
one = self.get_json('/alarms/%s' % alarms[0]['alarm_id'])
self.assertEquals(one['name'], 'name1')
self.assertEquals(one['counter_name'], 'meter.test')
self.assertEquals(one['alarm_id'], alarms[0]['alarm_id'])
self.assertEquals(one['repeat_actions'], alarms[0]['repeat_actions'])
self.assertEqual(one['name'], 'name1')
self.assertEqual(one['counter_name'], 'meter.test')
self.assertEqual(one['alarm_id'], alarms[0]['alarm_id'])
self.assertEqual(one['repeat_actions'], alarms[0]['repeat_actions'])
def test_post_invalid_alarm(self):
json = {
@ -108,7 +105,7 @@ class TestAlarms(FunctionalTest,
self.post_json('/alarms', params=json, expect_errors=True, status=400,
headers=self.auth_headers)
alarms = list(self.conn.get_alarms())
self.assertEquals(3, len(alarms))
self.assertEqual(3, len(alarms))
def test_post_alarm(self):
json = {
@ -122,7 +119,7 @@ class TestAlarms(FunctionalTest,
self.post_json('/alarms', params=json, status=200,
headers=self.auth_headers)
alarms = list(self.conn.get_alarms())
self.assertEquals(4, len(alarms))
self.assertEqual(4, len(alarms))
for alarm in alarms:
if alarm.name == 'added_alarm':
self.assertEqual(alarm.repeat_actions, True)
@ -139,15 +136,15 @@ class TestAlarms(FunctionalTest,
q=[{'field': 'name',
'value': 'name1',
}])
self.assertEquals(1, len(data))
self.assertEqual(1, len(data))
alarm_id = data[0]['alarm_id']
self.put_json('/alarms/%s' % alarm_id,
params=json,
headers=self.auth_headers)
alarm = list(self.conn.get_alarms(alarm_id=alarm_id))[0]
self.assertEquals(alarm.name, json['name'])
self.assertEquals(alarm.repeat_actions, json['repeat_actions'])
self.assertEqual(alarm.name, json['name'])
self.assertEqual(alarm.repeat_actions, json['repeat_actions'])
def test_put_alarm_wrong_field(self):
# Note: wsme will ignore unknown fields so will just not appear in
@ -161,27 +158,27 @@ class TestAlarms(FunctionalTest,
'value': 'name1',
}],
headers=self.auth_headers)
self.assertEquals(1, len(data))
self.assertEqual(1, len(data))
resp = self.put_json('/alarms/%s' % data[0]['alarm_id'],
params=json,
expect_errors=True,
headers=self.auth_headers)
self.assertEquals(resp.status_code, 200)
self.assertEqual(resp.status_code, 200)
def test_delete_alarm(self):
data = self.get_json('/alarms')
self.assertEquals(3, len(data))
self.assertEqual(3, len(data))
self.delete('/alarms/%s' % data[0]['alarm_id'],
status=200)
alarms = list(self.conn.get_alarms())
self.assertEquals(2, len(alarms))
self.assertEqual(2, len(alarms))
def test_get_alarm_history(self):
data = self.get_json('/alarms')
history = self.get_json('/alarms/%s/history' % data[0]['alarm_id'])
self.assertEquals([], history)
self.assertEqual([], history)
def test_get_nonexistent_alarm_history(self):
response = self.get_json('/alarms/%s/history' % 'foobar',

24
tests/api/v2/test_list_events_scenarios.py
View File

@ -85,15 +85,15 @@ class TestListEvents(FunctionalTest,
def test_all(self):
data = self.get_json('/meters/instance')
self.assertEquals(2, len(data))
self.assertEqual(2, len(data))
def test_all_trailing_slash(self):
data = self.get_json('/meters/instance/')
self.assertEquals(2, len(data))
self.assertEqual(2, len(data))
def test_all_limit(self):
data = self.get_json('/meters/instance?limit=1')
self.assertEquals(1, len(data))
self.assertEqual(1, len(data))
def test_all_limit_negative(self):
self.assertRaises(webtest.app.AppError,
@ -102,63 +102,63 @@ class TestListEvents(FunctionalTest,
def test_all_limit_bigger(self):
data = self.get_json('/meters/instance?limit=42')
self.assertEquals(2, len(data))
self.assertEqual(2, len(data))
def test_empty_project(self):
data = self.get_json('/meters/instance',
q=[{'field': 'project_id',
'value': 'no-such-project',
}])
self.assertEquals([], data)
self.assertEqual([], data)
def test_by_project(self):
data = self.get_json('/meters/instance',
q=[{'field': 'project_id',
'value': 'project1',
}])
self.assertEquals(1, len(data))
self.assertEqual(1, len(data))
def test_empty_resource(self):
data = self.get_json('/meters/instance',
q=[{'field': 'resource_id',
'value': 'no-such-resource',
}])
self.assertEquals([], data)
self.assertEqual([], data)
def test_by_resource(self):
data = self.get_json('/meters/instance',
q=[{'field': 'resource_id',
'value': 'resource-id',
}])
self.assertEquals(1, len(data))
self.assertEqual(1, len(data))
def test_empty_source(self):
data = self.get_json('/meters/instance',
q=[{'field': 'source',
'value': 'no-such-source',
}])
self.assertEquals(0, len(data))
self.assertEqual(0, len(data))
def test_by_source(self):
data = self.get_json('/meters/instance',
q=[{'field': 'source',
'value': 'test_source',
}])
self.assertEquals(1, len(data))
self.assertEqual(1, len(data))
def test_empty_user(self):
data = self.get_json('/meters/instance',
q=[{'field': 'user_id',
'value': 'no-such-user',
}])
self.assertEquals([], data)
self.assertEqual([], data)
def test_by_user(self):
data = self.get_json('/meters/instance',
q=[{'field': 'user_id',
'value': 'user-id',
}])
self.assertEquals(1, len(data))
self.assertEqual(1, len(data))
def test_metadata(self):
data = self.get_json('/meters/instance',

103
tests/api/v2/test_list_meters_scenarios.py
View File

@ -40,7 +40,7 @@ class TestListEmptyMeters(FunctionalTest,
def test_empty(self):
data = self.get_json('/meters')
self.assertEquals([], data)
self.assertEqual([], data)
class TestListMeters(FunctionalTest,
@ -122,15 +122,14 @@ class TestListMeters(FunctionalTest,
def test_list_meters(self):
data = self.get_json('/meters')
self.assertEquals(4, len(data))
self.assertEquals(set(r['resource_id'] for r in data),
set(['resource-id',
'resource-id2',
'resource-id3',
'resource-id4']))
self.assertEquals(set(r['name'] for r in data),
set(['meter.test',
'meter.mine']))
self.assertEqual(4, len(data))
self.assertEqual(set(r['resource_id'] for r in data),
set(['resource-id',
'resource-id2',
'resource-id3',
'resource-id4']))
self.assertEqual(set(r['name'] for r in data),
set(['meter.test', 'meter.mine']))
def test_list_meters_with_dict_metadata(self):
data = self.get_json('/meters/meter.mine',
@ -139,8 +138,8 @@ class TestListMeters(FunctionalTest,
'op': 'eq',
'value': 'sub_prop_value',
}])
self.assertEquals(1, len(data))
self.assertEquals('resource-id4', data[0]['resource_id'])
self.assertEqual(1, len(data))
self.assertEqual('resource-id4', data[0]['resource_id'])
metadata = data[0]['resource_metadata']
self.assertIsNotNone(metadata)
# FIXME (flwang): Based on current implement, the metadata of
@ -154,11 +153,11 @@ class TestListMeters(FunctionalTest,
'op': 'eq',
'value': 'self.counter1',
}],)
self.assertEquals(1, len(data))
self.assertEquals(set(r['resource_id'] for r in data),
set(['resource-id']))
self.assertEquals(set(r['counter_name'] for r in data),
set(['meter.test']))
self.assertEqual(1, len(data))
self.assertEqual(set(r['resource_id'] for r in data),
set(['resource-id']))
self.assertEqual(set(r['counter_name'] for r in data),
set(['meter.test']))
def test_list_meters_multi_metadata_query(self):
data = self.get_json('/meters/meter.test',
@ -170,18 +169,18 @@ class TestListMeters(FunctionalTest,
'op': 'eq',
'value': 'test-server',
}],)
self.assertEquals(1, len(data))
self.assertEquals(set(r['resource_id'] for r in data),
set(['resource-id']))
self.assertEquals(set(r['counter_name'] for r in data),
set(['meter.test']))
self.assertEqual(1, len(data))
self.assertEqual(set(r['resource_id'] for r in data),
set(['resource-id']))
self.assertEqual(set(r['counter_name'] for r in data),
set(['meter.test']))
def test_with_resource(self):
data = self.get_json('/meters', q=[{'field': 'resource_id',
'value': 'resource-id',
}])
ids = set(r['name'] for r in data)
self.assertEquals(set(['meter.test']), ids)
self.assertEqual(set(['meter.test']), ids)
def test_with_resource_and_metadata_query(self):
data = self.get_json('/meters/meter.mine',
@ -193,21 +192,21 @@ class TestListMeters(FunctionalTest,
'op': 'eq',
'value': 'self.counter2',
}])
self.assertEquals(1, len(data))
self.assertEquals(set(r['resource_id'] for r in data),
set(['resource-id2']))
self.assertEquals(set(r['counter_name'] for r in data),
set(['meter.mine']))
self.assertEqual(1, len(data))
self.assertEqual(set(r['resource_id'] for r in data),
set(['resource-id2']))
self.assertEqual(set(r['counter_name'] for r in data),
set(['meter.mine']))
def test_with_source(self):
data = self.get_json('/meters', q=[{'field': 'source',
'value': 'test_source',
}])
ids = set(r['resource_id'] for r in data)
self.assertEquals(set(['resource-id',
'resource-id2',
'resource-id3',
'resource-id4']), ids)
self.assertEqual(set(['resource-id',
'resource-id2',
'resource-id3',
'resource-id4']), ids)
def test_with_source_and_metadata_query(self):
data = self.get_json('/meters/meter.mine',
@ -219,11 +218,10 @@ class TestListMeters(FunctionalTest,
'op': 'eq',
'value': 'self.counter2',
}])
self.assertEquals(1, len(data))
self.assertEquals(set(r['source'] for r in data),
set(['test_source']))
self.assertEquals(set(r['counter_name'] for r in data),
set(['meter.mine']))
self.assertEqual(1, len(data))
self.assertEqual(set(r['source'] for r in data), set(['test_source']))
self.assertEqual(set(r['counter_name'] for r in data),
set(['meter.mine']))
def test_with_source_non_existent(self):
data = self.get_json('/meters',
@ -241,13 +239,13 @@ class TestListMeters(FunctionalTest,
)
uids = set(r['user_id'] for r in data)
self.assertEquals(set(['user-id']), uids)
self.assertEqual(set(['user-id']), uids)
nids = set(r['name'] for r in data)
self.assertEquals(set(['meter.mine', 'meter.test']), nids)
self.assertEqual(set(['meter.mine', 'meter.test']), nids)
rids = set(r['resource_id'] for r in data)
self.assertEquals(set(['resource-id', 'resource-id2']), rids)
self.assertEqual(set(['resource-id', 'resource-id2']), rids)
def test_with_user_and_metadata_query(self):
data = self.get_json('/meters/meter.test',
@ -259,11 +257,10 @@ class TestListMeters(FunctionalTest,
'op': 'eq',
'value': 'self.counter1',
}])
self.assertEquals(1, len(data))
self.assertEquals(set(r['user_id'] for r in data),
set(['user-id']))
self.assertEquals(set(r['counter_name'] for r in data),
set(['meter.test']))
self.assertEqual(1, len(data))
self.assertEqual(set(r['user_id'] for r in data), set(['user-id']))
self.assertEqual(set(r['counter_name'] for r in data),
set(['meter.test']))
def test_with_user_non_existent(self):
data = self.get_json('/meters',
@ -271,7 +268,7 @@ class TestListMeters(FunctionalTest,
'value': 'user-id-foobar123',
}],
)
self.assertEquals(data, [])
self.assertEqual(data, [])
def test_with_project(self):
data = self.get_json('/meters',
@ -280,7 +277,7 @@ class TestListMeters(FunctionalTest,
}],
)
ids = set(r['resource_id'] for r in data)
self.assertEquals(set(['resource-id3', 'resource-id4']), ids)
self.assertEqual(set(['resource-id3', 'resource-id4']), ids)
def test_with_project_and_metadata_query(self):
data = self.get_json('/meters/meter.test',
@ -292,11 +289,11 @@ class TestListMeters(FunctionalTest,
'op': 'eq',
'value': 'self.counter1',
}])
self.assertEquals(1, len(data))
self.assertEquals(set(r['project_id'] for r in data),
set(['project-id']))
self.assertEquals(set(r['counter_name'] for r in data),
set(['meter.test']))
self.assertEqual(1, len(data))
self.assertEqual(set(r['project_id'] for r in data),
set(['project-id']))
self.assertEqual(set(r['counter_name'] for r in data),
set(['meter.test']))
def test_with_project_non_existent(self):
data = self.get_json('/meters',
@ -304,4 +301,4 @@ class TestListMeters(FunctionalTest,
'value': 'jd-was-here',
}],
)
self.assertEquals(data, [])
self.assertEqual(data, [])

24
tests/api/v2/test_list_resources_scenarios.py
View File

@ -42,7 +42,7 @@ class TestListResources(FunctionalTest,
def test_empty(self):
data = self.get_json('/resources')
self.assertEquals([], data)
self.assertEqual([], data)
def test_instance_no_metadata(self):
counter1 = sample.Sample(
@ -64,7 +64,7 @@ class TestListResources(FunctionalTest,
self.conn.record_metering_data(msg)
data = self.get_json('/resources')
self.assertEquals(1, len(data))
self.assertEqual(1, len(data))
def test_instances(self):
counter1 = sample.Sample(
@ -108,7 +108,7 @@ class TestListResources(FunctionalTest,
self.conn.record_metering_data(msg2)
data = self.get_json('/resources')
self.assertEquals(2, len(data))
self.assertEqual(2, len(data))
def test_instances_one(self):
counter1 = sample.Sample(
@ -152,7 +152,7 @@ class TestListResources(FunctionalTest,
self.conn.record_metering_data(msg2)
data = self.get_json('/resources/resource-id')
self.assertEquals('resource-id', data['resource_id'])
self.assertEqual('resource-id', data['resource_id'])
def test_with_source(self):
counter1 = sample.Sample(
@ -199,7 +199,7 @@ class TestListResources(FunctionalTest,
'value': 'test_list_resources',
}])
ids = [r['resource_id'] for r in data]
self.assertEquals(['resource-id'], ids)
self.assertEqual(['resource-id'], ids)
def test_with_invalid_resource_id(self):
counter1 = sample.Sample(
@ -243,13 +243,13 @@ class TestListResources(FunctionalTest,
self.conn.record_metering_data(msg2)
resp1 = self.get_json('/resources/resource-id-1')
self.assertEquals(resp1["resource_id"], "resource-id-1")
self.assertEqual(resp1["resource_id"], "resource-id-1")
resp2 = self.get_json('/resources/resource-id-2')
self.assertEquals(resp2["resource_id"], "resource-id-2")
self.assertEqual(resp2["resource_id"], "resource-id-2")
resp3 = self.get_json('/resources/resource-id-3', expect_errors=True)
self.assertEquals(resp3.status_code, 400)
self.assertEqual(resp3.status_code, 400)
def test_with_user(self):
counter1 = sample.Sample(
@ -296,7 +296,7 @@ class TestListResources(FunctionalTest,
'value': 'user-id',
}])
ids = [r['resource_id'] for r in data]
self.assertEquals(['resource-id'], ids)
self.assertEqual(['resource-id'], ids)
def test_with_project(self):
counter1 = sample.Sample(
@ -343,7 +343,7 @@ class TestListResources(FunctionalTest,
'value': 'project-id',
}])
ids = [r['resource_id'] for r in data]
self.assertEquals(['resource-id'], ids)
self.assertEqual(['resource-id'], ids)
def test_with_user_non_admin(self):
counter1 = sample.Sample(
@ -370,7 +370,7 @@ class TestListResources(FunctionalTest,
headers={"X-Roles": "Member",
"X-Project-Id": "project-id2"})
ids = set(r['resource_id'] for r in data)
self.assertEquals(set(['resource-id-alternate']), ids)
self.assertEqual(set(['resource-id-alternate']), ids)
def test_with_user_wrong_tenant(self):
counter1 = sample.Sample(
@ -397,7 +397,7 @@ class TestListResources(FunctionalTest,
headers={"X-Roles": "Member",
"X-Project-Id": "project-wrong"})
ids = set(r['resource_id'] for r in data)
self.assertEquals(set(), ids)
self.assertEqual(set(), ids)
def test_metadata(self):
counter1 = sample.Sample(

34
tests/api/v2/test_post_samples_scenarios.py
View File

@ -65,7 +65,7 @@ class TestPostSamples(FunctionalTest,
# source is generated if not provided.
s1[0]['source'] = '%s:openstack' % s1[0]['project_id']
self.assertEquals(s1, data.json)
self.assertEqual(s1, data.json)
def test_wrong_project_id(self):
"""Do not accept cross posting samples to different projects."""
@ -89,7 +89,7 @@ class TestPostSamples(FunctionalTest,
"bc23a9d531064583ace8f67dad60f6bb",
})
self.assertEquals(data.status_int, 400)
self.assertEqual(data.status_int, 400)
def test_multiple_samples(self):
"""Send multiple samples.
@ -121,14 +121,14 @@ class TestPostSamples(FunctionalTest,
for (k, v) in samples[x].iteritems():
if k == 'timestamp':
timestamp = timeutils.parse_isotime(data.json[x][k])
self.assertEquals(stamps[x].replace(tzinfo=None),
timestamp.replace(tzinfo=None))
self.assertEqual(stamps[x].replace(tzinfo=None),
timestamp.replace(tzinfo=None))
elif k == 'source':
self.assertEquals(data.json[x][k],
'%s:%s' % (samples[x]['project_id'],
samples[x]['source']))
self.assertEqual(data.json[x][k],
'%s:%s' % (samples[x]['project_id'],
samples[x]['source']))
else:
self.assertEquals(v, data.json[x][k])
self.assertEqual(v, data.json[x][k])
def test_missing_mandatory_fields(self):
"""Do not accept posting samples with missing mandatory fields."""
@ -151,7 +151,7 @@ class TestPostSamples(FunctionalTest,
print('posting without %s' % m)
data = self.post_json('/meters/my_counter_name/', s_broke,
expect_errors=True)
self.assertEquals(data.status_int, 400)
self.assertEqual(data.status_int, 400)
def test_multiple_sources(self):
"""Do not accept a single post of mixed sources."""
@ -176,7 +176,7 @@ class TestPostSamples(FunctionalTest,
'name2': 'value2'}}]
data = self.post_json('/meters/my_counter_name/', s1,
expect_errors=True)
self.assertEquals(data.status_int, 400)
self.assertEqual(data.status_int, 400)
def test_multiple_samples_some_null_sources(self):
"""Do accept a single post with some null sources
@ -203,13 +203,13 @@ class TestPostSamples(FunctionalTest,
'name2': 'value2'}}]
data = self.post_json('/meters/my_counter_name/', s1,
expect_errors=True)
self.assertEquals(data.status_int, 200)
self.assertEqual(data.status_int, 200)
for x in range(2):
for (k, v) in s1[x].iteritems():
if k == 'source':
self.assertEquals(data.json[x][k],
'%s:%s' % (s1[x]['project_id'],
'paperstack'))
self.assertEqual(data.json[x][k],
'%s:%s' % (s1[x]['project_id'],
'paperstack'))
def test_missing_project_user_id(self):
"""Ensure missing project & user IDs are defaulted appropriately.
@ -233,6 +233,6 @@ class TestPostSamples(FunctionalTest,
'X-User-Id': user_id,
})
self.assertEquals(data.status_int, 200)
self.assertEquals(data.json[0]['project_id'], project_id)
self.assertEquals(data.json[0]['user_id'], user_id)
self.assertEqual(data.status_int, 200)
self.assertEqual(data.json[0]['project_id'], project_id)
self.assertEqual(data.json[0]['user_id'], user_id)

22
tests/collector/test_service.py
View File

@ -237,11 +237,11 @@ class TestCollectorService(TestCollector):
self.srv._message_to_event(message)
self.assertFalse(mylog.exception.called)
events = self.srv.storage_conn.record_events.call_args[0]
self.assertEquals(1, len(events))
self.assertEqual(1, len(events))
event = events[0][0]
self.assertEquals("foo", event.event_name)
self.assertEquals(now, event.generated)
self.assertEquals(1, len(event.traits))
self.assertEqual("foo", event.event_name)
self.assertEqual(now, event.generated)
self.assertEqual(1, len(event.traits))
def test_message_to_event_bad_save(self):
cfg.CONF.set_override("store_events", True, group="collector")
@ -260,16 +260,16 @@ class TestCollectorService(TestCollector):
timeutils.set_time_override(now)
body = {"timestamp": str(modified)}
self.assertEquals(service.CollectorService._extract_when(body),
modified)
self.assertEqual(service.CollectorService._extract_when(body),
modified)
body = {"_context_timestamp": str(modified)}
self.assertEquals(service.CollectorService._extract_when(body),
modified)
self.assertEqual(service.CollectorService._extract_when(body),
modified)
then = now + datetime.timedelta(hours=1)
body = {"timestamp": str(modified), "_context_timestamp": str(then)}
self.assertEquals(service.CollectorService._extract_when(body),
modified)
self.assertEqual(service.CollectorService._extract_when(body),
modified)
self.assertEquals(service.CollectorService._extract_when({}), now)
self.assertEqual(service.CollectorService._extract_when({}), now)

11
tests/compute/pollsters/test_cpu.py
View File

@ -51,12 +51,11 @@ class TestCPUPollster(base.TestPollsterBase):
def _verify_cpu_metering(expected_time):
cache = {}
samples = list(pollster.get_samples(mgr, cache, self.instance))
self.assertEquals(len(samples), 1)
self.assertEqual(len(samples), 1)
self.assertEqual(set([s.name for s in samples]),
set(['cpu']))
assert samples[0].volume == expected_time
self.assertEquals(samples[0].resource_metadata.get('cpu_number'),
2)
self.assertEqual(samples[0].resource_metadata.get('cpu_number'), 2)
# ensure elapsed time between polling cycles is non-zero
time.sleep(0.001)
@ -75,6 +74,6 @@ class TestCPUPollster(base.TestPollsterBase):
cache = {}
samples = list(pollster.get_samples(mgr, cache, self.instance))
self.assertEquals(len(samples), 1)
self.assertEquals(samples[0].volume, 10 ** 6)
self.assertEquals(len(cache), 0)
self.assertEqual(len(samples), 1)
self.assertEqual(samples[0].volume, 10 ** 6)
self.assertEqual(len(cache), 0)

6
tests/compute/pollsters/test_diskio.py
View File

@ -56,9 +56,9 @@ class TestDiskPollsters(base.TestPollsterBase):
set([name]))
match = [s for s in samples if s.name == name]
self.assertEquals(len(match), 1, 'missing counter %s' % name)
self.assertEquals(match[0].volume, expected_volume)
self.assertEquals(match[0].type, 'cumulative')
self.assertEqual(len(match), 1, 'missing counter %s' % name)
self.assertEqual(match[0].volume, expected_volume)
self.assertEqual(match[0].type, 'cumulative')
def test_disk_read_requests(self):
self._check_get_samples(disk.ReadRequestsPollster,

4
tests/compute/pollsters/test_instance.py
View File

@ -38,7 +38,7 @@ class TestInstancePollster(base.TestPollsterBase):
mgr = manager.AgentManager()
pollster = pollsters_instance.InstancePollster()
samples = list(pollster.get_samples(mgr, {}, self.instance))
self.assertEquals(len(samples), 1)
self.assertEqual(len(samples), 1)
self.assertEqual(samples[0].name, 'instance')
@mock.patch('ceilometer.pipeline.setup_pipeline', mock.MagicMock())
@ -48,5 +48,5 @@ class TestInstancePollster(base.TestPollsterBase):
mgr = manager.AgentManager()
pollster = pollsters_instance.InstanceFlavorPollster()
samples = list(pollster.get_samples(mgr, {}, self.instance))
self.assertEquals(len(samples), 1)
self.assertEqual(len(samples), 1)
self.assertEqual(samples[0].name, 'instance:m1.small')

8
tests/compute/pollsters/test_net.py
View File

@ -83,10 +83,10 @@ class TestNetPollster(base.TestPollsterBase):
match = [s for s in samples
if s.resource_metadata['parameters']['ip'] == ip
]
self.assertEquals(len(match), 1, 'missing ip %s' % ip)
self.assertEquals(match[0].volume, expected_volume)
self.assertEquals(match[0].type, 'cumulative')
self.assertEquals(match[0].resource_id, expected_rid)
self.assertEqual(len(match), 1, 'missing ip %s' % ip)
self.assertEqual(match[0].volume, expected_volume)
self.assertEqual(match[0].type, 'cumulative')
self.assertEqual(match[0].resource_id, expected_rid)
for ip, volume, rid in expected:
_verify_vnic_metering(ip, volume, rid)

4
tests/compute/virt/libvirt/test_inspector.py
View File

@ -96,7 +96,7 @@ class TestLibvirtInspection(test_base.TestCase):
interfaces = list(self.inspector.inspect_vnics(self.instance_name))
self.assertEquals(len(interfaces), 3)
self.assertEqual(len(interfaces), 3)
vnic0, info0 = interfaces[0]
self.assertEqual(vnic0.name, 'vnet0')
self.assertEqual(vnic0.mac, 'fa:16:3e:71:ec:6d')
@ -158,7 +158,7 @@ class TestLibvirtInspection(test_base.TestCase):
disks = list(self.inspector.inspect_disks(self.instance_name))
self.assertEquals(len(disks), 1)
self.assertEqual(len(disks), 1)
disk0, info0 = disks[0]
self.assertEqual(disk0.device, 'vda')
self.assertEqual(info0.read_requests, 1L)

10
tests/image/test_notifications.py
View File

@ -98,7 +98,7 @@ class TestNotification(base.TestCase):
self.assertEqual(c.timestamp, NOW)
self.assertEqual(c.volume, volume)
metadata = c.resource_metadata
self.assertEquals(metadata.get('host'), u'images.example.com')
self.assertEqual(metadata.get('host'), u'images.example.com')
def test_image_download(self):
handler = notifications.ImageDownload()
@ -117,10 +117,10 @@ class TestNotification(base.TestCase):
serve = counters[0]
self._verify_common_counter(serve, 'image.serve', 42)
self.assertEqual(serve.project_id, fake_uuid('e'))
self.assertEquals(serve.resource_metadata.get('receiver_user_id'),
fake_uuid('d'))
self.assertEquals(serve.resource_metadata.get('receiver_tenant_id'),
fake_uuid('b'))
self.assertEqual(serve.resource_metadata.get('receiver_user_id'),
fake_uuid('d'))
self.assertEqual(serve.resource_metadata.get('receiver_tenant_id'),
fake_uuid('b'))
self.assertEqual(serve.type, sample.TYPE_DELTA)
def test_image_crud_on_update(self):

109
tests/storage/base.py
View File

@ -357,25 +357,24 @@ class ResourceTestPagination(DBTestBase):
results = self.conn.get_resources(limit=3, marker_pairs=marker_pairs,
sort_key='user_id',
sort_dir='asc')
self.assertEquals(['user-id-5', 'user-id-6', 'user-id-7'],
[i.user_id for i in results])
self.assertEqual(['user-id-5', 'user-id-6', 'user-id-7'],
[i.user_id for i in results])
marker_pairs = {'user_id': 'user-id-4'}
results = list(self.conn.get_resources(limit=2,
marker_pairs=marker_pairs,
sort_key='user_id',
sort_dir='desc'))
self.assertEquals(['user-id-3', 'user-id-2'],
[i.user_id for i in results])
self.assertEqual(['user-id-3', 'user-id-2'],
[i.user_id for i in results])
marker_pairs = {'project_id': 'project-id-5'}
results = list(self.conn.get_resources(limit=3,
marker_pairs=marker_pairs,
sort_key='user_id',
sort_dir='asc'))
self.assertEquals(['resource-id-6', 'resource-id-7',
'resource-id-8'],
[i.resource_id for i in results])
self.assertEqual(['resource-id-6', 'resource-id-7', 'resource-id-8'],
[i.resource_id for i in results])
class MeterTest(DBTestBase):
@ -431,28 +430,28 @@ class MeterTestPagination(DBTestBase):
marker_pairs = {'user_id': 'user-id-alternate'}
results = self.conn.get_meters(limit=3, marker_pairs=marker_pairs,
sort_key='user_id', sort_dir='desc')
self.assertEquals(['user-id-8', 'user-id-7', 'user-id-6'],
[i.user_id for i in results])
self.assertEqual(['user-id-8', 'user-id-7', 'user-id-6'],
[i.user_id for i in results])
marker_pairs = {'user_id': 'user-id-4'}
results = self.conn.get_meters(limit=3, marker_pairs=marker_pairs,
sort_key='user_id',
sort_dir='asc')
self.assertEquals(['user-id-5', 'user-id-6', 'user-id-7'],
[i.user_id for i in results])
self.assertEqual(['user-id-5', 'user-id-6', 'user-id-7'],
[i.user_id for i in results])
marker_pairs = {'user_id': 'user-id-4'}
results = list(self.conn.get_meters(limit=2,
marker_pairs=marker_pairs,
sort_key='user_id',
sort_dir='desc'))
self.assertEquals(['user-id-3', 'user-id-2'],
[i.user_id for i in results])
self.assertEqual(['user-id-3', 'user-id-2'],
[i.user_id for i in results])
marker_pairs = {'user_id': 'user-id'}
results = self.conn.get_meters(limit=3, marker_pairs=marker_pairs,
sort_key='user_id', sort_dir='desc')
self.assertEquals([], [i.user_id for i in results])
self.assertEqual([], [i.user_id for i in results])
class RawSampleTest(DBTestBase):
@ -823,8 +822,7 @@ class StatisticsTest(DBTestBase):
self.assertEqual(r.sum, 8)
self.assertEqual(r.period, 1800)
self.assertEqual(r.period_end,
r.period_start
+ datetime.timedelta(seconds=1800))
r.period_start + datetime.timedelta(seconds=1800))
self.assertEqual(r.duration, 0)
self.assertEqual(r.duration_start,
datetime.datetime(2012, 9, 25, 10, 30))
@ -974,29 +972,28 @@ class AlarmTest(AlarmTestBase):
def test_empty(self):
alarms = list(self.conn.get_alarms())
self.assertEquals([], alarms)
self.assertEqual([], alarms)
def test_add(self):
self.add_some_alarms()
alarms = list(self.conn.get_alarms())
self.assertEquals(len(alarms), 3)
self.assertEqual(len(alarms), 3)
def test_defaults(self):
self.add_some_alarms()
yellow = list(self.conn.get_alarms(name='yellow-alert'))[0]
self.assertEquals(yellow.evaluation_periods, 1)
self.assertEquals(yellow.period, 60)
self.assertEquals(yellow.enabled, True)
self.assertEquals(yellow.description,
'Alarm when test.five is lt %s' %
'a min of 10 over 60 seconds')
self.assertEquals(yellow.state, models.Alarm.ALARM_INSUFFICIENT_DATA)
self.assertEquals(yellow.ok_actions, [])
self.assertEquals(yellow.insufficient_data_actions, [])
self.assertEquals(yellow.matching_metadata,
{'key2': 'value2',
'user_metadata.key3': 'value3'})
self.assertEqual(yellow.evaluation_periods, 1)
self.assertEqual(yellow.period, 60)
self.assertEqual(yellow.enabled, True)
self.assertEqual(yellow.description,
'Alarm when test.five is lt '
'a min of 10 over 60 seconds')
self.assertEqual(yellow.state, models.Alarm.ALARM_INSUFFICIENT_DATA)
self.assertEqual(yellow.ok_actions, [])
self.assertEqual(yellow.insufficient_data_actions, [])
self.assertEqual(yellow.matching_metadata,
{'key2': 'value2', 'user_metadata.key3': 'value3'})
def test_update(self):
self.add_some_alarms()
@ -1006,11 +1003,10 @@ class AlarmTest(AlarmTestBase):
orange.matching_metadata = {'new': 'value',
'user_metadata.new2': 'value4'}
updated = self.conn.update_alarm(orange)
self.assertEquals(updated.enabled, False)
self.assertEquals(updated.state, models.Alarm.ALARM_INSUFFICIENT_DATA)
self.assertEquals(updated.matching_metadata,
{'new': 'value',
'user_metadata.new2': 'value4'})
self.assertEqual(updated.enabled, False)
self.assertEqual(updated.state, models.Alarm.ALARM_INSUFFICIENT_DATA)
self.assertEqual(updated.matching_metadata,
{'new': 'value', 'user_metadata.new2': 'value4'})
def test_update_llu(self):
llu = models.Alarm('llu',
@ -1022,14 +1018,14 @@ class AlarmTest(AlarmTestBase):
self.conn.update_alarm(updated)
all = list(self.conn.get_alarms())
self.assertEquals(len(all), 1)
self.assertEqual(len(all), 1)
def test_delete(self):
self.add_some_alarms()
victim = list(self.conn.get_alarms(name='orange-alert'))[0]
self.conn.delete_alarm(victim.alarm_id)
survivors = list(self.conn.get_alarms())
self.assertEquals(len(survivors), 2)
self.assertEqual(len(survivors), 2)
for s in survivors:
self.assertNotEquals(victim.name, s.name)
@ -1088,15 +1084,14 @@ class AlarmTestPagination(AlarmTestBase):
page = list(self.conn.get_alarms(limit=4,
marker_pairs=marker_pairs,
sort_key='name', sort_dir='desc'))
self.assertEquals(['red-alert', 'orange-alert'],
[i.name for i in page])
self.assertEqual(['red-alert', 'orange-alert'], [i.name for i in page])
marker_pairs = {'name': 'orange-alert'}
page1 = list(self.conn.get_alarms(limit=2,
sort_key='comparison_operator',
sort_dir='desc',
marker_pairs=marker_pairs))
self.assertEquals(['red-alert'], [i.name for i in page1])
self.assertEqual(['red-alert'], [i.name for i in page1])
class EventTestBase(test_db.TestBase):
@ -1125,42 +1120,42 @@ class EventTest(EventTestBase):
def test_string_traits(self):
model = models.Trait("Foo", models.Trait.TEXT_TYPE, "my_text")
trait = self.conn._make_trait(model, None)
self.assertEquals(trait.t_type, models.Trait.TEXT_TYPE)
self.assertEqual(trait.t_type, models.Trait.TEXT_TYPE)
self.assertIsNone(trait.t_float)
self.assertIsNone(trait.t_int)
self.assertIsNone(trait.t_datetime)
self.assertEquals(trait.t_string, "my_text")
self.assertEqual(trait.t_string, "my_text")
self.assertIsNotNone(trait.name)
def test_int_traits(self):
model = models.Trait("Foo", models.Trait.INT_TYPE, 100)
trait = self.conn._make_trait(model, None)
self.assertEquals(trait.t_type, models.Trait.INT_TYPE)
self.assertEqual(trait.t_type, models.Trait.INT_TYPE)
self.assertIsNone(trait.t_float)
self.assertIsNone(trait.t_string)
self.assertIsNone(trait.t_datetime)
self.assertEquals(trait.t_int, 100)
self.assertEqual(trait.t_int, 100)
self.assertIsNotNone(trait.name)
def test_float_traits(self):
model = models.Trait("Foo", models.Trait.FLOAT_TYPE, 123.456)
trait = self.conn._make_trait(model, None)
self.assertEquals(trait.t_type, models.Trait.FLOAT_TYPE)
self.assertEqual(trait.t_type, models.Trait.FLOAT_TYPE)
self.assertIsNone(trait.t_int)
self.assertIsNone(trait.t_string)
self.assertIsNone(trait.t_datetime)
self.assertEquals(trait.t_float, 123.456)
self.assertEqual(trait.t_float, 123.456)
self.assertIsNotNone(trait.name)
def test_datetime_traits(self):
now = datetime.datetime.utcnow()
model = models.Trait("Foo", models.Trait.DATETIME_TYPE, now)
trait = self.conn._make_trait(model, None)
self.assertEquals(trait.t_type, models.Trait.DATETIME_TYPE)
self.assertEqual(trait.t_type, models.Trait.DATETIME_TYPE)
self.assertIsNone(trait.t_int)
self.assertIsNone(trait.t_string)
self.assertIsNone(trait.t_float)
self.assertEquals(trait.t_datetime, utils.dt_to_decimal(now))
self.assertEqual(trait.t_datetime, utils.dt_to_decimal(now))
self.assertIsNotNone(trait.name)
def test_save_events_traits(self):
@ -1211,11 +1206,11 @@ class GetEventTest(EventTestBase):
def test_simple_get(self):
event_filter = storage.EventFilter(self.start, self.end)
events = self.conn.get_events(event_filter)
self.assertEquals(3, len(events))
self.assertEqual(3, len(events))
start_time = None
for i, name in enumerate(["Foo", "Bar", "Zoo"]):
self.assertEquals(events[i].event_name, name)
self.assertEquals(4, len(events[i].traits))
self.assertEqual(events[i].event_name, name)
self.assertEqual(4, len(events[i].traits))
# Ensure sorted results ...
if start_time is not None:
# Python 2.6 has no assertLess :(
@ -1225,15 +1220,15 @@ class GetEventTest(EventTestBase):
def test_simple_get_event_name(self):
event_filter = storage.EventFilter(self.start, self.end, "Bar")
events = self.conn.get_events(event_filter)
self.assertEquals(1, len(events))
self.assertEquals(events[0].event_name, "Bar")
self.assertEquals(4, len(events[0].traits))
self.assertEqual(1, len(events))
self.assertEqual(events[0].event_name, "Bar")
self.assertEqual(4, len(events[0].traits))
def test_get_event_trait_filter(self):
trait_filters = {'key': 'trait_B', 't_int': 101}
event_filter = storage.EventFilter(self.start, self.end,
traits=trait_filters)
events = self.conn.get_events(event_filter)
self.assertEquals(1, len(events))
self.assertEquals(events[0].event_name, "Bar")
self.assertEquals(4, len(events[0].traits))
self.assertEqual(1, len(events))
self.assertEqual(events[0].event_name, "Bar")
self.assertEqual(4, len(events[0].traits))

6
tests/storage/test_base.py
View File

@ -52,10 +52,10 @@ class BaseTest(test_base.TestCase):
def test_handle_sort_key(self):
sort_keys_alarm = base._handle_sort_key('alarm')
self.assertEquals(sort_keys_alarm, ['name', 'user_id', 'project_id'])
self.assertEqual(sort_keys_alarm, ['name', 'user_id', 'project_id'])
sort_keys_meter = base._handle_sort_key('meter', 'foo')
self.assertEquals(sort_keys_meter, ['foo', 'user_id', 'project_id'])
self.assertEqual(sort_keys_meter, ['foo', 'user_id', 'project_id'])
sort_keys_resource = base._handle_sort_key('resource', 'project_id')
self.assertEquals(sort_keys_resource, ['project_id', 'user_id'])
self.assertEqual(sort_keys_resource, ['project_id', 'user_id'])

4
tests/storage/test_impl_mongodb.py
View File

@ -66,7 +66,7 @@ class MongoDBConnection(MongoDBEngineTestBase):
marker=marker,
flag=flag)
expect = {'k3': {'$lt': 'v3'}, 'k2': {'eq': 'v2'}, 'k1': {'eq': 'v1'}}
self.assertEquals(ret, expect)
self.assertEqual(ret, expect)
class MongoDBTestMarkerBase(MongoDBEngineTestBase):
@ -174,7 +174,7 @@ class AlarmTest(base.AlarmTest, MongoDBEngineTestBase):
def test_alarm_get_old_matching_metadata_format(self):
self.prepare_old_matching_metadata_alarm()
old = list(self.conn.get_alarms(name='old-alert'))[0]
self.assertEquals(old.matching_metadata, {'key': 'value'})
self.assertEqual(old.matching_metadata, {'key': 'value'})
class CompatibilityTest(MongoDBEngineTestBase):

42
tests/test_pipeline.py
View File

@ -747,10 +747,10 @@ class TestPipeline(base.TestCase):
pipe.flush(None)
self.assertEqual(len(publisher.counters), 1)
cpu_mins = publisher.counters[-1]
self.assertEquals(getattr(cpu_mins, 'name'), 'cpu_mins')
self.assertEquals(getattr(cpu_mins, 'unit'), 'min')
self.assertEquals(getattr(cpu_mins, 'type'), sample.TYPE_CUMULATIVE)
self.assertEquals(getattr(cpu_mins, 'volume'), 20)
self.assertEqual(getattr(cpu_mins, 'name'), 'cpu_mins')
self.assertEqual(getattr(cpu_mins, 'unit'), 'min')
self.assertEqual(getattr(cpu_mins, 'type'), sample.TYPE_CUMULATIVE)
self.assertEqual(getattr(cpu_mins, 'volume'), 20)
def test_unit_identified_source_unit_conversion(self):
self.pipeline_cfg[0]['transformers'] = [
@ -798,14 +798,14 @@ class TestPipeline(base.TestCase):
publisher = pipeline_manager.pipelines[0].publishers[0]
self.assertEqual(len(publisher.counters), 2)
core_temp = publisher.counters[1]
self.assertEquals(getattr(core_temp, 'name'), 'core_temperature')
self.assertEquals(getattr(core_temp, 'unit'), '°F')
self.assertEquals(getattr(core_temp, 'volume'), 96.8)
self.assertEqual(getattr(core_temp, 'name'), 'core_temperature')
self.assertEqual(getattr(core_temp, 'unit'), '°F')
self.assertEqual(getattr(core_temp, 'volume'), 96.8)
amb_temp = publisher.counters[0]
self.assertEquals(getattr(amb_temp, 'name'), 'ambient_temperature')
self.assertEquals(getattr(amb_temp, 'unit'), '°F')
self.assertEquals(getattr(amb_temp, 'volume'), 88.8)
self.assertEquals(getattr(core_temp, 'volume'), 96.8)
self.assertEqual(getattr(amb_temp, 'name'), 'ambient_temperature')
self.assertEqual(getattr(amb_temp, 'unit'), '°F')
self.assertEqual(getattr(amb_temp, 'volume'), 88.8)
self.assertEqual(getattr(core_temp, 'volume'), 96.8)
def _do_test_rate_of_change_conversion(self, prev, curr, type, expected,
offset=1, weight=None):
@ -889,17 +889,17 @@ class TestPipeline(base.TestCase):
pipe.flush(None)
self.assertEqual(len(publisher.counters), 2)
cpu_util = publisher.counters[0]
self.assertEquals(getattr(cpu_util, 'name'), 'cpu_util')
self.assertEquals(getattr(cpu_util, 'resource_id'), 'test_resource')
self.assertEquals(getattr(cpu_util, 'unit'), '%')
self.assertEquals(getattr(cpu_util, 'type'), sample.TYPE_GAUGE)
self.assertEquals(getattr(cpu_util, 'volume'), expected)
self.assertEqual(getattr(cpu_util, 'name'), 'cpu_util')
self.assertEqual(getattr(cpu_util, 'resource_id'), 'test_resource')
self.assertEqual(getattr(cpu_util, 'unit'), '%')
self.assertEqual(getattr(cpu_util, 'type'), sample.TYPE_GAUGE)
self.assertEqual(getattr(cpu_util, 'volume'), expected)
cpu_util = publisher.counters[1]
self.assertEquals(getattr(cpu_util, 'name'), 'cpu_util')
self.assertEquals(getattr(cpu_util, 'resource_id'), 'test_resource2')
self.assertEquals(getattr(cpu_util, 'unit'), '%')
self.assertEquals(getattr(cpu_util, 'type'), sample.TYPE_GAUGE)
self.assertEquals(getattr(cpu_util, 'volume'), expected * 2)
self.assertEqual(getattr(cpu_util, 'name'), 'cpu_util')
self.assertEqual(getattr(cpu_util, 'resource_id'), 'test_resource2')
self.assertEqual(getattr(cpu_util, 'unit'), '%')
self.assertEqual(getattr(cpu_util, 'type'), sample.TYPE_GAUGE)
self.assertEqual(getattr(cpu_util, 'volume'), expected * 2)
def test_rate_of_change_conversion(self):
self._do_test_rate_of_change_conversion(120000000000,

2
tests/volume/test_notifications.py
View File

@ -80,7 +80,7 @@ class TestNotifications(base.TestCase):
self.assertEqual(c.resource_id, notification['payload']['volume_id'])
self.assertEqual(c.timestamp, notification['timestamp'])
metadata = c.resource_metadata
self.assertEquals(metadata.get('host'), notification['publisher_id'])
self.assertEqual(metadata.get('host'), notification['publisher_id'])
def test_volume_exists(self):
v = notifications.Volume()

2
tox.ini
View File

@ -22,7 +22,7 @@ commands = bash -x {toxinidir}/run-tests.sh --coverage
deps = pep8==1.4.5
pyflakes==0.7.2
flake8==2.0
hacking>=0.5.6,<0.7
hacking>=0.5.6,<0.8
commands =
flake8
flake8 --filename=ceilometer-* bin