#!/usr/bin/env python
# coding=utf-8
"""Test scriptworker.utils
"""
import asyncio
import os
import re
import shutil
import tempfile
import time

import mock
import pytest

import scriptworker.utils as utils
from scriptworker.exceptions import Download404, DownloadError, ScriptWorkerException, ScriptWorkerRetryException

from . import FakeResponse, touch

# constants helpers and fixtures {{{1
TEST_DATA_DIR = os.path.join(os.path.dirname(__file__), "data")

# from https://github.com/SecurityInnovation/PGPy/blob/develop/tests/test_01_types.py
text = {
    # some basic utf-8 test strings - these should all pass
    "english": "The quick brown fox jumped over the lazy dog",
    # this hiragana pangram comes from http://www.columbia.edu/~fdc/utf8/
    "hiragana": "いろはにほへど　ちりぬるを\n" "わがよたれぞ　つねならむ\n" "うゐのおくやま　けふこえて\n" "あさきゆめみじ　ゑひもせず",
    "poo": "Hello, \U0001F4A9!",
}

non_text = {
    "None": None,
    "dict": {"a": 1, 2: 3},
    "cyrillic": "грызть гранит науки".encode("iso8859_5"),
    "cp865": "Mit luftpudefartøj er fyldt med ål".encode("cp865"),
}

retry_count = {}


@pytest.fixture(scope="function")
def datestring():
    """Datestring constant."""
    return "2016-04-16T03:46:24.958Z"


async def fail_first(*args, **kwargs):
    global retry_count
    retry_count["fail_first"] += 1
    if retry_count["fail_first"] < 2:
        raise ScriptWorkerRetryException("first")
    return "yay"


async def always_fail(*args, **kwargs):
    global retry_count
    retry_count.setdefault("always_fail", 0)
    retry_count["always_fail"] += 1
    raise ScriptWorkerException("fail")


async def fake_sleep(*args, **kwargs):
    pass


# to_unicode {{{1
@pytest.mark.parametrize("text", [v for _, v in sorted(text.items())])
def test_text_to_unicode(text):
    assert text == utils.to_unicode(text)
    assert text == utils.to_unicode(text.encode("utf-8"))


@pytest.mark.parametrize("non_text", [v for _, v in sorted(non_text.items())])
def test_nontext_to_unicode(non_text):
    assert non_text == utils.to_unicode(non_text)


# datestring_to_timestamp {{{1
def test_datestring_to_timestamp(datestring):
    assert utils.datestring_to_timestamp(datestring) == 1460778384


# cleanup {{{1
def test_cleanup(rw_context):
    for name in "work_dir", "artifact_dir", "task_log_dir":
        path = rw_context.config[name]
        open(os.path.join(path, "tempfile"), "w").close()
        assert os.path.exists(os.path.join(path, "tempfile"))
    utils.cleanup(rw_context)
    for name in "work_dir", "artifact_dir":
        path = rw_context.config[name]
        assert os.path.exists(path)
        assert not os.path.exists(os.path.join(path, "tempfile"))
    # 2nd pass
    utils.rm(rw_context.config["work_dir"])
    utils.cleanup(rw_context)


# request and retry_request {{{1
@pytest.mark.asyncio
async def test_request(rw_context, fake_session):
    rw_context.session = fake_session
    result = await utils.request(rw_context, "url")
    assert result == "{}"


@pytest.mark.asyncio
async def test_request_json(rw_context, fake_session):
    rw_context.session = fake_session
    result = await utils.request(rw_context, "url", return_type="json")
    assert result == {}


@pytest.mark.asyncio
async def test_request_response(rw_context, fake_session):
    rw_context.session = fake_session
    result = await utils.request(rw_context, "url", return_type="response")
    assert isinstance(result, FakeResponse)


@pytest.mark.asyncio
async def test_request_retry(rw_context, fake_session_500):
    rw_context.session = fake_session_500
    with pytest.raises(ScriptWorkerRetryException):
        await utils.request(rw_context, "url")


@pytest.mark.asyncio
async def test_request_exception(rw_context, fake_session_500):
    rw_context.session = fake_session_500
    with pytest.raises(ScriptWorkerException):
        await utils.request(rw_context, "url", retry=())


@pytest.mark.asyncio
async def test_retry_request(rw_context, fake_session):
    rw_context.session = fake_session
    result = await utils.retry_request(rw_context, "url")
    assert result == "{}"


