← Back to branch summary

~mvo/click/acquire-asyncio

« back to all changes in this revision

Viewing changes to click/tests/test_acquire.py

  • Committer: Michael Vogt
  • Date: 2014-09-19 14:00:46 UTC
  • Revision ID: michael.vogt@ubuntu.com-20140919140046-s3cv307ihch9mk0e
fix py3 compat problems

Show diffs side-by-side

added added

removed removed

Lines of Context:
click/tests/test_acquire.py
24
24
 
25
25
import multiprocessing
26
26
import os.path
27
 
import SocketServer
28
27
 
29
28
from six.moves import (
30
29
    SimpleHTTPServer,
 
30
    socketserver,
31
31
)
32
32
from six import StringIO
33
33
 
41
41
    TestCase,
42
42
)
43
43
 
 
44
import click.acquire
 
45
click.acquire.acquire_methods_dir = os.path.join(
 
46
    os.path.dirname(__file__), "..", "..", "acquire")
 
47
os.environ["PYTHONPATH"] = os.path.abspath(
 
48
    os.path.join(os.path.dirname(__file__), "..", ".."))
 
49
 
44
50
# local httpd
45
51
LOCALHOST = "localhost"
46
52
PORT = 8128
51
57
        pass
52
58
 
53
59
 
54
 
class MySocketServer(SocketServer.TCPServer):
 
60
class MySocketServer(socketserver.TCPServer):
55
61
    allow_reuse_address = True
56
62
 
57
63
 
64
70
    def run(self):
65
71
        os.chdir(self.basedir)
66
72
        server = MySocketServer((LOCALHOST, PORT), MyHandler)
67
 
        server.handle_request()
 
73
        while True:
 
74
            server.handle_request()
68
75
        server.shutdown()
69
76
 
70
77
    def stop(self):
106
113
        self.assertEqual(number, 100)
107
114
 
108
115
    def test_read_messages_multiple(self):
109
 
        s = StringIO("100 something\nfoo: bar\n\n101 more\n\n\n")
 
116
        s = StringIO("100 something\nfoo: bar\n\n101 more\n\n")
110
117
        msg, number = read_message(s)
111
118
        self.assertEqual(number, 100)
112
119
        # and another one
113
120
        msg, number = read_message(s)
114
121
        self.assertEqual(number, 101)
115