import json
import logging
import os
from io import BytesIO
from io import StringIO
from pathlib import Path

import elasticsearch
import pytest
from freezegun import freeze_time

from pytest_benchmark import plugin
from pytest_benchmark.plugin import BenchmarkSession
from pytest_benchmark.plugin import pytest_benchmark_compare_machine_info
from pytest_benchmark.plugin import pytest_benchmark_generate_json
from pytest_benchmark.plugin import pytest_benchmark_group_stats
from pytest_benchmark.storage.elasticsearch import ElasticsearchStorage
from pytest_benchmark.storage.elasticsearch import _mask_hosts
from pytest_benchmark.utils import parse_elasticsearch_storage

try:
    import unittest.mock as mock
except ImportError:
    from unittest import mock

logger = logging.getLogger(__name__)

THIS = Path(__file__)
BENCHFILE = THIS.with_name('test_storage') / '0030_5b78858eb718649a31fb93d8dc96ca2cee41a4cd_20150815_030419_uncommitted-changes.json'
SAVE_DATA = json.loads(BENCHFILE.read_text(encoding='utf8'))
SAVE_DATA['machine_info'] = {'foo': 'bar'}
SAVE_DATA['commit_info'] = {'foo': 'bar'}

tmp = SAVE_DATA.copy()

ES_DATA = tmp.pop('benchmarks')[0]
ES_DATA.update(tmp)
ES_DATA['benchmark_id'] = 'FoobarOS_commitId'


class Namespace:
    def __init__(self, **kwargs):
        self.__dict__.update(kwargs)

    def __getitem__(self, item):
        return self.__dict__[item]


class LooseFileLike(BytesIO):
    def close(self):
        value = self.getvalue()
        super().close()
        self.getvalue = lambda: value


class MockStorage(ElasticsearchStorage):
    def __init__(self):
        self._es = mock.Mock(spec=elasticsearch.Elasticsearch)
        self._es_hosts = self._es_index = self._es_doctype = 'mocked'
        self.logger = logger
        self.default_machine_id = 'FoobarOS'


class MockSession(BenchmarkSession):
    def __init__(self):
        self.verbose = False
        self.quiet = False
        self.histogram = True
        self.benchmarks = []
        self.performance_regressions = []
        self.sort = 'min'
        self.compare = '0001'
        self.logger = logging.getLogger(__name__)
        self.machine_id = 'FoobarOS'
        self.machine_info = {'foo': 'bar'}
        self.save = self.autosave = self.json = False
        self.options = {
            'min_rounds': 123,
            'min_time': 234,
            'max_time': 345,
        }
        self.compare_fail = []
        self.config = Namespace(
            hook=Namespace(
                pytest_benchmark_group_stats=pytest_benchmark_group_stats,
                pytest_benchmark_generate_machine_info=lambda **kwargs: {'foo': 'bar'},
                pytest_benchmark_update_machine_info=lambda **kwargs: None,
                pytest_benchmark_compare_machine_info=pytest_benchmark_compare_machine_info,
                pytest_benchmark_generate_json=pytest_benchmark_generate_json,
                pytest_benchmark_update_json=lambda **kwargs: None,
                pytest_benchmark_generate_commit_info=lambda **kwargs: {'foo': 'bar'},
                pytest_benchmark_update_commit_info=lambda **kwargs: None,
            )
        )
        self.elasticsearch_host = 'localhost:9200'
        self.elasticsearch_index = 'benchmark'
        self.elasticsearch_doctype = 'benchmark'
        self.storage = MockStorage()
        self.group_by = 'group'
        self.columns = ['min', 'max', 'mean', 'stddev', 'median', 'iqr', 'outliers', 'rounds', 'iterations']
        self.benchmarks = []
        data = json.loads(BENCHFILE.read_text(encoding='utf8'))
        self.benchmarks.extend(
            Namespace(
                as_dict=lambda include_data=False, stats=True, flat=False, _bench=bench: dict(_bench, **_bench['stats'])
                if flat
                else dict(_bench),
                name=bench['name'],
                fullname=bench['fullname'],
                group=bench['group'],
                options=bench['options'],
                has_error=False,
                params=None,
                **bench['stats'],
            )
            for bench in data['benchmarks']
        )


