python/helpers/pydev/pydev_tests/test_jyserver.py

'''
@author Fabio Zadrozny
'''
import sys
import unittest
import socket
import urllib
import pytest
import pycompletionserver


IS_JYTHON = sys.platform.find('java') != -1
DEBUG = 0

def dbg(s):
    if DEBUG:
        sys.stdout.write('TEST %s\n' % s)

@pytest.mark.skipif(not IS_JYTHON, reason='Jython related test')
class TestJython(unittest.TestCase):

    def test_it(self):
        dbg('ok')

    
    def test_message(self):
        t = pycompletionserver.CompletionServer(0)
        t.exit_process_on_kill = False

        l = []
        l.append(('Def', 'description'  , 'args'))
        l.append(('Def1', 'description1', 'args1'))
        l.append(('Def2', 'description2', 'args2'))

        msg = t.processor.format_completion_message('test_jyserver.py', l)

        self.assertEqual('@@COMPLETIONS(test_jyserver.py,(Def,description,args),(Def1,description1,args1),(Def2,description2,args2))END@@', msg)

        l = []
        l.append(('Def', 'desc,,r,,i()ption', ''))
        l.append(('Def(1', 'descriptio(n1', ''))
        l.append(('De,f)2', 'de,s,c,ription2', ''))
        msg = t.processor.format_completion_message(None, l)
        expected = '@@COMPLETIONS(None,(Def,desc%2C%2Cr%2C%2Ci%28%29ption, ),(Def%281,descriptio%28n1, ),(De%2Cf%292,de%2Cs%2Cc%2Cription2, ))END@@'

        self.assertEqual(expected, msg)

    
    def test_completion_sockets_and_messages(self):
        dbg('test_completion_sockets_and_messages')
        t, socket = self.create_connections()
        self.socket = socket
        dbg('connections created')

        try:
            #now that we have the connections all set up, check the code completion messages.
            msg = urllib.quote_plus('math')

            toWrite = '@@IMPORTS:%sEND@@' % msg
            dbg('writing' + str(toWrite))
            socket.send(toWrite)  #math completions
            completions = self.read_msg()
            dbg(urllib.unquote_plus(completions))

            start = '@@COMPLETIONS('
            self.assertTrue(completions.startswith(start), '%s DOESNT START WITH %s' % (completions, start))
            self.assertTrue(completions.find('@@COMPLETIONS') != -1)
            self.assertTrue(completions.find('END@@') != -1)


            msg = urllib.quote_plus('__builtin__.str')
            toWrite = '@@IMPORTS:%sEND@@' % msg
            dbg('writing' + str(toWrite))
            socket.send(toWrite)  #math completions
            completions = self.read_msg()
            dbg(urllib.unquote_plus(completions))

            start = '@@COMPLETIONS('
            self.assertTrue(completions.startswith(start), '%s DOESNT START WITH %s' % (completions, start))
            self.assertTrue(completions.find('@@COMPLETIONS') != -1)
            self.assertTrue(completions.find('END@@') != -1)



        finally:
            try:
                self.send_kill_msg(socket)


                while not t.ended:
                    pass  #wait until it receives the message and quits.


                socket.close()
            except:
                pass

    def get_free_port(self):
        from _pydev_bundle.pydev_localhost import get_socket_name
        return get_socket_name(close=True)[1]

    def create_connections(self):
        '''
        Creates the connections needed for testing.
        '''
        server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
        server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
        server.bind((pycompletionserver.HOST, 0))
        server.listen(1)  #socket to receive messages.

        from thread import start_new_thread
        t = pycompletionserver.CompletionServer(server.getsockname()[1])
        t.exit_process_on_kill = False

        start_new_thread(t.run, ())

        sock, _addr = server.accept()

        return t, sock

    def read_msg(self):
        msg = '@@PROCESSING_END@@'
        while msg.startswith('@@PROCESSING'):
            msg = self.socket.recv(1024)
            if msg.startswith('@@PROCESSING:'):
                dbg('Status msg:' + str(msg))

        while msg.find('END@@') == -1:
            msg += self.socket.recv(1024)

        return msg

    def send_kill_msg(self, socket):
        socket.send(pycompletionserver.MSG_KILL_SERVER)




# Run for jython in command line:
# c:\bin\jython2.7.0\bin\jython.exe -m py.test tests\test_jyserver.py