<commit_before>from __future__ import absolute_import

from copy import deepcopy
import os
import pytest
import stat
import subprocess

from project.internal.test.tmpfile_utils import with_directory_contents
from project.internal.crypto import decrypt_string
from project.internal.prepare_ui import NotInteractivePrepareUI
from project.prepare import prepare, unprepare, UI_MODE_BROWSER, prepare_in_stages, _after_stage_success
from project.project import Project
from project.project_file import PROJECT_FILENAME
from project.local_state_file import LocalStateFile


def test_prepare_empty_directory():
    def prepare_empty(dirname):
        project = Project(dirname)
        environ = dict()
        result = prepare(project, environ=environ)
        assert result
        assert dict(PROJECT_DIR=project.directory_path) == result.environ
        assert dict() == environ

    with_directory_contents(dict(), prepare_empty)


def test_prepare_bad_ui_mode():
    def prepare_bad_ui_mode(dirname):
        with pytest.raises(ValueError) as excinfo:
            project = Project(dirname)
            environ = dict()
            prepare(project, ui_mode="BAD_UI_MODE", environ=environ)
        assert "invalid UI mode" in repr(excinfo.value)

    with_directory_contents(dict(), prepare_bad_ui_mode)


def test_unprepare_empty_directory():
    def unprepare_empty(dirname):
        project = Project(dirname)
        result = unprepare(project)
        assert result is None

    with_directory_contents(dict(), unprepare_empty)


def test_default_to_system_environ():
    def prepare_system_environ(dirname):
        project = Project(dirname)
        os_environ_copy = deepcopy(os.environ)
        result = prepare(project)
        assert project.directory_path == result.environ['PROJECT_DIR']
        # os.environ wasn't modified
        assert os_environ_copy == os.environ
        # result.environ inherits everything in os.environ
        for key in os_environ_copy:
            assert result.environ[key] == os.environ[key]

    with_directory_contents(dict(), prepare_system_environ)


def test_prepare_some_env_var_already_set():
    def prepare_some_env_var(dirname):
        project = Project(dirname)
        environ = dict(FOO='bar')
        result = prepare(project, environ=environ)
        assert result
        assert dict(FOO='bar', PROJECT_DIR=project.directory_path) == result.environ
        assert dict(FOO='bar') == environ

    with_directory_contents({PROJECT_FILENAME: """
runtime:
  FOO: {}
"""}, prepare_some_env_var)


def test_prepare_some_env_var_not_set():
    def prepare_some_env_var(dirname):
        project = Project(dirname)
        environ = dict(BAR='bar')
        result = prepare(project, environ=environ)
        assert not result
        assert dict(BAR='bar') == environ

    with_directory_contents({PROJECT_FILENAME: """
runtime:
  FOO: {}
"""}, prepare_some_env_var)


