← Back to branch summary

~ubuntu-branches/ubuntu/trusty/ceilometer/trusty-proposed

« back to all changes in this revision

Viewing changes to ceilometer/tests/alarm/partition/test_coordination.py

  • Committer: Package Import Robot
  • Author(s): Chuck Short, James Page, Chuck Short
  • Date: 2014-01-23 15:08:11 UTC
  • mfrom: (1.1.11)
  • Revision ID: package-import@ubuntu.com-20140123150811-1zaismsuyh1hcl8y
Tags: 2014.1~b2-0ubuntu1
https://launchpad.net/bugs/1261101
https://launchpad.net/bugs/1250002
https://launchpad.net/bugs/1224223
[ James Page ]
* d/control: Add python-jsonpath-rw to BD's.
* d/p/fix-setup-requirements.patch: Bump WebOb to support < 1.4.
 (LP: #1261101)

[ Chuck Short ]
* New upstream version.
* debian/control, debian/ceilometer-common.install: Split out
  ceilometer-alarm-evaluator and ceilometer-alarm-notifier into their
  own packages. (LP: #1250002)
* debian/ceilometer-agent-central.logrotate,
  debian/ceilometer-agent-compute.logrotate,
  debian/ceilometer-api.logrotate,
  debian/ceilometer-collector.logrotate: Add logrotate files, 
  thanks to Ahmed Rahal. (LP: #1224223)
* Fix typos in upstart files.

Show diffs side-by-side

added added

removed removed

Lines of Context:
ceilometer/tests/alarm/partition/test_coordination.py
18
18
"""Tests for ceilometer/alarm/partition/coordination.py
19
19
"""
20
20
import datetime
 
21
import logging
 
22
import StringIO
21
23
import uuid
22
24
 
23
25
import mock
42
44
        timeutils.utcnow.override_time = self.override_start
43
45
        self.partition_coordinator = coordination.PartitionCoordinator()
44
46
        self.partition_coordinator.coordination_rpc = mock.Mock()
 
47
        #add extra logger to check exception conditions and logged content
 
48
        self.output = StringIO.StringIO()
 
49
        self.str_handler = logging.StreamHandler(self.output)
 
50
        coordination.LOG.logger.addHandler(self.str_handler)
45
51
 
46
52
    def tearDown(self):
47
53
        super(TestCoordinate, self).tearDown()
48
54
        timeutils.utcnow.override_time = None
 
55
        # clean up the logger
 
56
        coordination.LOG.logger.removeHandler(self.str_handler)
 
57
        self.output.close()
49
58
 
50
59
    def _no_alarms(self):
51
60
        self.api_client.alarms.list.return_value = []
381
390
 
382
391
        alarms = self.partition_coordinator.assigned_alarms(self.api_client)
383
392
        self.assertEqual(alarms, self._current_alarms())
 
393
 
 
394
    def test__record_oldest(self):
 
395
        # Test when the partition to be recorded is the same as the oldest.
 
396
        self.partition_coordinator._record_oldest(
 
397
            self.partition_coordinator.oldest, True)
 
398
        self.assertIsNone(self.partition_coordinator.oldest)
 
399
 
 
400
    def test_check_mastership(self):
 
401
        # Test the method exception condition.
 
402
        self.partition_coordinator._is_master = mock.Mock(
 
403
            side_effect=Exception('Boom!'))
 
404
        self.partition_coordinator.check_mastership(10, None)
 
405
        self.assertTrue('mastership check failed' in self.output.getvalue())
 
406
 
 
407
    def test_report_presence(self):
 
408
        self.partition_coordinator.coordination_rpc.presence = mock.Mock(
 
409
            side_effect=Exception('Boom!'))
 
410
        self.partition_coordinator.report_presence()
 
411
        self.assertTrue('presence reporting failed' in self.output.getvalue())
 
412
 
 
413
    def test_assigned_alarms(self):
 
414
        api_client = mock.MagicMock()
 
415
        api_client.alarms = mock.Mock(side_effect=Exception('Boom!'))
 
416
        self.partition_coordinator.assignment = ['something']
 
417
        self.partition_coordinator.assigned_alarms(api_client)
 
418
        self.assertTrue('assignment retrieval failed' in
 
419
                        self.output.getvalue())
 
420
 
 
421
 
 
422
class TestPartitionIdentity(test.BaseTestCase):
 
423
    def setUp(self):
 
424
        super(TestPartitionIdentity, self).setUp()
 
425
        self.id_1st = coordination.PartitionIdentity(str(uuid.uuid4()), 1)
 
426
        self.id_2nd = coordination.PartitionIdentity(str(uuid.uuid4()), 2)
 
427
 
 
428
    def test_identity_ops(self):
 
429
        self.assertNotEqual(self.id_1st, 'Nothing')
 
430
        self.assertNotEqual(self.id_1st, self.id_2nd)
 
431
        self.assertTrue(self.id_1st < None)
 
432
        self.assertFalse(self.id_1st < 'Nothing')
 
433
        self.assertTrue(self.id_2nd > self.id_1st)