# calculate_sleep_time {{{1
@pytest.mark.parametrize("attempt", (-1, 0))
def test_calculate_no_sleep_time(attempt):
    assert utils.calculate_sleep_time(attempt) == 0


@pytest.mark.parametrize(
    "attempt,kwargs,min_expected,max_expected",
    (
        (1, {"delay_factor": 5.0, "randomization_factor": 0, "max_delay": 15}, 5.0, 5.0),
        (2, {"delay_factor": 5.0, "randomization_factor": 0.25, "max_delay": 15}, 10.0, 12.5),
        (3, {"delay_factor": 5.0, "randomization_factor": 0.25, "max_delay": 10}, 10.0, 10.0),
    ),
)
def test_calculate_sleep_time(attempt, kwargs, min_expected, max_expected):
    assert min_expected <= utils.calculate_sleep_time(attempt, **kwargs) <= max_expected


# retry_async {{{1
@pytest.mark.asyncio
async def test_retry_async_fail_first():
    global retry_count
    retry_count["fail_first"] = 0
    status = await utils.retry_async(fail_first, sleeptime_kwargs={"delay_factor": 0})
    assert status == "yay"
    assert retry_count["fail_first"] == 2


@pytest.mark.asyncio
async def test_retry_async_always_fail():
    global retry_count
    retry_count["always_fail"] = 0
    with mock.patch("asyncio.sleep", new=fake_sleep):
        with pytest.raises(ScriptWorkerException):
            status = await utils.retry_async(always_fail, sleeptime_kwargs={"delay_factor": 0})
            assert status is None
    assert retry_count["always_fail"] == 5


def test_retry_sync_fail_first_and_blocks_the_main_process():
    global retry_count
    retry_count["fail_first"] = 0

    def fail_first_sync(*args, **kwargs):
        global retry_count
        retry_count["fail_first"] += 1
        if retry_count["fail_first"] < 2:
            raise ScriptWorkerRetryException("first")
        return "yay"

    start_time = time.time()
    status = utils.retry_sync(fail_first_sync, sleeptime_kwargs={"delay_factor": 0.25, "randomization_factor": 0})
    elapsed_time = time.time() - start_time

    assert 0.4 <= elapsed_time <= 0.6
    assert status == "yay"
    assert retry_count["fail_first"] == 2


def test_retry_sync_always_fail():
    global retry_count
    retry_count["always_fail"] = 0

    def always_fail_sync(*args, **kwargs):
        global retry_count
        retry_count.setdefault("always_fail", 0)
        retry_count["always_fail"] += 1
        raise ScriptWorkerException("fail")

    with mock.patch("asyncio.sleep", new=fake_sleep):
        with pytest.raises(ScriptWorkerException):
            status = utils.retry_sync(always_fail_sync, sleeptime_kwargs={"delay_factor": 0})
            asser
t status is None
    assert retry_count["always_fail"] == 5


@pytest.mark.asyncio
async def test_retry_async_decorator_fail_first():
    global retry_count

    @utils.retry_async_decorator(sleeptime_kwargs={"delay_factor": 0})
    async def decorated_fail_first(*args, **kwargs):
        return await fail_first(*arg
s, **kwargs)

    retry_count["fail_first"] = 0
    status = await decorated_fail_first()
    assert status == "yay"
    assert retry_count["fail_first"] == 2


@pytest.mark.asyncio
async def test_retry_async_decorator_always_fail_async():
    global retry_count

    @utils.retry_async_decorator(sleeptime_kwargs={"delay_factor": 0})
    async def decorated_always_fail(*args, **kwargs):
        return await always_fail(*args, **kwargs)

    retry_count["always_fail"] = 0
    with mock.patch("asyncio.sleep", new=fake_sleep):
        with pytest.raises(ScriptWorkerException):
            await decorated_always_fail()

    assert retry_count["always_fail"] == 5


# create_temp_creds {{{1
def test_create_temp_creds():
    with mock.patch.object(utils, "createTemporaryCredentials") as p:
        p.return_value = {"one": b"one", "two": "two"}
        creds = utils.create_temp_creds("clientId", "accessToken", "start", "expires")
        assert p.called_once_with("clientId", "accessToken", "start", "expires", ["assume:project:taskcluster:worker-test-scopes"], name=None)
        assert creds == {"one": "one", "two": "two"}


# raise_future_exceptions {{{1
@pytest.mark.asyncio
@pytest.mark.parametrize("exc"