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

# Copyright 2015-2017 <NAME> <<EMAIL>>

# This file is part of toxicbuild.

# toxicbuild is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# toxicbuild is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.

# You should have received a copy of the GNU General Public License
# along with toxicbuild. If not, see <http://www.gnu.org/licenses/>.


import asyncio
import datetime
from concurrent import futures
import os
import subprocess
import time
from unittest import TestCase
from unittest.mock import patch, Mock, MagicMock
from toxicbuild.core import utils
from tests.unit.core import TEST_DATA_DIR
from tests import async_test


class UtilsTest(TestCase):

    @async_test
    def test_exec_cmd(self):
        out = yield from utils.exec_cmd('ls', cwd='.')
        self.assertTrue(out)

    @async_test
    def test_exec_cmd_with_error(self):
        with self.assertRaises(utils.ExecCmdError):
            # please, don't tell me you have a lsz command on your system.
            yield from utils.exec_cmd('lsz', cwd='.')

    @async_test
    def test_exec_cmd_with_timeout(self, *args, **kwargs):
        with self.assertRaises(asyncio.TimeoutError):
            yield from utils.exec_cmd('sleep 2', cwd='.', timeout=1)

        # wait here to avoid runtime error saying the loop is closed
        # when the process try to send its message to the caller
        time.sleep(1)

    @async_test
    def test_kill_group(self):
        proc = yield from utils._create_cmd_proc('sleep 2', cwd='.')
        try:
            f = proc.stdout.readline()
            yield from asyncio.wait_for(f, 1)
        except futures.TimeoutError:
            pass

        utils._kill_group(proc)
        procs = subprocess.check_output(['ps', 'aux']).decode()
        self.assertNotIn('sleep 2', procs)

    @async_test
    def test_exec_cmd_with_envvars(self):
        envvars = {'PATH': 'PATH:venv/bin',
                   'MYPROGRAMVAR': 'something'}

        cmd = 'echo $MYPROGRAMVAR'

        returned = yield from utils.exec_cmd(cmd, cwd='.', **envvars)

        self.assertEqual(returned, 'something')

    @async_test
    def test_exec_cmd_with_out_fn(self):
        envvars = {'PATH': 'PATH:venv/bin',
                   'MYPROGRAMVAR': 'something'}<commit_msg>correcting tests<commit_after># -*- coding: utf-8 -*-

# Copyright 2015-2017 <NAME> <<EMAIL>>

# This file is part of toxicbuild.

# toxicbuild is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.

# toxicbuild is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
# GNU General Public License for more details.

# You should have received a copy of the GNU General Public License
# along with toxicbuild. If not, see <http://www.gnu.org/licenses/>.


import asyncio
import datetime
from concurrent import futures
import os
import subprocess
import sys
import time
from unittest import TestCase
from unittest.mock import patch, Mock, MagicMock
from toxicbuild.core import utils
from tests.unit.core import TEST_DATA_DIR
from tests import async_test


class UtilsTest(TestCase):

    @async_test
    def test_exec_cmd(self):
        out = yield from utils.exec_cmd('ls', cwd='.')
        self.assertTrue(out)

    @async_test
    def test_exec_cmd_with_error(self):
        with self.assertRaises(utils.ExecCmdError):
            # please, don't tell me you have a lsz command on your system.
            yield from utils.exec_cmd('lsz', cwd='.')

    @async_test
    def test_exec_cmd_with_timeout(self, *args, **kwargs):
        with self.assertRaises(asyncio.TimeoutError):
            yield from utils.exec_cmd('sleep 2', cwd='.', timeout=1)

        # wait here to avoid runtime error saying the loop is closed
        # when the process try to send its message to the caller
        time.sleep(1)

    @async_test
    def test_kill_group(self):
        wait = sys.version_info.minor
        cmd = 'sleep {}'.format(wait)
        proc = yield from utils._create_cmd_proc(cmd, cwd='.')
        try:
            f = proc.stdout.readline()
            yield from asyncio.wait_for(f, 1)
        except futures.TimeoutError:
            pass

        utils._kill_group(proc)
        procs = subprocess.check_output(['ps', 'aux']).decode()
        self.assertNotIn(cmd, procs)

    @async_test
    def test_exec_cmd_with_envvars(self):
        envvars = {'PATH': 'PATH:venv/bin',
                   'MYPROGRAMVAR': 'something'}

        cmd = 'echo $MYPROGRAMVAR'

        returned = yield from utils.exec_cmd(cmd, cwd='.', **envvars)

        self.assertEqual(returned, 'something')

    @async_test
    def test_exec_cmd_with_out_fn(self):
        envvars = {'PATH': 'PATH:venv/bin',
                   'MYPROGRAMVAR': 'something'}
{'fraction_non_alphanumeric': 0.09321719792438844, 'fraction_numerical': 0.005559673832468495, 'mean_word_length': 2.980088495575221, 'pattern_counts': {'":': 0, '<': 11, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, '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': '5299238', 'n_tokens_mistral': 1602, 'n_tokens_neox': 1548, 'n_words': 537}