self_tests

#!/usr/bin/env python
# coding: utf-8

from __future__ import print_function

import contextlib
import os
import subprocess
import sys
import threading
import time
import unittest

import getaddons
import run_pylint
import travis_helpers
from test_server import main as test_server_main
from test_server import get_test_dependencies

try:
    import xmlrpc.client as xmlrpclib
except ImportError:
    import xmlrpclib

try:
    from StringIO import StringIO
except ImportError:
    from io import StringIO

PY3K = sys.version_info[0] == 3


EXPECTED_ERRORS = {
    'anomalous-backslash-in-string': 1,
    'assignment-from-none': 1,
    'dangerous-default-value': 1,
    'duplicate-key': 2,
    'pointless-statement': 1,
    'pointless-string-statement': 1,
    'print-used': 1,
    'redundant-keyword-arg': 1,
    'reimported': 4,
    'return-in-init': 1,
    'rst-syntax-error': 1,
    'too-few-format-args': 1,
    'unreachable': 1,
}
EXPECTED_ERRORS_PR = {
    'eval-referenced': 6,
    'eval-used': 1,
    'license-allowed': 1,
    'manifest-required-author': 1,
    'manifest-required-key': 1,
    'manifest-version-format': 4,
}

if PY3K:
    # python3 uses another pylint version
    # and pylint change these expected errors
    EXPECTED_ERRORS.update({
        'assignment-from-none': 2,
    })

@contextlib.contextmanager
def _patch_streams(out):
    sys.stderr = sys.stdout = out
    try:
        yield
    finally:
        sys.stderr = sys.__stderr__
        sys.stdout = sys.__stdout__


class TestServerThread(threading.Thread):
    def run(self):
        test_server_main()


