Python code coverage for Lib/test/test_xmlrpc.py

#	count	content
1	n/a	import base64
2	n/a	import datetime
3	n/a	import decimal
4	n/a	import sys
5	n/a	import time
6	n/a	import unittest
7	n/a	from unittest import mock
8	n/a	import xmlrpc.client as xmlrpclib
9	n/a	import xmlrpc.server
10	n/a	import http.client
11	n/a	import http, http.server
12	n/a	import socket
13	n/a	import re
14	n/a	import io
15	n/a	import contextlib
16	n/a	from test import support
17	n/a
18	n/a	try:
19	n/a	import gzip
20	n/a	except ImportError:
21	n/a	gzip = None
22	n/a	try:
23	n/a	import threading
24	n/a	except ImportError:
25	n/a	threading = None
26	n/a
27	n/a	alist = [{'astring': 'foo@bar.baz.spam',
28	n/a	'afloat': 7283.43,
29	n/a	'anint': 2**20,
30	n/a	'ashortlong': 2,
31	n/a	'anotherlist': ['.zyx.41'],
32	n/a	'abase64': xmlrpclib.Binary(b"my dog has fleas"),
33	n/a	'b64bytes': b"my dog has fleas",
34	n/a	'b64bytearray': bytearray(b"my dog has fleas"),
35	n/a	'boolean': False,
36	n/a	'unicode': '\u4000\u6000\u8000',
37	n/a	'ukey\u4000': 'regular value',
38	n/a	'datetime1': xmlrpclib.DateTime('20050210T11:41:23'),
39	n/a	'datetime2': xmlrpclib.DateTime(
40	n/a	(2005, 2, 10, 11, 41, 23, 0, 1, -1)),
41	n/a	'datetime3': xmlrpclib.DateTime(
42	n/a	datetime.datetime(2005, 2, 10, 11, 41, 23)),
43	n/a	}]
44	n/a
45	n/a	class XMLRPCTestCase(unittest.TestCase):
46	n/a
47	n/a	def test_dump_load(self):
48	n/a	dump = xmlrpclib.dumps((alist,))
49	n/a	load = xmlrpclib.loads(dump)
50	n/a	self.assertEqual(alist, load[0][0])
51	n/a
52	n/a	def test_dump_bare_datetime(self):
53	n/a	# This checks that an unwrapped datetime.date object can be handled
54	n/a	# by the marshalling code. This can't be done via test_dump_load()
55	n/a	# since with use_builtin_types set to 1 the unmarshaller would create
56	n/a	# datetime objects for the 'datetime[123]' keys as well
57	n/a	dt = datetime.datetime(2005, 2, 10, 11, 41, 23)
58	n/a	self.assertEqual(dt, xmlrpclib.DateTime('20050210T11:41:23'))
59	n/a	s = xmlrpclib.dumps((dt,))
60	n/a
61	n/a	result, m = xmlrpclib.loads(s, use_builtin_types=True)
62	n/a	(newdt,) = result
63	n/a	self.assertEqual(newdt, dt)
64	n/a	self.assertIs(type(newdt), datetime.datetime)
65	n/a	self.assertIsNone(m)
66	n/a
67	n/a	result, m = xmlrpclib.loads(s, use_builtin_types=False)
68	n/a	(newdt,) = result
69	n/a	self.assertEqual(newdt, dt)
70	n/a	self.assertIs(type(newdt), xmlrpclib.DateTime)
71	n/a	self.assertIsNone(m)
72	n/a
73	n/a	result, m = xmlrpclib.loads(s, use_datetime=True)
74	n/a	(newdt,) = result
75	n/a	self.assertEqual(newdt, dt)
76	n/a	self.assertIs(type(newdt), datetime.datetime)
77	n/a	self.assertIsNone(m)
78	n/a
79	n/a	result, m = xmlrpclib.loads(s, use_datetime=False)
80	n/a	(newdt,) = result
81	n/a	self.assertEqual(newdt, dt)
82	n/a	self.assertIs(type(newdt), xmlrpclib.DateTime)
83	n/a	self.assertIsNone(m)
84	n/a
85	n/a
86	n/a	def test_datetime_before_1900(self):
87	n/a	# same as before but with a date before 1900
88	n/a	dt = datetime.datetime(1, 2, 10, 11, 41, 23)
89	n/a	self.assertEqual(dt, xmlrpclib.DateTime('00010210T11:41:23'))
90	n/a	s = xmlrpclib.dumps((dt,))
91	n/a
92	n/a	result, m = xmlrpclib.loads(s, use_builtin_types=True)
93	n/a	(newdt,) = result
94	n/a	self.assertEqual(newdt, dt)
95	n/a	self.assertIs(type(newdt), datetime.datetime)
96	n/a	self.assertIsNone(m)
97	n/a
98	n/a	result, m = xmlrpclib.loads(s, use_builtin_types=False)
99	n/a	(newdt,) = result
100	n/a	self.assertEqual(newdt, dt)
101	n/a	self.assertIs(type(newdt), xmlrpclib.DateTime)
102	n/a	self.assertIsNone(m)
103	n/a
104	n/a	def test_bug_1164912 (self):
105	n/a	d = xmlrpclib.DateTime()
106	n/a	((new_d,), dummy) = xmlrpclib.loads(xmlrpclib.dumps((d,),
107	n/a	methodresponse=True))
108	n/a	self.assertIsInstance(new_d.value, str)
109	n/a
110	n/a	# Check that the output of dumps() is still an 8-bit string
111	n/a	s = xmlrpclib.dumps((new_d,), methodresponse=True)
112	n/a	self.assertIsInstance(s, str)
113	n/a
114	n/a	def test_newstyle_class(self):
115	n/a	class T(object):
116	n/a	pass
117	n/a	t = T()
118	n/a	t.x = 100
119	n/a	t.y = "Hello"
120	n/a	((t2,), dummy) = xmlrpclib.loads(xmlrpclib.dumps((t,)))
121	n/a	self.assertEqual(t2, t.__dict__)
122	n/a
123	n/a	def test_dump_big_long(self):
124	n/a	self.assertRaises(OverflowError, xmlrpclib.dumps, (2**99,))
125	n/a
126	n/a	def test_dump_bad_dict(self):
127	n/a	self.assertRaises(TypeError, xmlrpclib.dumps, ({(1,2,3): 1},))
128	n/a
129	n/a	def test_dump_recursive_seq(self):
130	n/a	l = [1,2,3]
131	n/a	t = [3,4,5,l]
132	n/a	l.append(t)
133	n/a	self.assertRaises(TypeError, xmlrpclib.dumps, (l,))
134	n/a
135	n/a	def test_dump_recursive_dict(self):
136	n/a	d = {'1':1, '2':1}
137	n/a	t = {'3':3, 'd':d}
138	n/a	d['t'] = t
139	n/a	self.assertRaises(TypeError, xmlrpclib.dumps, (d,))
140	n/a
141	n/a	def test_dump_big_int(self):
142	n/a	if sys.maxsize > 2**31-1:
143	n/a	self.assertRaises(OverflowError, xmlrpclib.dumps,
144	n/a	(int(2**34),))
145	n/a
146	n/a	xmlrpclib.dumps((xmlrpclib.MAXINT, xmlrpclib.MININT))
147	n/a	self.assertRaises(OverflowError, xmlrpclib.dumps,
148	n/a	(xmlrpclib.MAXINT+1,))
149	n/a	self.assertRaises(OverflowError, xmlrpclib.dumps,
150	n/a	(xmlrpclib.MININT-1,))
151	n/a
152	n/a	def dummy_write(s):
153	n/a	pass
154	n/a
155	n/a	m = xmlrpclib.Marshaller()
156	n/a	m.dump_int(xmlrpclib.MAXINT, dummy_write)
157	n/a	m.dump_int(xmlrpclib.MININT, dummy_write)
158	n/a	self.assertRaises(OverflowError, m.dump_int,
159	n/a	xmlrpclib.MAXINT+1, dummy_write)
160	n/a	self.assertRaises(OverflowError, m.dump_int,
161	n/a	xmlrpclib.MININT-1, dummy_write)
162	n/a
163	n/a	def test_dump_double(self):
164	n/a	xmlrpclib.dumps((float(2 ** 34),))
165	n/a	xmlrpclib.dumps((float(xmlrpclib.MAXINT),
166	n/a	float(xmlrpclib.MININT)))
167	n/a	xmlrpclib.dumps((float(xmlrpclib.MAXINT + 42),
168	n/a	float(xmlrpclib.MININT - 42)))
169	n/a
170	n/a	def dummy_write(s):
171	n/a	pass
172	n/a
173	n/a	m = xmlrpclib.Marshaller()
174	n/a	m.dump_double(xmlrpclib.MAXINT, dummy_write)
175	n/a	m.dump_double(xmlrpclib.MININT, dummy_write)
176	n/a	m.dump_double(xmlrpclib.MAXINT + 42, dummy_write)
177	n/a	m.dump_double(xmlrpclib.MININT - 42, dummy_write)
178	n/a
179	n/a	def test_dump_none(self):
180	n/a	value = alist + [None]
181	n/a	arg1 = (alist + [None],)
182	n/a	strg = xmlrpclib.dumps(arg1, allow_none=True)
183	n/a	self.assertEqual(value,
184	n/a	xmlrpclib.loads(strg)[0][0])
185	n/a	self.assertRaises(TypeError, xmlrpclib.dumps, (arg1,))
186	n/a
187	n/a	def test_dump_encoding(self):
188	n/a	value = {'key\u20ac\xa4':
189	n/a	'value\u20ac\xa4'}
190	n/a	strg = xmlrpclib.dumps((value,), encoding='iso-8859-15')
191	n/a	strg = "<?xml version='1.0' encoding='iso-8859-15'?>" + strg
192	n/a	self.assertEqual(xmlrpclib.loads(strg)[0][0], value)
193	n/a	strg = strg.encode('iso-8859-15', 'xmlcharrefreplace')
194	n/a	self.assertEqual(xmlrpclib.loads(strg)[0][0], value)
195	n/a
196	n/a	strg = xmlrpclib.dumps((value,), encoding='iso-8859-15',
197	n/a	methodresponse=True)
198	n/a	self.assertEqual(xmlrpclib.loads(strg)[0][0], value)
199	n/a	strg = strg.encode('iso-8859-15', 'xmlcharrefreplace')
200	n/a	self.assertEqual(xmlrpclib.loads(strg)[0][0], value)
201	n/a
202	n/a	methodname = 'method\u20ac\xa4'
203	n/a	strg = xmlrpclib.dumps((value,), encoding='iso-8859-15',
204	n/a	methodname=methodname)
205	n/a	self.assertEqual(xmlrpclib.loads(strg)[0][0], value)
206	n/a	self.assertEqual(xmlrpclib.loads(strg)[1], methodname)
207	n/a
208	n/a	def test_dump_bytes(self):
209	n/a	sample = b"my dog has fleas"
210	n/a	self.assertEqual(sample, xmlrpclib.Binary(sample))
211	n/a	for type_ in bytes, bytearray, xmlrpclib.Binary:
212	n/a	value = type_(sample)
213	n/a	s = xmlrpclib.dumps((value,))
214	n/a
215	n/a	result, m = xmlrpclib.loads(s, use_builtin_types=True)
216	n/a	(newvalue,) = result
217	n/a	self.assertEqual(newvalue, sample)
218	n/a	self.assertIs(type(newvalue), bytes)
219	n/a	self.assertIsNone(m)
220	n/a
221	n/a	result, m = xmlrpclib.loads(s, use_builtin_types=False)
222	n/a	(newvalue,) = result
223	n/a	self.assertEqual(newvalue, sample)
224	n/a	self.assertIs(type(newvalue), xmlrpclib.Binary)
225	n/a	self.assertIsNone(m)
226	n/a
227	n/a	def test_loads_unsupported(self):
228	n/a	ResponseError = xmlrpclib.ResponseError
229	n/a	data = '<params><param><value><spam/></value></param></params>'
230	n/a	self.assertRaises(ResponseError, xmlrpclib.loads, data)
231	n/a	data = ('<params><param><value><array>'
232	n/a	'<value><spam/></value>'
233	n/a	'</array></value></param></params>')
234	n/a	self.assertRaises(ResponseError, xmlrpclib.loads, data)
235	n/a	data = ('<params><param><value><struct>'
236	n/a	'<member><name>a</name><value><spam/></value></member>'
237	n/a	'<member><name>b</name><value><spam/></value></member>'
238	n/a	'</struct></value></param></params>')
239	n/a	self.assertRaises(ResponseError, xmlrpclib.loads, data)
240	n/a
241	n/a	def check_loads(self, s, value, **kwargs):
242	n/a	dump = '<params><param><value>%s</value></param></params>' % s
243	n/a	result, m = xmlrpclib.loads(dump, **kwargs)
244	n/a	(newvalue,) = result
245	n/a	self.assertEqual(newvalue, value)
246	n/a	self.assertIs(type(newvalue), type(value))
247	n/a	self.assertIsNone(m)
248	n/a
249	n/a	def test_load_standard_types(self):
250	n/a	check = self.check_loads
251	n/a	check('string', 'string')
252	n/a	check('<string>string</string>', 'string')
253	n/a	check('<string>ð”˜ð”«ð”¦ð” ð”¬ð”¡ð”¢ string</string>', 'ð”˜ð”«ð”¦ð” ð”¬ð”¡ð”¢ string')
254	n/a	check('<int>2056183947</int>', 2056183947)
255	n/a	check('<int>-2056183947</int>', -2056183947)
256	n/a	check('<i4>2056183947</i4>', 2056183947)
257	n/a	check('<double>46093.78125</double>', 46093.78125)
258	n/a	check('<boolean>0</boolean>', False)
259	n/a	check('<base64>AGJ5dGUgc3RyaW5n/w==</base64>',
260	n/a	xmlrpclib.Binary(b'\x00byte string\xff'))
261	n/a	check('<base64>AGJ5dGUgc3RyaW5n/w==</base64>',
262	n/a	b'\x00byte string\xff', use_builtin_types=True)
263	n/a	check('<dateTime.iso8601>20050210T11:41:23</dateTime.iso8601>',
264	n/a	xmlrpclib.DateTime('20050210T11:41:23'))
265	n/a	check('<dateTime.iso8601>20050210T11:41:23</dateTime.iso8601>',
266	n/a	datetime.datetime(2005, 2, 10, 11, 41, 23),
267	n/a	use_builtin_types=True)
268	n/a	check('<array><data>'
269	n/a	'<value><int>1</int></value><value><int>2</int></value>'
270	n/a	'</data></array>', [1, 2])
271	n/a	check('<struct>'
272	n/a	'<member><name>b</name><value><int>2</int></value></member>'
273	n/a	'<member><name>a</name><value><int>1</int></value></member>'
274	n/a	'</struct>', {'a': 1, 'b': 2})
275	n/a
276	n/a	def test_load_extension_types(self):
277	n/a	check = self.check_loads
278	n/a	check('<nil/>', None)
279	n/a	check('<ex:nil/>', None)
280	n/a	check('<i1>205</i1>', 205)
281	n/a	check('<i2>20561</i2>', 20561)
282	n/a	check('<i8>9876543210</i8>', 9876543210)
283	n/a	check('<biginteger>98765432100123456789</biginteger>',
284	n/a	98765432100123456789)
285	n/a	check('<float>93.78125</float>', 93.78125)
286	n/a	check('<bigdecimal>9876543210.0123456789</bigdecimal>',
287	n/a	decimal.Decimal('9876543210.0123456789'))
288	n/a
289	n/a	def test_get_host_info(self):
290	n/a	# see bug #3613, this raised a TypeError
291	n/a	transp = xmlrpc.client.Transport()
292	n/a	self.assertEqual(transp.get_host_info("user@host.tld"),
293	n/a	('host.tld',
294	n/a	[('Authorization', 'Basic dXNlcg==')], {}))
295	n/a
296	n/a	def test_ssl_presence(self):
297	n/a	try:
298	n/a	import ssl
299	n/a	except ImportError:
300	n/a	has_ssl = False
301	n/a	else:
302	n/a	has_ssl = True
303	n/a	try:
304	n/a	xmlrpc.client.ServerProxy('https://localhost:9999').bad_function()
305	n/a	except NotImplementedError:
306	n/a	self.assertFalse(has_ssl, "xmlrpc client's error with SSL support")
307	n/a	except OSError:
308	n/a	self.assertTrue(has_ssl)
309	n/a
310	n/a	@unittest.skipUnless(threading, "Threading required for this test.")
311	n/a	def test_keepalive_disconnect(self):
312	n/a	class RequestHandler(http.server.BaseHTTPRequestHandler):
313	n/a	protocol_version = "HTTP/1.1"
314	n/a	handled = False
315	n/a
316	n/a	def do_POST(self):
317	n/a	length = int(self.headers.get("Content-Length"))
318	n/a	self.rfile.read(length)
319	n/a	if self.handled:
320	n/a	self.close_connection = True
321	n/a	return
322	n/a	response = xmlrpclib.dumps((5,), methodresponse=True)
323	n/a	response = response.encode()
324	n/a	self.send_response(http.HTTPStatus.OK)
325	n/a	self.send_header("Content-Length", len(response))
326	n/a	self.end_headers()
327	n/a	self.wfile.write(response)
328	n/a	self.handled = True
329	n/a	self.close_connection = False
330	n/a
331	n/a	def run_server():
332	n/a	server.socket.settimeout(float(1)) # Don't hang if client fails
333	n/a	server.handle_request() # First request and attempt at second
334	n/a	server.handle_request() # Retried second request
335	n/a
336	n/a	server = http.server.HTTPServer((support.HOST, 0), RequestHandler)
337	n/a	self.addCleanup(server.server_close)
338	n/a	thread = threading.Thread(target=run_server)
339	n/a	thread.start()
340	n/a	self.addCleanup(thread.join)
341	n/a	url = "http://{}:{}/".format(*server.server_address)
342	n/a	with xmlrpclib.ServerProxy(url) as p:
343	n/a	self.assertEqual(p.method(), 5)
344	n/a	self.assertEqual(p.method(), 5)
345	n/a
346	n/a	class HelperTestCase(unittest.TestCase):
347	n/a	def test_escape(self):
348	n/a	self.assertEqual(xmlrpclib.escape("a&b"), "a&b")
349	n/a	self.assertEqual(xmlrpclib.escape("a<b"), "a<b")
350	n/a	self.assertEqual(xmlrpclib.escape("a>b"), "a>b")
351	n/a
352	n/a	class FaultTestCase(unittest.TestCase):
353	n/a	def test_repr(self):
354	n/a	f = xmlrpclib.Fault(42, 'Test Fault')
355	n/a	self.assertEqual(repr(f), "<Fault 42: 'Test Fault'>")
356	n/a	self.assertEqual(repr(f), str(f))
357	n/a
358	n/a	def test_dump_fault(self):
359	n/a	f = xmlrpclib.Fault(42, 'Test Fault')
360	n/a	s = xmlrpclib.dumps((f,))
361	n/a	(newf,), m = xmlrpclib.loads(s)
362	n/a	self.assertEqual(newf, {'faultCode': 42, 'faultString': 'Test Fault'})
363	n/a	self.assertEqual(m, None)
364	n/a
365	n/a	s = xmlrpclib.Marshaller().dumps(f)
366	n/a	self.assertRaises(xmlrpclib.Fault, xmlrpclib.loads, s)
367	n/a
368	n/a	def test_dotted_attribute(self):
369	n/a	# this will raise AttributeError because code don't want us to use
370	n/a	# private methods
371	n/a	self.assertRaises(AttributeError,
372	n/a	xmlrpc.server.resolve_dotted_attribute, str, '__add')
373	n/a	self.assertTrue(xmlrpc.server.resolve_dotted_attribute(str, 'title'))
374	n/a
375	n/a	class DateTimeTestCase(unittest.TestCase):
376	n/a	def test_default(self):
377	n/a	with mock.patch('time.localtime') as localtime_mock:
378	n/a	time_struct = time.struct_time(
379	n/a	[2013, 7, 15, 0, 24, 49, 0, 196, 0])
380	n/a	localtime_mock.return_value = time_struct
381	n/a	localtime = time.localtime()
382	n/a	t = xmlrpclib.DateTime()
383	n/a	self.assertEqual(str(t),
384	n/a	time.strftime("%Y%m%dT%H:%M:%S", localtime))
385	n/a
386	n/a	def test_time(self):
387	n/a	d = 1181399930.036952
388	n/a	t = xmlrpclib.DateTime(d)
389	n/a	self.assertEqual(str(t),
390	n/a	time.strftime("%Y%m%dT%H:%M:%S", time.localtime(d)))
391	n/a
392	n/a	def test_time_tuple(self):
393	n/a	d = (2007,6,9,10,38,50,5,160,0)
394	n/a	t = xmlrpclib.DateTime(d)
395	n/a	self.assertEqual(str(t), '20070609T10:38:50')
396	n/a
397	n/a	def test_time_struct(self):
398	n/a	d = time.localtime(1181399930.036952)
399	n/a	t = xmlrpclib.DateTime(d)
400	n/a	self.assertEqual(str(t), time.strftime("%Y%m%dT%H:%M:%S", d))
401	n/a
402	n/a	def test_datetime_datetime(self):
403	n/a	d = datetime.datetime(2007,1,2,3,4,5)
404	n/a	t = xmlrpclib.DateTime(d)
405	n/a	self.assertEqual(str(t), '20070102T03:04:05')
406	n/a
407	n/a	def test_repr(self):
408	n/a	d = datetime.datetime(2007,1,2,3,4,5)
409	n/a	t = xmlrpclib.DateTime(d)
410	n/a	val ="<DateTime '20070102T03:04:05' at %#x>" % id(t)
411	n/a	self.assertEqual(repr(t), val)
412	n/a
413	n/a	def test_decode(self):
414	n/a	d = ' 20070908T07:11:13 '
415	n/a	t1 = xmlrpclib.DateTime()
416	n/a	t1.decode(d)
417	n/a	tref = xmlrpclib.DateTime(datetime.datetime(2007,9,8,7,11,13))
418	n/a	self.assertEqual(t1, tref)
419	n/a
420	n/a	t2 = xmlrpclib._datetime(d)
421	n/a	self.assertEqual(t2, tref)
422	n/a
423	n/a	def test_comparison(self):
424	n/a	now = datetime.datetime.now()
425	n/a	dtime = xmlrpclib.DateTime(now.timetuple())
426	n/a
427	n/a	# datetime vs. DateTime
428	n/a	self.assertTrue(dtime == now)
429	n/a	self.assertTrue(now == dtime)
430	n/a	then = now + datetime.timedelta(seconds=4)
431	n/a	self.assertTrue(then >= dtime)
432	n/a	self.assertTrue(dtime < then)
433	n/a
434	n/a	# str vs. DateTime
435	n/a	dstr = now.strftime("%Y%m%dT%H:%M:%S")
436	n/a	self.assertTrue(dtime == dstr)
437	n/a	self.assertTrue(dstr == dtime)
438	n/a	dtime_then = xmlrpclib.DateTime(then.timetuple())
439	n/a	self.assertTrue(dtime_then >= dstr)
440	n/a	self.assertTrue(dstr < dtime_then)
441	n/a
442	n/a	# some other types
443	n/a	dbytes = dstr.encode('ascii')
444	n/a	dtuple = now.timetuple()
445	n/a	with self.assertRaises(TypeError):
446	n/a	dtime == 1970
447	n/a	with self.assertRaises(TypeError):
448	n/a	dtime != dbytes
449	n/a	with self.assertRaises(TypeError):
450	n/a	dtime == bytearray(dbytes)
451	n/a	with self.assertRaises(TypeError):
452	n/a	dtime != dtuple
453	n/a	with self.assertRaises(TypeError):
454	n/a	dtime < float(1970)
455	n/a	with self.assertRaises(TypeError):
456	n/a	dtime > dbytes
457	n/a	with self.assertRaises(TypeError):
458	n/a	dtime <= bytearray(dbytes)
459	n/a	with self.assertRaises(TypeError):
460	n/a	dtime >= dtuple
461	n/a
462	n/a	class BinaryTestCase(unittest.TestCase):
463	n/a
464	n/a	# XXX What should str(Binary(b"\xff")) return? I'm chosing "\xff"
465	n/a	# for now (i.e. interpreting the binary data as Latin-1-encoded
466	n/a	# text). But this feels very unsatisfactory. Perhaps we should
467	n/a	# only define repr(), and return r"Binary(b'\xff')" instead?
468	n/a
469	n/a	def test_default(self):
470	n/a	t = xmlrpclib.Binary()
471	n/a	self.assertEqual(str(t), '')
472	n/a
473	n/a	def test_string(self):
474	n/a	d = b'\x01\x02\x03abc123\xff\xfe'
475	n/a	t = xmlrpclib.Binary(d)
476	n/a	self.assertEqual(str(t), str(d, "latin-1"))
477	n/a
478	n/a	def test_decode(self):
479	n/a	d = b'\x01\x02\x03abc123\xff\xfe'
480	n/a	de = base64.encodebytes(d)
481	n/a	t1 = xmlrpclib.Binary()
482	n/a	t1.decode(de)
483	n/a	self.assertEqual(str(t1), str(d, "latin-1"))
484	n/a
485	n/a	t2 = xmlrpclib._binary(de)
486	n/a	self.assertEqual(str(t2), str(d, "latin-1"))
487	n/a
488	n/a
489	n/a	ADDR = PORT = URL = None
490	n/a
491	n/a	# The evt is set twice. First when the server is ready to serve.
492	n/a	# Second when the server has been shutdown. The user must clear
493	n/a	# the event after it has been set the first time to catch the second set.
494	n/a	def http_server(evt, numrequests, requestHandler=None, encoding=None):
495	n/a	class TestInstanceClass:
496	n/a	def div(self, x, y):
497	n/a	return x // y
498	n/a
499	n/a	def _methodHelp(self, name):
500	n/a	if name == 'div':
501	n/a	return 'This is the div function'
502	n/a
503	n/a	class Fixture:
504	n/a	@staticmethod
505	n/a	def getData():
506	n/a	return '42'
507	n/a
508	n/a	def my_function():
509	n/a	'''This is my function'''
510	n/a	return True
511	n/a
512	n/a	class MyXMLRPCServer(xmlrpc.server.SimpleXMLRPCServer):
513	n/a	def get_request(self):
514	n/a	# Ensure the socket is always non-blocking. On Linux, socket
515	n/a	# attributes are not inherited like they are on *BSD and Windows.
516	n/a	s, port = self.socket.accept()
517	n/a	s.setblocking(True)
518	n/a	return s, port
519	n/a
520	n/a	if not requestHandler:
521	n/a	requestHandler = xmlrpc.server.SimpleXMLRPCRequestHandler
522	n/a	serv = MyXMLRPCServer(("localhost", 0), requestHandler,
523	n/a	encoding=encoding,
524	n/a	logRequests=False, bind_and_activate=False)
525	n/a	try:
526	n/a	serv.server_bind()
527	n/a	global ADDR, PORT, URL
528	n/a	ADDR, PORT = serv.socket.getsockname()
529	n/a	#connect to IP address directly. This avoids socket.create_connection()
530	n/a	#trying to connect to "localhost" using all address families, which
531	n/a	#causes slowdown e.g. on vista which supports AF_INET6. The server listens
532	n/a	#on AF_INET only.
533	n/a	URL = "http://%s:%d"%(ADDR, PORT)
534	n/a	serv.server_activate()
535	n/a	serv.register_introspection_functions()
536	n/a	serv.register_multicall_functions()
537	n/a	serv.register_function(pow)
538	n/a	serv.register_function(lambda x,y: x+y, 'add')
539	n/a	serv.register_function(lambda x: x, 'tÃªÅ¡t')
540	n/a	serv.register_function(my_function)
541	n/a	testInstance = TestInstanceClass()
542	n/a	serv.register_instance(testInstance, allow_dotted_names=True)
543	n/a	evt.set()
544	n/a
545	n/a	# handle up to 'numrequests' requests
546	n/a	while numrequests > 0:
547	n/a	serv.handle_request()
548	n/a	numrequests -= 1
549	n/a
550	n/a	except socket.timeout:
551	n/a	pass
552	n/a	finally:
553	n/a	serv.socket.close()
554	n/a	PORT = None
555	n/a	evt.set()
556	n/a
557	n/a	def http_multi_server(evt, numrequests, requestHandler=None):
558	n/a	class TestInstanceClass:
559	n/a	def div(self, x, y):
560	n/a	return x // y
561	n/a
562	n/a	def _methodHelp(self, name):
563	n/a	if name == 'div':
564	n/a	return 'This is the div function'
565	n/a
566	n/a	def my_function():
567	n/a	'''This is my function'''
568	n/a	return True
569	n/a
570	n/a	class MyXMLRPCServer(xmlrpc.server.MultiPathXMLRPCServer):
571	n/a	def get_request(self):
572	n/a	# Ensure the socket is always non-blocking. On Linux, socket
573	n/a	# attributes are not inherited like they are on *BSD and Windows.
574	n/a	s, port = self.socket.accept()
575	n/a	s.setblocking(True)
576	n/a	return s, port
577	n/a
578	n/a	if not requestHandler:
579	n/a	requestHandler = xmlrpc.server.SimpleXMLRPCRequestHandler
580	n/a	class MyRequestHandler(requestHandler):
581	n/a	rpc_paths = []
582	n/a
583	n/a	class BrokenDispatcher:
584	n/a	def _marshaled_dispatch(self, data, dispatch_method=None, path=None):
585	n/a	raise RuntimeError("broken dispatcher")
586	n/a
587	n/a	serv = MyXMLRPCServer(("localhost", 0), MyRequestHandler,
588	n/a	logRequests=False, bind_and_activate=False)
589	n/a	serv.socket.settimeout(3)
590	n/a	serv.server_bind()
591	n/a	try:
592	n/a	global ADDR, PORT, URL
593	n/a	ADDR, PORT = serv.socket.getsockname()
594	n/a	#connect to IP address directly. This avoids socket.create_connection()
595	n/a	#trying to connect to "localhost" using all address families, which
596	n/a	#causes slowdown e.g. on vista which supports AF_INET6. The server listens
597	n/a	#on AF_INET only.
598	n/a	URL = "http://%s:%d"%(ADDR, PORT)
599	n/a	serv.server_activate()
600	n/a	paths = ["/foo", "/foo/bar"]
601	n/a	for path in paths:
602	n/a	d = serv.add_dispatcher(path, xmlrpc.server.SimpleXMLRPCDispatcher())
603	n/a	d.register_introspection_functions()
604	n/a	d.register_multicall_functions()
605	n/a	serv.get_dispatcher(paths[0]).register_function(pow)
606	n/a	serv.get_dispatcher(paths[1]).register_function(lambda x,y: x+y, 'add')
607	n/a	serv.add_dispatcher("/is/broken", BrokenDispatcher())
608	n/a	evt.set()
609	n/a
610	n/a	# handle up to 'numrequests' requests
611	n/a	while numrequests > 0:
612	n/a	serv.handle_request()
613	n/a	numrequests -= 1
614	n/a
615	n/a	except socket.timeout:
616	n/a	pass
617	n/a	finally:
618	n/a	serv.socket.close()
619	n/a	PORT = None
620	n/a	evt.set()
621	n/a
622	n/a	# This function prevents errors like:
623	n/a	# <ProtocolError for localhost:57527/RPC2: 500 Internal Server Error>
624	n/a	def is_unavailable_exception(e):
625	n/a	'''Returns True if the given ProtocolError is the product of a server-side
626	n/a	exception caused by the 'temporarily unavailable' response sometimes
627	n/a	given by operations on non-blocking sockets.'''
628	n/a
629	n/a	# sometimes we get a -1 error code and/or empty headers
630	n/a	try:
631	n/a	if e.errcode == -1 or e.headers is None:
632	n/a	return True
633	n/a	exc_mess = e.headers.get('X-exception')
634	n/a	except AttributeError:
635	n/a	# Ignore OSErrors here.
636	n/a	exc_mess = str(e)
637	n/a
638	n/a	if exc_mess and 'temporarily unavailable' in exc_mess.lower():
639	n/a	return True
640	n/a
641	n/a	def make_request_and_skipIf(condition, reason):
642	n/a	# If we skip the test, we have to make a request because
643	n/a	# the server created in setUp blocks expecting one to come in.
644	n/a	if not condition:
645	n/a	return lambda func: func
646	n/a	def decorator(func):
647	n/a	def make_request_and_skip(self):
648	n/a	try:
649	n/a	xmlrpclib.ServerProxy(URL).my_function()
650	n/a	except (xmlrpclib.ProtocolError, OSError) as e:
651	n/a	if not is_unavailable_exception(e):
652	n/a	raise
653	n/a	raise unittest.SkipTest(reason)
654	n/a	return make_request_and_skip
655	n/a	return decorator
656	n/a
657	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
658	n/a	class BaseServerTestCase(unittest.TestCase):
659	n/a	requestHandler = None
660	n/a	request_count = 1
661	n/a	threadFunc = staticmethod(http_server)
662	n/a
663	n/a	def setUp(self):
664	n/a	# enable traceback reporting
665	n/a	xmlrpc.server.SimpleXMLRPCServer._send_traceback_header = True
666	n/a
667	n/a	self.evt = threading.Event()
668	n/a	# start server thread to handle requests
669	n/a	serv_args = (self.evt, self.request_count, self.requestHandler)
670	n/a	threading.Thread(target=self.threadFunc, args=serv_args).start()
671	n/a
672	n/a	# wait for the server to be ready
673	n/a	self.evt.wait()
674	n/a	self.evt.clear()
675	n/a
676	n/a	def tearDown(self):
677	n/a	# wait on the server thread to terminate
678	n/a	self.evt.wait()
679	n/a
680	n/a	# disable traceback reporting
681	n/a	xmlrpc.server.SimpleXMLRPCServer._send_traceback_header = False
682	n/a
683	n/a	class SimpleServerTestCase(BaseServerTestCase):
684	n/a	def test_simple1(self):
685	n/a	try:
686	n/a	p = xmlrpclib.ServerProxy(URL)
687	n/a	self.assertEqual(p.pow(6,8), 6**8)
688	n/a	except (xmlrpclib.ProtocolError, OSError) as e:
689	n/a	# ignore failures due to non-blocking socket 'unavailable' errors
690	n/a	if not is_unavailable_exception(e):
691	n/a	# protocol error; provide additional information in test output
692	n/a	self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
693	n/a
694	n/a	def test_nonascii(self):
695	n/a	start_string = 'P\N{LATIN SMALL LETTER Y WITH CIRCUMFLEX}t'
696	n/a	end_string = 'h\N{LATIN SMALL LETTER O WITH HORN}n'
697	n/a	try:
698	n/a	p = xmlrpclib.ServerProxy(URL)
699	n/a	self.assertEqual(p.add(start_string, end_string),
700	n/a	start_string + end_string)
701	n/a	except (xmlrpclib.ProtocolError, OSError) as e:
702	n/a	# ignore failures due to non-blocking socket 'unavailable' errors
703	n/a	if not is_unavailable_exception(e):
704	n/a	# protocol error; provide additional information in test output
705	n/a	self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
706	n/a
707	n/a	def test_client_encoding(self):
708	n/a	start_string = '\u20ac'
709	n/a	end_string = '\xa4'
710	n/a
711	n/a	try:
712	n/a	p = xmlrpclib.ServerProxy(URL, encoding='iso-8859-15')
713	n/a	self.assertEqual(p.add(start_string, end_string),
714	n/a	start_string + end_string)
715	n/a	except (xmlrpclib.ProtocolError, socket.error) as e:
716	n/a	# ignore failures due to non-blocking socket unavailable errors.
717	n/a	if not is_unavailable_exception(e):
718	n/a	# protocol error; provide additional information in test output
719	n/a	self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
720	n/a
721	n/a	def test_nonascii_methodname(self):
722	n/a	try:
723	n/a	p = xmlrpclib.ServerProxy(URL, encoding='ascii')
724	n/a	self.assertEqual(p.tÃªÅ¡t(42), 42)
725	n/a	except (xmlrpclib.ProtocolError, socket.error) as e:
726	n/a	# ignore failures due to non-blocking socket unavailable errors.
727	n/a	if not is_unavailable_exception(e):
728	n/a	# protocol error; provide additional information in test output
729	n/a	self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
730	n/a
731	n/a	# [ch] The test 404 is causing lots of false alarms.
732	n/a	def XXXtest_404(self):
733	n/a	# send POST with http.client, it should return 404 header and
734	n/a	# 'Not Found' message.
735	n/a	conn = httplib.client.HTTPConnection(ADDR, PORT)
736	n/a	conn.request('POST', '/this-is-not-valid')
737	n/a	response = conn.getresponse()
738	n/a	conn.close()
739	n/a
740	n/a	self.assertEqual(response.status, 404)
741	n/a	self.assertEqual(response.reason, 'Not Found')
742	n/a
743	n/a	def test_introspection1(self):
744	n/a	expected_methods = set(['pow', 'div', 'my_function', 'add', 'tÃªÅ¡t',
745	n/a	'system.listMethods', 'system.methodHelp',
746	n/a	'system.methodSignature', 'system.multicall',
747	n/a	'Fixture'])
748	n/a	try:
749	n/a	p = xmlrpclib.ServerProxy(URL)
750	n/a	meth = p.system.listMethods()
751	n/a	self.assertEqual(set(meth), expected_methods)
752	n/a	except (xmlrpclib.ProtocolError, OSError) as e:
753	n/a	# ignore failures due to non-blocking socket 'unavailable' errors
754	n/a	if not is_unavailable_exception(e):
755	n/a	# protocol error; provide additional information in test output
756	n/a	self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
757	n/a
758	n/a
759	n/a	def test_introspection2(self):
760	n/a	try:
761	n/a	# test _methodHelp()
762	n/a	p = xmlrpclib.ServerProxy(URL)
763	n/a	divhelp = p.system.methodHelp('div')
764	n/a	self.assertEqual(divhelp, 'This is the div function')
765	n/a	except (xmlrpclib.ProtocolError, OSError) as e:
766	n/a	# ignore failures due to non-blocking socket 'unavailable' errors
767	n/a	if not is_unavailable_exception(e):
768	n/a	# protocol error; provide additional information in test output
769	n/a	self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
770	n/a
771	n/a	@make_request_and_skipIf(sys.flags.optimize >= 2,
772	n/a	"Docstrings are omitted with -O2 and above")
773	n/a	def test_introspection3(self):
774	n/a	try:
775	n/a	# test native doc
776	n/a	p = xmlrpclib.ServerProxy(URL)
777	n/a	myfunction = p.system.methodHelp('my_function')
778	n/a	self.assertEqual(myfunction, 'This is my function')
779	n/a	except (xmlrpclib.ProtocolError, OSError) as e:
780	n/a	# ignore failures due to non-blocking socket 'unavailable' errors
781	n/a	if not is_unavailable_exception(e):
782	n/a	# protocol error; provide additional information in test output
783	n/a	self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
784	n/a
785	n/a	def test_introspection4(self):
786	n/a	# the SimpleXMLRPCServer doesn't support signatures, but
787	n/a	# at least check that we can try making the call
788	n/a	try:
789	n/a	p = xmlrpclib.ServerProxy(URL)
790	n/a	divsig = p.system.methodSignature('div')
791	n/a	self.assertEqual(divsig, 'signatures not supported')
792	n/a	except (xmlrpclib.ProtocolError, OSError) as e:
793	n/a	# ignore failures due to non-blocking socket 'unavailable' errors
794	n/a	if not is_unavailable_exception(e):
795	n/a	# protocol error; provide additional information in test output
796	n/a	self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
797	n/a
798	n/a	def test_multicall(self):
799	n/a	try:
800	n/a	p = xmlrpclib.ServerProxy(URL)
801	n/a	multicall = xmlrpclib.MultiCall(p)
802	n/a	multicall.add(2,3)
803	n/a	multicall.pow(6,8)
804	n/a	multicall.div(127,42)
805	n/a	add_result, pow_result, div_result = multicall()
806	n/a	self.assertEqual(add_result, 2+3)
807	n/a	self.assertEqual(pow_result, 6**8)
808	n/a	self.assertEqual(div_result, 127//42)
809	n/a	except (xmlrpclib.ProtocolError, OSError) as e:
810	n/a	# ignore failures due to non-blocking socket 'unavailable' errors
811	n/a	if not is_unavailable_exception(e):
812	n/a	# protocol error; provide additional information in test output
813	n/a	self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
814	n/a
815	n/a	def test_non_existing_multicall(self):
816	n/a	try:
817	n/a	p = xmlrpclib.ServerProxy(URL)
818	n/a	multicall = xmlrpclib.MultiCall(p)
819	n/a	multicall.this_is_not_exists()
820	n/a	result = multicall()
821	n/a
822	n/a	# result.results contains;
823	n/a	# [{'faultCode': 1, 'faultString': '<class \'exceptions.Exception\'>:'
824	n/a	# 'method "this_is_not_exists" is not supported'>}]
825	n/a
826	n/a	self.assertEqual(result.results[0]['faultCode'], 1)
827	n/a	self.assertEqual(result.results[0]['faultString'],
828	n/a	'<class \'Exception\'>:method "this_is_not_exists" '
829	n/a	'is not supported')
830	n/a	except (xmlrpclib.ProtocolError, OSError) as e:
831	n/a	# ignore failures due to non-blocking socket 'unavailable' errors
832	n/a	if not is_unavailable_exception(e):
833	n/a	# protocol error; provide additional information in test output
834	n/a	self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
835	n/a
836	n/a	def test_dotted_attribute(self):
837	n/a	# Raises an AttributeError because private methods are not allowed.
838	n/a	self.assertRaises(AttributeError,
839	n/a	xmlrpc.server.resolve_dotted_attribute, str, '__add')
840	n/a
841	n/a	self.assertTrue(xmlrpc.server.resolve_dotted_attribute(str, 'title'))
842	n/a	# Get the test to run faster by sending a request with test_simple1.
843	n/a	# This avoids waiting for the socket timeout.
844	n/a	self.test_simple1()
845	n/a
846	n/a	def test_allow_dotted_names_true(self):
847	n/a	# XXX also need allow_dotted_names_false test.
848	n/a	server = xmlrpclib.ServerProxy("http://%s:%d/RPC2" % (ADDR, PORT))
849	n/a	data = server.Fixture.getData()
850	n/a	self.assertEqual(data, '42')
851	n/a
852	n/a	def test_unicode_host(self):
853	n/a	server = xmlrpclib.ServerProxy("http://%s:%d/RPC2" % (ADDR, PORT))
854	n/a	self.assertEqual(server.add("a", "\xe9"), "a\xe9")
855	n/a
856	n/a	def test_partial_post(self):
857	n/a	# Check that a partial POST doesn't make the server loop: issue #14001.
858	n/a	conn = http.client.HTTPConnection(ADDR, PORT)
859	n/a	conn.request('POST', '/RPC2 HTTP/1.0\r\nContent-Length: 100\r\n\r\nbye')
860	n/a	conn.close()
861	n/a
862	n/a	def test_context_manager(self):
863	n/a	with xmlrpclib.ServerProxy(URL) as server:
864	n/a	server.add(2, 3)
865	n/a	self.assertNotEqual(server('transport')._connection,
866	n/a	(None, None))
867	n/a	self.assertEqual(server('transport')._connection,
868	n/a	(None, None))
869	n/a
870	n/a	def test_context_manager_method_error(self):
871	n/a	try:
872	n/a	with xmlrpclib.ServerProxy(URL) as server:
873	n/a	server.add(2, "a")
874	n/a	except xmlrpclib.Fault:
875	n/a	pass
876	n/a	self.assertEqual(server('transport')._connection,
877	n/a	(None, None))
878	n/a
879	n/a
880	n/a	class SimpleServerEncodingTestCase(BaseServerTestCase):
881	n/a	@staticmethod
882	n/a	def threadFunc(evt, numrequests, requestHandler=None, encoding=None):
883	n/a	http_server(evt, numrequests, requestHandler, 'iso-8859-15')
884	n/a
885	n/a	def test_server_encoding(self):
886	n/a	start_string = '\u20ac'
887	n/a	end_string = '\xa4'
888	n/a
889	n/a	try:
890	n/a	p = xmlrpclib.ServerProxy(URL)
891	n/a	self.assertEqual(p.add(start_string, end_string),
892	n/a	start_string + end_string)
893	n/a	except (xmlrpclib.ProtocolError, socket.error) as e:
894	n/a	# ignore failures due to non-blocking socket unavailable errors.
895	n/a	if not is_unavailable_exception(e):
896	n/a	# protocol error; provide additional information in test output
897	n/a	self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
898	n/a
899	n/a
900	n/a	class MultiPathServerTestCase(BaseServerTestCase):
901	n/a	threadFunc = staticmethod(http_multi_server)
902	n/a	request_count = 2
903	n/a	def test_path1(self):
904	n/a	p = xmlrpclib.ServerProxy(URL+"/foo")
905	n/a	self.assertEqual(p.pow(6,8), 6**8)
906	n/a	self.assertRaises(xmlrpclib.Fault, p.add, 6, 8)
907	n/a
908	n/a	def test_path2(self):
909	n/a	p = xmlrpclib.ServerProxy(URL+"/foo/bar")
910	n/a	self.assertEqual(p.add(6,8), 6+8)
911	n/a	self.assertRaises(xmlrpclib.Fault, p.pow, 6, 8)
912	n/a
913	n/a	def test_path3(self):
914	n/a	p = xmlrpclib.ServerProxy(URL+"/is/broken")
915	n/a	self.assertRaises(xmlrpclib.Fault, p.add, 6, 8)
916	n/a
917	n/a	#A test case that verifies that a server using the HTTP/1.1 keep-alive mechanism
918	n/a	#does indeed serve subsequent requests on the same connection
919	n/a	class BaseKeepaliveServerTestCase(BaseServerTestCase):
920	n/a	#a request handler that supports keep-alive and logs requests into a
921	n/a	#class variable
922	n/a	class RequestHandler(xmlrpc.server.SimpleXMLRPCRequestHandler):
923	n/a	parentClass = xmlrpc.server.SimpleXMLRPCRequestHandler
924	n/a	protocol_version = 'HTTP/1.1'
925	n/a	myRequests = []
926	n/a	def handle(self):
927	n/a	self.myRequests.append([])
928	n/a	self.reqidx = len(self.myRequests)-1
929	n/a	return self.parentClass.handle(self)
930	n/a	def handle_one_request(self):
931	n/a	result = self.parentClass.handle_one_request(self)
932	n/a	self.myRequests[self.reqidx].append(self.raw_requestline)
933	n/a	return result
934	n/a
935	n/a	requestHandler = RequestHandler
936	n/a	def setUp(self):
937	n/a	#clear request log
938	n/a	self.RequestHandler.myRequests = []
939	n/a	return BaseServerTestCase.setUp(self)
940	n/a
941	n/a	#A test case that verifies that a server using the HTTP/1.1 keep-alive mechanism
942	n/a	#does indeed serve subsequent requests on the same connection
943	n/a	class KeepaliveServerTestCase1(BaseKeepaliveServerTestCase):
944	n/a	def test_two(self):
945	n/a	p = xmlrpclib.ServerProxy(URL)
946	n/a	#do three requests.
947	n/a	self.assertEqual(p.pow(6,8), 6**8)
948	n/a	self.assertEqual(p.pow(6,8), 6**8)
949	n/a	self.assertEqual(p.pow(6,8), 6**8)
950	n/a	p("close")()
951	n/a
952	n/a	#they should have all been handled by a single request handler
953	n/a	self.assertEqual(len(self.RequestHandler.myRequests), 1)
954	n/a
955	n/a	#check that we did at least two (the third may be pending append
956	n/a	#due to thread scheduling)
957	n/a	self.assertGreaterEqual(len(self.RequestHandler.myRequests[-1]), 2)
958	n/a
959	n/a
960	n/a	#test special attribute access on the serverproxy, through the __call__
961	n/a	#function.
962	n/a	class KeepaliveServerTestCase2(BaseKeepaliveServerTestCase):
963	n/a	#ask for two keepalive requests to be handled.
964	n/a	request_count=2
965	n/a
966	n/a	def test_close(self):
967	n/a	p = xmlrpclib.ServerProxy(URL)
968	n/a	#do some requests with close.
969	n/a	self.assertEqual(p.pow(6,8), 6**8)
970	n/a	self.assertEqual(p.pow(6,8), 6**8)
971	n/a	self.assertEqual(p.pow(6,8), 6**8)
972	n/a	p("close")() #this should trigger a new keep-alive request
973	n/a	self.assertEqual(p.pow(6,8), 6**8)
974	n/a	self.assertEqual(p.pow(6,8), 6**8)
975	n/a	self.assertEqual(p.pow(6,8), 6**8)
976	n/a	p("close")()
977	n/a
978	n/a	#they should have all been two request handlers, each having logged at least
979	n/a	#two complete requests
980	n/a	self.assertEqual(len(self.RequestHandler.myRequests), 2)
981	n/a	self.assertGreaterEqual(len(self.RequestHandler.myRequests[-1]), 2)
982	n/a	self.assertGreaterEqual(len(self.RequestHandler.myRequests[-2]), 2)
983	n/a
984	n/a
985	n/a	def test_transport(self):
986	n/a	p = xmlrpclib.ServerProxy(URL)
987	n/a	#do some requests with close.
988	n/a	self.assertEqual(p.pow(6,8), 6**8)
989	n/a	p("transport").close() #same as above, really.
990	n/a	self.assertEqual(p.pow(6,8), 6**8)
991	n/a	p("close")()
992	n/a	self.assertEqual(len(self.RequestHandler.myRequests), 2)
993	n/a
994	n/a	#A test case that verifies that gzip encoding works in both directions
995	n/a	#(for a request and the response)
996	n/a	@unittest.skipIf(gzip is None, 'requires gzip')
997	n/a	class GzipServerTestCase(BaseServerTestCase):
998	n/a	#a request handler that supports keep-alive and logs requests into a
999	n/a	#class variable
1000	n/a	class RequestHandler(xmlrpc.server.SimpleXMLRPCRequestHandler):
1001	n/a	parentClass = xmlrpc.server.SimpleXMLRPCRequestHandler
1002	n/a	protocol_version = 'HTTP/1.1'
1003	n/a
1004	n/a	def do_POST(self):
1005	n/a	#store content of last request in class
1006	n/a	self.__class__.content_length = int(self.headers["content-length"])
1007	n/a	return self.parentClass.do_POST(self)
1008	n/a	requestHandler = RequestHandler
1009	n/a
1010	n/a	class Transport(xmlrpclib.Transport):
1011	n/a	#custom transport, stores the response length for our perusal
1012	n/a	fake_gzip = False
1013	n/a	def parse_response(self, response):
1014	n/a	self.response_length=int(response.getheader("content-length", 0))
1015	n/a	return xmlrpclib.Transport.parse_response(self, response)
1016	n/a
1017	n/a	def send_content(self, connection, body):
1018	n/a	if self.fake_gzip:
1019	n/a	#add a lone gzip header to induce decode error remotely
1020	n/a	connection.putheader("Content-Encoding", "gzip")
1021	n/a	return xmlrpclib.Transport.send_content(self, connection, body)
1022	n/a
1023	n/a	def setUp(self):
1024	n/a	BaseServerTestCase.setUp(self)
1025	n/a
1026	n/a	def test_gzip_request(self):
1027	n/a	t = self.Transport()
1028	n/a	t.encode_threshold = None
1029	n/a	p = xmlrpclib.ServerProxy(URL, transport=t)
1030	n/a	self.assertEqual(p.pow(6,8), 6**8)
1031	n/a	a = self.RequestHandler.content_length
1032	n/a	t.encode_threshold = 0 #turn on request encoding
1033	n/a	self.assertEqual(p.pow(6,8), 6**8)
1034	n/a	b = self.RequestHandler.content_length
1035	n/a	self.assertTrue(a>b)
1036	n/a	p("close")()
1037	n/a
1038	n/a	def test_bad_gzip_request(self):
1039	n/a	t = self.Transport()
1040	n/a	t.encode_threshold = None
1041	n/a	t.fake_gzip = True
1042	n/a	p = xmlrpclib.ServerProxy(URL, transport=t)
1043	n/a	cm = self.assertRaisesRegex(xmlrpclib.ProtocolError,
1044	n/a	re.compile(r"\b400\b"))
1045	n/a	with cm:
1046	n/a	p.pow(6, 8)
1047	n/a	p("close")()
1048	n/a
1049	n/a	def test_gzip_response(self):
1050	n/a	t = self.Transport()
1051	n/a	p = xmlrpclib.ServerProxy(URL, transport=t)
1052	n/a	old = self.requestHandler.encode_threshold
1053	n/a	self.requestHandler.encode_threshold = None #no encoding
1054	n/a	self.assertEqual(p.pow(6,8), 6**8)
1055	n/a	a = t.response_length
1056	n/a	self.requestHandler.encode_threshold = 0 #always encode
1057	n/a	self.assertEqual(p.pow(6,8), 6**8)
1058	n/a	p("close")()
1059	n/a	b = t.response_length
1060	n/a	self.requestHandler.encode_threshold = old
1061	n/a	self.assertTrue(a>b)
1062	n/a
1063	n/a
1064	n/a	@unittest.skipIf(gzip is None, 'requires gzip')
1065	n/a	class GzipUtilTestCase(unittest.TestCase):
1066	n/a
1067	n/a	def test_gzip_decode_limit(self):
1068	n/a	max_gzip_decode = 20 * 1024 * 1024
1069	n/a	data = b'\0' * max_gzip_decode
1070	n/a	encoded = xmlrpclib.gzip_encode(data)
1071	n/a	decoded = xmlrpclib.gzip_decode(encoded)
1072	n/a	self.assertEqual(len(decoded), max_gzip_decode)
1073	n/a
1074	n/a	data = b'\0' * (max_gzip_decode + 1)
1075	n/a	encoded = xmlrpclib.gzip_encode(data)
1076	n/a
1077	n/a	with self.assertRaisesRegex(ValueError,
1078	n/a	"max gzipped payload length exceeded"):
1079	n/a	xmlrpclib.gzip_decode(encoded)
1080	n/a
1081	n/a	xmlrpclib.gzip_decode(encoded, max_decode=-1)
1082	n/a
1083	n/a
1084	n/a	#Test special attributes of the ServerProxy object
1085	n/a	class ServerProxyTestCase(unittest.TestCase):
1086	n/a	def setUp(self):
1087	n/a	unittest.TestCase.setUp(self)
1088	n/a	if threading:
1089	n/a	self.url = URL
1090	n/a	else:
1091	n/a	# Without threading, http_server() and http_multi_server() will not
1092	n/a	# be executed and URL is still equal to None. 'http://' is a just
1093	n/a	# enough to choose the scheme (HTTP)
1094	n/a	self.url = 'http://'
1095	n/a
1096	n/a	def test_close(self):
1097	n/a	p = xmlrpclib.ServerProxy(self.url)
1098	n/a	self.assertEqual(p('close')(), None)
1099	n/a
1100	n/a	def test_transport(self):
1101	n/a	t = xmlrpclib.Transport()
1102	n/a	p = xmlrpclib.ServerProxy(self.url, transport=t)
1103	n/a	self.assertEqual(p('transport'), t)
1104	n/a
1105	n/a
1106	n/a	# This is a contrived way to make a failure occur on the server side
1107	n/a	# in order to test the _send_traceback_header flag on the server
1108	n/a	class FailingMessageClass(http.client.HTTPMessage):
1109	n/a	def get(self, key, failobj=None):
1110	n/a	key = key.lower()
1111	n/a	if key == 'content-length':
1112	n/a	return 'I am broken'
1113	n/a	return super().get(key, failobj)
1114	n/a
1115	n/a
1116	n/a	@unittest.skipUnless(threading, 'Threading required for this test.')
1117	n/a	class FailingServerTestCase(unittest.TestCase):
1118	n/a	def setUp(self):
1119	n/a	self.evt = threading.Event()
1120	n/a	# start server thread to handle requests
1121	n/a	serv_args = (self.evt, 1)
1122	n/a	threading.Thread(target=http_server, args=serv_args).start()
1123	n/a
1124	n/a	# wait for the server to be ready
1125	n/a	self.evt.wait()
1126	n/a	self.evt.clear()
1127	n/a
1128	n/a	def tearDown(self):
1129	n/a	# wait on the server thread to terminate
1130	n/a	self.evt.wait()
1131	n/a	# reset flag
1132	n/a	xmlrpc.server.SimpleXMLRPCServer._send_traceback_header = False
1133	n/a	# reset message class
1134	n/a	default_class = http.client.HTTPMessage
1135	n/a	xmlrpc.server.SimpleXMLRPCRequestHandler.MessageClass = default_class
1136	n/a
1137	n/a	def test_basic(self):
1138	n/a	# check that flag is false by default
1139	n/a	flagval = xmlrpc.server.SimpleXMLRPCServer._send_traceback_header
1140	n/a	self.assertEqual(flagval, False)
1141	n/a
1142	n/a	# enable traceback reporting
1143	n/a	xmlrpc.server.SimpleXMLRPCServer._send_traceback_header = True
1144	n/a
1145	n/a	# test a call that shouldn't fail just as a smoke test
1146	n/a	try:
1147	n/a	p = xmlrpclib.ServerProxy(URL)
1148	n/a	self.assertEqual(p.pow(6,8), 6**8)
1149	n/a	except (xmlrpclib.ProtocolError, OSError) as e:
1150	n/a	# ignore failures due to non-blocking socket 'unavailable' errors
1151	n/a	if not is_unavailable_exception(e):
1152	n/a	# protocol error; provide additional information in test output
1153	n/a	self.fail("%s\n%s" % (e, getattr(e, "headers", "")))
1154	n/a
1155	n/a	def test_fail_no_info(self):
1156	n/a	# use the broken message class
1157	n/a	xmlrpc.server.SimpleXMLRPCRequestHandler.MessageClass = FailingMessageClass
1158	n/a
1159	n/a	try:
1160	n/a	p = xmlrpclib.ServerProxy(URL)
1161	n/a	p.pow(6,8)
1162	n/a	except (xmlrpclib.ProtocolError, OSError) as e:
1163	n/a	# ignore failures due to non-blocking socket 'unavailable' errors
1164	n/a	if not is_unavailable_exception(e) and hasattr(e, "headers"):
1165	n/a	# The two server-side error headers shouldn't be sent back in this case
1166	n/a	self.assertTrue(e.headers.get("X-exception") is None)
1167	n/a	self.assertTrue(e.headers.get("X-traceback") is None)
1168	n/a	else:
1169	n/a	self.fail('ProtocolError not raised')
1170	n/a
1171	n/a	def test_fail_with_info(self):
1172	n/a	# use the broken message class
1173	n/a	xmlrpc.server.SimpleXMLRPCRequestHandler.MessageClass = FailingMessageClass
1174	n/a
1175	n/a	# Check that errors in the server send back exception/traceback
1176	n/a	# info when flag is set
1177	n/a	xmlrpc.server.SimpleXMLRPCServer._send_traceback_header = True
1178	n/a
1179	n/a	try:
1180	n/a	p = xmlrpclib.ServerProxy(URL)
1181	n/a	p.pow(6,8)
1182	n/a	except (xmlrpclib.ProtocolError, OSError) as e:
1183	n/a	# ignore failures due to non-blocking socket 'unavailable' errors
1184	n/a	if not is_unavailable_exception(e) and hasattr(e, "headers"):
1185	n/a	# We should get error info in the response
1186	n/a	expected_err = "invalid literal for int() with base 10: 'I am broken'"
1187	n/a	self.assertEqual(e.headers.get("X-exception"), expected_err)
1188	n/a	self.assertTrue(e.headers.get("X-traceback") is not None)
1189	n/a	else:
1190	n/a	self.fail('ProtocolError not raised')
1191	n/a
1192	n/a
1193	n/a	@contextlib.contextmanager
1194	n/a	def captured_stdout(encoding='utf-8'):
1195	n/a	"""A variation on support.captured_stdout() which gives a text stream
1196	n/a	having a `buffer` attribute.
1197	n/a	"""
1198	n/a	orig_stdout = sys.stdout
1199	n/a	sys.stdout = io.TextIOWrapper(io.BytesIO(), encoding=encoding)
1200	n/a	try:
1201	n/a	yield sys.stdout
1202	n/a	finally:
1203	n/a	sys.stdout = orig_stdout
1204	n/a
1205	n/a
1206	n/a	class CGIHandlerTestCase(unittest.TestCase):
1207	n/a	def setUp(self):
1208	n/a	self.cgi = xmlrpc.server.CGIXMLRPCRequestHandler()
1209	n/a
1210	n/a	def tearDown(self):
1211	n/a	self.cgi = None
1212	n/a
1213	n/a	def test_cgi_get(self):
1214	n/a	with support.EnvironmentVarGuard() as env:
1215	n/a	env['REQUEST_METHOD'] = 'GET'
1216	n/a	# if the method is GET and no request_text is given, it runs handle_get
1217	n/a	# get sysout output
1218	n/a	with captured_stdout(encoding=self.cgi.encoding) as data_out:
1219	n/a	self.cgi.handle_request()
1220	n/a
1221	n/a	# parse Status header
1222	n/a	data_out.seek(0)
1223	n/a	handle = data_out.read()
1224	n/a	status = handle.split()[1]
1225	n/a	message = ' '.join(handle.split()[2:4])
1226	n/a
1227	n/a	self.assertEqual(status, '400')
1228	n/a	self.assertEqual(message, 'Bad Request')
1229	n/a
1230	n/a
1231	n/a	def test_cgi_xmlrpc_response(self):
1232	n/a	data = """<?xml version='1.0'?>
1233	n/a	<methodCall>
1234	n/a	<methodName>test_method</methodName>
1235	n/a	<params>
1236	n/a	<param>
1237	n/a	<value><string>foo</string></value>
1238	n/a	</param>
1239	n/a	<param>
1240	n/a	<value><string>bar</string></value>
1241	n/a	</param>
1242	n/a	</params>
1243	n/a	</methodCall>
1244	n/a	"""
1245	n/a
1246	n/a	with support.EnvironmentVarGuard() as env, \
1247	n/a	captured_stdout(encoding=self.cgi.encoding) as data_out, \
1248	n/a	support.captured_stdin() as data_in:
1249	n/a	data_in.write(data)
1250	n/a	data_in.seek(0)
1251	n/a	env['CONTENT_LENGTH'] = str(len(data))
1252	n/a	self.cgi.handle_request()
1253	n/a	data_out.seek(0)
1254	n/a
1255	n/a	# will respond exception, if so, our goal is achieved ;)
1256	n/a	handle = data_out.read()
1257	n/a
1258	n/a	# start with 44th char so as not to get http header, we just
1259	n/a	# need only xml
1260	n/a	self.assertRaises(xmlrpclib.Fault, xmlrpclib.loads, handle[44:])
1261	n/a
1262	n/a	# Also test the content-length returned by handle_request
1263	n/a	# Using the same test method inorder to avoid all the datapassing
1264	n/a	# boilerplate code.
1265	n/a	# Test for bug: http://bugs.python.org/issue5040
1266	n/a
1267	n/a	content = handle[handle.find("<?xml"):]
1268	n/a
1269	n/a	self.assertEqual(
1270	n/a	int(re.search(r'Content-Length: (\d+)', handle).group(1)),
1271	n/a	len(content))
1272	n/a
1273	n/a
1274	n/a	class UseBuiltinTypesTestCase(unittest.TestCase):
1275	n/a
1276	n/a	def test_use_builtin_types(self):
1277	n/a	# SimpleXMLRPCDispatcher.__init__ accepts use_builtin_types, which
1278	n/a	# makes all dispatch of binary data as bytes instances, and all
1279	n/a	# dispatch of datetime argument as datetime.datetime instances.
1280	n/a	self.log = []
1281	n/a	expected_bytes = b"my dog has fleas"
1282	n/a	expected_date = datetime.datetime(2008, 5, 26, 18, 25, 12)
1283	n/a	marshaled = xmlrpclib.dumps((expected_bytes, expected_date), 'foobar')
1284	n/a	def foobar(*args):
1285	n/a	self.log.extend(args)
1286	n/a	handler = xmlrpc.server.SimpleXMLRPCDispatcher(
1287	n/a	allow_none=True, encoding=None, use_builtin_types=True)
1288	n/a	handler.register_function(foobar)
1289	n/a	handler._marshaled_dispatch(marshaled)
1290	n/a	self.assertEqual(len(self.log), 2)
1291	n/a	mybytes, mydate = self.log
1292	n/a	self.assertEqual(self.log, [expected_bytes, expected_date])
1293	n/a	self.assertIs(type(mydate), datetime.datetime)
1294	n/a	self.assertIs(type(mybytes), bytes)
1295	n/a
1296	n/a	def test_cgihandler_has_use_builtin_types_flag(self):
1297	n/a	handler = xmlrpc.server.CGIXMLRPCRequestHandler(use_builtin_types=True)
1298	n/a	self.assertTrue(handler.use_builtin_types)
1299	n/a
1300	n/a	def test_xmlrpcserver_has_use_builtin_types_flag(self):
1301	n/a	server = xmlrpc.server.SimpleXMLRPCServer(("localhost", 0),
1302	n/a	use_builtin_types=True)
1303	n/a	server.server_close()
1304	n/a	self.assertTrue(server.use_builtin_types)
1305	n/a
1306	n/a
1307	n/a	@support.reap_threads
1308	n/a	def test_main():
1309	n/a	support.run_unittest(XMLRPCTestCase, HelperTestCase, DateTimeTestCase,
1310	n/a	BinaryTestCase, FaultTestCase, UseBuiltinTypesTestCase,
1311	n/a	SimpleServerTestCase, SimpleServerEncodingTestCase,
1312	n/a	KeepaliveServerTestCase1, KeepaliveServerTestCase2,
1313	n/a	GzipServerTestCase, GzipUtilTestCase,
1314	n/a	MultiPathServerTestCase, ServerProxyTestCase, FailingServerTestCase,
1315	n/a	CGIHandlerTestCase)
1316	n/a
1317	n/a
1318	n/a	if __name__ == "__main__":
1319	n/a	test_main()