<commit_before>
import unittest
import ctypes
import ctypes.util
import sys
import os

import mapi_structs


hostname = 'localhost'
port = int(os.environ.get('MAPIPORT', '50000'))
username = 'monetdb'
password = '<PASSWORD>'
lang = 'sql'
db = os.environ.get('TSTDB', 'demo')


class MapiBaseTests(unittest.TestCase):
    """ the base test class. Loads the mapi library """

    def __init__(self, *args):
        location = ctypes.util.find_library('libmapi')
        if not location:
            locations = {
                    #TODO: add more platforms
                    'darwin': '../.libs/libmapilite.dylib',
                    'linux2': '../.libs/libmapilite.so',
                    }
            location = locations[sys.platform]
        location = "/opt/monetdb/may-2009-sp2/lib/libMapi.dylib"
        self.libmapi = ctypes.cdll.LoadLibrary(location)
        unittest.TestCase.__init__(self, *args)

    def _connect(self):
        mapi_connect = self.libmapi.mapi_connect
        mapi_connect.restype = mapi_structs.Mapi
        mid = mapi_connect(hostname, port, username, password, lang, db)
        if mid.contents.error != 0:
            self.fail("can't connect to database:  %s" % mid.contents.errorstr)
        return mid

    def _disconnect(self, mid):
        self.libmapi.mapi_disconnect(mid)
        self.libmapi.mapi_destroy(mid)



class MapiConnectTests(MapiBaseTests):
    """ here we test the connect and disconnect
    functions"""

    def test_mapi_connect(self):
        mid = self._connect()

        self.assertEqual(mid.contents.hostname, hostname)
        self.assertEqual(mid.contents.username, username)
        self.assertEqual(mid.contents.error, 0, mid.contents.errorstr +
                ". would like 0, got %s" % mid.contents.error)
        self.assertEqual(mid.contents.errorstr, None)
        self.assertEqual(mid.contents.connected, 0)
        #self.assertEqual(mid.contents.port, port) # merovingian redirects

    def test_mapi_mapi(self):
        mid = self._connect()

        self.assertEqual(mid.contents.hostname, hostname)
        #self.assertEqual(mid.contents.port, port) # merovingian redirects
        self.assertEqual(mid.contents.username, username)
        self.assertEqual(mid.contents.error, 0)
        self.assertEqual(mid.contents.errorstr, None)
        self.assertEqual(mid.contents.connected, 0)

    def mapi_destroy(self):
        self.fail("not yet implemented")


class MapiFunctionsTests(MapiBaseTests):
    """ Here we test everything _after_ connecting """

    def setUp(self):
        """ called for every test before the test is run """
        self.connection = self._connect()

    def tearDown(self):
        """ called for every test after the test is run """
        self._disconnect(self.connection)
        self.connection = None

    def test_mapi_start_talking(self):
        self.fail("not yet implemented")

    def mapi_resolve(self):
        self.fail("not yet implemented")

    def mapi_embedded_init(self):
        self.fail("not yet implemented")

    def mapi_disconnect(self):
        self.fail("not yet implemented")

    def mapi_reconnect(self):
        self.fail("not yet implemented")

    def mapi_ping(self):
        self.fail("not yet implemented")

    def mapi_error(self):
        self.fail("not yet implemented")

    def mapi_error_str(self):
        self.fail("not yet implemented")

    def mapi_noexplain(self):
        self.fail("not yet implemented")

    def mapi_explain(self):
        self.fail("not yet implemented")

    def mapi_explain_query(self):
        self.fail("not yet implemented")

    def mapi_explain_result(self):
        self.fail("not yet implemented")

    def mapi_output(self):
        self.fail("not yet implemented")
<commit_msg>basic connection tests are finished<commit_after>
import unittest
import ctypes
import ctypes.util
import sys
import os

import mapi_structs


hostname = 'localhost'
port = int(os.environ.get('MAPIPORT', '50000'))
username = 'monetdb'
password = '<PASSWORD>'
lang = 'sql'
db = os.environ.get('TSTDB', 'demo')


