Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
all = list(self.conn.get_alarms())
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.assertEqual(len(survivors), 2)
for s in survivors:
self.assertNotEquals(victim.name, s.name)
class AlarmTestPagination(AlarmTestBase,
tests_db.MixinTestsWithBackendScenarios):
def test_get_alarm_all_limit(self):
self.add_some_alarms()
pagination = Pagination(limit=2)
alarms = list(self.conn.get_alarms(pagination=pagination))
self.assertEqual(len(alarms), 2)
pagination = Pagination(limit=1)
alarms = list(self.conn.get_alarms(pagination=pagination))
self.assertEqual(len(alarms), 1)
def test_get_alarm_all_marker(self):
self.add_some_alarms()
pagination = Pagination(marker_value='orange-alert')
alarms = list(self.conn.get_alarms(pagination=pagination))
'counter-3', 'counter-4', 'counter-5', 'counter-6',
'counter-7', 'counter-8']
for resource in resources:
self.assertIn(resource.metadata['tag'], expected_tags)
def test_get_resources_most_recent_metadata_single(self):
resource = list(
self.conn.get_resources(resource='resource-id-alternate')
)[0]
expected_tag = 'self.counter3'
self.assertEqual(resource.metadata['tag'], expected_tag)
class ResourceTestPagination(DBTestBase,
tests_db.MixinTestsWithBackendScenarios):
def test_get_resource_all_limit(self):
pagination = Pagination(limit=8)
results = list(self.conn.get_resources(pagination=pagination))
self.assertEqual(len(results), 8)
pagination = Pagination(limit=5)
results = list(self.conn.get_resources(pagination=pagination))
self.assertEqual(len(results), 5)
def test_get_resources_all_marker(self):
pagination = Pagination(primary_sort_dir='asc', sort_keys=['user_id'],
sort_dirs=['asc'],
marker_value='resource-id-4')
results = list(self.conn.get_resources(pagination=pagination))
self.assertEqual(len(results), 5)
self.msgs.append(
self.create_and_store_sample(
timestamp=datetime.datetime(*ts),
user_id='user-id-%s' % i,
project_id='project-id-%s' % i,
resource_id='resource-id-%s' % i,
metadata={
'display_name': 'test-server',
'tag': 'counter-%s' % i
},
source='test')
)
class UserTest(DBTestBase,
tests_db.MixinTestsWithBackendScenarios):
def test_get_users(self):
users = self.conn.get_users()
expected = set(['user-id', 'user-id-alternate', 'user-id-2',
'user-id-3', 'user-id-4', 'user-id-5', 'user-id-6',
'user-id-7', 'user-id-8'])
self.assertEqual(set(users), expected)
def test_get_users_by_source(self):
users = self.conn.get_users(source='test-1')
self.assertEqual(list(users), ['user-id'])
class ProjectTest(DBTestBase,
tests_db.MixinTestsWithBackendScenarios):
self.assertEqual(r['period'], 7200)
self.assertEqual(r['period_end'], u'2013-08-01T18:00:00')
else:
self.assertNotEqual([grp, period_start],
[{'project_id': 'project-1'},
u'2013-08-01T16:00:00'])
self.assertNotEqual([grp, period_start],
[{'project_id': 'project-2'},
u'2013-08-01T10:00:00'])
self.assertNotEqual([grp, period_start],
[{'project_id': 'project-2'},
u'2013-08-01T14:00:00'])
class TestGroupBySource(base.FunctionalTest,
tests_db.MixinTestsWithBackendScenarios):
# FIXME(terriyu): We have to put test_group_by_source in its own class
# because SQLAlchemy currently doesn't support group by source statistics.
# When group by source is supported in SQLAlchemy, this test should be
# moved to TestGroupByInstance with all the other group by statistics
# tests.
scenarios = [
('mongodb',
dict(database_connection=tests_db.MongoDBFakeConnectionUrl())),
('hbase', dict(database_connection='hbase://__test__')),
('db2', dict(database_connection=tests_db.DB2FakeConnectionUrl())),
]
PATH = '/meters/instance/statistics'
"""
import datetime
import testscenarios
from ceilometer.publisher import rpc
from ceilometer import sample
from ceilometer.tests import api as tests_api
from ceilometer.tests import db as tests_db
load_tests = testscenarios.load_tests_apply_scenarios
class TestSumResourceVolume(tests_api.TestBase,
tests_db.MixinTestsWithBackendScenarios):
def setUp(self):
super(TestSumResourceVolume, self).setUp()
for i in range(3):
s = sample.Sample(
'volume.size',
'gauge',
'GiB',
5 + i,
'user-id',
'project1',
'resource-id',
timestamp=datetime.datetime(2012, 9, 25, 10 + i, 30 + i),
resource_metadata={'display_name': 'test-volume',
'tag': 'self.sample',
},
load_tests = testscenarios.load_tests_apply_scenarios
LOG = logging.getLogger(__name__)
class TestListEmptyResources(tests_api.TestBase,
tests_db.MixinTestsWithBackendScenarios):
def test_empty(self):
data = self.get('/resources')
self.assertEqual({'resources': []}, data)
class TestListResourcesBase(tests_api.TestBase,
tests_db.MixinTestsWithBackendScenarios):
def setUp(self):
super(TestListResourcesBase, self).setUp()
for cnt in [
sample.Sample(
'instance',
'cumulative',
'',
1,
'user-id',
'project-id',
'resource-id',
timestamp=datetime.datetime(2012, 7, 2, 10, 40),
resource_metadata={'display_name': 'test-server',
'tag': 'self.sample'},
},
{'field': 'timestamp',
'op': 'ge',
'value': '2012-09-25T11:30:00',
},
{'field': 'timestamp',
'op': 'le',
'value': '2012-09-25T11:32:00',
},
])
self.assertEqual(data[0]['max'], 6)
self.assertEqual(data[0]['count'], 1)
class TestMaxResourceVolume(base.FunctionalTest,
tests_db.MixinTestsWithBackendScenarios):
PATH = '/meters/volume.size/statistics'
def setUp(self):
super(TestMaxResourceVolume, self).setUp()
for i in range(3):
s = sample.Sample(
'volume.size',
'gauge',
'GiB',
5 + i,
'user-id',
'project1',
'resource-id',
timestamp=datetime.datetime(2012, 9, 25, 10 + i, 30 + i),
resource_metadata={'display_name': 'test-volume',
headers={"X-Roles": "Member",
"X-Project-Id": "project-id"},
start_timestamp=datetime.datetime(
2012, 7, 2, 10, 40).isoformat(),
end_timestamp=datetime.datetime(
2012, 7, 2, 10, 41).isoformat())
ids = set(r['resource_id'] for r in data['resources'])
self.assertEqual(set(['resource-id']), ids)
def test_with_project_non_existent(self):
data = self.get('/projects/jd-was-here/resources')
self.assertEqual(data['resources'], [])
class TestListResourcesMetaquery(TestListResourcesBase,
tests_db.MixinTestsWithBackendScenarios):
def test_metaquery1(self):
q = '/sources/test_list_resources/resources'
data = self.get('%s?metadata.display_name=test-server' % q)
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.assertEqual(2, len(data['resources']))
def test_metaquery2(self):
q = '/sources/test_list_resources/resources'
data = self.get('%s?metadata.tag=self.sample4' % q)
load_tests = testscenarios.load_tests_apply_scenarios
LOG = logging.getLogger(__name__)
class TestListEmptyMeters(tests_api.TestBase,
tests_db.MixinTestsWithBackendScenarios):
def test_empty(self):
data = self.get('/meters')
self.assertEqual({'meters': []}, data)
class TestListMeters(tests_api.TestBase,
tests_db.MixinTestsWithBackendScenarios):
def setUp(self):
super(TestListMeters, self).setUp()
for cnt in [
sample.Sample(
'meter.test',
'cumulative',
'',
1,
'user-id',
'project-id',
'resource-id',
timestamp=datetime.datetime(2012, 7, 2, 10, 40),
resource_metadata={'display_name': 'test-server',
'tag': 'self.sample'},
source='test_list_resources'),
load_tests = testscenarios.load_tests_apply_scenarios
LOG = logging.getLogger(__name__)
class TestListEmptyAlarms(FunctionalTest,
tests_db.MixinTestsWithBackendScenarios):
def test_empty(self):
data = self.get_json('/alarms')
self.assertEqual([], data)
class TestAlarms(FunctionalTest,
tests_db.MixinTestsWithBackendScenarios):
def setUp(self):
super(TestAlarms, self).setUp()
self.auth_headers = {'X-User-Id': str(uuid.uuid4()),
'X-Project-Id': str(uuid.uuid4())}
for alarm in [Alarm(name='name1',
type='threshold',
enabled=True,
alarm_id='a',
description='a',
state='insufficient data',
state_timestamp=None,
timestamp=None,
ok_actions=[],
insufficient_data_actions=[],
alarm_actions=[],