class MainTest(unittest.TestCase):
    """PYLINT TEST CASES """
    def setUp(self):
        super(MainTest, self).setUp()
        self.repo_dir = os.environ.get("TRAVIS_BUILD_DIR", ".")
        self.repo_dir_with_subfolders = os.path.join(
            self.repo_dir, "tests", "test_repo_with_subfolders")
        self.exclude = os.environ.get("EXCLUDE")
        self.addons_list = ['test_module', 'second_module']
        self.to_preinstall = get_test_dependencies(
            self.repo_dir, self.addons_list)

        self.git_work_dir = os.environ.get('TRAVIS_BUILD_DIR', False)
        self.pylint_rcfile = os.path.join(
            os.path.dirname(os.path.realpath(__file__)),
            'cfg', "travis_run_pylint.cfg")
        self.pylint_rcfile_pr = os.path.join(
            os.path.dirname(os.path.realpath(__file__)),
            'cfg', "travis_run_pylint_pr.cfg")
        self.modules_cmd = run_pylint.get_modules_cmd(self.git_work_dir)
        self.beta_msgs = run_pylint.get_beta_msgs()
        self.errors_dict = EXPECTED_ERRORS.copy()
        self.maxDiff = None

    def test_modules_to_preinstall(self):
        res = [x for x in self.to_preinstall if x in self.addons_list]
        self.assertFalse(res, "Should not preinstall modules to test!")

    def test_get_addons(self):
        self.assertEquals(getaddons.main(), 1)
        self.assertEquals(
            getaddons.main(["getaddons.py", self.repo_dir]), [self.repo_dir])
        self.assertEquals(
            len(getaddons.main(["getaddons.py", "-m", self.repo_dir])), 6)
        self.assertEquals(
            len(getaddons.main(
                ["getaddons.py", "-m", self.repo_dir + "/" if
                 self.repo_dir[-1] == '/' else self.repo_dir])), 6)
        self.assertEquals(
            len(getaddons.main(
                ["getaddons.py", "-m", "--only-applications",
                 self.repo_dir])), 0)
        self.assertEquals(
            len(getaddons.main(
                ["getaddons.py", "-m", "--only-localization",
                 self.repo_dir])), 0)
        self.assertEquals(
            len(getaddons.main(
                ["getaddons.py", "-m", "--exclude-applications",
                 "--exclude-localization", self.repo_dir])), 6)

    @unittest.skipIf(os.environ.get("EXCLUDE", False) is False, "Set EXCLUDE")
    def test_get_addons_exclude(self):
        self.assertEquals(
            len(getaddons.main(
                ["getaddons.py", "-m", "-e", self.exclude, self.repo_dir])), 3)
        self.assertEquals(
            len(getaddons.main(
                ["getaddons.py", "-m", self.repo_dir, "-e", self.exclude])), 6)
        self.assertEquals(
            getaddons.main(
                ["getaddons.py", "-e", self.exclude, self.repo_dir]),
            [self.repo_dir])

    def test_addons_path_order(self):
        addon_paths_alfanumerical_order_is = getaddons.get_addons(
            self.repo_dir_with_subfolders)
        addon_paths_alfanumerical_order_should = [
            '1_testfolder', '2_testfolder']
        combined = []
        for i, item in enumerate(addon_paths_alfanumerical_order_should):
            combined.append([addon_paths_alfanumerical_order_is[i], item])
        print(combined.__repr__())
        self.assertEqual(
            len(combined), len(addon_paths_alfanumerical_order_is))
        for ist, should in combined:
            self.assertTrue(ist.rstrip('/').rstrip('\\').endswith(should))

    def test_travis_helpers(self):
        self.assertEqual(travis_helpers.red(u'test'),
                         u"\033[1;31mtest\033[0;m")
        self.assertEqual(travis_helpers.green(u'test'),
                         u"\033[1;32mtest\033[0;m")
        self.assertEqual(travis_helpers.yellow(u'test'),
                         u"\033[1;33mtest\033[0;m")
        self.assertEqual(travis_helpers.yellow_light(u'test'),
                         u"\033[33mtest\033[0;m")

        self.assertEqual(
            travis_helpers.red(u'\ntest\nnewline'),
            u"\033[1;31m\033[0;m\n\033[1;31mtest\033"
            "[0;m\n\033[1;31mnewline\033[0;m")
        self.assertEqual(
            travis_helpers.green(u'\ntest\nnewline'),
            u"\033[1;32m\033[0;m\n\033[1;32mtest\033"
            "[0;m\n\033[1;32mnewline\033[0;m")
        self.assertEqual(
            travis_helpers.yellow(u'\ntest\nnewline'),
            u"\033[1;33m\033[0;m\n\033[1;33mtest\033"
            "[0;m\n\033[1;33mnewline\033[0;m")
        self.assertEqual(
            travis_helpers.yellow_light(u'\ntest\nnewline'),
            u"\033[33m\033[0;m\n\033[33mtest\033"
            "[0;m\n\033[33mnewline\033[0;m")

    @unittest.skipIf(os.environ.get('LINT_CHECK', 0) != '1', "Set LINT_CHECK")
    def test_pylint_check(self):
        """Testing empty paths and pylint_run fix of:
           https://www.mail-archive.com/code-quality@python.org/msg00294.html
        """
        pylint_rcfile = os.path.join(
            os.path.dirname(os.path.realpath(__file__)),
            'cfg', "travis_run_pylint.cfg")
        with _patch_streams(StringIO()):
            stats = run_pylint.main([
                "--config-file=" + pylint_rcfile,
                "--extra-params", "-d", "--extra-params",
                "all", "--extra-params", "-e",
                "--extra-params", "F0010,duplicate-key",
                "--path", self.repo_dir], standalone_mode=False)
        count_errors = run_pylint.get_count_fails(stats, list())
        self.assertEqual(2, count_errors)
        empty_path = os.path.join(self.repo_dir, 'empty_path')
        if not os.path.exists(empty_path):
            os.mkdir(empty_path)
        count_errors = run_pylint.main([
            "--config-file=" + pylint_rcfile,
            "--path", empty_path], standalone_mode=False)
        self.assertEqual(-1, sum(count_errors.values()))
        if os.environ.get('TRAVIS_PULL_REQUEST', 'false') == 'true':
            git_script_path = os.path.join(os.path.dirname(
                os.path.dirname(os.path.realpath(__file__))), 'git')
            pre_commit_returned = subprocess.call(os.path.join(
                git_script_path, 'pre-commit'))
            assert pre_commit_returned == 0, \
                "Git pre-commit script returned value != 0"

    def test_get_modules_changed(self):
        """Testing git run from getaddons"""
        self.assertIsNotNone(
            getaddons.get_modules_changed(self.repo_dir))

    def connection_test(self):
        username = "admin"
        password = "admin"
        database_name = "openerp_test"
        port = 8069
        host = '127.0.0.1'
        sock_common = xmlrpclib.ServerProxy(
            "http://%s:%d/xmlrpc/common" % (host, port))
        uid = sock_common.login(
            database_name, username, password)
        sock = xmlrpclib.ServerProxy(
            "http://%s:%d/xmlrpc/object" % (host, port))
        user_ids = sock.execute(
            database_name, uid, password, 'res.users',
            'search', [('login', '=', 'admin')])
        return user_ids

    @unittest.skipIf(
        (os.environ.get('INSTANCE_ALIVE', 0) != '1' or
         os.environ.get('TESTS', 0) != '1'), "Set LINT_CHECK")
    def test_instance_running(self):
        """Testing instance running"""
        thread = TestServerThread()
        thread.daemon = True
        thread.start()
        for __ in range(10):
            try:
                connection_result = self.connection_test()
            except:
                connection_result = None
            if connection_result:
                break
            time.sleep(2)
        self.assertNotEqual(None, connection_result, "Connection test failed")
        # On TravisCI tests, force stopping the alive server
        odoo_pid = open('/tmp/odoo.pid').read()
        subprocess.call(['kill', '-9', odoo_pid])

    @unittest.skipIf(os.environ.get('LINT_CHECK', 0) != '1', "Set LINT_CHECK")
    def test_check_ispr_without_changes(self):
        git_work_dir = os.path.join(self.git_work_dir, "tests", "test_repo")
        with _patch_streams(StringIO()):
            result = run_pylint.pylint_run(
                is_pr=True, version="7.0", dir=git_work_dir)
        # Expected vs found errors
        self.assertEqual(self.errors_dict, result)

    @unittest.skipIf(os.environ.get('LINT_CHECK', 0) != '1', "Set LINT_CHECK")
    def test_check_vmaster_nopr(self):
        self.errors_dict.update({
            'missing-manifest-dependency': 2,
            'missing-import-error': 2,
        })
        with _patch_streams(StringIO()):
            result = run_pylint.pylint_run(
                is_pr=False, version="master", dir=self.git_work_dir)

        # Expected vs found errors
        self.assertEqual(self.errors_dict, result)

    @unittest.skipIf(os.environ.get('LINT_CHECK', 0) != '1', "Set LINT_CHECK")
    def test_check_vmaster_ispr(self):
        self.errors_dict.update(EXPECTED_ERRORS_PR)
        self.errors_dict.update({
            'missing-manifest-dependency': 2,
            'missing-import-error': 2,
        })

        with _patch_streams(StringIO()):
            result = run_pylint.pylint_run(
                is_pr=True, version="master", dir=self.git_work_dir)
        # Expected vs found errors
        self.assertEqual(self.errors_dict, result)

    @unittest.skipIf(os.environ.get('LINT_CHECK', 0) != '1', "Set LINT_CHECK")
    def test_check_v7_ispr(self):
        self.errors_dict.update(EXPECTED_ERRORS_PR)
        self.errors_dict.update({
            'manifest-version-format': 6,
        })

        with _patch_streams(StringIO()):
            result = run_pylint.pylint_run(
                is_pr=True, version="7.0", dir=self.git_work_dir)
        # Expected vs found errors
        self.assertEqual(self.errors_dict, result)

    @unittest.skipIf(os.environ.get('LINT_CHECK', 0) != '1', "Set LINT_CHECK")
    def test_check_v7_nopr(self):
        with _patch_streams(StringIO()):
            result = run_pylint.pylint_run(
                is_pr=False, version="7.0", dir=self.git_work_dir)
        # Expected vs found errors
        self.assertEqual(self.errors_dict, result)

    @unittest.skipIf(os.environ.get('LINT_CHECK', 0) != '1', "Set LINT_CHECK")
    def test_check_v8_ispr(self):
        self.errors_dict.update(EXPECTED_ERRORS_PR)
        self.errors_dict.update({
            'missing-manifest-dependency': 2,
            'missing-import-error': 2,
        })

        with _patch_streams(StringIO()):
            result = run_pylint.pylint_run(
                is_pr=True, version="8.0", dir=self.git_work_dir)
        # Expected vs found errors
        self.assertEqual(self.errors_dict, result)

    @unittest.skipIf(os.environ.get('LINT_CHECK', 0) != '1', "Set LINT_CHECK")
    def test_check_v8_nopr(self):
        self.errors_dict.update({
            'missing-manifest-dependency': 2,
            'missing-import-error': 2,
        })
        with _patch_streams(StringIO()):
            result = run_pylint.pylint_run(
                is_pr=False, version="8.0", dir=self.git_work_dir)
        # Expected vs found errors
        self.assertEqual(self.errors_dict, result)

    @unittest.skipIf(os.environ.get('LINT_CHECK', 0) != '1', "Set LINT_CHECK")
    def test_check_without_version_ispr(self):
        self.errors_dict.update(EXPECTED_ERRORS_PR)
        self.errors_dict.update({
            'missing-manifest-dependency': 2,
            'missing-import-error': 2,
        })
        with _patch_streams(StringIO()):
            result = run_pylint.pylint_run(
                is_pr=True, version="", dir=self.git_work_dir)
        # Expected vs found errors
        self.assertEqual(self.errors_dict, result)

    @unittest.skipIf(os.environ.get('LINT_CHECK', 0) != '1', "Set LINT_CHECK")
    def test_check_without_version_nopr(self):
        self.errors_dict.update({
            'missing-manifest-dependency': 2,
            'missing-import-error': 2,
        })
        with _patch_streams(StringIO()):
            result = run_pylint.pylint_run(
                is_pr=False, version="", dir=self.git_work_dir)
        # Expected vs found errors
        self.assertEqual(self.errors_dict, result)


if __name__ == '__main__':
    suite = unittest.TestLoader().loadTestsFromTestCase(MainTest)
    result = unittest.TextTestRunner(verbosity=2).run(suite)
    sys.exit(not result.wasSuccessful())