# -*- coding: utf-8 -*-
from __future__ import unicode_literals

import inspect
import mock
import raven
import time
import os
import sys

from raven.utils.compat import PY2
from raven.base import Client, ClientState
from raven.exceptions import RateLimited
from raven.transport import AsyncTransport
from raven.transport.http import HTTPTransport
from raven.utils.stacks import iter_stack_frames
from raven.utils.testutils import TestCase


class TempStoreClient(Client):
    def __init__(self, **kwargs):
        self.events = []
        super(TempStoreClient, self).__init__(**kwargs)

    def is_enabled(self):
        return True

    def send(self, **kwargs):
        self.events.append(kwargs)


class ClientStateTest(TestCase):
    def test_should_try_online(self):
        state = ClientState()
        self.assertEquals(state.should_try(), True)

    def test_should_try_new_error(self):
        state = ClientState()
        state.status = state.ERROR
        state.last_check = time.time()
        state.retry_number = 1
        self.assertEquals(state.should_try(), False)

    def test_should_try_time_passed_error(self):
        state = ClientState()
        state.status = state.ERROR
        state.last_check = time.time() - 10
        state.retry_number = 1
        self.assertEquals(state.should_try(), True)

    def test_set_fail(self):
        state = ClientState()
        state.set_fail()
        self.assertEquals(state.status, state.ERROR)
        self.assertNotEquals(state.last_check, None)
        self.assertEquals(state.retry_number, 1)

    def test_set_success(self):
        state = ClientState()
        state.status = state.ERROR
        state.last_check = 'foo'
        state.retry_number = 0
        state.set_success()
        self.assertEquals(state.status, state.ONLINE)
        self.assertEquals(state.last_check, None)
        self.assertEquals(state.retry_number, 0)

    def test_should_try_retry_after(self):
        state = ClientState()
        state.status = state.ERROR
        state.last_check = time.time()
        state.retry_number = 1
        state.retry_after = 1
        self.assertFalse(state.should_try())

    def test_should_try_retry_after_passed(self):
        state = ClientState()
        state.status = state.ERROR
        state.last_check = time.time() - 1
        state.retry_number = 1
        state.retry_after = 1
        self.assertTrue(state.should_try())


