<commit_before>#!/usr/bin/env python
# Copyright (c) 2015 <NAME>
# See the LICENSE.txt file.

import tempfile
import os
import unittest
import subprocess
import contextlib
import signal

import six
from spm import run, pipe, empty_environ


class TempFileMixin(object):
    def setUp(self):
        super(TempFileMixin, self).setUp()
        self._tempfiles = []

    def get_temp_filename(self):
        fd, fname = tempfile.mkstemp()
        os.close(fd)
        return fname

    def tearDown(self):
        for fname in self._tempfiles:
            os.remove(fname)


class DeadLockMixin(object):
    @contextlib.contextmanager
    def assertDoesNotHang(self, timeout=2):

        def handler(signum, frame):
            assert False, "Hanged for more than {} seconds".format(timeout)

        signal.signal(signal.SIGALRM, handler)
        try:
            signal.alarm(timeout)
            yield
        finally:
            signal.signal(signal.SIGALRM, signal.SIG_DFL)


class RunTest(TempFileMixin, unittest.TestCase):
    def test_stdin_from_file(self):
        content = '__file_content__'

        fname = self.get_temp_filename()
        with open(fname, 'w') as file_:
            file_.write(content)

        cat = run('cat')
        cat.stdin = open(fname)

        assert cat.stdout.read().decode() == content

    def test_stdout_to_file(self):
        string = '__output__'

        fname = self.get_temp_filename()

        echo = run('echo', '-n', string)
        echo.stdout = open(fname, 'w')
        out, err = echo.wait()

        assert out is None

        with open(fname) as file_:
            assert six.u(file_.read()) == string

    def test_environment(self):
        env = run('env', env={'FOO': 'BAR'}).stdout.read().decode().split('\n')

        assert 'FOO=BAR' in env

    def test_empty_environment(self):
        env = run('env', env=empty_environ()).stdout.read().decode()

        assert env == ''

    def test_repr(self):
        """
        A user should be able to run str(Subprocess) in their shell prompt.
        """
        cmd_str = str(run('echo', '-n', 'foo"bar'))
        output = subprocess.check_output(cmd_str, shell=True).decode()

        assert output == 'foo"bar'

    def test_repr_env(self):
        cmd_str = str(run('env', env={'FOO': 'BAR'}))
        env = subprocess.check_output(cmd_str, shell=True).decode().split('\n')

        assert 'FOO=BAR' in env

    def test_empty_env(self):
        proc = run('env', env=empty_environ(foo='bar'))

        spm_run = set(proc.stdout.read().decode().split('\n'))
        sh_run = set(
            subprocess.check_output(str(proc), shell=True).decode().split('\n')
        )

        assert sh_run == spm_run

    def test_subprocess_failure(self):
        with self.assertRaises(subprocess.CalledProcessError):
            run('false').wait()

    def test_environement_on_pipe(self):
        proc = pipe(['env'], ['egrep', '^FOO='], env={'FOO': 'BAR'})

        assert proc.stdout.read().decode() == 'FOO=BAR\n'

    def test_pass_subprocess_to_pipe(self):
        proc = pipe(['echo', '-n', 'hello'], ['gzip'], run('zcat'))
        assert proc.stdout.read().decode() == 'hello'


class PipeTest(DeadLockMixin, TempFileMixin, unittest.TestCase):
    def test_stdin_from_file(self):
        content = '__file_content__'

        fname = self.get_temp_filename()
        with open(fname, 'w') as file_:
            file_.write(content)

        cat = run('gzip').pipe('zcat')
        cat.stdin = open(fname)

        assert cat.stdout.read().decode() == content

    def test_stdout_to_file(self):
        string = '__output__'
        fname = self.get_temp_filename()

        echo = run('echo', '-n', string)
        echo.stdout = open(fname, 'w')
        out, err = echo.wait()

        assert out is None

        with open(fname) as file_:
            assert six.u(file_.read()) == string
<commit_msg>Use printf for more consistent behavior accross platforms<commit_after>#!/usr/bin/env python
# Copyright (c) 2015 <NAME>
# See the LICENSE.txt file.

