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

   1 #!/usr/bin/env python
   2 # This is run by the "cover" script.
   3 import unittest
   4 import socket
   5 import signal
   6 import re
   7 import os
   8 import random
   9
  10 WWWROOT = "tmp.httpd.tests"
  11
  12 class Conn:
  13     def __init__(self):
  14         self.port = 12346
  15         self.s = socket.socket()
  16         self.s.connect(("0.0.0.0", self.port))
  17         # connect throws socket.error on connection refused
  18
  19     def get(self, url, http_ver="1.0", endl="\n", req_hdrs={}, method="GET"):
  20         req = method+" "+url
  21         if http_ver is not None:
  22             req += " HTTP/"+http_ver
  23         req += endl
  24         if http_ver is not None:
  25             req_hdrs["User-Agent"] = "test.py"
  26             req_hdrs["Connection"] = "close"
  27             for k,v in req_hdrs.items():
  28                 req += k+": "+v+endl
  29         req += endl # end of request
  30         self.s.send(req)
  31         ret = ""
  32         while True:
  33             signal.alarm(1) # don't wait forever
  34             r = self.s.recv(65536)
  35             signal.alarm(0)
  36             if r == "":
  37                 break
  38             else:
  39                 ret += r
  40         return ret
  41
  42 def parse(resp):
  43     """
  44     Parse response into status line, headers and body.
  45     """
  46     pos = resp.index("\r\n\r\n") # throws exception on failure
  47     head = resp[:pos]
  48     body = resp[pos+4:]
  49     status,head = head.split("\r\n", 1)
  50     hdrs = {}
  51     for line in head.split("\r\n"):
  52         k, v = line.split(": ", 1)
  53         hdrs[k] = v
  54     return (status, hdrs, body)
  55
  56 class TestHelper(unittest.TestCase):
  57     def assertContains(self, body, *strings):
  58         for s in strings:
  59             self.assertTrue(s in body,
  60                             msg="expected %s in %s"%(repr(s), repr(body)))
  61
  62     def assertIsIndex(self, body, path):
  63         self.assertContains(body,
  64             "<title>%s</title>\n"%path,
  65             "<h1>%s</h1>\n"%path,
  66             '<a href="..">..</a>/',
  67             'Generated by darkhttpd')
  68
  69     def assertIsInvalid(self, body, path):
  70         self.assertContains(body,
  71             "<title>400 Bad Request</title>",
  72             "<h1>Bad Request</h1>\n",
  73             "You requested an invalid URL: %s\n"%path,
  74             'Generated by darkhttpd')
  75
  76     def drive_range(self, range_in, range_out, len_out, data_out,
  77             status_out = "206 Partial Content"):
  78         resp = Conn().get(self.url, req_hdrs = {"Range": "bytes="+range_in})
  79         status, hdrs, body = parse(resp)
  80         self.assertContains(status, status_out)
  81         self.assertEquals(hdrs["Accept-Ranges"], "bytes")
  82         self.assertEquals(hdrs["Content-Range"], "bytes "+range_out)
  83         self.assertEquals(hdrs["Content-Length"], str(len_out))
  84         self.assertEquals(body, data_out)
  85
  86 class TestDirList(TestHelper):
  87     def setUp(self):
  88         self.fn = WWWROOT+"/escape#this"
  89         open(self.fn, "w").write("x"*12345)
  90
  91     def tearDown(self):
  92         os.unlink(self.fn)
  93
  94     def test_dirlist_escape(self):
  95         resp = Conn().get("/")
  96         status, hdrs, body = parse(resp)
  97         self.assertEquals(ord("#"), 0x23)
  98         self.assertContains(body, "escape%23this", "12345")
  99
 100 class TestCases(TestHelper):
 101     pass # these get autogenerated in setUpModule()
 102
 103 def nerf(s):
 104     return re.sub("[^a-zA-Z0-9]", "_", s)
 105
 106 def makeCase(name, url, hdr_checker=None, body_checker=None,
 107              req_hdrs={"User-Agent": "test.py"},
 108              http_ver=None, endl="\n"):
 109     def do_test(self):
 110         resp = Conn().get(url, http_ver, endl, req_hdrs)
 111         if http_ver is None:
 112             status = ""
 113             hdrs = {}
 114             body = resp
 115         else:
 116             status, hdrs, body = parse(resp)
 117
 118         if hdr_checker is not None and http_ver is not None:
 119             hdr_checker(self, hdrs)
 120
 121         if body_checker is not None:
 122             body_checker(self, body)
 123
 124         # FIXME: check status
 125         if http_ver is not None:
 126             prefix = "HTTP/1.1 " # should 1.0 stay 1.0?
 127             self.assertTrue(status.startswith(prefix),
 128                 msg="%s at start of %s"%(repr(prefix), repr(status)))
 129
 130     v = http_ver
 131     if v is None:
 132         v = "0.9"
 133     test_name = "_".join([
 134         "test",
 135         nerf(name),
 136         nerf("HTTP"+v),
 137         {"\n":"LF", "\r\n":"CRLF"}[endl],
 138     ])
 139     do_test.__name__ = test_name # hax
 140     setattr(TestCases, test_name, do_test)
 141
 142 def makeCases(name, url, hdr_checker=None, body_checker=None,
 143               req_hdrs={"User-Agent": "test.py"}):
 144     for http_ver in [None, "1.0", "1.1"]:
 145         for endl in ["\n", "\r\n"]:
 146             makeCase(name, url, hdr_checker, body_checker,
 147                      req_hdrs, http_ver, endl)
 148
 149 def makeSimpleCases(name, url, assert_name):
 150     makeCases(name, url, None,
 151         lambda self,body: getattr(self, assert_name)(body, url))
 152
 153 def setUpModule():
 154     for args in [
 155         ["index",                "/",               "assertIsIndex"],
 156         ["up dir",               "/dir/../",        "assertIsIndex"],
 157         ["extra slashes",        "//dir///..////",  "assertIsIndex"],
 158         ["no trailing slash",    "/dir/..",         "assertIsIndex"],
 159         ["no leading slash",     "dir/../",         "assertIsInvalid"],
 160         ["invalid up dir",       "/../",            "assertIsInvalid"],
 161         ["fancy invalid up dir", "/./dir/./../../", "assertIsInvalid"],
 162         ]:
 163         makeSimpleCases(*args)
 164
 165 class TestDirRedirect(TestHelper):
 166     def setUp(self):
 167         self.url = "/mydir"
 168         self.fn = WWWROOT + self.url
 169         os.mkdir(self.fn)
 170
 171     def tearDown(self):
 172         os.rmdir(self.fn)
 173
 174     def test_dir_redirect(self):
 175         resp = Conn().get(self.url)
 176         status, hdrs, body = parse(resp)
 177         self.assertContains(status, "301 Moved Permanently")
 178         self.assertEquals(hdrs["Location"], self.url+"/") # trailing slash
 179
 180 class TestFileGet(TestHelper):
 181     def setUp(self):
 182         self.datalen = 2345
 183         self.data = "".join(
 184             [chr(random.randint(0,255)) for _ in xrange(self.datalen)])
 185         self.url = "/data.jpeg"
 186         self.fn = WWWROOT + self.url
 187         open(self.fn, "w").write(self.data)
 188
 189     def tearDown(self):
 190         os.unlink(self.fn)
 191
 192     def get_helper(self, url):
 193         resp = Conn().get(url)
 194         status, hdrs, body = parse(resp)
 195         self.assertContains(status, "200 OK")
 196         self.assertEquals(hdrs["Accept-Ranges"], "bytes")
 197         self.assertEquals(hdrs["Content-Length"], str(self.datalen))
 198         self.assertEquals(hdrs["Content-Type"], "image/jpeg")
 199         self.assertContains(hdrs["Server"], "darkhttpd/")
 200         self.assertEquals(body, self.data)
 201
 202     def test_file_get(self):
 203         self.get_helper(self.url)
 204
 205     def test_file_get_redundant_dots(self):
 206         self.get_helper("/././." + self.url)
 207
 208     def test_file_head(self):
 209         resp = Conn().get(self.url, method="HEAD")
 210         status, hdrs, body = parse(resp)
 211         self.assertContains(status, "200 OK")
 212         self.assertEquals(hdrs["Accept-Ranges"], "bytes")
 213         self.assertEquals(hdrs["Content-Length"], str(self.datalen))
 214         self.assertEquals(hdrs["Content-Type"], "image/jpeg")
 215
 216     def test_if_modified_since(self):
 217         resp1 = Conn().get(self.url, method="HEAD")
 218         status, hdrs, body = parse(resp1)
 219         lastmod = hdrs["Last-Modified"]
 220
 221         resp2 = Conn().get(self.url, method="GET", req_hdrs =
 222             {"If-Modified-Since": lastmod })
 223         status, hdrs, body = parse(resp2)
 224         self.assertContains(status, "304 Not Modified")
 225         self.assertEquals(hdrs["Accept-Ranges"], "bytes")
 226         self.assertFalse(hdrs.has_key("Last-Modified"))
 227         self.assertFalse(hdrs.has_key("Content-Length"))
 228         self.assertFalse(hdrs.has_key("Content-Type"))
 229
 230     def test_range_single(self):
 231         self.drive_range("5-5", "5-5/%d" % self.datalen,
 232             1, self.data[5])
 233
 234     def test_range_single_first(self):
 235         self.drive_range("0-0", "0-0/%d" % self.datalen,
 236             1, self.data[0])
 237
 238     def test_range_single_last(self):
 239         self.drive_range("%d-%d"%(self.datalen-1, self.datalen-1),
 240         "%d-%d/%d"%(self.datalen-1, self.datalen-1, self.datalen),
 241         1, self.data[-1])
 242
 243     def test_range_single_bad(self):
 244         resp = Conn().get(self.url, req_hdrs = {"Range":
 245             "bytes=%d-%d"%(self.datalen, self.datalen)})
 246         status, hdrs, body = parse(resp)
 247         self.assertContains(status, "416 Requested Range Not Satisfiable")
 248
 249     def test_range_reasonable(self):
 250         self.drive_range("10-20", "10-20/%d" % self.datalen,
 251             20-10+1, self.data[10:20+1])
 252
 253     def test_range_start_given(self):
 254         self.drive_range("10-", "10-%d/%d" % (self.datalen-1, self.datalen),
 255             self.datalen-10, self.data[10:])
 256
 257     def test_range_end_given(self):
 258         self.drive_range("-25",
 259             "%d-%d/%d"%(self.datalen-25, self.datalen-1, self.datalen),
 260             25, self.data[-25:])
 261
 262     def test_range_beyond_end(self):
 263         # expecting same result as test_range_end_given
 264         self.drive_range("%d-%d"%(self.datalen-25, self.datalen*2),
 265             "%d-%d/%d"%(self.datalen-25, self.datalen-1, self.datalen),
 266             25, self.data[-25:])
 267
 268     def test_range_end_given_oversize(self):
 269         # expecting full file
 270         self.drive_range("-%d"%(self.datalen*3),
 271             "0-%d/%d"%(self.datalen-1, self.datalen),
 272             self.datalen, self.data)
 273
 274     def test_range_bad_start(self):
 275         resp = Conn().get(self.url, req_hdrs = {"Range": "bytes=%d-"%(
 276             self.datalen*2)})
 277         status, hdrs, body = parse(resp)
 278         self.assertContains(status, "416 Requested Range Not Satisfiable")
 279
 280     def test_range_backwards(self):
 281         resp = Conn().get(self.url, req_hdrs = {"Range": "bytes=20-10"})
 282         status, hdrs, body = parse(resp)
 283         self.assertContains(status, "416 Requested Range Not Satisfiable")
 284
 285 def make_large_file(fn, boundary, data):
 286     big = 1<<33
 287     assert big == 8589934592L
 288     assert str(big) == "8589934592"
 289
 290     f = open(fn, "w")
 291     pos = boundary - len(data)/2
 292     f.seek(pos)
 293     assert f.tell() == pos
 294     assert f.tell() < boundary
 295     f.write(data)
 296     filesize = f.tell()
 297     assert filesize == pos + len(data)
 298     assert filesize > boundary
 299     f.close()
 300     return (pos, filesize)
 301
 302 class TestLargeFile2G(TestHelper):
 303     BOUNDARY = 1<<31
 304
 305     def setUp(self):
 306         self.datalen = 4096
 307         self.data = "".join(
 308             [chr(random.randint(0,255)) for _ in xrange(self.datalen)])
 309         self.url = "/big.jpeg"
 310         self.fn = WWWROOT + self.url
 311         self.filepos, self.filesize = make_large_file(
 312             self.fn, self.BOUNDARY, self.data)
 313
 314     def tearDown(self):
 315         os.unlink(self.fn)
 316
 317     def drive_start(self, ofs):
 318         req_start = self.BOUNDARY + ofs
 319         req_end = req_start + self.datalen/4 - 1
 320         range_in = "%d-%d"%(req_start, req_end)
 321         range_out = "%s/%d"%(range_in, self.filesize)
 322
 323         data_start = req_start - self.filepos
 324         data_end = data_start + self.datalen/4
 325
 326         self.drive_range(range_in, range_out, self.datalen/4,
 327             self.data[data_start:data_end])
 328
 329     def test_largefile_head(self):
 330         resp = Conn().get(self.url, method="HEAD")
 331         status, hdrs, body = parse(resp)
 332         self.assertContains(status, "200 OK")
 333         self.assertEquals(hdrs["Accept-Ranges"], "bytes")
 334         self.assertEquals(hdrs["Content-Length"], str(self.filesize))
 335         self.assertEquals(hdrs["Content-Type"], "image/jpeg")
 336
 337     def test_largefile__3(self): self.drive_start(-3)
 338     def test_largefile__2(self): self.drive_start(-2)
 339     def test_largefile__1(self): self.drive_start(-1)
 340     def test_largefile_0(self): self.drive_start(0)
 341     def test_largefile_1(self): self.drive_start(1)
 342     def test_largefile_2(self): self.drive_start(2)
 343     def test_largefile_3(self): self.drive_start(3)
 344
 345 class TestLargeFile4G(TestLargeFile2G):
 346     BOUNDARY = 1<<32
 347
 348 if __name__ == '__main__':
 349     setUpModule()
 350     unittest.main()
 351
 352 # vim:set ts=4 sw=4 et: