unix4lyfe git - darkhttpd/blob - devel/test.py

   1 #!/usr/bin/env python
   2 import unittest
   3 import socket
   4 import signal
   5 import re
   6 import os
   7 import random
   8
   9 WWWROOT = "tmp.httpd.tests"
  10
  11 class Conn:
  12     def __init__(self):
  13         self.port = 12346
  14         self.s = socket.socket()
  15         self.s.connect(("0.0.0.0", self.port))
  16         # connect throws socket.error on connection refused
  17
  18     def get(self, url, http_ver="1.0", endl="\n", req_hdrs={}, method="GET"):
  19         req = method+" "+url
  20         if http_ver is not None:
  21             req += " HTTP/"+http_ver
  22         req += endl
  23         if http_ver is not None:
  24             req_hdrs["User-Agent"] = "test.py"
  25             req_hdrs["Connection"] = "close"
  26             for k,v in req_hdrs.items():
  27                 req += k+": "+v+endl
  28         req += endl # end of request
  29         self.s.send(req)
  30         ret = ""
  31         while True:
  32             signal.alarm(1) # don't wait forever
  33             r = self.s.recv(65536)
  34             signal.alarm(0)
  35             if r == "":
  36                 break
  37             else:
  38                 ret += r
  39         return ret
  40
  41 def parse(resp):
  42     """
  43     Parse response into status line, headers and body.
  44     """
  45     pos = resp.index("\r\n\r\n") # throws exception on failure
  46     head = resp[:pos]
  47     body = resp[pos+4:]
  48     status,head = head.split("\r\n", 1)
  49     hdrs = {}
  50     for line in head.split("\r\n"):
  51         k, v = line.split(": ", 1)
  52         hdrs[k] = v
  53     return (status, hdrs, body)
  54
  55 class TestHelper(unittest.TestCase):
  56     def assertContains(self, body, *strings):
  57         for s in strings:
  58             self.assertTrue(s in body,
  59                             msg="expected %s in %s"%(repr(s), repr(body)))
  60
  61     def assertIsIndex(self, body, path):
  62         self.assertContains(body,
  63             "<title>%s</title>\n"%path,
  64             "<h1>%s</h1>\n"%path,
  65             '<a href="..">..</a>/',
  66             'Generated by darkhttpd')
  67
  68     def assertIsInvalid(self, body, path):
  69         self.assertContains(body,
  70             "<title>400 Bad Request</title>",
  71             "<h1>Bad Request</h1>\n",
  72             "You requested an invalid URL: %s\n"%path,
  73             'Generated by darkhttpd')
  74
  75 class TestDirList(TestHelper):
  76     def setUp(self):
  77         self.fn = WWWROOT+"/escape#this"
  78         open(self.fn, "w").write("x"*12345)
  79
  80     def tearDown(self):
  81         os.unlink(self.fn)
  82
  83     def test_dirlist_escape(self):
  84         resp = Conn().get("/")
  85         status, hdrs, body = parse(resp)
  86         self.assertEquals(ord("#"), 0x23)
  87         self.assertContains(body, "escape%23this", "12345")
  88
  89 class TestCases(TestHelper):
  90     pass # these get autogenerated in setUpModule()
  91
  92 def nerf(s):
  93     return re.sub("[^a-zA-Z0-9]", "_", s)
  94
  95 def makeCase(name, url, hdr_checker=None, body_checker=None,
  96              req_hdrs={"User-Agent": "test.py"},
  97              http_ver=None, endl="\n"):
  98     def do_test(self):
  99         resp = Conn().get(url, http_ver, endl, req_hdrs)
 100         if http_ver is None:
 101             status = ""
 102             hdrs = {}
 103             body = resp
 104         else:
 105             status, hdrs, body = parse(resp)
 106
 107         if hdr_checker is not None and http_ver is not None:
 108             hdr_checker(self, hdrs)
 109
 110         if body_checker is not None:
 111             body_checker(self, body)
 112
 113         # FIXME: check status
 114         if http_ver is not None:
 115             prefix = "HTTP/1.1 " # should 1.0 stay 1.0?
 116             self.assertTrue(status.startswith(prefix),
 117                 msg="%s at start of %s"%(repr(prefix), repr(status)))
 118
 119     v = http_ver
 120     if v is None:
 121         v = "0.9"
 122     test_name = "_".join([
 123         "test",
 124         nerf(name),
 125         nerf("HTTP"+v),
 126         {"\n":"LF", "\r\n":"CRLF"}[endl],
 127     ])
 128     do_test.__name__ = test_name # hax
 129     setattr(TestCases, test_name, do_test)
 130
 131 def makeCases(name, url, hdr_checker=None, body_checker=None,
 132               req_hdrs={"User-Agent": "test.py"}):
 133     for http_ver in [None, "1.0", "1.1"]:
 134         for endl in ["\n", "\r\n"]:
 135             makeCase(name, url, hdr_checker, body_checker,
 136                      req_hdrs, http_ver, endl)
 137
 138 def makeSimpleCases(name, url, assert_name):
 139     makeCases(name, url, None,
 140         lambda self,body: getattr(self, assert_name)(body, url))
 141
 142 def setUpModule():
 143     for args in [
 144         ["index",                "/",               "assertIsIndex"],
 145         ["up dir",               "/dir/../",        "assertIsIndex"],
 146         ["extra slashes",        "//dir///..////",  "assertIsIndex"],
 147         ["no trailing slash",    "/dir/..",         "assertIsIndex"],
 148         ["no leading slash",     "dir/../",         "assertIsInvalid"],
 149         ["invalid up dir",       "/../",            "assertIsInvalid"],
 150         ["fancy invalid up dir", "/./dir/./../../", "assertIsInvalid"],
 151         ]:
 152         makeSimpleCases(*args)
 153
 154 class TestFileGet(TestHelper):
 155     def setUp(self):
 156         self.datalen = 2345
 157         self.data = "".join(
 158             [chr(random.randint(0,255)) for _ in xrange(self.datalen)])
 159         self.url = "/data.jpeg"
 160         self.fn = WWWROOT + self.url
 161         open(self.fn, "w").write(self.data)
 162
 163     def tearDown(self):
 164         os.unlink(self.fn)
 165
 166     def test_file_get(self):
 167         resp = Conn().get(self.url)
 168         status, hdrs, body = parse(resp)
 169         self.assertContains(status, "200 OK")
 170         self.assertEquals(hdrs["Accept-Ranges"], "bytes")
 171         self.assertEquals(hdrs["Content-Length"], str(self.datalen))
 172         self.assertEquals(hdrs["Content-Type"], "image/jpeg")
 173         self.assertEquals(body, self.data)
 174
 175     def test_file_head(self):
 176         resp = Conn().get(self.url, method="HEAD")
 177         status, hdrs, body = parse(resp)
 178         self.assertContains(status, "200 OK")
 179         self.assertEquals(hdrs["Accept-Ranges"], "bytes")
 180         self.assertEquals(hdrs["Content-Length"], str(self.datalen))
 181         self.assertEquals(hdrs["Content-Type"], "image/jpeg")
 182
 183     def test_if_modified_since(self):
 184         resp1 = Conn().get(self.url, method="HEAD")
 185         status, hdrs, body = parse(resp1)
 186         lastmod = hdrs["Last-Modified"]
 187
 188         resp2 = Conn().get(self.url, method="GET", req_hdrs =
 189             {"If-Modified-Since": lastmod })
 190         status, hdrs, body = parse(resp2)
 191         self.assertContains(status, "304 Not Modified")
 192         self.assertEquals(hdrs["Accept-Ranges"], "bytes")
 193         self.assertFalse(hdrs.has_key("Last-Modified"))
 194         self.assertFalse(hdrs.has_key("Content-Length"))
 195         self.assertFalse(hdrs.has_key("Content-Type"))
 196
 197     def drive_range(self, range_in, range_out, len_out, data_out,
 198             status_out = "206 Partial Content"):
 199         resp = Conn().get(self.url, req_hdrs = {"Range": "bytes="+range_in})
 200         status, hdrs, body = parse(resp)
 201         self.assertContains(status, status_out)
 202         self.assertEquals(hdrs["Accept-Ranges"], "bytes")
 203         self.assertEquals(hdrs["Content-Range"], "bytes "+range_out)
 204         self.assertEquals(hdrs["Content-Length"], str(len_out))
 205         self.assertEquals(body, data_out)
 206
 207     def test_range_reasonable(self):
 208         self.drive_range("10-20", "10-20/%d" % self.datalen,
 209             20-10+1, self.data[10:20+1])
 210
 211     def test_range_tail(self):
 212         self.drive_range("10-", "10-%d/%d" % (self.datalen-1, self.datalen),
 213             self.datalen-10, self.data[10:])
 214
 215     def test_range_negative(self):
 216         self.drive_range("-25", "%d-%d/%d" % (
 217             self.datalen-25, self.datalen-1, self.datalen),
 218             25, self.data[-25:])
 219
 220     def test_range_bad_end(self):
 221         # expecting same result as test_range_negative
 222         self.drive_range("%d-%d"%(self.datalen-25, self.datalen*2),
 223             "%d-%d/%d"%(self.datalen-25, self.datalen-1, self.datalen),
 224             25, self.data[-25:])
 225
 226     def test_range_bad_start(self):
 227         resp = Conn().get(self.url, req_hdrs = {"Range": "bytes=%d-"%(
 228             self.datalen*2)})
 229         status, hdrs, body = parse(resp)
 230         self.assertContains(status, "416 Requested Range Not Satisfiable")
 231
 232     def test_range_backwards(self):
 233         resp = Conn().get(self.url, req_hdrs = {"Range": "bytes=20-10"})
 234         status, hdrs, body = parse(resp)
 235         self.assertContains(status, "416 Requested Range Not Satisfiable")
 236
 237 if __name__ == '__main__':
 238     setUpModule()
 239     unittest.main()
 240
 241 # vim:set ts=4 sw=4 et: