← Back to branch summary

~ahasenack/landscape-client/landscape-client-11.02-0ubuntu0.8.04.1

« back to all changes in this revision

Viewing changes to landscape/broker/tests/test_transport.py

  • Committer: Andreas Hasenack
  • Date: 2011-05-05 14:12:15 UTC
  • Revision ID: andreas@canonical.com-20110505141215-5ymuyyh5es9pwa6p
Added hardy files.

Show diffs side-by-side

added added

removed removed

Lines of Context:
landscape/broker/tests/test_transport.py
 
1
import os
 
2
 
 
3
from landscape import VERSION
 
4
from landscape.broker.transport import HTTPTransport
 
5
from landscape.lib.fetch import PyCurlError
 
6
from landscape.lib import bpickle
 
7
 
 
8
from landscape.tests.helpers import LandscapeTest, LogKeeperHelper
 
9
 
 
10
from twisted.web import server, resource
 
11
from twisted.internet import reactor
 
12
from twisted.internet.ssl import DefaultOpenSSLContextFactory
 
13
from twisted.internet.threads import deferToThread
 
14
 
 
15
 
 
16
def sibpath(path):
 
17
    return os.path.join(os.path.dirname(__file__), path)
 
18
 
 
19
 
 
20
PRIVKEY = sibpath("private.ssl")
 
21
PUBKEY = sibpath("public.ssl")
 
22
BADPRIVKEY = sibpath("badprivate.ssl")
 
23
BADPUBKEY = sibpath("badpublic.ssl")
 
24
 
 
25
 
 
26
class DataCollectingResource(resource.Resource):
 
27
    request = content = None
 
28
    def getChild(self, request, name):
 
29
        return self
 
30
 
 
31
    def render(self, request):
 
32
        self.request = request
 
33
        self.content = request.content.read()
 
34
        return bpickle.dumps("Great.")
 
35
 
 
36
 
 
37
class HTTPTransportTest(LandscapeTest):
 
38
 
 
39
    helpers = [LogKeeperHelper]
 
40
 
 
41
    def setUp(self):
 
42
        super(HTTPTransportTest, self).setUp()
 
43
        self.ports = []
 
44
 
 
45
    def tearDown(self):
 
46
        super(HTTPTransportTest, self).tearDown()
 
47
        for port in self.ports:
 
48
            port.stopListening()
 
49
 
 
50
    def test_get_url(self):
 
51
        url = "http://example/ooga"
 
52
        transport = HTTPTransport(url)
 
53
        self.assertEquals(transport.get_url(), url)
 
54
 
 
55
    def test_set_url(self):
 
56
        transport = HTTPTransport("http://example/ooga")
 
57
        transport.set_url("http://example/message-system")
 
58
        self.assertEquals(transport.get_url(), "http://example/message-system")
 
59
 
 
60
    def test_request_data(self):
 
61
        """
 
62
        When a request is sent with HTTPTransport.exchange, it should
 
63
        include the (optional) computer ID, a user agent, and the
 
64
        message API version as HTTP headers, and the payload as a
 
65
        bpickled request body.
 
66
        """
 
67
        r = DataCollectingResource()
 
68
        port = reactor.listenTCP(0, server.Site(r), interface="127.0.0.1")
 
69
        self.ports.append(port)
 
70
        transport = HTTPTransport("http://localhost:%d/"
 
71
                                  % (port.getHost().port,))
 
72
        result = deferToThread(transport.exchange, "HI", computer_id="34",
 
73
                               message_api="X.Y")
 
74
 
 
75
        def got_result(ignored):
 
76
            self.assertEquals(r.request.received_headers["x-computer-id"], "34")
 
77
            self.assertEquals(r.request.received_headers["user-agent"],
 
78
                              "landscape-client/%s" % (VERSION,))
 
79
            self.assertEquals(r.request.received_headers["x-message-api"],
 
80
                              "X.Y")
 
81
            self.assertEquals(bpickle.loads(r.content), "HI")
 
82
        result.addCallback(got_result)
 
83
        return result
 
84
 
 
85
    def test_ssl_verification_positive(self):
 
86
        """
 
87
        The client transport should complete an upload of messages to
 
88
        a host which provides SSL data which can be verified by the
 
89
        public key specified.
 
90
        """
 
91
        r = DataCollectingResource()
 
92
        context_factory = DefaultOpenSSLContextFactory(PRIVKEY,
 
93
                                                       PUBKEY)
 
94
        port = reactor.listenSSL(0, server.Site(r), context_factory,
 
95
                                 interface="127.0.0.1")
 
96
        self.ports.append(port)
 
97
        transport = HTTPTransport("https://localhost:%d/"
 
98
                                  % (port.getHost().port,),
 
99
                                  pubkey=PUBKEY)
 
100
        result = deferToThread(transport.exchange, "HI", computer_id="34",
 
101
                               message_api="X.Y")
 
102
 
 
103
        def got_result(ignored):
 
104
            self.assertEquals(r.request.received_headers["x-computer-id"], "34")
 
105
            self.assertEquals(r.request.received_headers["user-agent"],
 
106
                              "landscape-client/%s" % (VERSION,))
 
107
            self.assertEquals(r.request.received_headers["x-message-api"],
 
108
                              "X.Y")
 
109
            self.assertEquals(bpickle.loads(r.content), "HI")
 
110
        result.addCallback(got_result)
 
111
        return result
 
112
 
 
113
 
 
114
    def test_ssl_verification_negative(self):
 
115
        """
 
116
        If the SSL server provides a key which is not verified by the
 
117
        specified public key, then the client should immediately end
 
118
        the connection without uploading any message data.
 
119
        """
 
120
        self.log_helper.ignore_errors(PyCurlError)
 
121
        r = DataCollectingResource()
 
122
        context_factory = DefaultOpenSSLContextFactory(BADPRIVKEY,
 
123
                                                       BADPUBKEY)
 
124
        port = reactor.listenSSL(0, server.Site(r), context_factory,
 
125
                                 interface="127.0.0.1")
 
126
        self.ports.append(port)
 
127
        transport = HTTPTransport("https://localhost:%d/"
 
128
                                  % (port.getHost().port,),
 
129
                                  pubkey=PUBKEY)
 
130
 
 
131
        result = deferToThread(transport.exchange, "HI", computer_id="34",
 
132
                               message_api="X.Y")
 
133
        def got_result(ignored):
 
134
            self.assertEquals(r.request, None)
 
135
            self.assertEquals(r.content, None)
 
136
            self.assertTrue("server certificate verification failed"
 
137
                            in self.logfile.getvalue())
 
138
        result.addCallback(got_result)
 
139
        return result