class MapiBaseTests(unittest.TestCase):
    """ the base test class. Loads the mapi library """

    def __init__(self, *args):
        location = ctypes.util.find_library('libmapilite')
        if not location:
            locations = {
                    #TODO: add more platforms
                    'darwin': '../.libs/libmapilite.dylib',
                    'linux2': '../.libs/libmapilite.so',
                    }
            location = locations[sys.platform]
        self.libmapi = ctypes.cdll.LoadLibrary(location)
        unittest.TestCase.__init__(self, *args)

    def _connect(self):
        mapi_connect = self.libmapi.mapi_connect
        mapi_connect.restype = mapi_structs.Mapi
        mid = mapi_connect(hostname, port, username, password, lang, db)
        self.assertEqual(mid.contents.error, 0, \
            "can't connect to database:  %s" % mid.contents.errorstr)
        return mid

    def _disconnect(self, mid):
        self.assertEqual(self.libmapi.mapi_disconnect(mid), 0)
        self.assertEqual(self.libmapi.mapi_destroy(mid), 0)



class MapiConnectTests(MapiBaseTests):
    """ here we test the connect and disconnect
    functions"""

    def test_mapi_connect(self):
        mid = self._connect()

        self.assertEqual(mid.contents.hostname, hostname)
        self.assertEqual(mid.contents.username, username)
        self.assertEqual(mid.contents.error, 0,
                "%s. would like 0, got %s" % (mid.contents.errorstr, mid.contents.error))
        self.assertEqual(mid.contents.errorstr, None)
        self.assertEqual(mid.contents.connected, 1)
        self.assertEqual(mid.contents.port, port)
        self._disconnect(mid)

    def mapi_mapi(self): # not a external used function
        mid = self._connect()
        self.assertEqual(mid.contents.hostname, hostname)
        self.assertEqual(mid.contents.port, port)
        self.assertEqual(mid.contents.username, username)
        self.assertEqual(mid.contents.error, 0)
        self.assertEqual(mid.contents.errorstr, None)
        self.assertEqual(mid.contents.connected, 1)

    def test_mapi_disconnect(self):
        mid = self._connect()
        mapi_disconnect = self.libmapi.mapi_disconnect
        message = mapi_disconnect(mid)
        self.assertEqual(message, 0)
        self.assertEqual(mid.contents.connected, 0)
        self.libmapi.mapi_destroy(mid)

    def test_mapi_destroy(self):
        mid = self._connect()
        self.libmapi.mapi_disconnect(mid)
        mapi_destroy = self.libmapi.mapi_destroy
        message = mapi_destroy(mid)
        self.assertEqual(message, 0)
        self.assertEqual(mid.contents.connected, 0)

    def test_mapi_reconnect(self):
        mid = self._connect()
        mapi_reconnect = self.libmapi.mapi_reconnect
        message = mapi_reconnect(mid)
        self.assertEqual(message, 0)
        self.assertEqual(mid.contents.connected, 1)
        # also test when disconnected
        self.libmapi.mapi_disconnect(mid)
        message = mapi_reconnect(mid)
        self.assertEqual(message, 0)
        self.assertEqual(mid.contents.connected, 1)
        self._disconnect(mid)

    def test_mapi_ping(self):
        mid = self._connect()
        mapi_ping = self.libmapi.mapi_ping
        message = mapi_ping(mid)
        self.assertEqual(message, 0)
        # also test when disconnected
        self.libmapi.mapi_disconnect(mid)
        message = mapi_ping(mid)
        self.assertEqual(message, -1)
        self._disconnect(mid)


class MapiFunctionsTests(MapiBaseTests):
    """ Here we test everything _after_ connecting """

    def setUp(self):
        """ called for every test before the test is run """
        self.mid = self._connect()

    def tearDown(self):
        """ called for every test after the test is run """
        self._disconnect(self.mid)
        self.mid = None

    def mapi_start_talking(self):
        self.fail("not yet implemented")

    def mapi_resolve(self):
        self.fail("not yet implemented")

    def mapi_embedded_init(self):
        self.fail("not yet implemented")

    def mapi_error(self):
        self.fail("not yet implemented")

    def mapi_error_str(self):
        self.fail("not yet implemented")

    def mapi_noexplain(self):
        self.fail("not yet implemented")

    def mapi_explain(self):
        self.fail("not yet implemented")

    def mapi_explain_query(self):
        self.fail("not yet implemented")

    def mapi_explain_result(self):
        self.fail("not yet implemented")

    def mapi_output(self):
        self.fail("not yet implemented")

{'fraction_non_alphanumeric': 0.1097645429362881, 'fraction_numerical': 0.004732225300092336, 'mean_word_length': 2.6607520067596115, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 5, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '1773363', 'n_tokens_mistral': 2663, 'n_tokens_neox': 2566, 'n_words': 554}