← Back to branch summary

~ahasenack/landscape-client/landscape-client-1.5.5-0ubuntu0.9.04.0

« back to all changes in this revision

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

  • Committer: Bazaar Package Importer
  • Author(s): Rick Clark
  • Date: 2008-09-08 16:35:57 UTC
  • mfrom: (1.1.1 upstream)
  • Revision ID: james.westby@ubuntu.com-20080908163557-l3ixzj5dxz37wnw2
Tags: 1.0.18-0ubuntu1
New upstream release 

Show diffs side-by-side

added added

removed removed

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