def test_prepare_with_app_entry():
    def prepare_with_app_entry(dirname):
        os.chmod(os.path.join(dirname, "echo.py"), stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)

        project = Project(dirname)
        environ = deepcopy(os.environ)
        environ['FOO'] = 'bar'
        environ['CONDA_DEFAULT_ENV'] = '/someplace'
        result = prepare(project, environ=environ)
        assert result

        command = result.command_exec_info
        assert 'FOO' in command.env
        assert command.cwd == project.directory_path
        assert command.args == ['%s/echo.py' % dirname, '/someplace', 'foo', 'bar']
        p = command.popen(stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        (out, err) = p.communicate()
        assert out.decode() == ("['%s/echo.py', '/someplace', 'foo', 'bar']\n" % dirname)
        assert err.decode() == ""

    with_directory_contents(
        {PROJECT_FILENAME: """
runtime:
  FOO: {}

app:
  entry: echo.py ${PREFIX} foo bar
""",
         "echo.py": """#!/usr/bin/env python
from __future__ import print_function
import sys
print(repr(sys.argv))
"""}, prepare_with_app_entry)


def test_update_environ():
    def prepare_then_update_environ(dirname):
        project = Project(dirname)
        environ = dict(FOO='bar')
        result = prepare(project, environ=environ)
        assert result

        other = dict(BAR='baz')
        result.update_environ(other)
        assert dict(FOO='bar', BAR='baz', PROJECT_DIR=dirname) == other

    with_directory_contents({PROJECT_FILENAME: """
runtime:
  FOO: {}
"""}, prepare_then_update_environ)


def test_attempt_to_grab_result_early():
    def early_result_grab(dirname):
        project = Project(dirname)
        first_stage = prepare_in_stages(project)
        with pytest.raises(RuntimeError) as excinfo:
            first_stage.result
        assert "result property isn't available" in repr(excinfo.value)

    with_directory_contents(dict(), early_result_grab)


def test_skip_after_success_function_when_stage_fails():
    def check_no_after_success_on_failed(dirname):
        project = Project(dirname)
        first_stage = prepare_in_stages(project, environ=dict())

        def after():
            raise RuntimeError("should not have been called")

        stage = _after_stage_success(first_stage, after)
        while stage is not None:
            next_stage = stage.execute(NotInteractivePrepareUI())
            result = stage.result
            if result.failed:
                assert stage.failed
                break
            else:
                assert not stage.failed
            stage = next_stage
        assert result.failed

    with_directory_contents({PROJECT_FILENAME: """
runtime:
  FOO: {}
"""}, check_no_after_success_on_failed)


def test_prepare_with_browser(monkeypatch):
    from tornado.ioloop import IOLoop
    io_loop = IOLoop()

    http_results = {}

    def mock_open_new_tab(url):
        from project.internal.test.http_utils import http_get_async, http_post_async
        from tornado import gen

        @gen.coroutine
        def do_http():
            http_results['get'] = yield http_get_async(url)
            http_results['post'] = yield http_post_async(url, body="")

        io_loop.add_callback(do_http)
<commit_msg>Rework _after_stage_success tests in test_prepare.py<commit_after>from __future__ import absolute_import

from copy import deepcopy
import os
import pytest
import stat
import subprocess

from project.internal.test.tmpfile_utils import with_directory_contents
from project.internal.crypto import decrypt_string
from project.prepare import (prepare, unprepare, UI_MODE_BROWSER, prepare_in_stages, PrepareSuccess, PrepareFailure,
                             _after_stage_success, _FunctionPrepareStage)
from project.project import Project
from project.project_file import PROJECT_FILENAME
from project.local_state_file import LocalStateFile


def test_prepare_empty_directory():
    def prepare_empty(dirname):
        project = Project(dirname)
        environ = dict()
        result = prepare(project, environ=environ)
        assert result
        assert dict(PROJECT_DIR=project.directory_path) == result.environ
        assert dict() == environ

    with_directory_contents(dict(), prepare_empty)


def test_prepare_bad_ui_mode():
    def prepare_bad_ui_mode(dirname):
        with pytest.raises(ValueError) as excinfo:
            project = Project(dirname)
            environ = dict()
            prepare(project, ui_mode="BAD_UI_MODE", environ=environ)
        assert "invalid UI mode" in repr(excinfo.value)

    with_directory_contents(dict(), prepare_bad_ui_mode)


def test_unprepare_empty_directory():
    def unprepare_empty(dirname):
        project = Project(dirname)
        result = unprepare(project)
        assert result is None

    with_directory_contents(dict(), unprepare_empty)


def test_default_to_system_environ():
    def prepare_system_environ(dirname):
        project = Project(dirname)
        os_environ_copy = deepcopy(os.environ)
        result = prepare(project)
        assert project.directory_path == result.environ['PROJECT_DIR']
        # os.environ wasn't modified
        assert os_environ_copy == os.environ
        # result.environ inherits everything in os.environ
        for key in os_environ_copy:
            assert result.environ[key] == os.environ[key]

    with_directory_contents(dict(), prepare_system_environ)


def test_prepare_some_env_var_already_set():
    def prepare_some_env_var(dirname):
        project = Project(dirname)
        environ = dict(FOO='bar')
        result = prepare(project, environ=environ)
        assert result
        assert dict(FOO='bar', PROJECT_DIR=project.directory_path) == result.environ
        assert dict(FOO='bar') == environ

    with_directory_contents({PROJECT_FILENAME: """
runtime:
  FOO: {}
"""}, prepare_some_env_var)


def test_prepare_some_env_var_not_set():
    def prepare_some_env_var(dirname):
        project = Project(dirname)
        environ = dict(BAR='bar')
        result = prepare(project, environ=environ)
        assert not result
        assert dict(BAR='bar') == environ

    with_directory_contents({PROJECT_FILENAME: """
runtime:
  FOO: {}
"""}, prepare_some_env_var)


def test_prepare_with_app_entry():
    def prepare_with_app_entry(dirname):
        os.chmod(os.path.join(dirname, "echo.py"), stat.S_IRUSR | stat.S_IWUSR | stat.S_IXUSR)

        project = Project(dirname)
        environ = deepcopy(os.environ)
        environ['FOO'] = 'bar'
        environ['CONDA_DEFAULT_ENV'] = '/someplace'
        result = prepare(project, environ=environ)
        assert result

        command = result.command_exec_info
        assert 'FOO' in command.env
        assert command.cwd == project.directory_path
        assert command.args == ['%s/echo.py' % dirname, '/someplace', 'foo', 'bar']
        p = command.popen(stdout=subprocess.PIPE, stderr=subprocess.PIPE)
        (out, err) = p.communicate()
        assert out.decode() == ("['%s/echo.py', '/someplace', 'foo', 'bar']\n" % dirname)
        assert err.decode() == ""

    with_directory_contents(
        {PROJECT_FILENAME: """
runtime:
  FOO: {}

app:
  entry: echo.py ${PREFIX} foo bar
""",
         "echo.py": """#!/usr/bin/env python
from __future__ import print_function
import sys
print(repr(sys.argv))
"""}, prepare_with_app_entry)


def test_update_environ():
    def prepare_then_update_environ(dirname):
        project = Project(dirname)
        environ = dict(FOO='bar')
        result = prepare(project, environ=environ)
        assert result

        other = dict(BAR='baz')
        result.update_environ(other)
        assert dict(FOO='bar', BAR='baz', PROJECT_DIR=dirname) == other

    with_directory_contents({PROJECT_FILENAME: """
runtime:
  FOO: {}
"""}, prepare_then_update_environ)


def test_attempt_to_grab_result_early():
    def early_result_grab(dirname):
        project = Project(dirname)
        first_stage = prepare_in_stages(project)
        with pytest.raises(RuntimeError) as excinfo:
            first_stage.result
        assert "result property isn't available" in repr(excinfo.value)

    with_directory_contents(dict(), early_result_grab)


def test_skip_after_success_function_when_second_stage_fails():
    state = {'state': 'start'}

    def do_first(stage):
        assert state['state'] == 'start'
        state['state'] = 'first'
        stage._result = PrepareSuccess(logs=[], command_exec_info=None, environ=dict())

        def last(stage):
            assert state['state'] == 'first'
            state['state'] = 'second'
            stage._result = PrepareFailure(logs=[], errors=[])
            return None

        return _FunctionPrepareStage("second", last)

    first_stage = _FunctionPrepareStage("first", do_first)

    def after():
        raise RuntimeError("should not have been called")

    stage = _after_stage_success(first_stage, after)
    while stage is not None:
        next_stage = stage.execute()
        result = stage.result
        if result.failed:
            assert stage.failed
            break
        else:
            assert not stage.failed
        stage = next_stage
    assert result.failed
    assert state['state'] == 'second'


def test_run_after_success_function_when_second_stage_succeeds():
    state = {'state': 'start'}

    def do_first(stage):
        assert state['state'] == 'start'
        state['state'] = 'first'
        stage._result = PrepareSuccess(logs=[], command_exec_info=None, environ=dict())

        def last(stage):
            assert state['state'] == 'first'
            state['state'] = 'second'
            stage._result = PrepareSuccess(logs=[], command_exec_info=None, environ=dict())
            return None

        return _FunctionPrepareStage("second", last)

    first_stage = _FunctionPrepareStage("first", do_first)

    def after():
        assert state['state'] == 'second'
        state['state'] = 'after'

    stage = _after_stage_success(first_stage, after)
    while stage is not None:
        next_stage = stage.execute()
        result = stage.result
        if result.failed:
            assert stage.failed
            break
        else:
            assert not stage.failed
        stage = next_stage
    assert not result.failed
    assert state['state'] == 'after'


def test_prepare_with_browser(monkeypatch):
    from tornado.ioloop import IOLoop
    io_loop = IOLoop()

    http_results = {}

    def mock_open_new_tab(url):
        from project.internal.test.http_utils import http_get_async, http_post_async
        from tornado import gen

        @gen.coroutine
        def do_http():
            http_results['get'] = yield http_get_async(url)
            http_results['post'] = yield http_post_async(url, body="")

        io_loop.add_callback(do_http)

{'fraction_non_alphanumeric': 0.10190158223254464, 'fraction_numerical': 0.0, 'mean_word_length': 3.186873290793072, 'pattern_counts': {'":': 2, '<': 3, '<?xml version=': 0, '>': 3, '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': '4983955', 'n_tokens_mistral': 4083, 'n_tokens_neox': 4058, 'n_words': 862}