test_cgi.py 15 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427
  1. from test.test_support import run_unittest, check_warnings
  2. import cgi
  3. import os
  4. import sys
  5. import tempfile
  6. import unittest
  7. from collections import namedtuple
  8. class HackedSysModule:
  9. # The regression test will have real values in sys.argv, which
  10. # will completely confuse the test of the cgi module
  11. argv = []
  12. stdin = sys.stdin
  13. cgi.sys = HackedSysModule()
  14. try:
  15. from cStringIO import StringIO
  16. except ImportError:
  17. from StringIO import StringIO
  18. class ComparableException:
  19. def __init__(self, err):
  20. self.err = err
  21. def __str__(self):
  22. return str(self.err)
  23. def __cmp__(self, anExc):
  24. if not isinstance(anExc, Exception):
  25. return -1
  26. x = cmp(self.err.__class__, anExc.__class__)
  27. if x != 0:
  28. return x
  29. return cmp(self.err.args, anExc.args)
  30. def __getattr__(self, attr):
  31. return getattr(self.err, attr)
  32. def do_test(buf, method):
  33. env = {}
  34. if method == "GET":
  35. fp = None
  36. env['REQUEST_METHOD'] = 'GET'
  37. env['QUERY_STRING'] = buf
  38. elif method == "POST":
  39. fp = StringIO(buf)
  40. env['REQUEST_METHOD'] = 'POST'
  41. env['CONTENT_TYPE'] = 'application/x-www-form-urlencoded'
  42. env['CONTENT_LENGTH'] = str(len(buf))
  43. else:
  44. raise ValueError, "unknown method: %s" % method
  45. try:
  46. return cgi.parse(fp, env, strict_parsing=1)
  47. except StandardError, err:
  48. return ComparableException(err)
  49. parse_strict_test_cases = [
  50. ("", ValueError("bad query field: ''")),
  51. ("&", ValueError("bad query field: ''")),
  52. ("&&", ValueError("bad query field: ''")),
  53. (";", ValueError("bad query field: ''")),
  54. (";&;", ValueError("bad query field: ''")),
  55. # Should the next few really be valid?
  56. ("=", {}),
  57. ("=&=", {}),
  58. ("=;=", {}),
  59. # This rest seem to make sense
  60. ("=a", {'': ['a']}),
  61. ("&=a", ValueError("bad query field: ''")),
  62. ("=a&", ValueError("bad query field: ''")),
  63. ("=&a", ValueError("bad query field: 'a'")),
  64. ("b=a", {'b': ['a']}),
  65. ("b+=a", {'b ': ['a']}),
  66. ("a=b=a", {'a': ['b=a']}),
  67. ("a=+b=a", {'a': [' b=a']}),
  68. ("&b=a", ValueError("bad query field: ''")),
  69. ("b&=a", ValueError("bad query field: 'b'")),
  70. ("a=a+b&b=b+c", {'a': ['a b'], 'b': ['b c']}),
  71. ("a=a+b&a=b+a", {'a': ['a b', 'b a']}),
  72. ("x=1&y=2.0&z=2-3.%2b0", {'x': ['1'], 'y': ['2.0'], 'z': ['2-3.+0']}),
  73. ("x=1;y=2.0&z=2-3.%2b0", {'x': ['1'], 'y': ['2.0'], 'z': ['2-3.+0']}),
  74. ("x=1;y=2.0;z=2-3.%2b0", {'x': ['1'], 'y': ['2.0'], 'z': ['2-3.+0']}),
  75. ("Hbc5161168c542333633315dee1182227:key_store_seqid=400006&cuyer=r&view=bustomer&order_id=0bb2e248638833d48cb7fed300000f1b&expire=964546263&lobale=en-US&kid=130003.300038&ss=env",
  76. {'Hbc5161168c542333633315dee1182227:key_store_seqid': ['400006'],
  77. 'cuyer': ['r'],
  78. 'expire': ['964546263'],
  79. 'kid': ['130003.300038'],
  80. 'lobale': ['en-US'],
  81. 'order_id': ['0bb2e248638833d48cb7fed300000f1b'],
  82. 'ss': ['env'],
  83. 'view': ['bustomer'],
  84. }),
  85. ("group_id=5470&set=custom&_assigned_to=31392&_status=1&_category=100&SUBMIT=Browse",
  86. {'SUBMIT': ['Browse'],
  87. '_assigned_to': ['31392'],
  88. '_category': ['100'],
  89. '_status': ['1'],
  90. 'group_id': ['5470'],
  91. 'set': ['custom'],
  92. })
  93. ]
  94. def first_elts(list):
  95. return map(lambda x:x[0], list)
  96. def first_second_elts(list):
  97. return map(lambda p:(p[0], p[1][0]), list)
  98. def gen_result(data, environ):
  99. fake_stdin = StringIO(data)
  100. fake_stdin.seek(0)
  101. form = cgi.FieldStorage(fp=fake_stdin, environ=environ)
  102. result = {}
  103. for k, v in dict(form).items():
  104. result[k] = isinstance(v, list) and form.getlist(k) or v.value
  105. return result
  106. class CgiTests(unittest.TestCase):
  107. def test_escape(self):
  108. self.assertEqual("test & string", cgi.escape("test & string"))
  109. self.assertEqual("&lt;test string&gt;", cgi.escape("<test string>"))
  110. self.assertEqual("&quot;test string&quot;", cgi.escape('"test string"', True))
  111. def test_strict(self):
  112. for orig, expect in parse_strict_test_cases:
  113. # Test basic parsing
  114. d = do_test(orig, "GET")
  115. self.assertEqual(d, expect, "Error parsing %s" % repr(orig))
  116. d = do_test(orig, "POST")
  117. self.assertEqual(d, expect, "Error parsing %s" % repr(orig))
  118. env = {'QUERY_STRING': orig}
  119. fcd = cgi.FormContentDict(env)
  120. sd = cgi.SvFormContentDict(env)
  121. fs = cgi.FieldStorage(environ=env)
  122. if isinstance(expect, dict):
  123. # test dict interface
  124. self.assertEqual(len(expect), len(fcd))
  125. self.assertItemsEqual(expect.keys(), fcd.keys())
  126. self.assertItemsEqual(expect.values(), fcd.values())
  127. self.assertItemsEqual(expect.items(), fcd.items())
  128. self.assertEqual(fcd.get("nonexistent field", "default"), "default")
  129. self.assertEqual(len(sd), len(fs))
  130. self.assertItemsEqual(sd.keys(), fs.keys())
  131. self.assertEqual(fs.getvalue("nonexistent field", "default"), "default")
  132. # test individual fields
  133. for key in expect.keys():
  134. expect_val = expect[key]
  135. self.assertTrue(fcd.has_key(key))
  136. self.assertItemsEqual(fcd[key], expect[key])
  137. self.assertEqual(fcd.get(key, "default"), fcd[key])
  138. self.assertTrue(fs.has_key(key))
  139. if len(expect_val) > 1:
  140. single_value = 0
  141. else:
  142. single_value = 1
  143. try:
  144. val = sd[key]
  145. except IndexError:
  146. self.assertFalse(single_value)
  147. self.assertEqual(fs.getvalue(key), expect_val)
  148. else:
  149. self.assertTrue(single_value)
  150. self.assertEqual(val, expect_val[0])
  151. self.assertEqual(fs.getvalue(key), expect_val[0])
  152. self.assertItemsEqual(sd.getlist(key), expect_val)
  153. if single_value:
  154. self.assertItemsEqual(sd.values(),
  155. first_elts(expect.values()))
  156. self.assertItemsEqual(sd.items(),
  157. first_second_elts(expect.items()))
  158. def test_weird_formcontentdict(self):
  159. # Test the weird FormContentDict classes
  160. env = {'QUERY_STRING': "x=1&y=2.0&z=2-3.%2b0&1=1abc"}
  161. expect = {'x': 1, 'y': 2.0, 'z': '2-3.+0', '1': '1abc'}
  162. d = cgi.InterpFormContentDict(env)
  163. for k, v in expect.items():
  164. self.assertEqual(d[k], v)
  165. for k, v in d.items():
  166. self.assertEqual(expect[k], v)
  167. self.assertItemsEqual(expect.values(), d.values())
  168. def test_log(self):
  169. cgi.log("Testing")
  170. cgi.logfp = StringIO()
  171. cgi.initlog("%s", "Testing initlog 1")
  172. cgi.log("%s", "Testing log 2")
  173. self.assertEqual(cgi.logfp.getvalue(), "Testing initlog 1\nTesting log 2\n")
  174. if os.path.exists("/dev/null"):
  175. cgi.logfp = None
  176. cgi.logfile = "/dev/null"
  177. cgi.initlog("%s", "Testing log 3")
  178. cgi.log("Testing log 4")
  179. def test_fieldstorage_readline(self):
  180. # FieldStorage uses readline, which has the capacity to read all
  181. # contents of the input file into memory; we use readline's size argument
  182. # to prevent that for files that do not contain any newlines in
  183. # non-GET/HEAD requests
  184. class TestReadlineFile:
  185. def __init__(self, file):
  186. self.file = file
  187. self.numcalls = 0
  188. def readline(self, size=None):
  189. self.numcalls += 1
  190. if size:
  191. return self.file.readline(size)
  192. else:
  193. return self.file.readline()
  194. def __getattr__(self, name):
  195. file = self.__dict__['file']
  196. a = getattr(file, name)
  197. if not isinstance(a, int):
  198. setattr(self, name, a)
  199. return a
  200. f = TestReadlineFile(tempfile.TemporaryFile())
  201. f.write('x' * 256 * 1024)
  202. f.seek(0)
  203. env = {'REQUEST_METHOD':'PUT'}
  204. fs = cgi.FieldStorage(fp=f, environ=env)
  205. # if we're not chunking properly, readline is only called twice
  206. # (by read_binary); if we are chunking properly, it will be called 5 times
  207. # as long as the chunksize is 1 << 16.
  208. self.assertGreater(f.numcalls, 2)
  209. def test_fieldstorage_invalid(self):
  210. fs = cgi.FieldStorage()
  211. self.assertFalse(fs)
  212. self.assertRaises(TypeError, bool(fs))
  213. self.assertEqual(list(fs), list(fs.keys()))
  214. fs.list.append(namedtuple('MockFieldStorage', 'name')('fieldvalue'))
  215. self.assertTrue(fs)
  216. def test_fieldstorage_multipart(self):
  217. #Test basic FieldStorage multipart parsing
  218. env = {'REQUEST_METHOD':'POST', 'CONTENT_TYPE':'multipart/form-data; boundary=---------------------------721837373350705526688164684', 'CONTENT_LENGTH':'558'}
  219. postdata = """-----------------------------721837373350705526688164684
  220. Content-Disposition: form-data; name="id"
  221. 1234
  222. -----------------------------721837373350705526688164684
  223. Content-Disposition: form-data; name="title"
  224. -----------------------------721837373350705526688164684
  225. Content-Disposition: form-data; name="file"; filename="test.txt"
  226. Content-Type: text/plain
  227. Testing 123.
  228. -----------------------------721837373350705526688164684
  229. Content-Disposition: form-data; name="submit"
  230. Add\x20
  231. -----------------------------721837373350705526688164684--
  232. """
  233. fs = cgi.FieldStorage(fp=StringIO(postdata), environ=env)
  234. self.assertEqual(len(fs.list), 4)
  235. expect = [{'name':'id', 'filename':None, 'value':'1234'},
  236. {'name':'title', 'filename':None, 'value':''},
  237. {'name':'file', 'filename':'test.txt','value':'Testing 123.\n'},
  238. {'name':'submit', 'filename':None, 'value':' Add '}]
  239. for x in range(len(fs.list)):
  240. for k, exp in expect[x].items():
  241. got = getattr(fs.list[x], k)
  242. self.assertEqual(got, exp)
  243. def test_fieldstorage_multipart_maxline(self):
  244. # Issue #18167
  245. maxline = 1 << 16
  246. self.maxDiff = None
  247. def check(content):
  248. data = """
  249. ---123
  250. Content-Disposition: form-data; name="upload"; filename="fake.txt"
  251. Content-Type: text/plain
  252. %s
  253. ---123--
  254. """.replace('\n', '\r\n') % content
  255. environ = {
  256. 'CONTENT_LENGTH': str(len(data)),
  257. 'CONTENT_TYPE': 'multipart/form-data; boundary=-123',
  258. 'REQUEST_METHOD': 'POST',
  259. }
  260. self.assertEqual(gen_result(data, environ), {'upload': content})
  261. check('x' * (maxline - 1))
  262. check('x' * (maxline - 1) + '\r')
  263. check('x' * (maxline - 1) + '\r' + 'y' * (maxline - 1))
  264. _qs_result = {
  265. 'key1': 'value1',
  266. 'key2': ['value2x', 'value2y'],
  267. 'key3': 'value3',
  268. 'key4': 'value4'
  269. }
  270. def testQSAndUrlEncode(self):
  271. data = "key2=value2x&key3=value3&key4=value4"
  272. environ = {
  273. 'CONTENT_LENGTH': str(len(data)),
  274. 'CONTENT_TYPE': 'application/x-www-form-urlencoded',
  275. 'QUERY_STRING': 'key1=value1&key2=value2y',
  276. 'REQUEST_METHOD': 'POST',
  277. }
  278. v = gen_result(data, environ)
  279. self.assertEqual(self._qs_result, v)
  280. def testQSAndFormData(self):
  281. data = """
  282. ---123
  283. Content-Disposition: form-data; name="key2"
  284. value2y
  285. ---123
  286. Content-Disposition: form-data; name="key3"
  287. value3
  288. ---123
  289. Content-Disposition: form-data; name="key4"
  290. value4
  291. ---123--
  292. """
  293. environ = {
  294. 'CONTENT_LENGTH': str(len(data)),
  295. 'CONTENT_TYPE': 'multipart/form-data; boundary=-123',
  296. 'QUERY_STRING': 'key1=value1&key2=value2x',
  297. 'REQUEST_METHOD': 'POST',
  298. }
  299. v = gen_result(data, environ)
  300. self.assertEqual(self._qs_result, v)
  301. def testQSAndFormDataFile(self):
  302. data = """
  303. ---123
  304. Content-Disposition: form-data; name="key2"
  305. value2y
  306. ---123
  307. Content-Disposition: form-data; name="key3"
  308. value3
  309. ---123
  310. Content-Disposition: form-data; name="key4"
  311. value4
  312. ---123
  313. Content-Disposition: form-data; name="upload"; filename="fake.txt"
  314. Content-Type: text/plain
  315. this is the content of the fake file
  316. ---123--
  317. """
  318. environ = {
  319. 'CONTENT_LENGTH': str(len(data)),
  320. 'CONTENT_TYPE': 'multipart/form-data; boundary=-123',
  321. 'QUERY_STRING': 'key1=value1&key2=value2x',
  322. 'REQUEST_METHOD': 'POST',
  323. }
  324. result = self._qs_result.copy()
  325. result.update({
  326. 'upload': 'this is the content of the fake file\n'
  327. })
  328. v = gen_result(data, environ)
  329. self.assertEqual(result, v)
  330. def test_deprecated_parse_qs(self):
  331. # this func is moved to urlparse, this is just a sanity check
  332. with check_warnings(('cgi.parse_qs is deprecated, use urlparse.'
  333. 'parse_qs instead', PendingDeprecationWarning)):
  334. self.assertEqual({'a': ['A1'], 'B': ['B3'], 'b': ['B2']},
  335. cgi.parse_qs('a=A1&b=B2&B=B3'))
  336. def test_deprecated_parse_qsl(self):
  337. # this func is moved to urlparse, this is just a sanity check
  338. with check_warnings(('cgi.parse_qsl is deprecated, use urlparse.'
  339. 'parse_qsl instead', PendingDeprecationWarning)):
  340. self.assertEqual([('a', 'A1'), ('b', 'B2'), ('B', 'B3')],
  341. cgi.parse_qsl('a=A1&b=B2&B=B3'))
  342. def test_parse_header(self):
  343. self.assertEqual(
  344. cgi.parse_header("text/plain"),
  345. ("text/plain", {}))
  346. self.assertEqual(
  347. cgi.parse_header("text/vnd.just.made.this.up ; "),
  348. ("text/vnd.just.made.this.up", {}))
  349. self.assertEqual(
  350. cgi.parse_header("text/plain;charset=us-ascii"),
  351. ("text/plain", {"charset": "us-ascii"}))
  352. self.assertEqual(
  353. cgi.parse_header('text/plain ; charset="us-ascii"'),
  354. ("text/plain", {"charset": "us-ascii"}))
  355. self.assertEqual(
  356. cgi.parse_header('text/plain ; charset="us-ascii"; another=opt'),
  357. ("text/plain", {"charset": "us-ascii", "another": "opt"}))
  358. self.assertEqual(
  359. cgi.parse_header('attachment; filename="silly.txt"'),
  360. ("attachment", {"filename": "silly.txt"}))
  361. self.assertEqual(
  362. cgi.parse_header('attachment; filename="strange;name"'),
  363. ("attachment", {"filename": "strange;name"}))
  364. self.assertEqual(
  365. cgi.parse_header('attachment; filename="strange;name";size=123;'),
  366. ("attachment", {"filename": "strange;name", "size": "123"}))
  367. self.assertEqual(
  368. cgi.parse_header('form-data; name="files"; filename="fo\\"o;bar"'),
  369. ("form-data", {"name": "files", "filename": 'fo"o;bar'}))
  370. def test_main():
  371. run_unittest(CgiTests)
  372. if __name__ == '__main__':
  373. test_main()