from __future__ import print_function
import unittest
import sys, os, time
from ctypes import *
from czmq import *

# http://python3porting.com/problems.html#common-migration-problems
PY3 = sys.version_info > (3,)

if PY3:
    # Python 3 does not have a "cmp" builtin.
    def cmp(a, b):
        return (a > b) - (a < b)

# Part test, part example of how to use python bindings.
# Note the del statements are just to explicitly test functionality, it is generally not required.

class TestCZMQ(unittest.TestCase):
    verbose = int(os.environ.get('VERBOSE', '0'))

    def test_zdir(self):
        older = Zdir(b'.', None)
        self.assertTrue(older)
        if self.verbose:
            older.print(0)

        newer = Zdir(b'..', None)
        self.assertTrue(newer)

        patches = Zdir.diff(older, newer, b'/')
        self.assertTrue(patches)

        while patches.size():
            patch = ZdirPatch(patches.pop(), True)

        nosuch = Zdir(b"does-not-exist", None)
        self.assertFalse(nosuch)

        # zdir_watch test:
        watch = Zactor(zactor_fn(lib.zdir_watch), None)
        self.assertTrue(watch)

        if self.verbose:
            watch.sock().send(b"s", b"VERBOSE")
            self.assertEqual(watch.sock().wait(), 0)

        # need to create a file in the test directory we're watching
        # in order to ensure the directory exists
        initfile = Zfile(b"./zdir-test-dir", b"initial_file")
        self.assertTrue(initfile)
        initfile.output()
        if PY3:
            with initfile.handle() as hdl:  # Avoid ResourceWarning: unclosed file.
                hdl.write(b"initial file\n")
        else:
            initfile.handle().write(b"initial file\n")
        initfile.close()

        try:
            stable_age = float(file_stable_age_msec()) / 1000.0 + 0.050
        except Exception:
            stable_age = 5.050
        time.sleep(stable_age) # wait for initial file to become 'stable'

        watch.sock().send(b"si", b"TIMEOUT", 100)
        self.assertEqual(watch.sock().wait(), 0)

        watch.sock().send(b"ss", b"SUBSCRIBE", b"zdir-test-dir")
        self.assertEqual(watch.sock().wait(), 0)

        watch.sock().send(b"ss", b"UNSUBSCRIBE", b"zdir-test-dir")
        self.assertEqual(watch.sock().wait(), 0)

        watch.sock().send(b"ss", b"SUBSCRIBE", b"zdir-test-dir")
        self.assertEqual(watch.sock().wait(), 0)

        newfile = Zfile(b"zdir-test-dir", b"test_abc")
        newfile.output()
        if PY3:
            with newfile.handle() as hdl:  # Avoid ResourceWarning: unclosed file.
                hdl.write(b"test file\n")
        else:
            newfile.handle().write(b"test file\n")
        newfile.close()

        #watch_poll = Zpoller(watch, None)

        # poll for a certain timeout before giving up and failing the test.
        #assert(watch_poll.wait(1001) == watch)

        # wait for notification of the file being added
        path = c_char_p()
        patches = zlist_p()
        rc = watch.sock().recv(b"sp", byref(path), byref(patches))
        self.assertEqual(rc, 0)
        patches = Zlist(patches, True)

        self.assertEqual(string_at(path), b"zdir-test-dir")
        libc.free (path)

        self.assertEqual(patches.size(), 1)

        patch = ZdirPatch(patches.pop(), True)
        self.assertEqual(patch.path(), b"zdir-test-dir")

        patch_file = patch.file()
        self.assertEqual(patch_file.filename(b""), b"zdir-test-dir/test_abc")

        del patch
        del patches

        # remove the file
        newfile.remove()
        del newfile

        # poll for a certain timeout before giving up and failing the test.
        #assert(watch_poll.wait(1001) == watch)

        # wait for notification of the file being removed
        path = c_char_p()
        patches = zlist_p()
        rc = watch.sock().recv(b"sp", byref(path), byref(patches))
        self.assertEqual(rc, 0)
        patches = Zlist(patches, True)

        self.assertEqual(string_at(path), b"zdir-test-dir")
        libc.free (path)

        self.assertEqual(patches.size(), 1)

        patch = ZdirPatch(patches.pop(), True)
        self.assertEqual(patch.path(), b"zdir-test-dir")

        patch_file = patch.file()
        self.assertEqual(patch_file.filename(b""), b"zdir-test-dir/test_abc")

        del patch
        del patches

        #zpoller_destroy (&watch_poll)
        del watch

        # clean up by removing the test directory.
        testdir = Zdir(b"zdir-test-dir", None)
        testdir.remove(True)
        del testdir

    def test_zdir_patch(self):
        file = Zfile(b".", b"bilbo")
        self.assertTrue(file)
        patch = ZdirPatch(b".", file, ZdirPatch.CREATE, b"/")
        self.assertTrue(patch)
        del file

        file = patch.file()
        self.assertTrue(file)
        self.assertEqual(file.filename(b"."), b"bilbo")
        self.assertEqual(patch.vpath(), b"/bilbo")
        del patch

    def test_zfile(self):
        file = Zfile(None, b"bilbo")
        self.assertTrue(file)
        self.assertEqual(file.filename(b"."), b"bilbo")
        self.assertFalse(file.is_readable())
        del file

        # Create a test file in some random subdirectory
        file = Zfile(b"./this/is/a/test", b"bilbo")
        self.assertTrue(file)
        self.assertEqual(file.output(), 0)
        #chunk = Zchunk.new(None, 100)
        #self.assertTrue(chunk)
        #chunk.fill(0, 100)

        # Write 100 bytes at position 1,000,000 in the file
        #self.assertEqual(file.write(chunk, 1000000), 0)
        #del chunk
        file.close()
        self.assertTrue(file.is_readable())
        #self.assertEqual(file.cursize(), 1000100)
        #self.assertFalse(file.is_stable())
        self.assertTrue(file.digest())

        # Now truncate file from outside
        f = open('./this/is/a/test/bilbo', 'wb')
        f.truncate()
        f.write(b"Hello, World\n")
        f.close()
        del f

        try:
            stable_age = float(file_stable_age_msec()) / 1000.0 + 0.050
        except Exception:
            stable_age = 5.050

        self.assertTrue(file.has_changed())
        self.assertFalse(file.is_stable())
        time.sleep(stable_age) # just over a threshold value passes...
        self.assertTrue(file.has_changed())
        self.assertFalse(file.is_stable())

        file.restat() # restat.
        self.assertTrue(file.is_stable())
        self.assertFalse(file.has_changed())

        self.assertEqual(file.digest(), b"4AB299C8AD6ED14F31923DD94F8B5F5CB89DFB54")

        # Check we can read from file
        self.assertEqual(file.input(), 0)
        #chunk = file.read(1000100, 0)
        #self.assertTrue(chunk)
        #self.assertEqual(chunk.size(), 13)
        #del chunk
        file.close()

        # Check we can read lines from file
        self.assertEqual(file.input(), 0)
        line = file.readln()
        self.assertEqual(line, b"Hello, World")
        line = file.readln()
        self.assertIsNone(line)
        file.close()

        # Try some fun with symbolic links
        link = Zfile(b"./this/is/a/test", b"bilbo.ln")
        self.assertTrue(link)
        self.assertEqual(link.output(), 0)
        if PY3:
            with link.handle() as hdl:  # Avoid ResourceWarning: unclosed file.
                hdl.write(b"./this/is/a/test/bilbo\n")
        else:
            link.handle().write(b"./this/is/a/test/bilbo\n")

        del link

        link = Zfile(b"./this/is/a/test", b"bilbo.ln")
        self.assertTrue(link)
        self.assertEqual(link.input(), 0)
        #chunk = link.read(1000100, 0)
        #self.assertTrue(chunk)
        #self.assertEqual(chunk.size(), 13)
        #del chunk
        del link

        # Remove file and directory
        dir = Zdir(b"./this", None)
        self.assertTrue(dir)
        self.assertEqual(dir.cursize(), 26)
        dir.remove(True)
        self.assertEqual(dir.cursize(), 0)
        del dir

        # Check we can no longer read from file
        self.assertTrue(file.is_readable())
        file.restat()
        self.assertFalse(file.is_readable())
        self.assertEqual(file.input(), -1)
        del file

    def test_zframe(self):
        # Create two PAIR sockets and connect over inproc
        output = Zsock.new_pair (b"@inproc://zframe.test")
        self.assertTrue(output)
        input = Zsock.new_pair (b">inproc://zframe.test")
        self.assertTrue(input)

        # Send five different frames, test Zframe.MORE
        for frame_nbr in range(5):
            frame = Zframe(b"Hello", 5)
            self.assertTrue(frame)
            self.assertEqual(Zframe.send(frame, output, Zframe.MORE), 0)

        # Send same frame five times, test ZFRAME_REUSE
        frame = Zframe(b"Hello", 5)
        self.assertTrue(frame)
        for frame_nbr in range(5):
            self.assertEqual(Zframe.send(frame, output, Zframe.MORE + Zframe.REUSE), 0)
        self.assertTrue(frame)
        copy = frame.dup()
        self.assertTrue(frame.eq(copy))
        self.assertEqual(copy.size(), 5)
        del copy
        self.assertFalse(frame.eq(None))

        # Test zframe_new_empty
        frame = Zframe.new_empty()
        self.assertTrue(frame)
        self.assertEqual(frame.size(), 0)
        del frame

        # Send END frame
        frame = Zframe(b"NOT", 3)
        self.assertTrue(frame)
        frame.reset(b"END", 3)
        string = frame.strhex()
        self.assertEqual(string, b"454E44")
        string = frame.strdup()
        self.assertEqual(string, b"END")
        self.assertEqual(Zframe.send(frame, output, 0), 0)

        # Read and count until we receive END
        frame_nbr = 0
        while True:
            frame = Zframe.recv(input)
            if frame.streq(b"END"):
                break

            self.assertTrue(frame.more())
            frame.set_more(0)
            self.assertEqual(frame.more(), 0)
            frame_nbr += 1
        self.assertEqual(frame_nbr, 10)

    def test_zhash(self):
        hash = Zhash()
        self.assertTrue(hash)
        self.assertEqual(hash.size(), 0)
        self.assertFalse(hash.first())
        self.assertFalse(hash.cursor())

        # Insert some items
        # Note the bindings for zhash are pretty pointless for actually storing
        # things - we have to keep our ctypes buffer objects around to keep the
        # memory alive
        buffer1 = create_string_buffer(b"dead beef")
        self.assertEqual(hash.insert(b"DEADBEEF", buffer1), 0)
        self.assertEqual(string_at(hash.first()), b"dead beef")
        self.assertEqual(hash.cursor(), b"DEADBEEF")

        buffer2 = create_string_buffer(b"a bad cafe")
        self.assertEqual(hash.insert(b"ABADCAFE", buffer2), 0)
        buffer3 = create_string_buffer(b"coded bad")
        self.assertEqual(hash.insert(b"C0DEDBAD", buffer3), 0)
        buffer4 = create_string_buffer(b"dead food")
        self.assertEqual(hash.insert(b"DEADF00D", b"dead food"), 0)
        self.assertEqual(hash.size(), 4)

        # Look for existing items
        self.assertEqual(string_at(hash.lookup(b"DEADBEEF")), b"dead beef")
        self.assertEqual(string_at(hash.lookup(b"ABADCAFE")), b"a bad cafe")
        self.assertEqual(string_at(hash.lookup(b"DEADF00D")), b"dead food")

        # Look for non-existent items
        self.assertFalse(hash.lookup(b"foo"))

        # Try to insert duplicate items
        self.assertEqual(hash.insert(b"DEADBEEF", b"foo"), -1)
        self.assertEqual(string_at(hash.lookup(b"DEADBEEF")), b"dead beef")

        # Some rename tests

        # Valid rename, key is now LIVEBEEF
        self.assertEqual(hash.rename(b"DEADBEEF", b"LIVEBEEF"), 0)
        self.assertEqual(string_at(hash.lookup(b"LIVEBEEF")), b"dead beef")

        # Trying to rename an unknown item to a non-existent key
        self.assertEqual(hash.rename(b"WHATBEEF", b"NONESUCH"), -1)

        # Trying to rename an unknown item to an existing key
        self.assertEqual(hash.rename(b"WHATBEEF", b"LIVEBEEF"), -1)
        self.assertEqual(string_at(hash.lookup(b"LIVEBEEF")), b"dead beef")

        # Trying to rename an existing item to another existing item
        self.assertEqual(hash.rename(b"LIVEBEEF", b"ABADCAFE"), -1)
        self.assertEqual(string_at(hash.lookup(b"LIVEBEEF")), b"dead beef")
        self.assertEqual(string_at(hash.lookup(b"ABADCAFE")), b"a bad cafe")

        # Test keys method
        keys = hash.keys()
        self.assertEqual(keys.size(), 4)
        del keys

        # Test dup method
        copy = hash.dup()
        self.assertEqual(copy.size(), 4)
        self.assertEqual(string_at(copy.lookup(b"LIVEBEEF")), b"dead beef")
        del copy

        # Test pack/unpack methods
        frame = hash.pack()
        copy = Zhash.unpack(frame)
        del frame
        self.assertEqual(copy.size(), 4)
        self.assertEqual(string_at(copy.lookup(b"LIVEBEEF")), b"dead beef")
        del copy

        # Test save and load
        hash.comment(b"This is a test file")
        hash.comment(b"Created by %s", b"czmq_selftest")
        hash.save(b".cache")
        copy = Zhash()
        self.assertTrue(copy)
        copy.load(b".cache")
        self.assertEqual(string_at(copy.lookup(b"LIVEBEEF")), b"dead beef")
        del copy
        os.remove(b".cache")

        # Delete a item
        hash.delete(b"LIVEBEEF")
        self.assertFalse(hash.lookup(b"LIVEBEEF"))
        self.assertEqual(hash.size(), 3)

        # Test autofree automatically copies and frees string values
        # This means the python string doesn't need to persist, like above
        hash = Zhash()
        self.assertTrue(hash)
        hash.autofree()
        self.assertEqual(hash.insert(b"key1", b"This is a string"), 0)
        self.assertEqual(hash.insert(b"key2", b"Ring a ding ding"), 0)
        self.assertEqual(string_at(hash.lookup(b"key1")), b"This is a string")
        self.assertEqual(string_at(hash.lookup(b"key2")), b"Ring a ding ding")
        del hash

    def test_zlist(self):
        def _zlist_free(data):
            x = Zlist(c_void_p(data), True)
            del x
        zlist_free = zlist_free_fn(_zlist_free)

        def _compare(item1, item2):
            return cmp(string_at(item1), string_at(item2)) > 0
        zlist_compare = zlist_compare_fn(_compare)

        list = Zlist()
        self.assertTrue(list)
        self.assertEqual(list.size(), 0)

        # Three items we'll use as test data
        # List items are void *, not particularly strings
        cheese = create_string_buffer(b"boursin")
        bread = create_string_buffer(b"baguette")
        wine = create_string_buffer(b"bordeaux")

        list.append(cheese)
        self.assertEqual(list.size(), 1)
        list.append(bread)
        self.assertEqual(list.size(), 2)
        list.append(wine)
        self.assertEqual(list.size(), 3)

        self.assertEqual(list.head().value, addressof(cheese))
        self.assertEqual(list.next().value, addressof(cheese))

        self.assertEqual(list.first().value, addressof(cheese))
        self.assertEqual(list.tail().value, addressof(wine))
        self.assertEqual(list.next().value, addressof(bread))

        self.assertEqual(list.first().value, addressof(cheese))
        self.assertEqual(list.next().value, addressof(bread))
        self.assertEqual(list.next().value, addressof(wine))
        self.assertFalse(list.next())
        # After we reach end of list, next wraps around
        self.assertEqual(list.next().value, addressof(cheese))
        self.assertEqual(list.size(), 3)

        list.remove(wine)
        self.assertEqual(list.size(), 2)

        self.assertEqual(list.first().value, addressof(cheese))
        list.remove(cheese)
        self.assertEqual(list.size(), 1)
        self.assertEqual(list.first().value, addressof(bread))

        list.remove(bread)
        self.assertEqual(list.size(), 0)

        list.append(cheese)
        list.append(bread)
        self.assertEqual(list.last().value, addressof(bread))
        list.remove(bread)
        self.assertEqual(list.last().value, addressof(cheese))
        list.remove(cheese)
        self.assertFalse(list.last())

        list.push(cheese)
        self.assertEqual(list.size(), 1)
        self.assertEqual(list.first().value, addressof(cheese))

        list.push(bread)
        self.assertEqual(list.size(), 2)
        self.assertEqual(list.first().value, addressof(bread))
        self.assertEqual(list.item().value, addressof(bread))

        list.append(wine)
        self.assertEqual(list.size(), 3)
        self.assertEqual(list.first().value, addressof(bread))

        sub_list = list.dup()
        self.assertTrue(sub_list)
        self.assertEqual(sub_list.size(), 3)

        list.sort(zlist_compare)
        self.assertEqual(list.pop().value, addressof(bread))
        self.assertEqual(list.pop().value, addressof(wine))
        self.assertEqual(list.pop().value, addressof(cheese))
        self.assertEqual(list.size(), 0)

        self.assertEqual(sub_list.size(), 3)
        list.push(sub_list)
        sub_list_2 = sub_list.dup()
        list.append(sub_list_2)

        ret = list.freefn(sub_list, zlist_free, False)
        sub_list.allow_destruct = False # Can't double-free
        self.assertEqual(ret.value, cast(sub_list._as_parameter_, c_void_p).value)

        ret = list.freefn(sub_list_2, zlist_free, True)
        sub_list_2.allow_destruct = False # Can't double-free
        self.assertEqual(ret.value, cast(sub_list_2._as_parameter_, c_void_p).value)
        del list

        # Test autofree functionality
        list = Zlist()
        self.assertTrue(list)
        list.autofree()
        list.push(b"bread")
        list.append(b"cheese")
        self.assertEqual(list.size(), 2)
        self.assertEqual(string_at(list.first()), b"bread")
        item = list.pop()
        self.assertEqual(string_at(item), b"bread")
        libc.free(item)
        item = list.pop()
        self.assertEqual(string_at(item), b"cheese")
        libc.free(item)

    def test_zloop(self):
        def _cancel_timer_event(loop, timer_id, arg):
            # We are handling timer 2, and will cancel timer 1
            self.assertIsInstance(arg, int)
            return Zloop(loop, False).timer_end(arg)
        cancel_timer_event = zloop_timer_fn(_cancel_timer_event)

        def _timer_event(loop, timer_id, arg):
            output.send(b's', b"PING")
            return 0
        timer_event = zloop_timer_fn(_timer_event)

        def _socket_event(loop, handle, arg):
            # Just end the reactor
            return -1
        socket_event = zloop_reader_fn(_socket_event)

        def _timer_event3(loop, timer_id, called):
            c_bool_p(called).contents = True
            # end the reactor
            return -1
        timer_event3 = zloop_timer_fn(_timer_event3)

        # Create two PAIR sockets and connect over inproc
        output = Zsock.new_pair(b"@inproc://zloop.test")
        self.assertTrue(output)
        input = Zsock.new_pair(b">inproc://zloop.test")
        self.assertTrue(input)

        loop = Zloop()
        self.assertTrue(loop)
        loop.set_verbose(self.verbose)

        # Create a timer that will be cancelled
        timer_id = loop.timer(1000, 1, timer_event, None)
        loop.timer(5, 1, cancel_timer_event, timer_id)

        # After 20 msecs, send a ping message to output3
        loop.timer(20, 1, timer_event, output)

        # Set up some tickets that will never expire
        loop.set_ticket_delay(10000)
        ticket1 = loop.ticket(timer_event, None)
        ticket2 = loop.ticket(timer_event, None)
        ticket3 = loop.ticket(timer_event, None)

        # When we get the ping message, end the reactor
        self.assertEqual(loop.reader(input, socket_event, None), 0)
        loop.reader_set_tolerant(input)
        loop.start()

        loop.ticket_delete(ticket1)
        loop.ticket_delete(ticket2)
        loop.ticket_delete(ticket3)
        del loop

        # Check whether loop properly ignores zsys_interrupted flag
        # when asked to
        #loop = Zloop()

        #timer_event_called = c_bool(False)
        #loop.timer(1, 1, timer_event3, byref(timer_event_called))

        #zsys_interrupted = 1
        #zloop_start (loop)
        ## zloop returns immediately without giving any handler a chance to run
        #self.assertFalse(timer_event_called)

        #zloop_ignore_interrupts (loop)
        #zloop_start (loop)
        # zloop runs the handler which will terminate the loop
        #self.assertTrue(timer_event_called)
        #zsys_interrupted = 0

        # cleanup
        #del loop
        #self.assertFalse(loop)

    def test_zmsg(self):
        # Create two PAIR sockets and connect over inproc
        output = Zsock.new_pair(b"@inproc://zmsg.test")
        self.assertTrue(output)
        input_ = Zsock.new_pair(b">inproc://zmsg.test")
        self.assertTrue(input_)

        # Test send and receive of single-frame message
        msg = Zmsg()
        self.assertTrue(msg)
        frame = Zframe(b"Hello", 5)
        self.assertTrue(frame)
        msg.prepend(frame)
        self.assertEqual(msg.size(), 1)
        self.assertEqual(msg.content_size(), 5)
        self.assertEqual(Zmsg.send(msg, output), 0)
        self.assertFalse(msg)

        msg = Zmsg.recv(input_)
        self.assertTrue(msg)
        self.assertEqual(msg.size(), 1)
        self.assertEqual(msg.content_size(), 5)
        del msg

        # Test send and receive of multi-frame message
        msg = Zmsg()
        self.assertTrue(msg)
        self.assertEqual(msg.addmem(b"Frame0", 6), 0)
        self.assertEqual(msg.addmem(b"Frame1", 6), 0)
        self.assertEqual(msg.addmem(b"Frame2", 6), 0)
        self.assertEqual(msg.addmem(b"Frame3", 6), 0)
        self.assertEqual(msg.addmem(b"Frame4", 6), 0)
        self.assertEqual(msg.addmem(b"Frame5", 6), 0)
        self.assertEqual(msg.addmem(b"Frame6", 6), 0)
        self.assertEqual(msg.addmem(b"Frame7", 6), 0)
        self.assertEqual(msg.addmem(b"Frame8", 6), 0)
        self.assertEqual(msg.addmem(b"Frame9", 6), 0)
        copy = msg.dup()
        self.assertTrue(copy)
        self.assertEqual(Zmsg.send(copy, output), 0)
        self.assertEqual(Zmsg.send(msg, output), 0)

        copy = Zmsg.recv(input_)
        self.assertTrue(copy)
        self.assertEqual(copy.size(), 10)
        self.assertEqual(copy.content_size(), 60)
        del copy

        msg = Zmsg.recv(input_)
        self.assertTrue(msg)
        self.assertEqual(msg.size(), 10)
        self.assertEqual(msg.content_size(), 60)

        # Save to a file, read back
        f = open("zmsg.test", "w")
        self.assertEqual(msg.save(f), 0)
        f.close()

        f = open ("zmsg.test", "r")
        self.assertEqual(msg.save(f), -1)
        f.close()
        del msg

        f = open ("zmsg.test", "r")
        msg = Zmsg.load(f)
        self.assertTrue(msg)
        f.close()
        os.remove ("zmsg.test")
        self.assertEqual(msg.size(), 10)
        self.assertEqual(msg.content_size(), 60)

        # Remove all frames except first and last
        for frame_nbr in range(8):
            msg.first()
            frame = msg.next()
            msg.remove(frame)

        # Test message frame manipulation
        self.assertEqual(msg.size(), 2)
        frame = msg.last()
        self.assertTrue(frame.streq(b"Frame9"))
        self.assertEqual(msg.content_size(), 12)
        frame = Zframe(b"Address", 7)
        self.assertTrue(frame)
        msg.prepend(frame)
        self.assertEqual(msg.size(), 3)
        self.assertEqual(msg.addstr(b"Body"), 0)
        self.assertEqual(msg.size(), 4)
        frame = msg.pop()
        del frame
        self.assertEqual(msg.size(), 3)
        body = msg.popstr()
        self.assertEqual(body, b"Frame0")
        del msg

        # Test encoding/decoding
        msg = Zmsg()
        self.assertTrue(msg)
        blank = create_string_buffer(100000)
        self.assertEqual(msg.addmem(blank, 0), 0)
        self.assertEqual(msg.addmem(blank, 1), 0)
        self.assertEqual(msg.addmem(blank, 253), 0)
        self.assertEqual(msg.addmem(blank, 254), 0)
        self.assertEqual(msg.addmem(blank, 255), 0)
        self.assertEqual(msg.addmem(blank, 256), 0)
        self.assertEqual(msg.addmem(blank, 65535), 0)
        self.assertEqual(msg.addmem(blank, 65536), 0)
        self.assertEqual(msg.addmem(blank, 65537), 0)
        del blank
        self.assertEqual(msg.size(), 9)
        frame = msg.encode()
        del msg
        msg = Zmsg.decode(frame)
        self.assertTrue(msg)
        del msg

        # Test submessages
        msg = Zmsg()
        self.assertTrue(msg)
        submsg = Zmsg()
        msg.pushstr(b"matr")
        submsg.pushstr(b"joska")
        self.assertEqual(msg.addmsg(submsg), 0)
        self.assertFalse(submsg)
        submsg = msg.popmsg()
        self.assertFalse(submsg)   # string "matr" is not encoded zmsg_t, so was discarded
        submsg = msg.popmsg()
        self.assertTrue(submsg)
        body = submsg.popstr()
        self.assertEqual(body, b"joska")
        del submsg
        frame = msg.pop()
        self.assertFalse(frame)
        del msg

        # Test comparison of two messages
        msg = Zmsg()
        msg.addstr(b"One")
        msg.addstr(b"Two")
        msg.addstr(b"Three")
        msg_other = Zmsg()
        msg_other.addstr(b"One")
        msg_other.addstr(b"Two")
        msg_other.addstr(b"One-Hundred")
        msg_dup = msg.dup()
        empty_msg = Zmsg()
        empty_msg_2 = Zmsg()
        self.assertTrue(msg.eq(msg_dup))
        self.assertFalse(msg.eq(msg_other))
        self.assertTrue(empty_msg.eq(empty_msg_2))
        self.assertFalse(msg.eq(None))
        del msg
        del msg_other
        del msg_dup
        del empty_msg

        # Test signal messages
        msg = Zmsg.new_signal(0)
        self.assertEqual(msg.signal(), 0)
        del msg
        msg = Zmsg.new_signal(-1)
        self.assertEqual(msg.signal(), 255)
        del msg

        # Now try methods on an empty message
        msg = Zmsg()
        self.assertTrue(msg)
        self.assertEqual(msg.size(), 0)
        self.assertFalse(msg.first())
        self.assertFalse(msg.last())
        self.assertFalse(msg.next())
        self.assertFalse(msg.pop())
        # Sending an empty message is valid and destroys the message
        self.assertEqual(Zmsg.send(msg, output), 0)
        self.assertFalse(msg)

    def test_zsock(self):
        writer = Zsock.new_push (b"@tcp://127.0.0.1:5560")
        self.assertTrue(writer)
        self.assertTrue(Zsock.resolve(writer) != writer)
        self.assertEqual(writer.type_str(), b"PUSH")

        # Check unbind
        self.assertEqual(writer.unbind(b"tcp://127.0.0.1:%d", 5560), 0)

        # In some cases and especially when running under Valgrind, doing
        # a bind immediately after an unbind causes an EADDRINUSE error.
        # Even a short sleep allows the OS to release the port for reuse.
        time.sleep(0.1)

        # Bind again
        self.assertEqual(writer.bind(b"tcp://127.0.0.1:%d", 5560), 5560)
        self.assertEqual(writer.endpoint(), b"tcp://127.0.0.1:5560")
