blob: c86283a3c6c3a24cfeaee45b689afa3c7f0c4c64 [file] [log] [blame]
#!/usr/bin/python
"""Fake minissdpd for unit tests.
"""
import BaseHTTPServer
import socket
import SocketServer
import sys
import threading
text_device_xml = """<root>
<specVersion><major>1</major><minor>0</minor></specVersion>
<device><friendlyName>Test Device</friendlyName>
<manufacturer>Google Fiber</manufacturer>
<modelDescription>Unit Test</modelDescription>
<modelName>ssdptax</modelName>
</device></root>"""
email_address_xml = """<root>
<specVersion><major>1</major><minor>0</minor></specVersion>
<device><friendlyName>FOOBAR: foo@example.com:</friendlyName>
<manufacturer>Google Fiber</manufacturer>
<modelDescription>Unit Test</modelDescription>
<modelName>ssdptax</modelName>
</device></root>"""
no_friendlyname_xml = """<root>
<specVersion><major>1</major><minor>0</minor></specVersion>
<device></device></root>"""
ssdp_device_xml = """<root>
<specVersion><major>1</major><minor>0</minor></specVersion>
<device><friendlyName>Test Device</friendlyName>
<manufacturer>Google Fiber</manufacturer>
<modelDescription>Unit Test</modelDescription>
<modelName>ssdptax multicast</modelName>
</device></root>"""
notify_template = 'NOTIFY\r\nHOST:239.255.255.250:1900\r\nLOCATION:%s\r\n'
notify_text = ['']
minissdpd_response = ['']
keep_running = [True]
class HttpHandler(BaseHTTPServer.BaseHTTPRequestHandler):
"""Respond to an HHTP GET for SSDP DeviceInfo."""
def do_GET(self):
self.send_response(200)
self.send_header('Content-type','text/xml')
self.end_headers()
if self.path.endswith('text_device_xml'):
self.wfile.write(text_device_xml)
if self.path.endswith('email_address_xml'):
self.wfile.write(email_address_xml)
if self.path.endswith('no_friendlyname_xml'):
self.wfile.write(no_friendlyname_xml)
if self.path.endswith('ssdp_device_xml'):
self.wfile.write(ssdp_device_xml)
class ThreadingHTTPServer(SocketServer.ThreadingMixIn, BaseHTTPServer.HTTPServer):
pass
class UnixHandler(SocketServer.StreamRequestHandler):
"""Respond to a command on MiniSSDPd's Unix socket."""
def handle(self):
data = self.request.recv(8192)
if 'quitquitquit' in data:
print 'Received quitquitquit, exiting...'
keep_running[0] = False
return
else:
self.request.sendall(bytearray(minissdpd_response[0]))
class UdpHandler(SocketServer.DatagramRequestHandler):
def handle(self):
self.request[1].sendto(bytearray(notify_text[0]), self.client_address)
class ThreadingUdpServer(SocketServer.ThreadingUDPServer):
allow_reuse_address = True
def main():
socketpath = sys.argv[1]
testnum = int(sys.argv[2])
if testnum == 1:
pathend = 'text_device_xml'
if testnum == 2:
pathend = 'email_address_xml'
if testnum == 3:
pathend = 'no_friendlyname_xml'
if testnum == 4:
pathend = 'ssdp_device_xml'
h = ThreadingHTTPServer(("", 0), HttpHandler)
sn = h.socket.getsockname()
port = sn[1]
url = 'http://127.0.0.1:%d/%s' % (port, pathend)
st = 'server type'
uuid = 'uuid goes here'
if testnum == 4:
minissdpd_response[0] = [0]
else:
minissdpd_response[0] = [1]
minissdpd_response[0].extend([len(url)] + list(url))
minissdpd_response[0].extend([len(st)] + list(st))
minissdpd_response[0].extend([len(uuid)] + list(uuid))
notify_text[0] = notify_template % url
h_thread = threading.Thread(target=h.serve_forever)
h_thread.daemon = True
h_thread.start()
d = ThreadingUdpServer(('', 1900), UdpHandler)
d.socket.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP,
socket.inet_aton('239.255.255.250') + socket.inet_aton('0.0.0.0'))
d_thread = threading.Thread(target=d.serve_forever)
d_thread.daemon = True
d_thread.start()
u = SocketServer.UnixStreamServer(socketpath, UnixHandler)
while keep_running[0]:
u.handle_request()
if __name__ == '__main__':
main()