<commit_before># -*- coding: utf-8 -*-

import sys
import unittest
if sys.version_info < (3, 5):
    raise unittest.SkipTest('Python 3.5+ is required to run the async API tests.')
import os
import asyncio

from mygeotab.ext.async import API, run, from_credentials, server_call


class AsyncTestCase(unittest.TestCase):
    def __init__(self, methodName='runTest', loop=None):
        self.loop = loop or asyncio.get_event_loop()
        self._function_cache = {}
        super().__init__(methodName=methodName)


class TestAsyncCallApi(AsyncTestCase):
    def setUp(self):
        self.username = os.environ.get('MYGEOTAB_USERNAME')
        self.password = os.environ.get('<PASSWORD>')
        self.database = os.environ.get('MYGEOTAB_DATABASE')
        if self.username and self.password:
            self.api = API(self.username, password=<PASSWORD>, database=self.database, loop=self.loop, verify=False)
            self.api.authenticate()
        else:
            self.skipTest(
                'Can\'t make calls to the API without the MYGEOTAB_USERNAME and MYGEOTAB_PASSWORD environment '
                'variables being set')

    def test_get_version(self):
        version = run(self.api.call_async('GetVersion'), loop=self.loop)
        self.assertEqual(len(version), 1)
        version_split = version[0].split('.')
        self.assertEqual(len(version_split), 4)

    def test_get_user(self):
        user = run(self.api.get_async('User', name='{0}'.format(self.username)), loop=self.loop)
        self.assertEqual(len(user), 1)
        self.assertEqual(len(user[0]), 1)
        user = user[0][0]
        self.assertEqual(user['name'], self.username)

    def test_get_user_search(self):
        user = run(self.api.search_async('User', name='{0}'.format(self.username)), loop=self.loop)
        self.assertEqual(len(user), 1)
        self.assertEqual(len(user[0]), 1)
        user = user[0][0]
        self.assertEqual(user['name'], self.username)

    def test_multi_call(self):
        calls = [
            ['Get', dict(typeName='User', search=dict(name='{0}'.format(self.username)))],
            ['GetVersion']
        ]
        results = run(self.api.multi_call_async(calls), loop=self.loop)
        self.assertEqual(len(results), 1)
        results = results[0]
        self.assertEqual(len(results), 2)
        self.assertIsNotNone(results[0])
        self.assertEqual(len(results[0]), 1)
        self.assertIsNotNone(results[0][0]['name'], self.username)
        self.assertIsNotNone(results[1])
        version_split = results[1].split('.')
        self.assertEqual(len(version_split), 4)

    def test_pythonic_parameters(self):
        users = self.api.get('User')
        count_users = run(self.api.call_async('Get', type_name='User'), loop=self.loop)
        self.assertEqual(len(count_users), 1)
        self.assertGreaterEqual(len(count_users[0]), 1)
        self.assertEqual(len(count_users[0]), len(users))

    def test_api_from_credentials(self):
        new_api = from_credentials(self.api.credentials, loop=self.loop)
        users = run(new_api.get_async('User'), loop=self.loop)
        self.assertGreaterEqual(len(users), 1)

    def test_results_limit(self):
        users = run(self.api.get_async('User', resultsLimit=1), loop=self.loop)
        self.assertEqual(len(users), 1)

    def test_session_expired(self):
        credentials = self.api.credentials
        credentials.password = <PASSWORD>
        credentials.session_id = 'abc123'
        test_api = from_credentials(credentials, loop=self.loop)
        users = run(test_api.get_async('User'), loop=self.loop)
        self.assertGreaterEqual(len(users), 1)


class TestAsyncServerCallApi(AsyncTestCase):
    def test_get_version(self):
        results = run(server_call('GetVersion', server='my3.geotab.com', loop=self.loop), loop=self.loop)
        version_split = results[0].split('.')
        self.assertEqual(len(version_split), 4)

    def test_invalid_server_call(self):
        with self.assertRaises(Exception) as cm1:
            run(server_call(None, None, loop=self.loop), loop=self.loop)
        with self.assertRaises(Exception) as cm2:
            run(server_call('GetVersion', None, loop=self.loop), loop=self.loop)
        self.assertTrue('method' in str(cm1.exception))
        self.assertTrue('server' in str(cm2.exception))


if __name__ == '__main__':
    unittest.main()
<commit_msg>Verify all on async tests<commit_after># -*- coding: utf-8 -*-

import sys
import unittest
if sys.version_info < (3, 5):
    raise unittest.SkipTest('Python 3.5+ is required to run the async API tests.')