#endif

        reader = Zsock.new_pull (b">tcp://127.0.0.1:5560")
        self.assertTrue(reader)
        self.assertTrue(Zsock.resolve(reader) != reader)
        self.assertEqual(reader.type_str(), b"PULL")

        # Basic Hello, World
        writer.send(b's', b"Hello, World")
        msg = Zmsg.recv(reader)
        self.assertTrue(msg)
        string = msg.popstr()
        self.assertEqual(string, b"Hello, World")
        del msg

        # Test resolve FD
        #fd = Zsock.fd(reader)
        #self.assertEqual(zsock_resolve (fd), None)

        # Test binding to ephemeral ports, sequential and random
        DYNAMIC_FIRST = 0xc000
        DYNAMIC_LAST = 0xffff
        port = writer.bind(b"tcp://127.0.0.1:*")
        self.assertTrue(port >= DYNAMIC_FIRST and port <= DYNAMIC_LAST)
        port = writer.bind(b"tcp://127.0.0.1:*[50000-]")
        self.assertTrue(port >= 50000 and port <= DYNAMIC_LAST)
        port = writer.bind(b"tcp://127.0.0.1:*[-50001]")
        self.assertTrue(port >= DYNAMIC_FIRST and port <= 50001)
        port = writer.bind(b"tcp://127.0.0.1:*[60000-60050]")
        self.assertTrue(port >= 60000 and port <= 60050)

        port = writer.bind(b"tcp://127.0.0.1:!")
        self.assertTrue(port >= DYNAMIC_FIRST and port <= DYNAMIC_LAST)
        port = writer.bind(b"tcp://127.0.0.1:![50000-]")
        self.assertTrue(port >= 50000 and port <= DYNAMIC_LAST)
        port = writer.bind(b"tcp://127.0.0.1:![-50001]")
        self.assertTrue(port >= DYNAMIC_FIRST and port <= 50001)
        port = writer.bind(b"tcp://127.0.0.1:![60000-60050]")
        self.assertTrue(port >= 60000 and port <= 60050)

        # Test zsock_attach method
        server = Zsock.new_dealer(b"")
        self.assertTrue(server)
        self.assertEqual(server.attach(b"@inproc://myendpoint,tcp://127.0.0.1:5556,inproc://others", True), 0)
        self.assertEqual(server.attach(b"", False), 0)
        self.assertEqual(server.attach(None, True), 0)
        self.assertEqual(server.attach(b">a,@b, c,, ", False), -1)
        del server

        # Test zsock_endpoint method
        self.assertEqual(writer.bind(b"inproc://test.%s", b"writer"), 0)
        self.assertEqual(writer.endpoint(), b"inproc://test.writer")

        # Test error state when connecting to an invalid socket type
        # ('txp://' instead of 'tcp://', typo intentional)
        self.assertEqual(reader.connect(b"txp://127.0.0.1:5560"), -1)

        # Test signal/wait methods
        self.assertEqual(writer.signal(123), 0)
        self.assertEqual(reader.wait(), 123)

        # Test zsock_send/recv pictures
        number1 = 123
        number2 = 123 * 123
        number4 = 123 * 123 * 123
        number8 = 123 * 123 * 123 * 123

        #chunk = zchunk_new (b"HELLO", 5)
        #self.assertTrue(chunk)
        frame = Zframe(b"WORLD", 5)
        self.assertTrue(frame)
        #hash = zhashx_new ()
        #self.assertTrue(hash)
        #hash.autofree()
        #hash.insert(b"1", b"value A")
        #hash.insert(b"2", b"value B")
        original = create_string_buffer(b"pointer")

        # Test zsock_recv into each supported type
        #writer.send(b"i1248zsbcfhp",
        #            -12345, number1, number2, number4, number8,
        #            b"This is a string", b"ABCDE", 5, chunk, frame, hash, original)
        writer.send(b"i1248zsbfp",
                    -12345, number1, number2, number4, c_longlong(number8),
                    b"This is a string", b"ABCDE", 5, frame, original)
        del frame
        #del chunk
        #del hash

        integer = c_int()
        number1 = c_byte()
        number2 = c_short()
        number4 = c_int()
        number8 = c_longlong()
        string = c_char_p()
        data = POINTER(c_ubyte)()
        size = c_size_t()
        #chunk = zchunk_p()
        frame = Zframe(c_void_p(), True)
        #hash = POINTER(zhashx_p)
        pointer = c_void_p()
        #rc = reader.recv(b"i1248zsbcfhp",
        #                 byref(integer), byref(number1), byref(number2), byref(number4), byref(number8),
        #                 byref(string), byref(data), byref(size), byref(chunk), byref(frame), byref(hash), byref(pointer))
        rc = reader.recv(b"i1248zsbfp",
                         byref(integer), byref(number1), byref(number2), byref(number4), byref(number8),
                         byref(string), byref(data), byref(size), byref(frame._as_parameter_), byref(pointer))
        self.assertEqual(rc, 0)
        self.assertEqual(integer.value, -12345)
        self.assertEqual(number1.value, 123)
        self.assertEqual(number2.value, 123 * 123)
        self.assertEqual(number4.value, 123 * 123 * 123)
        self.assertEqual(number8.value, 123 * 123 * 123 * 123)
        self.assertEqual(string_at(string), b"This is a string")
        self.assertEqual(string_at(data, size.value), b"ABCDE")
        self.assertEqual(size.value, 5)
        #self.assertEqual(memcmp (chunk.data(), b"HELLO", 5), 0)
        #self.assertEqual(chunk.size(), 5)
        self.assertTrue(frame.streq(b"WORLD"))
        self.assertEqual(frame.size(), 5)
        #value = string_at(hash.lookup(b"1"))
        #self.assertEqual(value, b"value A")
        #value = string_at(hash.lookup(b"2"))
        #self.assertEqual(value, b"value B")
        self.assertEqual(addressof(original), pointer.value)
        libc.free (string)
        libc.free (data)
        del frame
        #del chunk
        #del hash

        # Test zsock_recv of short message this lets us return a failure
        # with a status code and then nothing else the receiver will get
        # the status code and None/zero for all other values
        writer.send(b"i", -1)
        #reader.recv(b"izsbcfp",
        #    byref(integer), byref(string), byref(data), byref(size), byref(chunk), byref(frame), byref(pointer))
        frame = Zframe(c_void_p(), True)
        reader.recv(b"izsbfp",
            byref(integer), byref(string), byref(data), byref(size), byref(frame._as_parameter_), byref(pointer))
        self.assertEqual(integer.value, -1)
        self.assertEqual(string.value, None)
        self.assertFalse(data)
        self.assertEqual(size.value, 0)
        #self.assertEqual(chunk, None)
        self.assertFalse(frame)
        self.assertEqual(pointer.value, None)

        msg = Zmsg()
        msg.addstr(b"frame 1")
        msg.addstr(b"frame 2")
        writer.send(b"szm", b"header", msg)
        del msg

        msg = Zmsg(c_void_p(), True)
        reader.recv(b"szm", byref(string), byref(msg._as_parameter_))

        self.assertEqual(string_at(string), b"header")
        self.assertEqual(msg.size(), 2)
        self.assertEqual(msg.popstr(), b"frame 1")
        self.assertEqual(msg.popstr(), b"frame 2")
        #zstr_free (byref(string))
        libc.free(string)
        del msg

        # Test zsock_recv with null arguments
        #chunk = zchunk_new (b"HELLO", 5)
        #self.assertTrue(chunk)
        frame = Zframe(b"WORLD", 5)
        self.assertTrue(frame)
        #writer.send(b"izsbcfp",
        #            -12345, b"This is a string", b"ABCDE", 5, chunk, frame, original)
        writer.send(b"izsbfp",
                    -12345, b"This is a string", b"ABCDE", 5, frame, original)
        del frame
        #del chunk
        #reader.recv(b"izsbcfp", byref(integer), None, None, None, byref(chunk), None, None)
        reader.recv(b"izsbcfp", byref(integer), None, None, None, None, None, None)
        self.assertEqual(integer.value, -12345)
        #self.assertEqual(chunk.data(), b"HELLO", 5), 0)
        #self.assertEqual(chunk.size(), 5)
        #del chunk

        # Test zsock_bsend/brecv pictures with binary encoding
        frame = Zframe(b"Hello", 5)
        #chunk = Zchunk(b"World", 5)

        msg = Zmsg()
        msg.addstr(b"Hello")
        msg.addstr(b"World")

        #writer.bsend(b"1248sSpcfm",
        #             number1, number2, number4, number8,
        #             b"Hello, World",
        #             b"Goodbye cruel World!",
        #             original,
        #             chunk, frame, msg)
        writer.bsend(b"1248sSpfm",
                     number1, number2, number4, number8,
                     b"Hello, World",
                     b"Goodbye cruel World!",
                     original,
                     frame, msg)
        #del chunk
        del frame
        del msg

        frame = Zframe(c_void_p(), True)
        msg = Zmsg(c_void_p(), True)
        longstr = c_char_p()
        #reader.brecv(b"1248sSpcfm",
        #             byref(number1), byref(number2), byref(number4), byref(number8),
        #             byref(string), byref(longstr),
        #             byref(pointer),
        #             byref(chunk), byref(frame._as_parameter_), byref(msg._as_parameter_))
        reader.brecv(b"1248sSpfm",
                     byref(number1), byref(number2), byref(number4), byref(number8),
                     byref(string), byref(longstr),
                     byref(pointer),
                     byref(frame._as_parameter_), byref(msg._as_parameter_))
        self.assertEqual(number1.value, 123)
        self.assertEqual(number2.value, 123 * 123)
        self.assertEqual(number4.value, 123 * 123 * 123)
        self.assertEqual(number8.value, 123 * 123 * 123 * 123)
        self.assertEqual(string_at(string), b"Hello, World")
        self.assertEqual(string_at(longstr), b"Goodbye cruel World!")
        self.assertEqual(pointer.value, addressof(original))
        libc.free(longstr)
        #zstr_free (byref(longstr))
        #del chunk
        del frame
        del msg

        # Check that we can send a zproto format message
        #writer.bsend(b"1111sS4", 0xAA, 0xA0, 0x02, 0x01, b"key", b"value", 1234)
        #gossip = zgossip_msg_new ()
        #gossip.recv(reader)
        #self.assertEqual(gossip.id(), ZGOSSIP_MSG_PUBLISH)
        #del gossip

    def test_zactor(self):
        def _echo_actor(pipe, arg):
            # Do some initialization
            pipe = Zsock(pipe, False) # We don't own the pipe, so False.
            arg = string_at(arg)
            self.assertEqual(arg, b"Hello, World")
            pipe.signal(0) # startup ok

            terminated = False
            while not terminated:
                msg = Zmsg.recv(pipe)
                if not msg:
                    break # Interrupted

                command = msg.popstr()
                if command == b"$TERM":
                    # All actors must handle $TERM in this way
                    terminated = True
                elif command == b"ECHO":
                    # This is an example command for our test actor
                    Zmsg.send(msg, pipe)
                else:
                    self.fail(b"Unexpected message " + command)
        echo_actor = zactor_fn(_echo_actor) # ctypes function reference must live as long as the actor.

        arg_buffer = create_string_buffer(b"Hello, World") # argument buffer must live as long as the actor.
        actor = Zactor(echo_actor, arg_buffer)
        self.assertTrue(actor)

        actor.sock().send(b"ss", b"ECHO", b"This is a string");

        result = c_char_p()
        actor.sock().recv(b"s", byref(result))

        self.assertEqual(string_at(result), b"This is a string")

        libc.free(result)

if __name__ == '__main__':
    unittest.main()