import os
import pytest
from aioresponses import aioresponses

from rasa.core import utils
from rasa.core.utils import EndpointConfig
from tests.core.utilities import latest_request, json_of_latest_request


@pytest.fixture(scope="session")
def loop():
    from pytest_sanic.plugin import loop as sanic_loop
    return utils.enable_async_loop_debugging(next(sanic_loop()))


def test_is_int():
    assert utils.is_int(1)
    assert utils.is_int(1.0)
    assert not utils.is_int(None)
    assert not utils.is_int(1.2)
    assert not utils.is_int("test")


def test_subsample_array_read_only():
    t = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    r = utils.subsample_array(t, 5,
                              can_modify_incoming_array=False)

    assert len(r) == 5
    assert set(r).issubset(t)


def test_subsample_array():
    t = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    # this will modify the original array and shuffle it
    r = utils.subsample_array(t, 5)

    assert len(r) == 5
    assert set(r).issubset(t)


def test_on_hot():
    r = utils.one_hot(4, 6)
    assert (r[[0, 1, 2, 3, 5]] == 0).all()
    assert r[4] == 1


def test_on_hot_out_of_range():
    with pytest.raises(ValueError):
        utils.one_hot(4, 3)


def test_list_routes(default_agent):
    from rasa.core import server
    app = server.create_app(default_agent, auth_token=None)

    routes = utils.list_routes(app)
    assert set(routes.keys()) == {'hello',
                                  'version',
                                  'execute_action',
                                  'append_event',
                                  'replace_events',
                                  'list_trackers',
                                  'retrieve_tracker',
                                  'retrieve_story',
                                  'respond',
                                  'predict',
                                  'parse',
                                  'train_stack',
                                  'evaluate_intents',
                                  'log_message',
                                  'load_model',
                                  'evaluate_stories',
                                  'get_domain',
                                  'continue_training',
                                  'status',
                                  'tracker_predict'}


def test_cap_length():
    assert utils.cap_length("mystring", 6) == "mys..."


def test_cap_length_without_ellipsis():
    assert utils.cap_length("mystring", 3,
                            append_ellipsis=False) == "mys"


def test_cap_length_with_short_string():
    assert utils.cap_length("my", 3) == "my"


def test_pad_list_to_size():
    assert (utils.pad_list_to_size(["e1", "e2"], 4, "other") ==
            ["e1", "e2", "other", "other"])


def test_read_lines():
    lines = utils.read_lines("data/test_stories/stories.md",
                             max_line_limit=2,
                             line_pattern=r"\*.*")

    lines = list(lines)

    assert len(lines) == 2


async def test_endpoint_config():
    with aioresponses() as mocked:
        endpoint = EndpointConfig(
            "https://example.com/",
            params={"A": "B"},
            headers={"X-Powered-By": "Rasa"},
            basic_auth={"username": "user",
                        "password": "<PASSWORD>"},
            token="mytoken",
            token_name="letoken",
            type="redis",
            port=6379,
            db=0,
            password="password",
            timeout=30000
        )

        mocked.post('https://example.com/test?A=B&P=1&letoken=mytoken',
                    payload={"ok": True},
                    repeat=True,
                    status=200)

        await endpoint.request("post", subpath="test",
                               content_type="application/text",
                               json={"c": "d"},
                               params={"P": "1"})

        r = latest_request(mocked, 'post',
                           "https://example.com/test?A=B&P=1&letoken=mytoken")

        assert r

        assert json_of_latest_request(r) == {"c": "d"}
        assert r[-1].kwargs.get("params", {}).get("A") == "B"
        assert r[-1].kwargs.get("params", {}).get("P") == "1"
        assert r[-1].kwargs.get("params", {}).get("letoken") == "mytoken"

        # unfortunately, the mock library won't report any headers stored on
        # the session object, so we need to verify them separately
        async with endpoint.session() as s:
            assert s._default_headers.get("X-Powered-By") == "Rasa"
            assert s._default_auth.login == "user"
            assert s._default_auth.password == "<PASSWORD>"


os.environ['USER_NAME'] = 'user'
os.environ['PASS'] = '<PASSWORD>'


def test_read_yaml_string():
    config_without_env_var = """
    user: user
    password: <PASSWORD>
    """
    r = utils.read_yaml_string(config_without_env_var)
    assert r['user'] == 'user' and r['password'] == '<PASSWORD>'


