#!/usr/bin/env python # vim:ts=4:sw=4:et: from __future__ import absolute_import from __future__ import division from __future__ import print_function # no unicode literals import binascii import collections import inspect try: import unittest2 as unittest except ImportError: import unittest import os import sys from pywatchman import ( bser, compat, client, pybser, load, SocketTimeout, Transport, WatchmanError, ) PILE_OF_POO = u"\U0001F4A9" NON_UTF8_STRING = b'\xff\xff\xff' class TestSocketTimeout(unittest.TestCase): def test_exception_handling(self): try: raise SocketTimeout('should not raise') except WatchmanError: pass class TestTransportErrorHandling(unittest.TestCase): def test_transport_error(self): buf = '{"foo":"bar"}' failAfterBytesRead = 5 class FakeFailingTransport(Transport): def __init__(self, sockpath, timeout): self.readBuf = buf self.readBufPos = 0 self.writeBuf = [] self.closed = False def close(self): self.closed = True def readBytes(self, size): readEnd = self.readBufPos + size if readEnd > failAfterBytesRead: raise IOError(23, 'fnord') elif readEnd > len(self.readBuf): return '' read = self.readBuf[self.readBufPos:self.readBufPos + size] self.readBufPos += size return read def write(self, buf): self.writeBuf.extend(buf) c = client( sockpath='', transport=FakeFailingTransport, sendEncoding='json', recvEncoding='json', ) try: c.query("foobarbaz") self.assertTrue(False, "expected a WatchmanError") except WatchmanError as e: self.assertIn('I/O error communicating with watchman daemon: ' + 'errno=23 errmsg=fnord, while executing ' + '(\'foobarbaz\',)', str(e)) except Exception as e: self.assertTrue(False, "expected a WatchmanError, but got " + str(e)) def expand_bser_mods(test_class): ''' A decorator function used to create a class for bser and pybser variants of the test. ''' # We do some rather hacky things here to define new test class types # in our caller's scope. This is needed so that the unittest TestLoader # will find the subclasses we define. caller_scope = inspect.currentframe().f_back.f_locals flavors = [(bser, 'Bser'), (pybser, 'PyBser')] for (mod, suffix) in flavors: def make_class(mod, suffix): subclass_name = test_class.__name__ + suffix # Define a new class that derives from the input class class MatrixTest(test_class): def init_bser_mod(self): self.bser_mod = mod # Set the name and module information on our new subclass MatrixTest.__name__ = subclass_name MatrixTest.__qualname__ = subclass_name MatrixTest.__module__ = test_class.__module__ caller_scope[subclass_name] = MatrixTest make_class(mod, suffix) class FakeFile(object): def __init__(self, data): self._data = data self._ptr = 0 def readinto(self, buf): l = len(buf) if len(self._data) - self._ptr < l: return None buf[:] = self._data[self._ptr:self._ptr + l] self._ptr += l return l @expand_bser_mods class TestBSERDump(unittest.TestCase): def setUp(self): self.init_bser_mod() def raw(self, structured_input, bser_output): enc = self.bser_mod.dumps(structured_input) self.assertEqual(enc, bser_output) def roundtrip(self, val, mutable=True, value_encoding=None, value_errors=None): enc = self.bser_mod.dumps(val) #print("# %s --> %s" % (repr(val), # binascii.hexlify(enc).decode('ascii'))) dec = self.bser_mod.loads(enc, mutable, value_encoding=value_encoding, value_errors=value_errors) self.assertEqual(val, dec) fp = FakeFile(enc) dec = self.bser_mod.load(fp, mutable, value_encoding=value_encoding, value_errors=value_errors) self.assertEqual(val, dec) def munged(self, val, munged, value_encoding=None, value_errors=None): enc = self.bser_mod.dumps(val) #print("# %s --> %s" % (repr(val), # binascii.hexlify(enc).decode('ascii'))) dec = self.bser_mod.loads(enc, value_encoding=value_encoding, value_errors=value_errors) self.assertEqual(munged, dec) def test_raw(self): self.raw({"name": "Tom"}, b'\x00\x01\x05\x10\x00\x00\x00\x01\x03\x01\x02\x03\x04name\x02' b'\x03\x03Tom') self.raw({"names": ["Tom", "Jerry"]}, b'\x00\x01\x05\x1c\x00\x00\x00\x01\x03\x01\x02\x03\x05names\x00' b'\x03\x02\x02\x03\x03Tom\x02\x03\x05Jerry') self.raw(["Tom", "Jerry"], b"\x00\x01\x05\x11\x00\x00\x00\x00\x03\x02\x02\x03\x03Tom\x02" b"\x03\x05Jerry") self.raw([1, 123, 12345, 1234567, 12345678912345678], b"\x00\x01\x05\x18\x00\x00\x00\x00\x03\x05\x03\x01\x03{\x0490" b"\x05\x87\xd6\x12\x00\x06N\xd6\x14^T\xdc+\x00") def test_int(self): self.roundtrip(1) self.roundtrip(0x100) self.roundtrip(0x10000) self.roundtrip(0x10000000) self.roundtrip(0x1000000000) def test_negative_int(self): self.roundtrip(-0x80) self.roundtrip(-0x8000) self.roundtrip(-0x80000000) self.roundtrip(-0x8000000000000000) def test_float(self): self.roundtrip(1.5) def test_bool(self): self.roundtrip(True) self.roundtrip(False) def test_none(self): self.roundtrip(None) def test_string(self): self.roundtrip(b"hello") # For Python 3, here we can only check that a Unicode string goes in, # not that a Unicode string comes out. self.munged(u'Hello', b'Hello') self.roundtrip(u'Hello', value_encoding='utf8') self.roundtrip(u'Hello', value_encoding='ascii') self.roundtrip(u'Hello' + PILE_OF_POO, value_encoding='utf8') # can't use the with form here because Python 2.6 self.assertRaises(UnicodeDecodeError, self.roundtrip, u'Hello' + PILE_OF_POO, value_encoding='ascii') self.munged(u'Hello' + PILE_OF_POO, u'Hello', value_encoding='ascii', value_errors='ignore') self.roundtrip(b'hello' + NON_UTF8_STRING) self.assertRaises(UnicodeDecodeError, self.roundtrip, b'hello' + NON_UTF8_STRING, value_encoding='utf8') self.munged(b'hello' + NON_UTF8_STRING, u'hello', value_encoding='utf8', value_errors='ignore') # TODO: test non-UTF8 strings with surrogateescape in Python 3 ustr = u'\xe4\xf6\xfc' self.munged(ustr, ustr.encode('utf-8')) def test_list(self): self.roundtrip([1, 2, 3]) self.roundtrip([1, b"helo", 2.5, False, None, True, 3]) def test_tuple(self): self.munged((1, 2, 3), [1, 2, 3]) self.roundtrip((1, 2, 3), mutable=False) def test_dict(self): self.roundtrip({"hello": b"there"}) self.roundtrip({"hello": u"there"}, value_encoding='utf8') self.roundtrip({"hello": u"there"}, value_encoding='ascii') self.roundtrip({"hello": u"there" + PILE_OF_POO}, value_encoding='utf8') # can't use the with form here because Python 2.6 self.assertRaises(UnicodeDecodeError, self.roundtrip, {"hello": u"there" + PILE_OF_POO}, value_encoding='ascii') self.munged({'Hello': u'there' + PILE_OF_POO}, {'Hello': u'there'}, value_encoding='ascii', value_errors='ignore') self.roundtrip({'Hello': b'there' + NON_UTF8_STRING}) self.assertRaises(UnicodeDecodeError, self.roundtrip, {"hello": b"there" + NON_UTF8_STRING}, value_encoding='utf8') self.munged({'Hello': b'there' + NON_UTF8_STRING}, {'Hello': u'there'}, value_encoding='utf8', value_errors='ignore') obj = self.bser_mod.loads(self.bser_mod.dumps({"hello": b"there"}), False) self.assertEqual(1, len(obj)) self.assertEqual(b'there', obj.hello) self.assertEqual(b'there', obj[u'hello']) if not compat.PYTHON3: self.assertEqual(b'there', obj[b'hello']) self.assertEqual(b'there', obj[0]) # make sure this doesn't crash self.assertRaises(Exception, lambda: obj[45.25]) hello, = obj # sequence/list assignment self.assertEqual(b'there', hello) def assertItemAttributes(self, dictish, attrish): self.assertEqual(len(dictish), len(attrish)) # Use items for compatibility across Python 2 and 3. for k, v in dictish.items(): self.assertEqual(v, getattr(attrish, k)) def test_template(self): # since we can't generate the template bser output, here's a # a blob from the C test suite in watchman templ = b"\x00\x01\x03\x28" + \ b"\x0b\x00\x03\x02\x02\x03\x04\x6e\x61\x6d\x65\x02" + \ b"\x03\x03\x61\x67\x65\x03\x03\x02\x03\x04\x66\x72" + \ b"\x65\x64\x03\x14\x02\x03\x04\x70\x65\x74\x65\x03" + \ b"\x1e\x0c\x03\x19" dec = self.bser_mod.loads(templ) exp = [ {"name": b"fred", "age": 20}, {"name": b"pete", "age": 30}, {"name": None, "age": 25} ] self.assertEqual(exp, dec) res = self.bser_mod.loads(templ, False) for i in range(0, len(exp)): self.assertItemAttributes(exp[i], res[i]) def test_pdu_info(self): enc = self.bser_mod.dumps(1) DEFAULT_BSER_VERSION = 1 DEFAULT_BSER_CAPABILITIES = 0 self.assertEqual((DEFAULT_BSER_VERSION, DEFAULT_BSER_CAPABILITIES, len(enc)), self.bser_mod.pdu_info(enc)) # try a bigger one; prove that we get the correct length # even though we receive just a portion of the complete # data enc = self.bser_mod.dumps([1, 2, 3, "hello there, much larger"]) self.assertEqual((DEFAULT_BSER_VERSION, DEFAULT_BSER_CAPABILITIES, len(enc)), self.bser_mod.pdu_info(enc[0:7])) def test_pdu_len(self): enc = self.bser_mod.dumps(1) self.assertEqual(len(enc), self.bser_mod.pdu_len(enc)) # try a bigger one; prove that we get the correct length # even though we receive just a portion of the complete # data enc = self.bser_mod.dumps([1, 2, 3, "hello there, much larger"]) self.assertEqual(len(enc), self.bser_mod.pdu_len(enc[0:7])) def test_garbage(self): # can't use the with form here because Python 2.6 self.assertRaises(ValueError, self.bser_mod.loads, b"\x00\x01\n") self.assertRaises(ValueError, self.bser_mod.loads, b'\x00\x01\x04\x01\x00\x02') self.assertRaises(ValueError, self.bser_mod.loads, b'\x00\x01\x07') self.assertRaises(ValueError, self.bser_mod.loads, b'\x00\x01\x03\x01\xff') self.assertRaises(ValueError, self.bser_mod.pdu_info, b'\x00\x02') if __name__ == '__main__': suite = load_tests(unittest.TestLoader()) unittest.TextTestRunner().run(suite)