~ubuntuone-control-tower/ubuntu-sso-client/stable-1-0

« back to all changes in this revision

Viewing changes to ubuntu_sso/utils/tests/test_oauth_headers.py

  • Committer: Tarmac
  • Author(s): Alejandro J. Cura
  • Date: 2011-12-02 19:30:27 UTC
  • mfrom: (646.1.3 timestamp-autofix-1-0)
  • Revision ID: tarmac-20111202193027-f1u4ma2pyar42qdc
Do a HEAD request on the server to get accurate timestamp (LP: #692597 & LP: #891644)

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# -*- coding: utf-8 -*-
 
2
 
 
3
# Author: Alejandro J. Cura <alecu@canonical.com>
 
4
#
 
5
# Copyright 2011 Canonical Ltd.
 
6
#
 
7
# This program is free software: you can redistribute it and/or modify it
 
8
# under the terms of the GNU General Public License version 3, as published
 
9
# by the Free Software Foundation.
 
10
#
 
11
# This program is distributed in the hope that it will be useful, but
 
12
# WITHOUT ANY WARRANTY; without even the implied warranties of
 
13
# MERCHANTABILITY, SATISFACTORY QUALITY, or FITNESS FOR A PARTICULAR
 
14
# PURPOSE.  See the GNU General Public License for more details.
 
15
#
 
16
# You should have received a copy of the GNU General Public License along
 
17
# with this program.  If not, see <http://www.gnu.org/licenses/>.
 
18
 
 
19
"""Tests for the oauth_headers helper function."""
 
20
 
 
21
import time
 
22
 
 
23
from twisted.application import internet, service
 
24
from twisted.internet import defer
 
25
from twisted.internet.threads import deferToThread
 
26
from twisted.trial.unittest import TestCase
 
27
from twisted.web import server, resource
 
28
 
 
29
from ubuntu_sso.utils import SyncTimestampChecker
 
30
 
 
31
 
 
32
class RootResource(resource.Resource):
 
33
    """A root resource that logs the number of calls."""
 
34
 
 
35
    isLeaf = True
 
36
 
 
37
    def __init__(self, *args, **kwargs):
 
38
        """Initialize this fake instance."""
 
39
        resource.Resource.__init__(self, *args, **kwargs)
 
40
        self.count = 0
 
41
        self.request_headers = []
 
42
 
 
43
    # pylint: disable=C0103
 
44
    def render_HEAD(self, request):
 
45
        """Increase the counter on each render."""
 
46
        self.count += 1
 
47
        self.request_headers.append(request.requestHeaders)
 
48
        return ""
 
49
 
 
50
 
 
51
class MockWebServer(object):
 
52
    """A mock webserver for testing."""
 
53
 
 
54
    def __init__(self):
 
55
        """Start up this instance."""
 
56
        # pylint: disable=E1101
 
57
        self.root = RootResource()
 
58
        site = server.Site(self.root)
 
59
        application = service.Application('web')
 
60
        self.service_collection = service.IServiceCollection(application)
 
61
        self.tcpserver = internet.TCPServer(0, site)
 
62
        self.tcpserver.setServiceParent(self.service_collection)
 
63
        self.service_collection.startService()
 
64
 
 
65
    def get_url(self):
 
66
        """Build the url for this mock server."""
 
67
        # pylint: disable=W0212
 
68
        port_num = self.tcpserver._port.getHost().port
 
69
        return "http://localhost:%d/" % port_num
 
70
 
 
71
    def stop(self):
 
72
        """Shut it down."""
 
73
        # pylint: disable=E1101
 
74
        self.service_collection.stopService()
 
75
 
 
76
 
 
77
class FakedError(Exception):
 
78
    """A mock, test, sample, and fake exception."""
 
79
 
 
80
 
 
81
class TimestampCheckerTestCase(TestCase):
 
82
    """Tests for the timestamp checker."""
 
83
 
 
84
    def setUp(self):
 
85
        """Initialize a fake webserver."""
 
86
        self.ws = MockWebServer()
 
87
        self.addCleanup(self.ws.stop)
 
88
        self.patch(SyncTimestampChecker, "SERVER_URL", self.ws.get_url())
 
89
 
 
90
    @defer.inlineCallbacks
 
91
    def test_returned_value_is_int(self):
 
92
        """The returned value is an integer."""
 
93
        checker = SyncTimestampChecker()
 
94
        timestamp = yield deferToThread(checker.get_faithful_time)
 
95
        self.assertEqual(type(timestamp), int)
 
96
 
 
97
    @defer.inlineCallbacks
 
98
    def test_first_call_does_head(self):
 
99
        """The first call gets the clock from our web."""
 
100
        checker = SyncTimestampChecker()
 
101
        yield deferToThread(checker.get_faithful_time)
 
102
        self.assertEqual(self.ws.root.count, 1)
 
103
 
 
104
    @defer.inlineCallbacks
 
105
    def test_second_call_is_cached(self):
 
106
        """For the second call, the time is cached."""
 
107
        checker = SyncTimestampChecker()
 
108
        yield deferToThread(checker.get_faithful_time)
 
109
        yield deferToThread(checker.get_faithful_time)
 
110
        self.assertEqual(self.ws.root.count, 1)
 
111
 
 
112
    @defer.inlineCallbacks
 
113
    def test_after_timeout_cache_expires(self):
 
114
        """After some time, the cache expires."""
 
115
        fake_timestamp = 1
 
116
        self.patch(time, "time", lambda: fake_timestamp)
 
117
        checker = SyncTimestampChecker()
 
118
        yield deferToThread(checker.get_faithful_time)
 
119
        fake_timestamp += SyncTimestampChecker.CHECKING_INTERVAL
 
120
        yield deferToThread(checker.get_faithful_time)
 
121
        self.assertEqual(self.ws.root.count, 2)
 
122
 
 
123
    @defer.inlineCallbacks
 
124
    def test_server_date_sends_nocache_headers(self):
 
125
        """Getting the server date sends the no-cache headers."""
 
126
        checker = SyncTimestampChecker()
 
127
        yield deferToThread(checker.get_server_time)
 
128
        assert len(self.ws.root.request_headers) == 1
 
129
        headers = self.ws.root.request_headers[0]
 
130
        result = headers.getRawHeaders("Cache-Control")
 
131
        self.assertEqual(result, ["no-cache"])
 
132
 
 
133
    @defer.inlineCallbacks
 
134
    def test_server_error_means_skew_not_updated(self):
 
135
        """When server can't be reached, the skew is not updated."""
 
136
        fake_timestamp = 1
 
137
        self.patch(time, "time", lambda: fake_timestamp)
 
138
        checker = SyncTimestampChecker()
 
139
 
 
140
        def failing_get_server_time():
 
141
            """Let's fail while retrieving the server time."""
 
142
            raise FakedError()
 
143
 
 
144
        self.patch(checker, "get_server_time", failing_get_server_time)
 
145
        yield deferToThread(checker.get_faithful_time)
 
146
        self.assertEqual(checker.skew, 0)
 
147
        self.assertEqual(checker.next_check,
 
148
                         fake_timestamp + SyncTimestampChecker.ERROR_INTERVAL)