import os
import asyncio

from mygeotab.ext.async import API, run, from_credentials, server_call


class AsyncTestCase(unittest.TestCase):
    def __init__(self, methodName='runTest', loop=None):
        self.loop = loop or asyncio.get_event_loop()
        self._function_cache = {}
        super().__init__(methodName=methodName)


class TestAsyncCallApi(AsyncTestCase):
    def setUp(self):
        self.username = os.environ.get('MYGEOTAB_USERNAME')
        self.password = os.environ.get('MYGEOTAB_PASSWORD')
        self.database = os.environ.get('MYGEOTAB_DATABASE')
        if self.username and self.password:
            self.api = API(self.username, password=<PASSWORD>, database=self.database, loop=self.loop, verify=True)
            self.api.authenticate()
        else:
            self.skipTest(
                'Can\'t make calls to the API without the MYGEOTAB_USERNAME and MYGEOTAB_PASSWORD environment '
                'variables being set')

    def test_get_version(self):
        version = run(self.api.call_async('GetVersion'), loop=self.loop)
        self.assertEqual(len(version), 1)
        version_split = version[0].split('.')
        self.assertEqual(len(version_split), 4)

    def test_get_user(self):
        user = run(self.api.get_async('User', name='{0}'.format(self.username)), loop=self.loop)
        self.assertEqual(len(user), 1)
        self.assertEqual(len(user[0]), 1)
        user = user[0][0]
        self.assertEqual(user['name'], self.username)

    def test_get_user_search(self):
        user = run(self.api.search_async('User', name='{0}'.format(self.username)), loop=self.loop)
        self.assertEqual(len(user), 1)
        self.assertEqual(len(user[0]), 1)
        user = user[0][0]
        self.assertEqual(user['name'], self.username)

    def test_multi_call(self):
        calls = [
            ['Get', dict(typeName='User', search=dict(name='{0}'.format(self.username)))],
            ['GetVersion']
        ]
        results = run(self.api.multi_call_async(calls), loop=self.loop)
        self.assertEqual(len(results), 1)
        results = results[0]
        self.assertEqual(len(results), 2)
        self.assertIsNotNone(results[0])
        self.assertEqual(len(results[0]), 1)
        self.assertIsNotNone(results[0][0]['name'], self.username)
        self.assertIsNotNone(results[1])
        version_split = results[1].split('.')
        self.assertEqual(len(version_split), 4)

    def test_pythonic_parameters(self):
        users = self.api.get('User')
        count_users = run(self.api.call_async('Get', type_name='User'), loop=self.loop)
        self.assertEqual(len(count_users), 1)
        self.assertGreaterEqual(len(count_users[0]), 1)
        self.assertEqual(len(count_users[0]), len(users))

    def test_api_from_credentials(self):
        new_api = from_credentials(self.api.credentials, loop=self.loop)
        users = run(new_api.get_async('User'), loop=self.loop)
        self.assertGreaterEqual(len(users), 1)

    def test_results_limit(self):
        users = run(self.api.get_async('User', resultsLimit=1), loop=self.loop)
        self.assertEqual(len(users), 1)

    def test_session_expired(self):
        credentials = self.api.credentials
        credentials.password = <PASSWORD>
        credentials.session_id = 'abc123'
        test_api = from_credentials(credentials, loop=self.loop)
        users = run(test_api.get_async('User'), loop=self.loop)
        self.assertGreaterEqual(len(users), 1)


class TestAsyncServerCallApi(AsyncTestCase):
    def test_get_version(self):
        results = run(server_call('GetVersion', server='my3.geotab.com', loop=self.loop), loop=self.loop)
        version_split = results[0].split('.')
        self.assertEqual(len(version_split), 4)

    def test_invalid_server_call(self):
        with self.assertRaises(Exception) as cm1:
            run(server_call(None, None, loop=self.loop), loop=self.loop)
        with self.assertRaises(Exception) as cm2:
            run(server_call('GetVersion', None, loop=self.loop), loop=self.loop)
        self.assertTrue('method' in str(cm1.exception))
        self.assertTrue('server' in str(cm2.exception))


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

{'fraction_non_alphanumeric': 0.13768115942028986, 'fraction_numerical': 0.011322463768115942, 'mean_word_length': 3.240518482957273, 'pattern_counts': {'":': 0, '<': 10, '<?xml version=': 0, '>': 8, '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': '321569', 'n_tokens_mistral': 2807, 'n_tokens_neox': 2761, 'n_words': 478}