~ubuntu-branches/debian/sid/checkbox-ng/sid

« back to all changes in this revision

Viewing changes to checkbox_ng/test_certification.py

  • Committer: Package Import Robot
  • Author(s): Sylvain Pineau
  • Date: 2014-01-21 21:18:23 UTC
  • Revision ID: package-import@ubuntu.com-20140121211823-iru2ukliuatzpo5p
Tags: upstream-0.1
ImportĀ upstreamĀ versionĀ 0.1

Show diffs side-by-side

added added

removed removed

Lines of Context:
 
1
# This file is part of Checkbox.
 
2
#
 
3
# Copyright 2012, 2013 Canonical Ltd.
 
4
# Written by:
 
5
#   Zygmunt Krynicki <zygmunt.krynicki@canonical.com>
 
6
#   Daniel Manrique <roadmr@ubuntu.com>
 
7
#
 
8
# Checkbox is free software: you can redistribute it and/or modify
 
9
# it under the terms of the GNU General Public License version 3,
 
10
# as published by the Free Software Foundation.
 
11
 
 
12
#
 
13
# Checkbox is distributed in the hope that it will be useful,
 
14
# but WITHOUT ANY WARRANTY; without even the implied warranty of
 
15
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
 
16
# GNU General Public License for more details.
 
17
#
 
18
# You should have received a copy of the GNU General Public License
 
19
# along with Checkbox.  If not, see <http://www.gnu.org/licenses/>.
 
20
 
 
21
"""
 
22
plainbox.impl.transport.test_certification
 
23
==========================================
 
24
 
 
25
Test definitions for plainbox.impl.certification module
 
26
"""
 
27
 
 
28
from io import BytesIO
 
29
from unittest import TestCase
 
30
 
 
31
from pkg_resources import resource_string
 
32
from plainbox.impl.applogic import PlainBoxConfig
 
33
from plainbox.vendor import mock
 
34
from plainbox.vendor.mock import MagicMock
 
35
from requests.exceptions import ConnectionError, InvalidSchema, HTTPError
 
36
import requests
 
37
 
 
38
from checkbox_ng.certification import CertificationTransport
 
39
from checkbox_ng.certification import InvalidSecureIDError
 
40
 
 
41
 
 
42
class CertificationTransportTests(TestCase):
 
43
 
 
44
    #URL are just here to exemplify, since we mock away all network access,
 
45
    #they're not really used.
 
46
    valid_url = "https://certification.canonical.com/submissions/submit"
 
47
    invalid_url = "htz://:3128"
 
48
    unreachable_url = "http://i.dont.exist"
 
49
    valid_secure_id = "a00D000000Kkk5j"
 
50
    valid_option_string = "secure_id={}".format(valid_secure_id)
 
51
 
 
52
    def setUp(self):
 
53
        self.sample_xml = BytesIO(resource_string(
 
54
            "plainbox", "test-data/xml-exporter/example-data.xml"
 
55
        ))
 
56
        self.patcher = mock.patch('requests.post')
 
57
        self.mock_requests = self.patcher.start()
 
58
 
 
59
    def test_parameter_parsing(self):
 
60
        #Makes sense since I'm overriding the base class's constructor.
 
61
        transport = CertificationTransport(self.valid_url,
 
62
                                           self.valid_option_string)
 
63
        self.assertEqual(self.valid_url, transport.url)
 
64
        self.assertEqual(self.valid_secure_id,
 
65
                         transport.options['secure_id'])
 
66
 
 
67
    def test_invalid_length_secure_id_are_rejected(self):
 
68
        for length in (14, 16, 20):
 
69
            dummy_id = "a" * length
 
70
            option_string = "secure_id={}".format(dummy_id)
 
71
            with self.assertRaises(InvalidSecureIDError):
 
72
                transport = CertificationTransport(self.valid_url,
 
73
                                                   option_string)
 
74
                self.assertIsInstance(CertificationTransport,
 
75
                                      transport)
 
76
 
 
77
    def test_invalid_characters_in_secure_id_are_rejected(self):
 
78
        option_string = "secure_id=aA0#"
 
79
        with self.assertRaises(InvalidSecureIDError):
 
80
                transport = CertificationTransport(self.valid_url,
 
81
                                                   option_string)
 
82
                self.assertIsInstance(CertificationTransport,
 
83
                                      transport)
 
84
 
 
85
    def test_invalid_url(self):
 
86
        transport = CertificationTransport(self.invalid_url,
 
87
                                           self.valid_option_string)
 