def test_read_yaml_string_with_env_var():
    config_with_env_var = """
    user: ${USER_NAME}
    password: ${<PASSWORD>}
    """
    r = utils.read_yaml_string(config_with_env_var)
    assert r['user'] == 'user' and r['password'] == '<PASSWORD>'


def test_read_yaml_string_with_multiple_env_vars_per_line():
    config_with_env_var = """
    user: ${USER_NAME} ${PASS}
    password: ${<PASSWORD>}
    """
    r = utils.read_yaml_string(config_with_env_var)
    assert r['user'] == 'user pass' and r['password'] == '<PASSWORD>'


def test_read_yaml_string_with_env_var_prefix():
    config_with_env_var_prefix = """
    user: db_${USER_NAME}
    password: <PASSWORD>}
    """
    r = utils.read_yaml_string(config_with_env_var_prefix)
    assert r['user'] == 'db_user' and r['password'] == '<PASSWORD>'


def test_read_yaml_string_with_env_var_postfix():
    config_with_env_var_postfix = """
    user: ${USER_NAME}_admin
    password: ${<PASSWORD>
    """
    r = utils.read_yaml_string(config_with_env_var_postfix)
    assert r['user'] == 'user_admin' and r['password'] == '<PASSWORD>'


def test_read_yaml_string_with_env_var_infix():
    config_with_env_var_infix = """
    user: db_${USER_NAME}_admin
    password: <PASSWORD>
    """
    r = utils.read_yaml_string(config_with_env_var_infix)
    assert r['user'] == 'db_user_admin' and r['password'] == '<PASSWORD>'


def test_read_yaml_string_with_env_var_not_exist():
    config_with_env_var_not_exist = """
    user: ${USER_NAME}
    password: ${PASSWORD}
    """
    with pytest.raises(ValueError):
        utils.read_yaml_string(config_with_env_var_not_exist)

import os
import pytest
from aioresponses import aioresponses

from rasa.core import utils
from rasa.core.utils import EndpointConfig
from tests.core.utilities import latest_request, json_of_latest_request


@pytest.fixture(scope="session")
def loop():
    from pytest_sanic.plugin import loop as sanic_loop
    return utils.enable_async_loop_debugging(next(sanic_loop()))


def test_is_int():
    assert utils.is_int(1)
    assert utils.is_int(1.0)
    assert not utils.is_int(None)
    assert not utils.is_int(1.2)
    assert not utils.is_int("test")


def test_subsample_array_read_only():
    t = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    r = utils.subsample_array(t, 5,
                              can_modify_incoming_array=False)

    assert len(r) == 5
    assert set(r).issubset(t)


def test_subsample_array():
    t = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
    # this will modify the original array and shuffle it
    r = utils.subsample_array(t, 5)

    assert len(r) == 5
    assert set(r).issubset(t)


def test_on_hot():
    r = utils.one_hot(4, 6)
    assert (r[[0, 1, 2, 3, 5]] == 0).all()
    assert r[4] == 1


def test_on_hot_out_of_range():
    with pytest.raises(ValueError):
        utils.one_hot(4, 3)


def test_list_routes(default_agent):
    from rasa.core import server
    app = server.create_app(default_agent, auth_token=None)

    routes = utils.list_routes(app)
    assert set(routes.keys()) == {'hello',
                                  'version',
                                  'execute_action',
                                  'append_event',
                                  'replace_events',
                                  'list_trackers',
                                  'retrieve_tracker',
                                  'retrieve_story',
                                  'respond',
                                  'predict',
                                  'parse',
                                  'train_stack',
                                  'evaluate_intents',
                                  'log_message',
                                  'load_model',
                                  'evaluate_stories',
                                  'get_domain',
                                  'continue_training',
                                  'status',
                                  'tracker_predict'}


def test_cap_length():
    assert utils.cap_length("mystring", 6) == "mys..."


def test_cap_length_without_ellipsis():
    assert utils.cap_length("mystring", 3,
                            append_ellipsis=False) == "mys"


def test_cap_length_with_short_string():
    assert utils.cap_length("my", 3) == "my"


def test_pad_list_to_size():
    assert (utils.pad_list_to_size(["e1", "e2"], 4, "other") ==
            ["e1", "e2", "other", "other"])


def test_read_lines():
    lines = utils.read_lines("data/test_stories/stories.md",
                             max_line_limit=2,
                             line_pattern=r"\*.*")

    lines = list(lines)

    assert len(lines) == 2


async def test_endpoint_config():
    with aioresponses() as mocked:
        endpoint = EndpointConfig(
            "https://example.com/",
            params={"A": "B"},
            headers={"X-Powered-By": "Rasa"},
            basic_auth={"username": "user",
                        "password": "<PASSWORD>"},
            token="mytoken",
            token_name="letoken",
            type="redis",
            port=6379,
            db=0,
            password="password",
            timeout=30000
        )

        mocked.post('https://example.com/test?A=B&P=1&letoken=mytoken',
                    payload={"ok": True},
                    repeat=True,
                    status=200)

        await endpoint.request("post", subpath="test",
                               content_type="application/text",
                               json={"c": "d"},
                               params={"P": "1"})

        r = latest_request(mocked, 'post',
                           "https://example.com/test?A=B&P=1&letoken=mytoken")

        assert r

        assert json_of_latest_request(r) == {"c": "d"}
        assert r[-1].kwargs.get("params", {}).get("A") == "B"
        assert r[-1].kwargs.get("params", {}).get("P") == "1"
        assert r[-1].kwargs.get("params", {}).get("letoken") == "mytoken"

        # unfortunately, the mock library won't report any headers stored on
        # the session object, so we need to verify them separately
        async with endpoint.session() as s:
            assert s._default_headers.get("X-Powered-By") == "Rasa"
            assert s._default_auth.login == "user"
            assert s._default_auth.password == "<PASSWORD>"


os.environ['USER_NAME'] = 'user'
os.environ['PASS'] = '<PASSWORD>'


def test_read_yaml_string():
    config_without_env_var = """
    user: user
    password: <PASSWORD>
    """
    r = utils.read_yaml_string(config_without_env_var)
    assert r['user'] == 'user' and r['password'] == '<PASSWORD>'


def test_read_yaml_string_with_env_var():
    config_with_env_var = """
    user: ${USER_NAME}
    password: ${<PASSWORD>}
    """
    r = utils.read_yaml_string(config_with_env_var)
    assert r['user'] == 'user' and r['password'] == '<PASSWORD>'


def test_read_yaml_string_with_multiple_env_vars_per_line():
    config_with_env_var = """
    user: ${USER_NAME} ${PASS}
    password: ${<PASSWORD>}
    """
    r = utils.read_yaml_string(config_with_env_var)
    assert r['user'] == 'user pass' and r['password'] == '<PASSWORD>'


def test_read_yaml_string_with_env_var_prefix():
    config_with_env_var_prefix = """
    user: db_${USER_NAME}
    password: <PASSWORD>}
    """
    r = utils.read_yaml_string(config_with_env_var_prefix)
    assert r['user'] == 'db_user' and r['password'] == '<PASSWORD>'


def test_read_yaml_string_with_env_var_postfix():
    config_with_env_var_postfix = """
    user: ${USER_NAME}_admin
    password: ${<PASSWORD>
    """
    r = utils.read_yaml_string(config_with_env_var_postfix)
    assert r['user'] == 'user_admin' and r['password'] == '<PASSWORD>'


def test_read_yaml_string_with_env_var_infix():
    config_with_env_var_infix = """
    user: db_${USER_NAME}_admin
    password: <PASSWORD>
    """
    r = utils.read_yaml_string(config_with_env_var_infix)
    assert r['user'] == 'db_user_admin' and r['password'] == '<PASSWORD>'


def test_read_yaml_string_with_env_var_not_exist():
    config_with_env_var_not_exist = """
    user: ${USER_NAME}
    password: ${PASSWORD}
    """
    with pytest.raises(ValueError):
        utils.read_yaml_string(config_with_env_var_not_exist)

# this will modify the original array and shuffle it # unfortunately, the mock library won't report any headers stored on # the session object, so we need to verify them separately user: user     password: <PASSWORD> user: ${USER_NAME}     password: ${<PASSWORD>} user: ${USER_NAME} ${PASS}     password: ${<PASSWORD>} user: db_${USER_NAME}     password: <PASSWORD>} user: ${USER_NAME}_admin     password: ${<PASSWORD> user: db_${USER_NAME}_admin     password: <PASSWORD> user: ${USER_NAME}     password: ${PASSWORD}