1#!/usr/bin/env python 2 3import pytest 4 5from circuits import Component 6from circuits.web import Controller 7from circuits.web.client import parse_url 8from circuits.net.sockets import TCPClient 9from circuits.net.events import connect, write 10 11 12class Client(Component): 13 14 def __init__(self, *args, **kwargs): 15 super(Client, self).__init__(*args, **kwargs) 16 self._buffer = [] 17 self.done = False 18 19 def read(self, data): 20 self._buffer.append(data) 21 if data.find(b"\r\n") != -1: 22 self.done = True 23 24 def buffer(self): 25 return b''.join(self._buffer) 26 27 28class Root(Controller): 29 30 def index(self): 31 return "Hello World!" 32 33 34def test(webapp): 35 transport = TCPClient() 36 client = Client() 37 client += transport 38 client.start() 39 40 host, port, resource, secure = parse_url(webapp.server.http.base) 41 client.fire(connect(host, port)) 42 assert pytest.wait_for(transport, "connected") 43 44 client.fire(write(b"GET / HTTP/1.1\r\n")) 45 client.fire(write(b"Content-Type: text/plain\r\n\r\n")) 46 assert pytest.wait_for(client, "done") 47 48 client.stop() 49 50 s = client.buffer().decode('utf-8').split('\r\n')[0] 51 assert s == "HTTP/1.1 200 OK" 52