88
        dummy_data = BytesIO(b"some data to send")
 
89
        requests.post.side_effect = InvalidSchema
 
90
 
 
91
        with self.assertRaises(InvalidSchema):
 
92
            result = transport.send(dummy_data)
 
93
            self.assertIsNotNone(result)
 
94
        requests.post.assert_called_with(self.invalid_url,
 
95
                                         files={'data': dummy_data},
 
96
                                         headers={'X_HARDWARE_ID':
 
97
                                         self.valid_secure_id},
 
98
                                         proxies=None)
 
99
 
 
100
    def test_valid_url_cant_connect(self):
 
101
        transport = CertificationTransport(self.unreachable_url,
 
102
                                           self.valid_option_string)
 
103
        dummy_data = BytesIO(b"some data to send")
 
104
        requests.post.side_effect = ConnectionError
 
105
 
 
106
        with self.assertRaises(ConnectionError):
 
107
            result = transport.send(dummy_data)
 
108
            self.assertIsNotNone(result)
 
109
        requests.post.assert_called_with(self.unreachable_url,
 
110
                                         files={'data': dummy_data},
 
111
                                         headers={'X_HARDWARE_ID':
 
112
                                                  self.valid_secure_id},
 
113
                                         proxies=None)
 
114
 
 
115
    def test_send_success(self):
 
116
        transport = CertificationTransport(self.valid_url,
 
117
                                           self.valid_option_string)
 
118
 
 
119
        requests.post.return_value = MagicMock(name='response')
 
120
        requests.post.return_value.status_code = 200
 
121
        requests.post.return_value.text = '{"id": 768}'
 
122
        result = transport.send(self.sample_xml)
 
123
 
 
124
        self.assertTrue(result)
 
125
 
 
126
    def test_send_failure(self):
 
127
        transport = CertificationTransport(self.valid_url,
 
128
                                           self.valid_option_string)
 
129
 
 
130
        requests.post.return_value = MagicMock(name='response')
 
131
        requests.post.return_value.status_code = 412
 
132
        requests.post.return_value.text = 'Some error'
 
133
        #Oops, raise_for_status doesn't get fooled by my mocking,
 
134
        #so I have to mock *that* method as well..
 
135
        requests.post.return_value.raise_for_status = MagicMock(
 
136
            side_effect=HTTPError)
 
137
 
 
138
        with self.assertRaises(HTTPError):
 
139
            result = transport.send(self.sample_xml)
 
140
            self.assertIsNotNone(result)
 
141
 
 
142
    def proxy_test(self, environment, proxies):
 
143
        test_environment = environment
 
144
        test_proxies = proxies
 
145
        test_config = PlainBoxConfig()
 
146
        test_config.environment = test_environment
 
147
 
 
148
        transport = CertificationTransport(self.valid_url,
 
149
                                           self.valid_option_string,
 
150
                                           config=test_config)
 
151
        dummy_data = BytesIO(b"some data to send")
 
152
 
 
153
        requests.post.return_value = MagicMock(name='response')
 
154
        requests.post.return_value.status_code = 200
 
155
        requests.post.return_value.text = '{"id": 768}'
 
156
        result = transport.send(dummy_data)
 
157
 
 
158
        self.assertTrue(result)
 
159
 
 
160
        requests.post.assert_called_with(self.valid_url,
 
161
                                         files={'data': dummy_data},
 
162
                                         headers={'X_HARDWARE_ID':
 
163
                                         self.valid_secure_id},
 
164
                                         proxies=test_proxies)
 
165
 
 
166
    def test_set_only_one_proxy(self):
 
167
        test_environment = {'http_proxy': "http://1.2.3.4:5"}
 
168
        test_proxies = {'http': "http://1.2.3.4:5"}
 
169
        self.proxy_test(test_environment, test_proxies)
 
170
 
 
171
    def test_set_two_proxies(self):
 
172
        test_environment = {'http_proxy': "http://1.2.3.4:5",
 
173
                            'https_proxy': "http://1.2.3.4:6"}
 
174
        test_proxies = {'http': "http://1.2.3.4:5",
 
175
                        'https': "http://1.2.3.4:6"}
 
176
        self.proxy_test(test_environment, test_proxies)
 
177
 
 
178
    def test_behavior_with_extraneous_environment(self):
 
179
        test_environment = {'http_proxy': "http://1.2.3.4:5",
 
180
                            'weird_value': 'What is this'}
 
181
        test_proxies = {'http': "http://1.2.3.4:5"}
 
182
        self.proxy_test(test_environment, test_proxies)
 
183