# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

import os
from cStringIO import StringIO
from functools import partial

import mozunit
import pytest
from moztest.selftest.output import get_mozharness_status, filter_action

from mozharness.base.log import INFO, WARNING, ERROR
from mozharness.mozilla.buildbot import TBPL_SUCCESS, TBPL_WARNING, TBPL_FAILURE

here = os.path.abspath(os.path.dirname(__file__))
get_mozharness_status = partial(get_mozharness_status, 'reftest')


def test_output_pass(runtests):
    status, lines = runtests('reftest-pass.list')
    assert status == 0

    tbpl_status, log_level = get_mozharness_status(lines, status)
    assert tbpl_status == TBPL_SUCCESS
    assert log_level in (INFO, WARNING)

    test_status = filter_action('test_status', lines)
    assert len(test_status) == 3
    assert all(t['status'] == 'PASS' for t in test_status)

    test_end = filter_action('test_end', lines)
    assert len(test_end) == 3
    assert all(t['status'] == 'OK' for t in test_end)


def test_output_fail(runtests):
    formatter = pytest.importorskip('output').ReftestFormatter()

    status, lines = runtests('reftest-fail.list')
    assert status == 0

    buf = StringIO()
    tbpl_status, log_level = get_mozharness_status(
        lines, status, formatter=formatter, buf=buf)

    assert tbpl_status == TBPL_WARNING
    assert log_level == WARNING

    test_status = filter_action('test_status', lines)
    assert len(test_status) == 3
    assert all(t['status'] == 'FAIL' for t in test_status)
    assert all('reftest_screenshots' in t['extra'] for t in test_status)

    test_end = filter_action('test_end', lines)
    assert len(test_end) == 3
    assert all(t['status'] == 'OK' for t in test_end)

    # ensure screenshots were printed
    formatted = buf.getvalue()
    assert 'REFTEST   IMAGE 1' in formatted
    assert 'REFTEST   IMAGE 2' in formatted


@pytest.mark.skip_mozinfo("!crashreporter")
def test_output_crash(runtests):
    status, lines = runtests('reftest-crash.list', environment=["MOZ_CRASHREPORTER_SHUTDOWN=1"])
    assert status == 1

    tbpl_status, log_level = get_mozharness_status(lines, status)
    assert tbpl_status == TBPL_FAILURE
    assert log_level == ERROR

    crash = filter_action('crash', lines)
    assert len(crash) == 1
    assert crash[0]['action'] == 'crash'
    assert crash[0]['signature']
    assert crash[0]['minidump_path']

    lines = filter_action('test_end', lines)
    assert len(lines) == 0


@pytest.mark.skip_mozinfo("!asan")
def test_output_asan(runtests):
    status, lines = runtests('reftest-crash.list', environment=["MOZ_CRASHREPORTER_SHUTDOWN=1"])
    assert status == 0

    tbpl_status, log_level = get_mozharness_status(lines, status)
    assert tbpl_status == TBPL_FAILURE
    assert log_level == ERROR

    crash = filter_action('crash', lines)
    assert len(crash) == 0

    process_output = filter_action('process_output', lines)
    assert any('ERROR: AddressSanitizer' in l['data'] for l in process_output)


@pytest.mark.skip_mozinfo("!debug")
def test_output_assertion(runtests):
    status, lines = runtests('reftest-assert.list')
    assert status == 0

    tbpl_status, log_level = get_mozharness_status(lines, status)
    assert tbpl_status == TBPL_WARNING
    assert log_level == WARNING

    test_status = filter_action('test_status', lines)
    assert len(test_status) == 1
    assert test_status[0]['status'] == 'PASS'

    test_end = filter_action('test_end', lines)
    assert len(test_end) == 1
    assert test_end[0]['status'] == 'OK'

    assertions = filter_action('assertion_count', lines)
    assert len(assertions) == 1
    assert assertions[0]['count'] == 1


@pytest.mark.skip_mozinfo("!debug")
def test_output_leak(monkeypatch, runtests):
    # Monkeypatch mozleak so we always process a failing leak log
    # instead of the actual one.
    import mozleak
    old_process_leak_log = mozleak.process_leak_log

    def process_leak_log(*args, **kwargs):
        return old_process_leak_log(os.path.join(here, 'files', 'leaks.log'), *args[1:], **kwargs)

    monkeypatch.setattr('mozleak.process_leak_log', process_leak_log)

    status, lines = runtests('reftest-pass.list')
    assert status == 0

    tbpl_status, log_level = get_mozharness_status(lines, status)
    assert tbpl_status == TBPL_FAILURE
    assert log_level == ERROR

    errors = filter_action('log', lines)
    errors = [e for e in errors if e['level'] == 'ERROR']
    assert len(errors) == 1
    assert 'leakcheck' in errors[0]['message']


if __name__ == '__main__':
    mozunit.main()