import tempfile
import os
import unittest
import subprocess
import contextlib
import signal

import six
from spm import run, pipe, empty_environ


class TempFileMixin(object):
    def setUp(self):
        super(TempFileMixin, self).setUp()
        self._tempfiles = []

    def get_temp_filename(self):
        fd, fname = tempfile.mkstemp()
        os.close(fd)
        return fname

    def tearDown(self):
        for fname in self._tempfiles:
            os.remove(fname)


class DeadLockMixin(object):
    @contextlib.contextmanager
    def assertDoesNotHang(self, timeout=2):

        def handler(signum, frame):
            assert False, "Hanged for more than {} seconds".format(timeout)

        signal.signal(signal.SIGALRM, handler)
        try:
            signal.alarm(timeout)
            yield
        finally:
            signal.signal(signal.SIGALRM, signal.SIG_DFL)


class RunTest(TempFileMixin, unittest.TestCase):
    def test_stdin_from_file(self):
        content = '__file_content__'

        fname = self.get_temp_filename()
        with open(fname, 'w') as file_:
            file_.write(content)

        cat = run('cat')
        cat.stdin = open(fname)

        assert cat.stdout.read().decode() == content

    def test_stdout_to_file(self):
        string = '__output__'

        fname = self.get_temp_filename()

        echo = run('printf', string)
        echo.stdout = open(fname, 'w')
        out, err = echo.wait()

        assert out is None

        with open(fname) as file_:
            assert six.u(file_.read()) == string

    def test_environment(self):
        env = run('env', env={'FOO': 'BAR'}).stdout.read().decode().split('\n')

        assert 'FOO=BAR' in env

    def test_empty_environment(self):
        env = run('env', env=empty_environ()).stdout.read().decode()

        assert env == ''

    def test_repr(self):
        """
        A user should be able to run str(Subprocess) in their shell prompt.
        """
        cmd_str = str(run('printf', '%s\n', 'foo"bar'))
        output = subprocess.check_output(cmd_str, shell=True).decode()

        assert output == 'foo"bar\n'

    def test_repr_env(self):
        cmd_str = str(run('env', env={'FOO': 'BAR'}))
        env = subprocess.check_output(cmd_str, shell=True).decode().split('\n')

        assert 'FOO=BAR' in env

    def test_empty_env(self):
        proc = run('env', env=empty_environ(foo='bar'))

        spm_run = set(proc.stdout.read().decode().split('\n'))
        sh_run = set(
            subprocess.check_output(str(proc), shell=True).decode().split('\n')
        )

        assert sh_run == spm_run

    def test_subprocess_failure(self):
        with self.assertRaises(subprocess.CalledProcessError):
            run('false').wait()

    def test_environement_on_pipe(self):
        proc = pipe(['env'], ['egrep', '^FOO='], env={'FOO': 'BAR'})

        assert proc.stdout.read().decode() == 'FOO=BAR\n'

    def test_pass_subprocess_to_pipe(self):
        proc = pipe(['printf', 'hello'], ['gzip'], run('zcat'))
        assert proc.stdout.read().decode() == 'hello'


class PipeTest(DeadLockMixin, TempFileMixin, unittest.TestCase):
    def test_stdin_from_file(self):
        content = '__file_content__'

        fname = self.get_temp_filename()
        with open(fname, 'w') as file_:
            file_.write(content)

        cat = run('gzip').pipe('zcat')
        cat.stdin = open(fname)

        assert cat.stdout.read().decode() == content

    def test_stdout_to_file(self):
        string = '__output__'
        fname = self.get_temp_filename()

        echo = run('printf', string)
        echo.stdout = open(fname, 'w')
        out, err = echo.wait()

        assert out is None

        with open(fname) as file_:
            assert six.u(file_.read()) == string

{'fraction_non_alphanumeric': 0.13795731707317074, 'fraction_numerical': 0.0012703252032520325, 'mean_word_length': 2.5225950782997764, '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': '7604938', 'n_tokens_mistral': 2538, 'n_tokens_neox': 2455, 'n_words': 544}