class ClientTest(TestCase):
    def setUp(self):
        self.client = TempStoreClient()

    def test_first_client_is_singleton(self):
        from raven import base
        base.Raven = None

        client = Client()
        client2 = Client()

        assert base.Raven is client
        assert client is not client2

    def test_client_picks_up_env_dsn(self):
        DSN = 'sync+http://public:secret@example.com/1'
        PUBLIC_DSN = '//public@example.com/1'
        with mock.patch.dict(os.environ, {'SENTRY_DSN': DSN}):
            client = Client()
            assert client.remote.get_public_dsn() == PUBLIC_DSN
            client = Client('')
            assert client.remote.get_public_dsn() == PUBLIC_DSN

    @mock.patch('raven.transport.http.HTTPTransport.send')
    @mock.patch('raven.base.ClientState.should_try')
    def test_send_remote_failover(self, should_try, send):
        should_try.return_value = True

        client = Client(
            dsn='sync+http://public:secret@example.com/1'
        )

        # test error
        send.side_effect = Exception()
        client.send_remote('sync+http://example.com/api/store', client.encode({}))
        self.assertEquals(client.state.status, client.state.ERROR)

        # test recovery
        send.side_effect = None
        client.send_remote('sync+http://example.com/api/store', client.encode({}))
        self.assertEquals(client.state.status, client.state.ONLINE)

    @mock.patch('raven.transport.http.HTTPTransport.send')
    @mock.patch('raven.base.ClientState.should_try')
    def test_send_remote_failover_with_retry_after(self, should_try, send):
        should_try.return_value = True

        client = Client(
            dsn='sync+http://public:secret@example.com/1'
        )

        # test error
        send.side_effect = RateLimited('foo', 5)
        client.send_remote('sync+http://example.com/api/1/store/', client.encode({}))
        self.assertEquals(client.state.status, client.state.ERROR)
        self.assertEqual(client.state.retry_after, 5)

        # test recovery
        send.side_effect = None
        client.send_remote('sync+http://example.com/api/1/store/', client.encode({}))
        self.assertEquals(client.state.status, client.state.ONLINE)
        self.assertEqual(client.state.retry_after, 0)

    @mock.patch('raven.conf.remote.RemoteConfig.get_transport')
    @mock.patch('raven.base.ClientState.should_try')
    def test_async_send_remote_failover(self, should_try, get_transport):
        should_try.return_value = True
        async_transport = AsyncTransport()
        async_transport.async_send = async_send = mock.Mock()
        get_transport.return_value = async_transport

        client = Client(
            dsn='http://public:secret@example.com/1',
        )

        # test immediate raise of error
        async_send.side_effect = Exception()
        client.send_remote('http://example.com/api/1/store/', client.encode({}))
        self.assertEquals(client.state.status, client.state.ERROR)

        # test recovery
        client.send_remote('http://example.com/api/1/store/', client.encode({}))
        success_cb = async_send.call_args[0][3]
        success_cb()
        self.assertEquals(client.state.status, client.state.ONLINE)

        # test delayed raise of error
        client.send_remote('http://example.com/api/1/store/', client.encode({}))
        failure_cb = async_send.call_args[0][4]
        failure_cb(Exception())
        self.assertEquals(client.state.status, client.state.ERROR)

    @mock.patch('raven.base.Client.send_remote')
    @mock.patch('raven.base.time.time')
    def test_send(self, time, send_remote):
        time.return_value = 1328055286.51
        client = Client(
            dsn='http://public:secret@example.com/1',
        )
        client.send(**{
            'foo': 'bar',
        })
        send_remote.assert_called_once_with(
            url='http://example.com/api/1/store/',
            data=client.encode({'foo': 'bar'}),
            headers={
                'User-Agent': 'raven-python/%s' % (raven.VERSION,),
                'Content-Type': 'application/octet-stream',
                'Content-Encoding': client.get_content_encoding(),
                'X-Sentry-Auth': (
                    'Sentry sentry_timestamp=1328055286.51, '
                    'sentry_client=raven-python/%s, sentry_version=6, '
                    'sentry_key=public, '
                    'sentry_secret=secret' % (raven.VERSION,))
            },
        )

    @mock.patch('raven.base.Client.send_remote')
    @mock.patch('raven.base.time.time')
    def test_send_with_auth_header(self, time, send_remote):
        time.return_value = 1328055286.51
        client = Client(
            dsn='http://public:secret@example.com/1',
        )
        client.send(auth_header='foo', **{
            'foo': 'bar',
        })
        send_remote.assert_called_once_with(
            url='http://example.com/api/1/store/',
            data=client.encode({'foo': 'bar'}),
            headers={
                'User-Agent': 'raven-python/%s' % (raven.VERSION,),
                'Content-Type': 'application/octet-stream',
                'Content-Encoding': client.get_content_encoding(),
                'X-Sentry-Auth': 'foo',
            },
        )

    @mock.patch('raven.transport.http.HTTPTransport.send')
    @mock.patch('raven.base.ClientState.should_try')
    def test_raise_exception_on_send_error(self, should_try, _send_remote):
        should_try.return_value = True
        client = Client(
            dsn='sync+http://public:secret@example.com/1',
        )

        # Test for the default behaviour in which a send error is handled by the client
        _send_remote.side_effect = Exception()
        client.capture('Message', data={}, date=None, time_spent=10,
                       extra={}, stack=None, tags=None, message='Test message')
        assert client.state.status == client.state.ERROR

        # Test for the case in which a send error is raised to the calling frame.
        client = Client(
            dsn='sync+http://public:secret@example.com/1',
            raise_send_errors=True,
        )
        with self.assertRaises(Exception):
            client.capture('Message', data={}, date=None, time_spent=10,
                           extra={}, stack=None, tags=None, message='Test message')

    def test_encode_decode(self):
        data = {'foo': 'bar'}
        encoded = self.client.encode(data)
        self.assertTrue(type(encoded), str)
        self.assertEquals(data, self.client.decode(encoded))

    def test_get_public_dsn(self):
        client = Client('http://public:secret@example.com/1')
        public_dsn = client.get_public_dsn()
        self.assertEquals(public_dsn, '//public@example.com/1')

    def test_explicit_message_on_message_event(self):
        self.client.captureMessage(message='test', data={
            'message': 'foo'
        })

        self.assertEquals(len(self.client.events), 1)
        event = self.client.events.pop(0)
        self.assertEquals(event['message'], 'foo')

    def test_message_from_kwargs(self):
        try:
            raise ValueError('foo')
        except ValueError:
            self.client.captureException(message='test', data={})

        self.assertEquals(len(self.client.events), 1)
        event = self.client.events.pop(0)
        self.assertEquals(event['message'], 'test')

    def test_explicit_message_on_exception_event(self):
        try:
            raise ValueError('foo')
        except ValueError:
            self.client.captureException(data={'message': 'foobar'})

        self.assertEquals(len(self.client.events), 1)
        event = self.client.events.pop(0)
        self.assertEquals(event['message'], 'foobar')

    def test_exception_event(self):
        try:
            raise ValueError('foo')
        except ValueError:
            self.client.captureException()

        self.assertEquals(len(self.client.events), 1)
        event = self.client.events.pop(0)
        self.assertEquals(event['message'], 'ValueError: foo')
        self.assertTrue('exception' in event)
        exc = event['exception']['values'][-1]
        self.assertEquals(exc['type'], 'ValueError')
        self.assertEquals(exc['value'], 'foo')
        self.assertEquals(exc['module'], ValueError.__module__)  # this differs in some Python versions
        assert 'stacktrace' not in event
        stacktrace = exc['stacktrace']
        self.assertEquals(len(stacktrace['frames']), 1)
        frame = stacktrace['frames'][0]
        self.assertEquals(frame['abs_path'], __file__.replace('.pyc', '.py'))
        self.assertEquals(frame['filename'], 'tests/base/tests.py')
        self.assertEquals(frame['module'], __name__)
        self.assertEquals(frame['function'], 'test_exception_event')
        self.assertTrue('timestamp' in event)

    def test_exception_nan_in_vars(self):
        try:
            foo = float("nan")  # noqa
            raise ValueError("foo")
        except ValueError:
            self.client.captureException()

        event, = self.client.events
        exc, = event['exception']['values']
        frame, = exc['stacktrace']['frames']
        assert frame['vars']['foo'] == "nan"

    def test_exception_event_true_exc_info(self):
        try:
            raise ValueError('foo')
        except ValueError:
            self.client.captureException(exc_info=True)

        self.assertEquals(len(self.client.events), 1)
        event = self.client.events.pop(0)
        self.assertEquals(event['message'], 'ValueError: foo')
        self.assertTrue('exception' in event)
        exc = event['exception']['values'][-1]
        stacktrace = exc['stacktrace']
        self.assertEquals(len(stacktrace['frames']), 1)
        frame = stacktrace['frames'][0]
        self.assertEquals(frame['abs_path'], __file__.replace('.pyc', '.py'))
        self.assertEquals(frame['filename'], 'tests/base/tests.py')
        self.assertEquals(frame['module'], __name__)

    def test_exception_event_ignore_string(self):
        class Foo(Exception):
            pass

        client = TempStoreClient(ignore_exceptions=['Foo'])
        try:
            raise Foo()
        except Foo:
            client.captureException()

        self.assertEquals(len(client.events), 0)

    def test_exception_event_ignore_class(self):
        class Foo(Exception):
            pass

        client = TempStoreClient(ignore_exceptions=[Foo])
        try:
            raise Foo()
        except Foo:
            client.captureException()

        self.assertEquals(len(client.events), 0)

    def test_exception_event_ignore_child(self):
        class Foo(Exception):
            pass

        class Bar(Foo):
            pass

        client = TempStoreClient(ignore_exceptions=[Foo])
        try:
            raise Bar()
        except Bar:
            client.captureException()

        self.assertEquals(len(client.events), 0)

    def test_decorator_preserves_function(self):
        @self.client.capture_exceptions
        def test1():
            return 'foo'

        self.assertEquals(test1(), 'foo')

    class DecoratorTestException(Exception):
        pass

    def test_decorator_functionality(self):
        @self.client.capture_exceptions
        def test2():
            raise self.DecoratorTestException()

        try:
            test2()
        except self.DecoratorTestException:
            pass

        self.assertEquals(len(self.client.events), 1)
        event = self.client.events.pop(0)
        self.assertEquals(event['message'], 'DecoratorTestException')
        exc = event['exception']['values'][-1]
        self.assertEquals(exc['type'], 'DecoratorTestException')
        self.assertEquals(exc['module'], self.DecoratorTestException.__module__)
        stacktrace = exc['stacktrace']
        # this is a wrapped class object with __call__ so three frames are expected
        self.assertEquals(len(stacktrace['frames']), 3)
        frame = stacktrace['frames'][-1]
        self.assertEquals(frame['module'], __name__)
        self.assertEquals(frame['function'], 'test2')

    def test_decorator_filtering(self):
        @self.client.capture_exceptions(self.DecoratorTestException)
        def test3():
            raise Exception()

        try:
            test3()
        except Exception:
            pass

        self.assertEquals(len(self.client.events), 0)

    def test_context_manager_functionality(self):
        def test4():
            raise self.DecoratorTestException()

        try:
            with self.client.capture_exceptions():
                test4()
        except self.DecoratorTestException:
            pass

        self.assertEquals(len(self.client.events), 1)
        event = self.client.events.pop(0)
        self.assertEquals(event['message'], 'DecoratorTestException')
        exc = event['exception']['values'][-1]
        self.assertEquals(exc['type'], 'DecoratorTestException')
        self.assertEquals(exc['module'], self.DecoratorTestException.__module__)
        stacktrace = exc['stacktrace']
        # three frames are expected: test4, `with` block and context manager internals
        self.assertEquals(len(stacktrace['frames']), 3)
        frame = stacktrace['frames'][-1]
        self.assertEquals(frame['module'], __name__)
        self.assertEquals(frame['function'], 'test4')

    def test_content_manager_filtering(self):
        def test5():
            raise Exception()

        try:
            with self.client.capture_exceptions(self.DecoratorTestException):
                test5()
        except Exception:
            pass

        self.assertEquals(len(self.client.events), 0)

    def test_message_event(self):
        self.client.captureMessage(message='test')

        self.assertEquals(len(self.client.events), 1)
        event = self.client.events.pop(0)
        self.assertEquals(event['message'], 'test')
        assert 'stacktrace' not in event
        self.assertTrue('timestamp' in event)

    def test_fingerprint(self):
        self.client.captureMessage(
            message='test',
            fingerprint=['{{ default }}', 'foobar'],
        )

        assert len(self.client.events) == 1
        event = self.client.events.pop(0)
        assert event['fingerprint'] == ['{{ default }}', 'foobar']

    def test_context(self):
        self.client.context.merge({
            'tags': {'foo': 'bar'},
        })
        try:
            raise ValueError('foo')
        except ValueError:
            self.client.captureException()
        else:
            self.fail('Exception should have been raised')

        assert len(self.client.events) == 1
        event = self.client.events.pop(0)
        assert event['tags'] == {'foo': 'bar'}

    def test_stack_explicit_frames(self):
        def bar():
            return inspect.stack()

        frames = bar()

        self.client.captureMessage('test', stack=iter_stack_frames(frames))

        self.assertEquals(len(self.client.events), 1)
        event = self.client.events.pop(0)
        self.assertEquals(event['message'], 'test')
        assert 'stacktrace' in event
        self.assertEquals(len(frames), len(event['stacktrace']['frames']))
        for frame, frame_i in zip(frames[::-1], event['stacktrace']['frames']):
            self.assertEquals(frame[0].f_code.co_filename, frame_i['abs_path'])
            self.assertEquals(frame[0].f_code.co_name, frame_i['function'])

    def test_stack_auto_frames(self):
        self.client.captureMessage('test', stack=True)

        self.assertEquals(len(self.client.events), 1)
        event = self.client.events.pop(0)
        self.assertEquals(event['message'], 'test')
        self.assertTrue('stacktrace' in event)
        self.assertTrue('timestamp' in event)

    def test_site(self):
        self.client.captureMessage(message='test', data={'site': 'test'})

        self.assertEquals(len(self.client.events), 1)
        event = self.client.events.pop(0)
        assert 'site' in event['tags']
        assert event['tags']['site'] == 'test'

    def test_implicit_site(self):
        self.client = TempStoreClient(site='foo')
        self.client.captureMessage(message='test')

        self.assertEquals(len(self.client.events), 1)
        event = self.client.events.pop(0)
        assert 'site' in event['tags']
        assert event['tags']['site'] == 'foo'

    def test_logger(self):
        self.client.captureMessage(message='test', data={'logger': 'test'})

        self.assertEquals(len(self.client.events), 1)
        event = self.client.events.pop(0)
        self.assertEquals(event['logger'], 'test')
        self.assertTrue('timestamp' in event)

    def test_tags(self):
        self.client.captureMessage(message='test', tags={'logger': 'test'})

        self.assertEquals(len(self.client.events), 1)
        event = self.client.events.pop(0)
        self.assertEquals(event['tags'], {'logger': 'test'})

    def test_client_extra_context(self):
        self.client.extra = {
            'foo': 'bar',
            'logger': 'baz',
        }
        self.client.captureMessage(message='test', extra={'logger': 'test'})

        self.assertEquals(len(self.client.events), 1)
        event = self.client.events.pop(0)
        if not PY2:
            expected = {'logger': "'test'", 'foo': "'bar'"}
        else:
            expected = {'logger': "u'test'", 'foo': "u'bar'"}
        self.assertEquals(event['extra'], expected)

    def test_sample_rate(self):
        self.client.sample_rate = 0.0
        self.client.captureMessage(message='test')
        self.assertEquals(len(self.client.events), 0)

    def test_sample_rate_per_message(self):
        self.client.sample_rate = 1
        self.client.captureMessage(message='test', sample_rate=0.0)
        self.assertEquals(len(self.client.events), 0)

        self.client.sample_rate = 0
        self.client.captureMessage(message='test', sample_rate=1.0)
        self.assertEquals(len(self.client.events), 1)

    def test_transport_registration(self):
        client = Client('http://public:secret@example.com/1',
                        transport=HTTPTransport)
        assert type(client.remote.get_transport()) is HTTPTransport

        client = Client('sync+http://public:secret@example.com/1')
        assert type(client.remote.get_transport()) is HTTPTransport

    def test_marks_in_app_frames_for_stacktrace(self):
        client = TempStoreClient(
            include_paths=['foo'],
            exclude_paths=['foo.bar'],
        )
        client.captureMessage('hello', data={
            'stacktrace': {
                'frames': [
                    {'module': 'foo'},
                    {'module': 'bar'},
                    {'module': 'foo.bar'},
                    {'module': 'foo.baz'},
                ]
            }
        })

        event = client.events.pop(0)
        frames = event['stacktrace']['frames']
        assert frames[0]['in_app']
        assert not frames[1]['in_app']
        assert not frames[2]['in_app']
        assert frames[3]['in_app']

    def test_marks_in_app_frames_for_exception(self):
        client = TempStoreClient(
            include_paths=['foo'],
            exclude_paths=['foo.bar'],
        )
        client.captureMessage('hello', data={
            'exception': {
                'values': [{
                    'stacktrace': {
                        'frames': [
                            {'module': 'foo'},
                            {'module': 'bar'},
                            {'module': 'foo.bar'},
                            {'module': 'foo.baz'},
                        ]
                    }
                }]
            }
        })

        event = client.events.pop(0)
        frames = event['exception']['values'][-1]['stacktrace']['frames']
        assert frames[0]['in_app']
        assert not frames[1]['in_app']
        assert not frames[2]['in_app']
        assert frames[3]['in_app']

    def test_captures_last_event_id(self):
        client = TempStoreClient()
        result = client.captureMessage('hello')

        assert result == client.last_event_id

    def test_no_sys_argv(self):
        # if the python interpreter is started from C, sys.argv might not be available
        # see https://github.com/getsentry/raven-python/issues/918
        argv = sys.argv
        try:
            del sys.argv
            Client()
        finally:
            sys.argv = argv

    def test_repos_configuration(self):
        client = Client(repos={
            '/foo/bar': {
                'name': 'repo',
            },
            'raven': {
                'name': 'getsentry/raven-python',
            },
        })
        assert client.repos == {
            '/foo/bar': {
                'name': 'repo',
            },
            os.path.abspath(raven.__file__): {
                'name': 'getsentry/raven-python',
            },
        }