← Back to branch summary

~soren/surveilr/client

« back to all changes in this revision

Viewing changes to surveilr/tests/test_utils.py

  • Committer: Tarmac
  • Author(s): Soren Hansen
  • Date: 2011-12-12 10:57:49 UTC
  • mfrom: (13.1.5 plugins)
  • Revision ID: tarmac-20111212105749-rfu6wudnfr3x0ets
Plugin service support

Show diffs side-by-side

added added

removed removed

Lines of Context:
surveilr/tests/test_utils.py
 
1
"""
 
2
    Surveilr - Log aggregation, analysis and visualisation
 
3
 
 
4
    Copyright (C) 2011  Linux2Go
 
5
 
 
6
    This program is free software: you can redistribute it and/or
 
7
    modify it under the terms of the GNU Affero General Public License
 
8
    as published by the Free Software Foundation, either version 3 of
 
9
    the License, or (at your option) any later version.
 
10
 
 
11
    This program is distributed in the hope that it will be useful,
 
12
    but WITHOUT ANY WARRANTY; without even the implied warranty of
 
13
    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
14
    GNU Affero General Public License for more details.
 
15
 
 
16
    You should have received a copy of the GNU Affero General Public
 
17
    License along with this program.  If not, see
 
18
    <http://www.gnu.org/licenses/>.
 
19
 
 
20
    Plugin tests
 
21
"""
 
22
 
 
23
import json
 
24
import time
 
25
import mock
 
26
import unittest
 
27
 
 
28
from surveilr import models
 
29
from surveilr import utils
 
30
 
 
31
 
 
32
class UtilsTests(unittest.TestCase):
 
33
    def test_enhance(self):
 
34
        url = 'http://some-url:345/fooo'
 
35
        timestamp = int(time.time() * 1000)
 
36
        metrics = {'time': 773,
 
37
                   'size': 12554}
 
38
        new_state = ['somestate']
 
39
 
 
40
        # Create the user, service and log entry
 
41
        user = models.User()
 
42
        user.save()
 
43
        user_id = user.key
 
44
 
 
45
        service = models.Service()
 
46
        service.user = [user]
 
47
        service.plugins = [{'url': url}]
 
48
        service.save()
 
49
        service_id = service.key
 
50
 
 
51
        log_entry = models.LogEntry()
 
52
        log_entry.service = [service]
 
53
 
 
54
        log_entry.timestamp = timestamp
 
55
        log_entry.metrics = metrics.copy()
 
56
        log_entry.save()
 
57
 
 
58
        with mock.patch('surveilr.utils.httplib2') as httplib2:
 
59
            # Mock out the http calls
 
60
            request_call = httplib2.Http.return_value.request
 
61
            request_call.return_value = (None,
 
62
                                         json.dumps({'state': new_state}))
 
63
 
 
64
            # Exercise the SUT
 
65
            utils.enhance_data_point(log_entry)
 
66
 
 
67
            # Verify everything
 
68
            self.assertEquals(request_call.call_args[0], (url,), 'Wrong url')
 
69
            self.assertEquals(len(request_call.call_args[1]), 2,
 
70
                              'Wrong number of kwargs')
 
71
 
 
72
            self.assertIn('method', request_call.call_args[1],
 
73
                          'No method passed to request call')
 
74
            self.assertEquals(request_call.call_args[1]['method'], 'POST',
 
75
                              'method was not set to "POST"')
 
76
 
 
77
            self.assertIn('body', request_call.call_args[1],
 
78
                          'No body passed to request call')
 
79
            body = request_call.call_args[1]['body']
 
80
 
 
81
            self.assertDictEqual(json.loads(body), {'timestamp': timestamp,
 
82
                                                    'service_id': service_id,
 
83
                                                    'user_id': user_id,
 
84
                                                    'metrics': metrics,
 
85
                                                    'saved_state': None})
 
86
 
 
87
            self.assertEquals(len(service.plugins), 1)
 
88
            self.assertDictEqual(service.plugins[0],
 
89
                                 {'url': url, 'saved_state': new_state})
 
90
 
 
91
    def test_truncate(self):
 
92
        self.assertEquals(utils.truncate(133, 1), 133)
 
93
        self.assertEquals(utils.truncate(133, 10), 130)
 
94
        self.assertEquals(utils.truncate(133, 20), 120)
 
95
        self.assertEquals(utils.truncate(133, 100), 100)
 
96
        self.assertEquals(utils.truncate(133, 200), 0)