← Back to branch summary

~sophie-middleton08/maus/devel

« back to all changes in this revision

Viewing changes to tests/integration/test_distributed_processing/_test_online_okay.py

  • Committer: Chris Rogers
  • Date: 2012-11-06 12:04:39 UTC
  • mfrom: (659.1.45 release-candidate)
  • Revision ID: chris.rogers@stfc.ac.uk-20121106120439-e6znfg5kfg850l38
Tags: MAUS-v0.4.0
MAUS-v0.4.0

Show diffs side-by-side

added added

removed removed

Lines of Context:
tests/integration/test_distributed_processing/_test_online_okay.py
 
1
#  This file is part of MAUS: http://micewww.pp.rl.ac.uk:8080/projects/maus
 
2
# 
 
3
#  MAUS is free software: you can redistribute it and/or modify
 
4
#  it under the terms of the GNU General Public License as published by
 
5
#  the Free Software Foundation, either version 3 of the License, or
 
6
#  (at your option) any later version.
 
7
# 
 
8
#  MAUS is distributed in the hope that it will be useful,
 
9
#  but WITHOUT ANY WARRANTY; without even the implied warranty of
 
10
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
11
#  GNU General Public License for more details.
 
12
# 
 
13
#  You should have received a copy of the GNU General Public License
 
14
#  along with MAUS.  If not, see <http://www.gnu.org/licenses/>.
 
15
 
 
16
"""
 
17
Check that celery is alive on the test server. Note that this should not be run
 
18
as part of the regular test script - only should be run by online machine (and
 
19
online-capable machines)
 
20
"""
 
21
 
 
22
import unittest
 
23
import celery.task.control
 
24
import pymongo
 
25
import pymongo.errors
 
26
 
 
27
class OnlineOkayTest(unittest.TestCase): # pylint: disable=R0904, C0301
 
28
    """
 
29
    Check that celery nodes are available and RabbitMQ is alive or fail
 
30
    """
 
31
    def test_celery_is_alive(self):
 
32
        """
 
33
        _test_online_okay: Check that celery can be accessed 
 
34
 
 
35
        Fail an assertion if celery.task.control.inspect() fails (RabbitMQ is
 
36
        down) or no active nodes
 
37
        """
 
38
        try:
 
39
            active_nodes = celery.task.control.inspect().active()
 
40
        except: #pylint: disable=W0702
 
41
            self.assertTrue(False, "Failed to inspect celery workers")
 
42
        self.assertNotEqual(active_nodes, None)
 
43
 
 
44
    def test_mongodb_is_alive(self):
 
45
        """
 
46
        _test_online_okay: Check that mongodb can be accessed
 
47
 
 
48
        Fail an assertion if mongodb can't be contacted on localhost        
 
49
        """
 
50
        try:
 
51
            test_connx = pymongo.Connection("localhost", 27017)
 
52
        except pymongo.errors.AutoReconnect: # pylint: disable=W0702
 
53
            self.assertTrue(False, "MongoDB server is not accessible")
 
54
        test_connx.disconnect()
 
55
 
 
56
if __name__ == "__main__":
 
57
    unittest.main()
 
58
 
 
59