← Back to branch summary

~unity-2d-team/unity-2d/Shell-MultiMonitor

« back to all changes in this revision

Viewing changes to grackle/tests/test_client.py

  • Committer: Aaron Bentley
  • Date: 2012-01-10 11:06:36 UTC
  • Revision ID: aaron@canonical.com-20120110110636-fztfdo4gn1qfqty3
Add skeleton grackle-put-message.

Show diffs side-by-side

added added

removed removed

Lines of Context:
grackle/tests/test_client.py
1
 
from BaseHTTPServer import (
2
 
    HTTPServer,
3
 
    BaseHTTPRequestHandler,
4
 
    )
5
 
import httplib
6
 
import os
7
 
from signal import SIGKILL
8
 
from StringIO import StringIO
9
 
from unittest import TestCase
10
 
 
11
 
from testtools import ExpectedException
12
 
 
13
 
from grackle.client import (
14
 
    GrackleClient,
15
 
    message_path,
16
 
    )
17
 
 
18
 
 
19
 
class Forked:
20
 
 
21
 
    def __init__(self, func_or_method, *args):
22
 
        self.func_or_method = func_or_method
23
 
        self.pid = None
24
 
        self.args = args
25
 
 
26
 
    def __enter__(self):
27
 
        pid = os.fork()
28
 
        if pid != 0:
29
 
            self.pid = pid
30
 
            return
31
 
        self.func_or_method(*self.args)
32
 
 
33
 
 
34
 
    def __exit__(self, exc_type, exc_val, traceback):
35
 
        os.kill(self.pid, SIGKILL)
36
 
 
37
 
 
38
 
class FakeGrackleRequestHandler(BaseHTTPRequestHandler):
39
 
 
40
 
    def do_PUT(self):
41
 
        message = self.rfile.read(int(self.headers['content-length']))
42
 
        if message == 'This is a message':
43
 
            self.send_response(httplib.CREATED)
44
 
            self.end_headers()
45
 
            self.wfile.close()
46
 
        else:
47
 
            self.send_error(httplib.BAD_REQUEST)
48
 
 
49
 
 
50
 
def run_service(port):
51
 
    service = HTTPServer(('', port), FakeGrackleRequestHandler)
52
 
    service.serve_forever()
53
 
 
54
 
 
55
 
 
56
 
class TestPutMessage(TestCase):
57
 
 
58
 
    def test_put_message(self):
59
 
        client = GrackleClient('localhost', 8435)
60
 
        with Forked(run_service, client.port):
61
 
            client.put_message('arch1', 'asdf', StringIO('This is a message'))
62
 
            with ExpectedException(Exception, 'wtf'):
63
 
                client.put_message('arch1', 'asdf',
64
 
                    StringIO('This is not a message'))
65
 
 
66
 
 
67
 
class TestMessagePath(TestCase):
68
 
 
69
 
    def test_message_path(self):
70
 
        self.assertEqual('/my%20/%3C', message_path('my ', '<'))