try:
    text_type = unicode
except NameError:
    text_type = str


def force_text(text):
    if isinstance(text, text_type):
        return text
    else:
        return text.decode('utf-8')


def force_bytes(text):
    if isinstance(text, text_type):
        return text.encode('utf-8')
    else:
        return text


def make_logger(sess):
    output = StringIO()
    sess.logger = Namespace(
        info=lambda text, **opts: output.write(force_text(text) + '\n'),
        error=lambda text: output.write(force_text(text) + '\n'),
    )
    sess.storage.logger = Namespace(
        info=lambda text, **opts: output.write(force_text(text) + '\n'),
        error=lambda text: output.write(force_text(text) + '\n'),
    )
    return output


@pytest.fixture
def sess():
    return MockSession()


@pytest.fixture
def logger_output(sess):
    return make_logger(sess)


@freeze_time('2015-08-15T00:04:18.687119')
def test_handle_saving(sess, logger_output, monkeypatch):
    monkeypatch.setattr(plugin, '__version__', '2.5.0')
    sess.save = 'commitId'
    sess.autosave = True
    sess.json = None
    sess.save_data = False
    sess.handle_saving()
    sess.storage._es.index.assert_called_with(
        index='mocked',
        doc_type='mocked',
        body=ES_DATA,
        id='FoobarOS_commitId_tests/test_normal.py::test_xfast_parametrized[0]',
    )


def test_parse_with_no_creds():
    string = 'https://example.org,another.org'
    hosts, _, _, _ = parse_elasticsearch_storage(string)
    assert len(hosts) == 2
    assert 'https://example.org' in hosts
    assert 'https://another.org' in hosts


def test_parse_with_creds_in_first_host_of_url():
    string = 'https://user:pass@example.org,another.org'
    hosts, _, _, _ = parse_elasticsearch_storage(string)
    assert len(hosts) == 2
    assert 'https://user:pass@example.org' in hosts
    assert 'https://another.org' in hosts


def test_parse_with_creds_in_second_host_of_url():
    string = 'https://example.org,user:pass@another.org'
    hosts, _, _, _ = parse_elasticsearch_storage(string)
    assert len(hosts) == 2
    assert 'https://example.org' in hosts
    assert 'https://user:pass@another.org' in hosts


def test_parse_with_creds_in_netrc(tmpdir):
    netrc_file = os.path.join(tmpdir.strpath, 'netrc')
    with open(netrc_file, 'w') as f:
        f.write('machine example.org login user1 password pass1\n')
        f.write('machine another.org login user2 password pass2\n')
    string = 'https://example.org,another.org'
    hosts, _, _, _ = parse_elasticsearch_storage(string, netrc_file=netrc_file)
    assert len(hosts) == 2
    assert 'https://user1:pass1@example.org' in hosts
    assert 'https://user2:pass2@another.org' in hosts


def test_parse_url_creds_supersedes_netrc_creds(tmpdir):
    netrc_file = os.path.join(tmpdir.strpath, 'netrc')
    with open(netrc_file, 'w') as f:
        f.write('machine example.org login user1 password pass1\n')
        f.write('machine another.org login user2 password pass2\n')
    string = 'https://user3:pass3@example.org,another.org'
    hosts, _, _, _ = parse_elasticsearch_storage(string, netrc_file=netrc_file)
    assert len(hosts) == 2
    assert 'https://user3:pass3@example.org' in hosts  # superseded by creds in url
    assert 'https://user2:pass2@another.org' in hosts  # got creds from netrc file


def test__mask_hosts():
    hosts = ['https://user1:pass1@example.org', 'https://user2:pass2@another.org']
    masked_hosts = _mask_hosts(hosts)
    assert len(masked_hosts) == len(hosts)
    assert 'https://***:***@example.org' in masked_hosts
    assert 'https://***:***@another.org' in masked_hosts