1#! /usr/bin/env python
2# -*- coding: utf-8 -*-
3# vi:ts=4:et
4
5from . import localhost
6import flaky
7import os.path
8import pycurl
9import unittest
10try:
11    import json
12except ImportError:
13    import simplejson as json
14try:
15    import urllib.parse as urllib_parse
16except ImportError:
17    import urllib as urllib_parse
18
19from . import appmanager
20from . import util
21
22setup_module, teardown_module = appmanager.setup(('app', 8380))
23
24@flaky.flaky(max_runs=3)
25class PostTest(unittest.TestCase):
26    def setUp(self):
27        self.curl = util.DefaultCurl()
28
29    def tearDown(self):
30        self.curl.close()
31
32    def test_post_single_field(self):
33        pf = {'field1': 'value1'}
34        self.urlencode_and_check(pf)
35
36    def test_post_multiple_fields(self):
37        pf = {'field1':'value1', 'field2':'value2 with blanks', 'field3':'value3'}
38        self.urlencode_and_check(pf)
39
40    def test_post_fields_with_ampersand(self):
41        pf = {'field1':'value1', 'field2':'value2 with blanks and & chars',
42              'field3':'value3'}
43        self.urlencode_and_check(pf)
44
45    def urlencode_and_check(self, pf):
46        self.curl.setopt(pycurl.URL, 'http://%s:8380/postfields' % localhost)
47        postfields = urllib_parse.urlencode(pf)
48        self.curl.setopt(pycurl.POSTFIELDS, postfields)
49
50        # But directly passing urlencode result into setopt call:
51        #self.curl.setopt(pycurl.POSTFIELDS, urllib_parse.urlencode(pf))
52        # produces:
53        # {'\x00\x00\x00\x00\x80\x00\x00\x00\x00\x00\x00\x00\x00': ''}
54        # Traceback (most recent call last):
55        #   File "/usr/local/bin/bottle.py", line 744, in _handle
56        #     return route.call(**args)
57        #   File "/usr/local/bin/bottle.py", line 1479, in wrapper
58        #     rv = callback(*a, **ka)
59        #   File "/home/pie/apps/pycurl/tests/app.py", line 21, in postfields
60        #     return json.dumps(dict(bottle.request.forms))
61        #   File "/usr/local/lib/python2.7/json/__init__.py", line 231, in dumps
62        #     return _default_encoder.encode(obj)
63        #   File "/usr/local/lib/python2.7/json/encoder.py", line 201, in encode
64        #     chunks = self.iterencode(o, _one_shot=True)
65        #   File "/usr/local/lib/python2.7/json/encoder.py", line 264, in iterencode
66        #     return _iterencode(o, 0)
67        # UnicodeDecodeError: 'utf8' codec can't decode byte 0x80 in position 4: invalid start byte
68
69        #self.curl.setopt(pycurl.VERBOSE, 1)
70        sio = util.BytesIO()
71        self.curl.setopt(pycurl.WRITEFUNCTION, sio.write)
72        self.curl.perform()
73        self.assertEqual(200, self.curl.getinfo(pycurl.HTTP_CODE))
74        body = sio.getvalue().decode()
75        returned_fields = json.loads(body)
76        self.assertEqual(pf, returned_fields)
77
78    def test_post_with_null_byte(self):
79        send = [
80            ('field3', (pycurl.FORM_CONTENTS, 'this is wei\000rd, but null-bytes are okay'))
81        ]
82        expect = {
83            'field3': 'this is wei\000rd, but null-bytes are okay',
84        }
85        self.check_post(send, expect, 'http://%s:8380/postfields' % localhost)
86
87    def test_post_file(self):
88        path = os.path.join(os.path.dirname(__file__), '..', 'README.rst')
89        f = open(path)
90        try:
91            contents = f.read()
92        finally:
93            f.close()
94        send = [
95            #('field2', (pycurl.FORM_FILE, 'test_post.py', pycurl.FORM_FILE, 'test_post2.py')),
96            ('field2', (pycurl.FORM_FILE, path)),
97        ]
98        expect = [{
99            'name': 'field2',
100            'filename': 'README.rst',
101            'data': contents,
102        }]
103        self.check_post(send, expect, 'http://%s:8380/files' % localhost)
104
105    def test_post_byte_buffer(self):
106        contents = util.b('hello, world!')
107        send = [
108            ('field2', (pycurl.FORM_BUFFER, 'uploaded.file', pycurl.FORM_BUFFERPTR, contents)),
109        ]
110        expect = [{
111            'name': 'field2',
112            'filename': 'uploaded.file',
113            'data': 'hello, world!',
114        }]
115        self.check_post(send, expect, 'http://%s:8380/files' % localhost)
116
117    def test_post_unicode_buffer(self):
118        contents = util.u('hello, world!')
119        send = [
120            ('field2', (pycurl.FORM_BUFFER, 'uploaded.file', pycurl.FORM_BUFFERPTR, contents)),
121        ]
122        expect = [{
123            'name': 'field2',
124            'filename': 'uploaded.file',
125            'data': 'hello, world!',
126        }]
127        self.check_post(send, expect, 'http://%s:8380/files' % localhost)
128
129    def test_post_tuple_of_tuples_of_tuples(self):
130        contents = util.u('hello, world!')
131        send = (
132            ('field2', (pycurl.FORM_BUFFER, 'uploaded.file', pycurl.FORM_BUFFERPTR, contents)),
133        )
134        expect = [{
135            'name': 'field2',
136            'filename': 'uploaded.file',
137            'data': 'hello, world!',
138        }]
139        self.check_post(send, expect, 'http://%s:8380/files' % localhost)
140
141    def test_post_tuple_of_lists_of_tuples(self):
142        contents = util.u('hello, world!')
143        send = (
144            ['field2', (pycurl.FORM_BUFFER, 'uploaded.file', pycurl.FORM_BUFFERPTR, contents)],
145        )
146        expect = [{
147            'name': 'field2',
148            'filename': 'uploaded.file',
149            'data': 'hello, world!',
150        }]
151        self.check_post(send, expect, 'http://%s:8380/files' % localhost)
152
153    def test_post_tuple_of_tuple_of_lists(self):
154        contents = util.u('hello, world!')
155        send = (
156            ('field2', [pycurl.FORM_BUFFER, 'uploaded.file', pycurl.FORM_BUFFERPTR, contents]),
157        )
158        expect = [{
159            'name': 'field2',
160            'filename': 'uploaded.file',
161            'data': 'hello, world!',
162        }]
163        self.check_post(send, expect, 'http://%s:8380/files' % localhost)
164
165    def test_post_list_of_tuple_of_tuples(self):
166        contents = util.u('hello, world!')
167        send = [
168            ('field2', (pycurl.FORM_BUFFER, 'uploaded.file', pycurl.FORM_BUFFERPTR, contents)),
169        ]
170        expect = [{
171            'name': 'field2',
172            'filename': 'uploaded.file',
173            'data': 'hello, world!',
174        }]
175        self.check_post(send, expect, 'http://%s:8380/files' % localhost)
176
177    def test_post_list_of_list_of_lists(self):
178        contents = util.u('hello, world!')
179        send = [
180            ['field2', [pycurl.FORM_BUFFER, 'uploaded.file', pycurl.FORM_BUFFERPTR, contents]],
181        ]
182        expect = [{
183            'name': 'field2',
184            'filename': 'uploaded.file',
185            'data': 'hello, world!',
186        }]
187        self.check_post(send, expect, 'http://%s:8380/files' % localhost)
188
189    # XXX this test takes about a second to run, check keep-alives?
190    def check_post(self, send, expect, endpoint):
191        self.curl.setopt(pycurl.URL, endpoint)
192        self.curl.setopt(pycurl.HTTPPOST, send)
193        #self.curl.setopt(pycurl.VERBOSE, 1)
194        sio = util.BytesIO()
195        self.curl.setopt(pycurl.WRITEFUNCTION, sio.write)
196        self.curl.perform()
197        self.assertEqual(200, self.curl.getinfo(pycurl.HTTP_CODE))
198        body = sio.getvalue().decode()
199        returned_fields = json.loads(body)
200        self.assertEqual(expect, returned_fields)
201