backend/src/appenlight/tests/test_unit.py

# -*- coding: utf-8 -*-

# Copyright 2010 - 2017 RhodeCode GmbH and the AppEnlight project authors
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
#   http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import copy
import logging
import mock
import pyramid
import pytest
import sqlalchemy as sa
import webob

from datetime import datetime
from pyramid import testing


from appenlight.models import DBSession
from appenlight.lib.ext_json import json


log = logging.getLogger(__name__)


class DummyContext(object):
    pass


@pytest.mark.usefixtures("base_app")
class BasicTest(object):
    pass


@pytest.mark.usefixtures("base_app")
class TestMigration(object):
    def test_migration(self):
        assert 1 == 1


class TestSentryProto_7(object):
    def test_log_payload(self):
        import appenlight.tests.payload_examples as payload_examples
        from appenlight.lib.enums import ParsedSentryEventType
        from appenlight.lib.utils.sentry import parse_sentry_event

        event_dict, event_type = parse_sentry_event(
            payload_examples.SENTRY_LOG_PAYLOAD_7
        )
        assert ParsedSentryEventType.LOG == event_type
        assert event_dict["log_level"] == "CRITICAL"
        assert event_dict["message"] == "TEST from django logging"
        assert event_dict["namespace"] == "testlogger"
        assert event_dict["request_id"] == "9a6172f2e6d2444582f83a6c333d9cfb"
        assert event_dict["server"] == "ergo-virtual-machine"
        assert event_dict["date"] == datetime.utcnow().date().strftime(
            "%Y-%m-%dT%H:%M:%SZ"
        )
        tags = [
            ("site", "example.com"),
            ("sys.argv", ["'manage.py'", "'runserver'"]),
            ("price", 6),
            ("tag", "'extra'"),
            ("dupa", True),
            ("project", "sentry"),
            ("sentry_culprit", "testlogger in index"),
            ("sentry_language", "python"),
            ("sentry_release", "test"),
        ]
        assert sorted(event_dict["tags"]) == sorted(tags)

    def test_report_payload(self):
        import appenlight.tests.payload_examples as payload_examples
        from appenlight.lib.enums import ParsedSentryEventType
        from appenlight.lib.utils.sentry import parse_sentry_event

        utcnow = datetime.utcnow().date().strftime("%Y-%m-%dT%H:%M:%SZ")
        event_dict, event_type = parse_sentry_event(
            payload_examples.SENTRY_PYTHON_PAYLOAD_7
        )
        assert ParsedSentryEventType.ERROR_REPORT == event_type
        assert event_dict["client"] == "sentry"
        assert (
            event_dict["error"] == "Exception: test 500 "
            "\u0142\xf3\u201c\u0107\u201c\u0107\u017c\u0105"
        )
        assert event_dict["language"] == "python"
        assert event_dict["ip"] == "127.0.0.1"
        assert event_dict["request_id"] == "9fae652c8c1c4d6a8eee09260f613a98"
        assert event_dict["server"] == "ergo-virtual-machine"
        assert event_dict["start_time"] == utcnow
        assert event_dict["url"] == "http://127.0.0.1:8000/error"
        assert (
            event_dict["user_agent"] == "Mozilla/5.0 (X11; Linux x86_64) "
            "AppleWebKit/537.36 (KHTML, "
            "like Gecko) Chrome/47.0.2526.106 "
            "Safari/537.36"
        )
        assert event_dict["view_name"] == "djangoapp.views in error"
        tags = [("site", "example.com"), ("sentry_release", "test")]
        assert sorted(event_dict["tags"]) == sorted(tags)
        extra = [("sys.argv", ["'manage.py'", "'runserver'"]), ("project", "sentry")]
        assert sorted(event_dict["extra"]) == sorted(extra)
        request = event_dict["request"]
        assert request["url"] == "http://127.0.0.1:8000/error"
        assert request["cookies"] == {"appenlight": "X"}
        assert request["data"] is None
        assert request["method"] == "GET"
        assert request["query_string"] == ""
        assert request["env"] == {
            "REMOTE_ADDR": "127.0.0.1",
            "SERVER_NAME": "localhost",
            "SERVER_PORT": "8000",
        }
        assert request["headers"] == {
            "Accept": "text/html,application/xhtml+xml,"
            "application/xml;q=0.9,image/webp,*/*;q=0.8",
            "Accept-Encoding": "gzip, deflate, sdch",
            "Accept-Language": "en-US,en;q=0.8,pl;q=0.6",
            "Connection": "keep-alive",
            "Content-Length": "",
            "Content-Type": "text/plain",
            "Cookie": "appenlight=X",
            "Dnt": "1",
            "Host": "127.0.0.1:8000",
            "Upgrade-Insecure-Requests": "1",
            "User-Agent": "Mozilla/5.0 (X11; Linux x86_64) "
            "AppleWebKit/537.36 (KHTML, like Gecko) "
            "Chrome/47.0.2526.106 Safari/537.36",
        }
        traceback = event_dict["traceback"]
        assert (
            traceback[0]["cline"] == "response = wrapped_callback(request, "
            "*callback_args, **callback_kwargs)"
        )
        assert traceback[0]["file"] == "django/core/handlers/base.py"
        assert traceback[0]["fn"] == "get_response"
        assert traceback[0]["line"] == 111
        assert traceback[0]["module"] == "django.core.handlers.base"

        assert (
            traceback[1]["cline"] == "raise Exception(u'test 500 "
            "\u0142\xf3\u201c\u0107\u201c\u0107"
            "\u017c\u0105')"
        )
        assert traceback[1]["file"] == "djangoapp/views.py"
        assert traceback[1]["fn"] == "error"
        assert traceback[1]["line"] == 84
        assert traceback[1]["module"] == "djangoapp.views"
        assert sorted(traceback[1]["vars"]) == sorted(
            [
                ("c", "<sqlite3.Cursor object at 0x7fe7c82af8f0>"),
                ("request", "<WSGIRequest at 0x140633490316304>"),
                ("conn", "<sqlite3.Connection object at 0x7fe7c8b23bf8>"),
            ]
        )


class TestAPIReports_0_5_Validation(object):
    @pytest.mark.parametrize("dummy_json", ["", {}, [], None])
    def test_no_payload(self, dummy_json):
        import colander
        from appenlight.validators import ReportListSchema_0_5

        utcnow = datetime.utcnow()
        schema = ReportListSchema_0_5().bind(utcnow=utcnow)
        with pytest.raises(colander.Invalid):
            schema.deserialize(dummy_json)

    def test_minimal_payload(self):
        dummy_json = [{}]
        import colander
        from appenlight.validators import ReportListSchema_0_5

        utcnow = datetime.utcnow()
        schema = ReportListSchema_0_5().bind(utcnow=utcnow)
        with pytest.raises(colander.Invalid):
            schema.deserialize(dummy_json)

    def test_minimal_payload(self):
        dummy_json = [{"report_details": [{}]}]
        from appenlight.validators import ReportListSchema_0_5

        utcnow = datetime.utcnow()
        schema = ReportListSchema_0_5().bind(utcnow=utcnow)

        deserialized = schema.deserialize(dummy_json)

        expected_deserialization = [
            {
                "language": "unknown",
                "server": "unknown",
                "occurences": 1,
                "priority": 5,
                "view_name": "",
                "client": "unknown",
                "http_status": 200,
                "error": "",
                "tags": None,
                "username": "",
                "traceback": None,
                "extra": None,
                "url": "",
                "ip": None,
                "start_time": utcnow,
                "group_string": None,
                "request": {},
                "request_stats": None,
                "end_time": None,
                "request_id": "",
                "message": "",
                "slow_calls": [],
                "user_agent": "",
            }
        ]
        assert deserialized == expected_deserialization

    def test_full_payload(self):
        import appenlight.tests.payload_examples as payload_examples
        from appenlight.validators import ReportListSchema_0_5

        PYTHON_PAYLOAD = copy.deepcopy(payload_examples.PYTHON_PAYLOAD_0_5)
        utcnow = datetime.utcnow()
        schema = ReportListSchema_0_5().bind(utcnow=utcnow)
        PYTHON_PAYLOAD["tags"] = [
            ("foo", 1),
            ("action", "test"),
            ("baz", 1.1),
            ("date", utcnow.strftime("%Y-%m-%dT%H:%M:%S.0")),
        ]
        dummy_json = [PYTHON_PAYLOAD]
        deserialized = schema.deserialize(dummy_json)[0]
        assert deserialized["error"] == PYTHON_PAYLOAD["error"]
        assert deserialized["language"] == PYTHON_PAYLOAD["language"]
        assert deserialized["server"] == PYTHON_PAYLOAD["server"]
        assert deserialized["priority"] == PYTHON_PAYLOAD["priority"]
        assert deserialized["view_name"] == PYTHON_PAYLOAD["view_name"]
        assert deserialized["client"] == PYTHON_PAYLOAD["client"]
        assert deserialized["http_status"] == PYTHON_PAYLOAD["http_status"]
        assert deserialized["error"] == PYTHON_PAYLOAD["error"]
        assert deserialized["occurences"] == PYTHON_PAYLOAD["occurences"]
        assert deserialized["username"] == PYTHON_PAYLOAD["username"]
        assert deserialized["traceback"] == PYTHON_PAYLOAD["traceback"]
        assert deserialized["url"] == PYTHON_PAYLOAD["url"]
        assert deserialized["ip"] == PYTHON_PAYLOAD["ip"]
        assert (
            deserialized["start_time"].strftime("%Y-%m-%dT%H:%M:%S.0")
            == PYTHON_PAYLOAD["start_time"]
        )
        assert deserialized["ip"] == PYTHON_PAYLOAD["ip"]
        assert deserialized["group_string"] is None
        assert deserialized["request_stats"] == PYTHON_PAYLOAD["request_stats"]
        assert (
            deserialized["end_time"].strftime("%Y-%m-%dT%H:%M:%S.0")
            == PYTHON_PAYLOAD["end_time"]
        )
        assert deserialized["request_id"] == PYTHON_PAYLOAD["request_id"]
        assert deserialized["message"] == PYTHON_PAYLOAD["message"]
        assert deserialized["user_agent"] == PYTHON_PAYLOAD["user_agent"]
        assert (
            deserialized["slow_calls"][0]["start"].strftime("%Y-%m-%dT%H:%M:%S.0")
            == PYTHON_PAYLOAD["slow_calls"][0]["start"]
        )
        assert (
            deserialized["slow_calls"][0]["end"].strftime("%Y-%m-%dT%H:%M:%S.0")
            == PYTHON_PAYLOAD["slow_calls"][0]["end"]
        )
        assert (
            deserialized["slow_calls"][0]["statement"]
            == PYTHON_PAYLOAD["slow_calls"][0]["statement"]
        )
        assert (
            deserialized["slow_calls"][0]["parameters"]
            == PYTHON_PAYLOAD["slow_calls"][0]["parameters"]
        )
        assert (
            deserialized["slow_calls"][0]["type"]
            == PYTHON_PAYLOAD["slow_calls"][0]["type"]
        )
        assert (
            deserialized["slow_calls"][0]["subtype"]
            == PYTHON_PAYLOAD["slow_calls"][0]["subtype"]
        )
        assert deserialized["slow_calls"][0]["location"] == ""
        assert deserialized["tags"] == [
            ("foo", 1),
            ("action", "test"),
            ("baz", 1.1),
            ("date", utcnow.strftime("%Y-%m-%dT%H:%M:%S.0")),
        ]


@pytest.mark.usefixtures("log_schema")
class TestAPILogsValidation(object):
    @pytest.mark.parametrize("dummy_json", ["", {}, [], None])
    def test_no_payload(self, dummy_json, log_schema):
        import colander

        with pytest.raises(colander.Invalid):
            log_schema.deserialize(dummy_json)

    def test_minimal_payload(self, log_schema):
        dummy_json = [{}]
        deserialized = log_schema.deserialize(dummy_json)[0]
        expected = {
            "log_level": "UNKNOWN",
            "namespace": "",
            "server": "unknown",
            "request_id": "",
            "primary_key": None,
            "date": datetime.utcnow(),
            "message": "",
            "tags": None,
        }
        assert deserialized["log_level"] == expected["log_level"]
        assert deserialized["message"] == expected["message"]
        assert deserialized["namespace"] == expected["namespace"]
        assert deserialized["request_id"] == expected["request_id"]
        assert deserialized["server"] == expected["server"]
        assert deserialized["tags"] == expected["tags"]
        assert deserialized["primary_key"] == expected["primary_key"]

    def test_normal_payload(self, log_schema):
        import appenlight.tests.payload_examples as payload_examples

        deserialized = log_schema.deserialize(payload_examples.LOG_EXAMPLES)[0]
        expected = payload_examples.LOG_EXAMPLES[0]
        assert deserialized["log_level"] == expected["log_level"]
        assert deserialized["message"] == expected["message"]
        assert deserialized["namespace"] == expected["namespace"]
        assert deserialized["request_id"] == expected["request_id"]
        assert deserialized["server"] == expected["server"]
        assert deserialized["date"].strftime("%Y-%m-%dT%H:%M:%S.%f") == expected["date"]
        assert deserialized["tags"][0][0] == "tag_name"
        assert deserialized["tags"][0][1] == "tag_value"
        assert deserialized["tags"][1][0] == "tag_name2"
        assert deserialized["tags"][1][1] == 2

    def test_normal_payload_date_without_microseconds(self, log_schema):
        import appenlight.tests.payload_examples as payload_examples

        LOG_EXAMPLE = copy.deepcopy(payload_examples.LOG_EXAMPLES)
        LOG_EXAMPLE[0]["date"] = datetime.utcnow().strftime("%Y-%m-%dT%H:%M:%S")
        deserialized = log_schema.deserialize(LOG_EXAMPLE)
        assert (
            deserialized[0]["date"].strftime("%Y-%m-%dT%H:%M:%S")
            == LOG_EXAMPLE[0]["date"]
        )

    def test_normal_payload_date_without_seconds(self, log_schema):
        import appenlight.tests.payload_examples as payload_examples

        LOG_EXAMPLE = copy.deepcopy(payload_examples.LOG_EXAMPLES)
        LOG_EXAMPLE[0]["date"] = datetime.utcnow().date().strftime("%Y-%m-%dT%H:%M")
        deserialized = log_schema.deserialize(LOG_EXAMPLE)
        assert (
            deserialized[0]["date"].strftime("%Y-%m-%dT%H:%M") == LOG_EXAMPLE[0]["date"]
        )

    def test_payload_empty_date(self, log_schema):
        import appenlight.tests.payload_examples as payload_examples

        LOG_EXAMPLE = copy.deepcopy(payload_examples.LOG_EXAMPLES)
        LOG_EXAMPLE[0]["date"] = None
        deserialized = log_schema.deserialize(LOG_EXAMPLE)
        assert deserialized[0]["date"].strftime("%Y-%m-%dT%H:%M") is not None

    def test_payload_no_date(self, log_schema):
        import appenlight.tests.payload_examples as payload_examples

        LOG_EXAMPLE = copy.deepcopy(payload_examples.LOG_EXAMPLES)
        LOG_EXAMPLE[0].pop("date", None)
        deserialized = log_schema.deserialize(LOG_EXAMPLE)
        assert deserialized[0]["date"].strftime("%Y-%m-%dT%H:%M") is not None


@pytest.mark.usefixtures("general_metrics_schema")
class TestAPIGeneralMetricsValidation(object):
    @pytest.mark.parametrize("dummy_json", ["", {}, [], None])
    def test_no_payload(self, dummy_json, general_metrics_schema):
        import colander

        with pytest.raises(colander.Invalid):
            general_metrics_schema.deserialize(dummy_json)

    def test_minimal_payload(self, general_metrics_schema):
        dummy_json = [{"tags": [["counter_a", 15.5], ["counter_b", 63]]}]
        deserialized = general_metrics_schema.deserialize(dummy_json)[0]
        expected = {
            "namespace": "",
            "server_name": "unknown",
            "tags": [("counter_a", 15.5), ("counter_b", 63)],
            "timestamp": datetime.utcnow(),
        }
        assert deserialized["namespace"] == expected["namespace"]
        assert deserialized["server_name"] == expected["server_name"]
        assert deserialized["tags"] == expected["tags"]

    def test_normal_payload(self, general_metrics_schema):
        import appenlight.tests.payload_examples as payload_examples

        dummy_json = [payload_examples.METRICS_PAYLOAD]
        deserialized = general_metrics_schema.deserialize(dummy_json)[0]
        expected = {
            "namespace": "some.monitor",
            "server_name": "server.name",
            "tags": [("usage_foo", 15.5), ("usage_bar", 63)],
            "timestamp": datetime.utcnow(),
        }
        assert deserialized["namespace"] == expected["namespace"]
        assert deserialized["server_name"] == expected["server_name"]
        assert deserialized["tags"] == expected["tags"]


@pytest.mark.usefixtures("request_metrics_schema")
class TestAPIRequestMetricsValidation(object):
    @pytest.mark.parametrize("dummy_json", ["", {}, [], None])
    def test_no_payload(self, dummy_json, request_metrics_schema):
        import colander

        with pytest.raises(colander.Invalid):
            print(request_metrics_schema.deserialize(dummy_json))

    def test_normal_payload(self, request_metrics_schema):
        import appenlight.tests.payload_examples as payload_examples

        dummy_json = payload_examples.REQUEST_METRICS_EXAMPLES
        deserialized = request_metrics_schema.deserialize(dummy_json)[0]
        expected = {
            "metrics": [
                (
                    "dir/module:func",
                    {
                        "custom": 0.0,
                        "custom_calls": 0.0,
                        "main": 0.01664,
                        "nosql": 0.00061,
                        "nosql_calls": 23.0,
                        "remote": 0.0,
                        "remote_calls": 0.0,
                        "requests": 1,
                        "sql": 0.00105,
                        "sql_calls": 2.0,
                        "tmpl": 0.0,
                        "tmpl_calls": 0.0,
                    },
                ),
                (
                    "SomeView.function",
                    {
                        "custom": 0.0,
                        "custom_calls": 0.0,
                        "main": 0.647261,
                        "nosql": 0.306554,
                        "nosql_calls": 140.0,
                        "remote": 0.0,
                        "remote_calls": 0.0,
                        "requests": 28,
                        "sql": 0.0,
                        "sql_calls": 0.0,
                        "tmpl": 0.0,
                        "tmpl_calls": 0.0,
                    },
                ),
            ],
            "server": "some.server.hostname",
            "timestamp": datetime.utcnow(),
        }
        assert deserialized["server"] == expected["server"]
        metric = deserialized["metrics"][0]
        expected_metric = expected["metrics"][0]
        assert metric[0] == expected_metric[0]
        assert sorted(metric[1].items()) == sorted(expected_metric[1].items())


@pytest.mark.usefixtures("default_application")
@pytest.mark.usefixtures("base_app", "with_migrations", "clean_tables")
class TestAPIReportsView(object):
    def test_no_json_payload(self, default_application):
        import colander
        from appenlight.models.services.application import ApplicationService
        from appenlight.views.api import reports_create

        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        request = testing.DummyRequest(headers={"Content-Type": "application/json"})
        request.unsafe_json_body = ""
        request.context = context
        route = mock.Mock()
        route.name = "api_reports"
        request.matched_route = route
        with pytest.raises(colander.Invalid):
            response = reports_create(request)

    def test_single_proper_json_0_5_payload(self):
        import appenlight.tests.payload_examples as payload_examples
        from appenlight.views.api import reports_create
        from appenlight.models.services.application import ApplicationService
        from appenlight.models.report_group import ReportGroup

        route = mock.Mock()
        route.name = "api_reports"
        request = pyramid.threadlocal.get_current_request()
        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        request.context = context
        request.matched_route = route
        PYTHON_PAYLOAD = payload_examples.PYTHON_PAYLOAD_0_5
        request.unsafe_json_body = [copy.deepcopy(PYTHON_PAYLOAD)]
        reports_create(request)
        query = DBSession.query(ReportGroup)
        report = query.first()
        assert query.count() == 1
        assert report.total_reports == 1

    def test_grouping_0_5(self):
        import appenlight.tests.payload_examples as payload_examples
        from appenlight.views.api import reports_create
        from appenlight.models.services.application import ApplicationService
        from appenlight.models.report_group import ReportGroup

        route = mock.Mock()
        route.name = "api_reports"
        request = pyramid.threadlocal.get_current_request()
        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        request.context = context
        request.matched_route = route
        PYTHON_PAYLOAD = payload_examples.PYTHON_PAYLOAD_0_5
        request.unsafe_json_body = [
            copy.deepcopy(PYTHON_PAYLOAD),
            copy.deepcopy(PYTHON_PAYLOAD),
        ]
        reports_create(request)
        query = DBSession.query(ReportGroup)
        report = query.first()
        assert query.count() == 1
        assert report.total_reports == 2

    def test_grouping_different_reports_0_5(self):
        import appenlight.tests.payload_examples as payload_examples
        from appenlight.views.api import reports_create
        from appenlight.models.services.application import ApplicationService
        from appenlight.models.report_group import ReportGroup

        route = mock.Mock()
        route.name = "api_reports"
        request = pyramid.threadlocal.get_current_request()
        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        request.context = context
        request.matched_route = route
        PYTHON_PAYLOAD = payload_examples.PYTHON_PAYLOAD_0_5
        PARSED_REPORT_404 = payload_examples.PARSED_REPORT_404
        request.unsafe_json_body = [
            copy.deepcopy(PYTHON_PAYLOAD),
            copy.deepcopy(PARSED_REPORT_404),
        ]
        reports_create(request)
        query = DBSession.query(ReportGroup)
        report = query.first()
        assert query.count() == 2
        assert report.total_reports == 1


@pytest.mark.usefixtures("default_application")
@pytest.mark.usefixtures("base_app", "with_migrations", "clean_tables")
class TestAirbrakeXMLView(object):
    def test_normal_payload_parsing(self):
        import datetime
        import defusedxml.ElementTree as ElementTree
        import appenlight.tests.payload_examples as payload_examples
        from appenlight.lib.utils.airbrake import parse_airbrake_xml
        from appenlight.validators import ReportListSchema_0_5

        context = DummyContext()
        request = testing.DummyRequest(headers={"Content-Type": "application/xml"})
        request.context = context
        request.context.possibly_public = False
        root = ElementTree.fromstring(payload_examples.AIRBRAKE_RUBY_EXAMPLE)
        request.context.airbrake_xml_etree = root
        error_dict = parse_airbrake_xml(request)
        schema = ReportListSchema_0_5().bind(utcnow=datetime.datetime.utcnow())
        deserialized_report = schema.deserialize([error_dict])[0]
        assert deserialized_report["client"] == "Airbrake Notifier"
        assert (
            deserialized_report["error"]
            == "NameError: undefined local variable or method `sdfdfdf' for #<#<Class:0x000000039a8b90>:0x00000002c53df0>"
        )
        assert deserialized_report["http_status"] == 500
        assert deserialized_report["language"] == "unknown"
        assert deserialized_report["message"] == ""
        assert deserialized_report["occurences"] == 1
        assert deserialized_report["priority"] == 5
        d_request = deserialized_report["request"]
        assert d_request["GET"] == {"test": "1234"}
        assert d_request["action_dispatch.request.parameters"] == {
            "action": "index",
            "controller": "welcome",
            "test": "1234",
        }
        assert deserialized_report["request_id"] == "c11b2267f3ad8b00a1768cae35559fa1"
        assert deserialized_report["server"] == "ergo-desktop"
        assert deserialized_report["traceback"][0] == {
            "cline": "block in start_thread",
            "file": "/home/ergo/.rbenv/versions/1.9.3-p327/lib/ruby/1.9.1/webrick/server.rb",
            "fn": "block in start_thread",
            "line": "191",
            "module": "",
            "vars": {},
        }
        assert deserialized_report["traceback"][-1] == {
            "cline": "_app_views_welcome_index_html_erb___2570061166873166679_31748940",
            "file": "[PROJECT_ROOT]/app/views/welcome/index.html.erb",
            "fn": "_app_views_welcome_index_html_erb___2570061166873166679_31748940",
            "line": "3",
            "module": "",
            "vars": {},
        }
        assert (
            deserialized_report["url"] == "http://0.0.0.0:3000/welcome/index?test=1234"
        )
        assert deserialized_report["view_name"] == "welcome:index"

    def test_normal_payload_view(self):
        import defusedxml.ElementTree as ElementTree
        import appenlight.tests.payload_examples as payload_examples

        from appenlight.models.services.application import ApplicationService
        from appenlight.views.api import airbrake_xml_compat

        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        request = testing.DummyRequest(headers={"Content-Type": "application/xml"})
        request.context = context
        request.context.possibly_public = False
        root = ElementTree.fromstring(payload_examples.AIRBRAKE_RUBY_EXAMPLE)
        request.context.airbrake_xml_etree = root
        route = mock.Mock()
        route.name = "api_airbrake"
        request.matched_route = route
        result = airbrake_xml_compat(request)
        assert "<notice><id>" in result


@pytest.mark.usefixtures("default_application")
@pytest.mark.usefixtures("base_app", "with_migrations", "clean_tables")
class TestAPILogView(object):
    def test_no_json_payload(self, base_app):
        import colander
        from appenlight.models.services.application import ApplicationService
        from appenlight.views.api import logs_create

        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        request = testing.DummyRequest(headers={"Content-Type": "application/json"})
        request.context = context
        request.registry = base_app.registry
        request.unsafe_json_body = ""
        route = mock.Mock()
        route.name = "api_logs"
        request.matched_route = route
        with pytest.raises(colander.Invalid):
            response = logs_create(request)

    def test_single_json_payload(self):
        import appenlight.tests.payload_examples as payload_examples
        from appenlight.models.log import Log
        from appenlight.views.api import logs_create
        from appenlight.models.services.application import ApplicationService

        route = mock.Mock()
        route.name = "api_logs"
        request = pyramid.threadlocal.get_current_request()
        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        request.context = context
        request.matched_route = route
        request.unsafe_json_body = [copy.deepcopy(payload_examples.LOG_EXAMPLES[0])]
        logs_create(request)
        query = DBSession.query(Log)
        log = query.first()
        assert query.count() == 1
        assert log.message == "OMG ValueError happened"

    def test_multiple_json_payload(self):
        import appenlight.tests.payload_examples as payload_examples
        from appenlight.models.log import Log
        from appenlight.views.api import logs_create
        from appenlight.models.services.application import ApplicationService

        route = mock.Mock()
        route.name = "api_logs"
        request = pyramid.threadlocal.get_current_request()
        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        request.context = context
        request.matched_route = route
        LOG_PAYLOAD = payload_examples.LOG_EXAMPLES[0]
        LOG_PAYLOAD2 = payload_examples.LOG_EXAMPLES[1]
        request.unsafe_json_body = copy.deepcopy([LOG_PAYLOAD, LOG_PAYLOAD2])
        logs_create(request)
        query = DBSession.query(Log).order_by(sa.asc(Log.log_id))
        assert query.count() == 2
        assert query[0].message == "OMG ValueError happened"
        assert query[1].message == "OMG ValueError happened2"

    def test_public_key_rewriting(self):
        import appenlight.tests.payload_examples as payload_examples
        from appenlight.models.log import Log
        from appenlight.views.api import logs_create
        from appenlight.models.services.application import ApplicationService

        route = mock.Mock()
        route.name = "api_logs"
        request = pyramid.threadlocal.get_current_request()
        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        request.context = context
        request.matched_route = route

        LOG_PAYLOAD = copy.deepcopy(payload_examples.LOG_EXAMPLES[0])
        LOG_PAYLOAD2 = copy.deepcopy(payload_examples.LOG_EXAMPLES[1])
        LOG_PAYLOAD["primary_key"] = "X2"
        LOG_PAYLOAD2["primary_key"] = "X2"
        request.unsafe_json_body = [LOG_PAYLOAD, LOG_PAYLOAD2]
        logs_create(request)

        query = DBSession.query(Log).order_by(sa.asc(Log.log_id))
        assert query.count() == 1
        assert query[0].message == "OMG ValueError happened2"


@pytest.mark.usefixtures("default_application")
@pytest.mark.usefixtures("base_app", "with_migrations", "clean_tables")
class TestAPIGeneralMetricsView(object):
    def test_no_json_payload(self, base_app):
        import colander
        from appenlight.models.services.application import ApplicationService
        from appenlight.views.api import general_metrics_create

        route = mock.Mock()
        route.name = "api_general_metrics"
        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        request = testing.DummyRequest(headers={"Content-Type": "application/json"})
        request.context = context
        request.registry = base_app.registry
        request.unsafe_json_body = ""
        request.matched_route = route
        with pytest.raises(colander.Invalid):
            general_metrics_create(request)

    def test_single_json_payload(self):
        import appenlight.tests.payload_examples as payload_examples
        from appenlight.models.metric import Metric
        from appenlight.views.api import general_metrics_create
        from appenlight.models.services.application import ApplicationService

        route = mock.Mock()
        route.name = "api_general_metric"
        request = pyramid.threadlocal.get_current_request()
        request.matched_route = route
        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        request.context = context
        request.unsafe_json_body = payload_examples.METRICS_PAYLOAD
        general_metrics_create(request)
        query = DBSession.query(Metric)
        metric = query.first()
        assert query.count() == 1
        assert metric.namespace == "some.monitor"

    def test_multiple_json_payload(self):
        import appenlight.tests.payload_examples as payload_examples
        from appenlight.models.metric import Metric
        from appenlight.views.api import general_metrics_create
        from appenlight.models.services.application import ApplicationService

        route = mock.Mock()
        route.name = "api_general_metrics"
        request = pyramid.threadlocal.get_current_request()
        request.matched_route = route
        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        request.context = context
        request.unsafe_json_body = [
            copy.deepcopy(payload_examples.METRICS_PAYLOAD),
            copy.deepcopy(payload_examples.METRICS_PAYLOAD),
        ]
        general_metrics_create(request)
        query = DBSession.query(Metric)
        metric = query.first()
        assert query.count() == 2
        assert metric.namespace == "some.monitor"


class TestGroupingMessageReplacements(object):
    def replace_default_repr_python(self):
        test_str = """
        ConnectionError: ConnectionError((<urllib3.connection.HTTPConnection object at 0x7f87a0ba9fd0>, 'Connection to domain.gr timed out. (connect timeout=10)')) caused by: ConnectTimeoutError((<urllib3.connection.HTTPConnection object at 0x7f87a0ba9fd0>, 'Connection to domain.gr timed out. (connect timeout=10)'))
        """
        regex = r"<(.*?) object at (.*?)>"


class TestRulesKeyGetter(object):
    def test_default_dict_getter_top_key(self):
        from appenlight.lib.rule import Rule

        struct = {"a": {"b": "b", "c": {"d": "d", "g": {"h": "h"}}, "e": "e"}, "f": "f"}
        result = Rule.default_dict_struct_getter(struct, "a")
        assert result == struct["a"]

    def test_default_dict_getter_sub_key(self):
        from appenlight.lib.rule import Rule

        struct = {"a": {"b": "b", "c": {"d": "d", "g": {"h": "h"}}, "e": "e"}, "f": "f"}
        result = Rule.default_dict_struct_getter(struct, "a:b")
        assert result == struct["a"]["b"]
        result = Rule.default_dict_struct_getter(struct, "a:c:d")
        assert result == struct["a"]["c"]["d"]

    def test_default_obj_getter_top_key(self):
        from appenlight.lib.rule import Rule

        class TestStruct(object):
            def __init__(self, a, b):
                self.a = a
                self.b = b

        struct = TestStruct(a="a", b=TestStruct(a="x", b="y"))
        result = Rule.default_obj_struct_getter(struct, "a")
        assert result == struct.a

    def test_default_obj_getter_sub_key(self):
        from appenlight.lib.rule import Rule

        class TestStruct(object):
            def __init__(self, name, a, b):
                self.name = name
                self.a = a
                self.b = b

            def __repr__(self):
                return "<obj {}>".format(self.name)

        c = TestStruct("c", a=5, b="z")
        b = TestStruct("b", a=c, b="y")
        struct = TestStruct("a", a="a", b=b)
        result = Rule.default_obj_struct_getter(struct, "b:b")
        assert result == struct.b.b
        result = Rule.default_obj_struct_getter(struct, "b:a:b")
        assert result == struct.b.a.b


@pytest.mark.usefixtures("report_type_matrix")
class TestRulesParsing:
    @pytest.mark.parametrize(
        "op, struct_value, test_value, match_result",
        [
            ("eq", 500, 500, True),
            ("eq", 600, 500, False),
            ("eq", 300, 500, False),
            ("eq", "300", 500, False),
            ("eq", "600", 500, False),
            ("eq", "500", 500, True),
            ("ne", 500, 500, False),
            ("ne", 600, 500, True),
            ("ne", 300, 500, True),
            ("ne", "300", 500, True),
            ("ne", "600", 500, True),
            ("ne", "500", 500, False),
            ("ge", 500, 500, True),
            ("ge", 600, 500, True),
            ("ge", 499, 500, False),
            ("gt", 499, 500, False),
            ("gt", 500, 500, False),
            ("gt", 501, 500, True),
            ("le", 499, 500, True),
            ("le", 500, 500, True),
            ("le", 501, 500, False),
            ("lt", 499, 500, True),
            ("lt", 500, 500, False),
            ("lt", 501, 500, False),
        ],
    )
    def test_single_op_int(
        self, op, struct_value, test_value, match_result, report_type_matrix
    ):
        from appenlight.lib.rule import Rule

        rule_config = {"op": op, "field": "http_status", "value": test_value}
        rule = Rule(rule_config, report_type_matrix)

        data = {"http_status": struct_value}
        assert rule.match(data) is match_result

    @pytest.mark.parametrize(
        "op, struct_value, test_value, match_result",
        [
            ("ge", "500.01", 500, True),
            ("ge", "500.01", 500.02, False),
            ("le", "500.01", 500.02, True),
        ],
    )
    def test_single_op_float(
        self, op, struct_value, test_value, match_result, report_type_matrix
    ):
        from appenlight.lib.rule import Rule

        rule_config = {"op": op, "field": "duration", "value": test_value}
        rule = Rule(rule_config, report_type_matrix)

        data = {"duration": struct_value}
        assert rule.match(data) is match_result

    @pytest.mark.parametrize(
        "op, struct_value, test_value, match_result",
        [
            ("contains", "foo bar baz", "foo", True),
            ("contains", "foo bar baz", "bar", True),
            ("contains", "foo bar baz", "dupa", False),
            ("startswith", "foo bar baz", "foo", True),
            ("startswith", "foo bar baz", "bar", False),
            ("endswith", "foo bar baz", "baz", True),
            ("endswith", "foo bar baz", "bar", False),
        ],
    )
    def test_single_op_string(
        self, op, struct_value, test_value, match_result, report_type_matrix
    ):
        from appenlight.lib.rule import Rule

        rule_config = {"op": op, "field": "error", "value": test_value}
        rule = Rule(rule_config, report_type_matrix)

        data = {"error": struct_value}
        assert rule.match(data) is match_result

    @pytest.mark.parametrize(
        "field, value, s_type",
        [
            ("field_unicode", 500, str),
            ("field_unicode", 500.0, str),
            ("field_unicode", "500", str),
            ("field_int", "500", int),
            ("field_int", 500, int),
            ("field_int", 500.0, int),
            ("field_float", "500", float),
            ("field_float", 500, float),
            ("field_float", 500.0, float),
        ],
    )
    def test_type_normalization(self, field, value, s_type):
        from appenlight.lib.rule import Rule

        type_matrix = {
            "field_unicode": {"type": "unicode"},
            "field_float": {"type": "float"},
            "field_int": {"type": "int"},
        }

        rule = Rule({}, type_matrix)
        n_value = rule.normalized_type(field, value)
        assert isinstance(n_value, s_type) is True


@pytest.mark.usefixtures("report_type_matrix")
class TestNestedRuleParsing:
    @pytest.mark.parametrize(
        "data, result",
        [
            ({"http_status": 501, "group": {"priority": 7, "occurences": 11}}, False),
            ({"http_status": 101, "group": {"priority": 7, "occurences": 11}}, False),
            ({"http_status": 500, "group": {"priority": 1, "occurences": 11}}, False),
            ({"http_status": 101, "group": {"priority": 3, "occurences": 5}}, True),
        ],
    )
    def test_NOT_rule(self, data, result, report_type_matrix):
        from appenlight.lib.rule import Rule

        rule_config = {
            "field": "__NOT__",
            "rules": [
                {"op": "ge", "field": "group:occurences", "value": "10"},
                {"op": "ge", "field": "group:priority", "value": "4"},
            ],
        }

        rule = Rule(rule_config, report_type_matrix)
        assert rule.match(data) is result

    @pytest.mark.parametrize(
        "data, result",
        [
            ({"http_status": 501, "group": {"priority": 7, "occurences": 11}}, True),
            ({"http_status": 101, "group": {"priority": 7, "occurences": 11}}, True),
            ({"http_status": 500, "group": {"priority": 1, "occurences": 1}}, True),
            ({"http_status": 101, "group": {"priority": 3, "occurences": 11}}, False),
        ],
    )
    def test_nested_OR_AND_rule(self, data, result, report_type_matrix):
        from appenlight.lib.rule import Rule

        rule_config = {
            "field": "__OR__",
            "rules": [
                {
                    "field": "__AND__",
                    "rules": [
                        {"op": "ge", "field": "group:occurences", "value": "10"},
                        {"op": "ge", "field": "group:priority", "value": "4"},
                    ],
                },
                {"op": "eq", "field": "http_status", "value": "500"},
            ],
        }

        rule = Rule(rule_config, report_type_matrix)
        assert rule.match(data) is result

    @pytest.mark.parametrize(
        "data, result",
        [
            ({"http_status": 501, "group": {"priority": 7, "occurences": 11}}, True),
            ({"http_status": 101, "group": {"priority": 7, "occurences": 11}}, True),
            ({"http_status": 500, "group": {"priority": 1, "occurences": 1}}, True),
            ({"http_status": 101, "group": {"priority": 3, "occurences": 1}}, False),
        ],
    )
    def test_nested_OR_OR_rule(self, data, result, report_type_matrix):
        from appenlight.lib.rule import Rule

        rule_config = {
            "field": "__OR__",
            "rules": [
                {
                    "field": "__OR__",
                    "rules": [
                        {"op": "ge", "field": "group:occurences", "value": "10"},
                        {"op": "ge", "field": "group:priority", "value": "4"},
                    ],
                },
                {"op": "eq", "field": "http_status", "value": "500"},
            ],
        }

        rule = Rule(rule_config, report_type_matrix)
        assert rule.match(data) is result

    @pytest.mark.parametrize(
        "data, result",
        [
            ({"http_status": 500, "group": {"priority": 7, "occurences": 11}}, True),
            ({"http_status": 101, "group": {"priority": 7, "occurences": 11}}, False),
            ({"http_status": 500, "group": {"priority": 1, "occurences": 1}}, False),
            ({"http_status": 101, "group": {"priority": 3, "occurences": 1}}, False),
        ],
    )
    def test_nested_AND_AND_rule(self, data, result, report_type_matrix):
        from appenlight.lib.rule import Rule

        rule_config = {
            "field": "__AND__",
            "rules": [
                {
                    "field": "__AND__",
                    "rules": [
                        {"op": "ge", "field": "group:occurences", "value": "10"},
                        {"op": "ge", "field": "group:priority", "value": "4"},
                    ],
                },
                {"op": "eq", "field": "http_status", "value": "500"},
            ],
        }

        rule = Rule(rule_config, report_type_matrix)
        assert rule.match(data) is result

    @pytest.mark.parametrize(
        "data, result",
        [
            (
                {
                    "http_status": 500,
                    "group": {"priority": 7, "occurences": 11},
                    "url_path": "/test/register",
                    "error": "foo test bar",
                },
                True,
            ),
            (
                {
                    "http_status": 500,
                    "group": {"priority": 7, "occurences": 11},
                    "url_path": "/test/register",
                    "error": "foo INVALID bar",
                },
                False,
            ),
        ],
    )
    def test_nested_AND_AND_AND_rule(self, data, result, report_type_matrix):
        from appenlight.lib.rule import Rule

        rule_config = {
            "field": "__AND__",
            "rules": [
                {
                    "field": "__AND__",
                    "rules": [
                        {"op": "ge", "field": "group:occurences", "value": "10"},
                        {
                            "field": "__AND__",
                            "rules": [
                                {
                                    "op": "endswith",
                                    "field": "url_path",
                                    "value": "register",
                                },
                                {"op": "contains", "field": "error", "value": "test"},
                            ],
                        },
                    ],
                },
                {"op": "eq", "field": "http_status", "value": "500"},
            ],
        }

        rule = Rule(rule_config, report_type_matrix)
        assert rule.match(data) is result

    @pytest.mark.parametrize(
        "data, result",
        [
            (
                {
                    "http_status": 500,
                    "group": {"priority": 7, "occurences": 11},
                    "url_path": 6,
                    "error": 3,
                },
                False,
            ),
            (
                {
                    "http_status": 500,
                    "group": {"priority": 7, "occurences": 11},
                    "url_path": "/test/register",
                    "error": "foo INVALID bar",
                },
                True,
            ),
        ],
    )
    def test_nested_AND_AND_OR_rule(self, data, result, report_type_matrix):
        from appenlight.lib.rule import Rule

        rule_config = {
            "field": "__AND__",
            "rules": [
                {
                    "field": "__AND__",
                    "rules": [
                        {"op": "ge", "field": "group:occurences", "value": "10"},
                        {
                            "field": "__OR__",
                            "rules": [
                                {
                                    "op": "endswith",
                                    "field": "url_path",
                                    "value": "register",
                                },
                                {"op": "contains", "field": "error", "value": "test"},
                            ],
                        },
                    ],
                },
                {"op": "eq", "field": "http_status", "value": "500"},
            ],
        }

        rule = Rule(rule_config, report_type_matrix)
        assert rule.match(data) is result

    @pytest.mark.parametrize(
        "op, field, value, should_fail",
        [
            ("eq", "http_status", "1", False),
            ("ne", "http_status", "1", False),
            ("ne", "http_status", "foo", True),
            ("startswith", "http_status", "1", True),
            ("eq", "group:priority", "1", False),
            ("ne", "group:priority", "1", False),
            ("ge", "group:priority", "1", False),
            ("le", "group:priority", "1", False),
            ("startswith", "group:priority", "1", True),
            ("eq", "url_domain", "1", False),
            ("ne", "url_domain", "1", False),
            ("startswith", "url_domain", "1", False),
            ("endswith", "url_domain", "1", False),
            ("contains", "url_domain", "1", False),
            ("ge", "url_domain", "1", True),
            ("eq", "url_path", "1", False),
            ("ne", "url_path", "1", False),
            ("startswith", "url_path", "1", False),
            ("endswith", "url_path", "1", False),
            ("contains", "url_path", "1", False),
            ("ge", "url_path", "1", True),
            ("eq", "error", "1", False),
            ("ne", "error", "1", False),
            ("startswith", "error", "1", False),
            ("endswith", "error", "1", False),
            ("contains", "error", "1", False),
            ("ge", "error", "1", True),
            ("ge", "url_path", "1", True),
            ("eq", "tags:server_name", "1", False),
            ("ne", "tags:server_name", "1", False),
            ("startswith", "tags:server_name", "1", False),
            ("endswith", "tags:server_name", "1", False),
            ("contains", "tags:server_name", "1", False),
            ("ge", "tags:server_name", "1", True),
            ("contains", "traceback", "1", False),
            ("ge", "traceback", "1", True),
            ("eq", "group:occurences", "1", False),
            ("ne", "group:occurences", "1", False),
            ("ge", "group:occurences", "1", False),
            ("le", "group:occurences", "1", False),
            ("contains", "group:occurences", "1", True),
        ],
    )
    def test_rule_validation(self, op, field, value, should_fail, report_type_matrix):
        import colander
        from appenlight.validators import build_rule_schema

        rule_config = {"op": op, "field": field, "value": value}

        schema = build_rule_schema(rule_config, report_type_matrix)
        if should_fail:
            with pytest.raises(colander.Invalid):
                schema.deserialize(rule_config)
        else:
            schema.deserialize(rule_config)

    def test_nested_proper_rule_validation(self, report_type_matrix):
        from appenlight.validators import build_rule_schema

        rule_config = {
            "field": "__AND__",
            "rules": [
                {
                    "field": "__AND__",
                    "rules": [
                        {"op": "ge", "field": "group:occurences", "value": "10"},
                        {
                            "field": "__OR__",
                            "rules": [
                                {
                                    "op": "endswith",
                                    "field": "url_path",
                                    "value": "register",
                                },
                                {"op": "contains", "field": "error", "value": "test"},
                            ],
                        },
                    ],
                },
                {"op": "eq", "field": "http_status", "value": "500"},
            ],
        }

        schema = build_rule_schema(rule_config, report_type_matrix)
        deserialized = schema.deserialize(rule_config)

    def test_nested_bad_rule_validation(self, report_type_matrix):
        import colander
        from appenlight.validators import build_rule_schema

        rule_config = {
            "field": "__AND__",
            "rules": [
                {
                    "field": "__AND__",
                    "rules": [
                        {"op": "ge", "field": "group:occurences", "value": "10"},
                        {
                            "field": "__OR__",
                            "rules": [
                                {"op": "gt", "field": "url_path", "value": "register"},
                                {"op": "contains", "field": "error", "value": "test"},
                            ],
                        },
                    ],
                },
                {"op": "eq", "field": "http_status", "value": "500"},
            ],
        }

        schema = build_rule_schema(rule_config, report_type_matrix)
        with pytest.raises(colander.Invalid):
            deserialized = schema.deserialize(rule_config)

    def test_config_manipulator(self):
        from appenlight.lib.rule import Rule

        type_matrix = {
            "a": {"type": "int", "ops": ("eq", "ne", "ge", "le")},
            "b": {"type": "int", "ops": ("eq", "ne", "ge", "le")},
        }
        rule_config = {
            "field": "__OR__",
            "rules": [
                {
                    "field": "__OR__",
                    "rules": [{"op": "ge", "field": "a", "value": "10"}],
                },
                {"op": "eq", "field": "b", "value": "500"},
            ],
        }

        def rule_manipulator(rule):
            if "value" in rule.config:
                rule.config["value"] = "1"

        rule = Rule(rule_config, type_matrix, config_manipulator=rule_manipulator)
        rule.match({"a": 1, "b": "2"})
        assert rule.config["rules"][0]["rules"][0]["value"] == "1"
        assert rule.config["rules"][1]["value"] == "1"
        assert rule.type_matrix["b"]["type"] == "int"

    def test_dynamic_config_manipulator(self):
        from appenlight.lib.rule import Rule

        rule_config = {
            "field": "__OR__",
            "rules": [
                {
                    "field": "__OR__",
                    "rules": [{"op": "ge", "field": "a", "value": "10"}],
                },
                {"op": "eq", "field": "b", "value": "500"},
            ],
        }

        def rule_manipulator(rule):
            rule.type_matrix = {
                "a": {"type": "int", "ops": ("eq", "ne", "ge", "le")},
                "b": {"type": "unicode", "ops": ("eq", "ne", "ge", "le")},
            }

            if "value" in rule.config:
                if rule.config["field"] == "a":
                    rule.config["value"] = "1"
                elif rule.config["field"] == "b":
                    rule.config["value"] = "2"

        rule = Rule(rule_config, {}, config_manipulator=rule_manipulator)
        rule.match({"a": 11, "b": "55"})
        assert rule.config["rules"][0]["rules"][0]["value"] == "1"
        assert rule.config["rules"][1]["value"] == "2"
        assert rule.type_matrix["b"]["type"] == "unicode"


@pytest.mark.usefixtures("base_app", "with_migrations")
class TestViewsWithForms(object):
    def test_bad_csrf(self):
        from appenlight.forms import CSRFException
        from appenlight.views.index import register

        post_data = {"dupa": "dupa"}
        request = testing.DummyRequest(post=post_data)
        request.POST = webob.multidict.MultiDict(request.POST)
        with pytest.raises(CSRFException):
            register(request)

    def test_proper_csrf(self):
        from appenlight.views.index import register

        request = pyramid.threadlocal.get_current_request()
        post_data = {"dupa": "dupa", "csrf_token": request.session.get_csrf_token()}
        request = testing.DummyRequest(post=post_data)
        request.POST = webob.multidict.MultiDict(request.POST)
        result = register(request)
        assert result["form"].errors["email"][0] == "This field is required."


@pytest.mark.usefixtures("base_app", "with_migrations", "default_data")
class TestRegistration(object):
    def test_invalid_form(self):
        from appenlight.views.index import register

        request = pyramid.threadlocal.get_current_request()
        post_data = {
            "user_name": "",
            "user_password": "",
            "email": "",
            "csrf_token": request.session.get_csrf_token(),
        }
        request = testing.DummyRequest(post=post_data)
        request.POST = webob.multidict.MultiDict(request.POST)
        result = register(request)
        assert result["form"].errors["user_name"][0] == "This field is required."

    def test_valid_form(self):
        from appenlight.views.index import register
        from ziggurat_foundations.models.services.user import UserService

        request = pyramid.threadlocal.get_current_request()
        post_data = {
            "user_name": "foo",
            "user_password": "barr",
            "email": "[e-mail]",
            "csrf_token": request.session.get_csrf_token(),
        }
        request = testing.DummyRequest(post=post_data)
        request.add_flash_to_headers = mock.Mock()
        request.POST = webob.multidict.MultiDict(request.POST)
        assert UserService.by_user_name("foo") is None
        register(request)
        user = UserService.by_user_name("foo")
        assert user.user_name == "foo"
        assert len(user.user_password) >= 60


@pytest.mark.usefixtures("base_app", "with_migrations", "clean_tables", "default_user")
class TestApplicationCreation(object):
    def test_wrong_data(self):
        import appenlight.views.applications as applications
        from ziggurat_foundations.models.services.user import UserService

        request = pyramid.threadlocal.get_current_request()
        request.user = UserService.by_user_name("testuser")
        request.unsafe_json_body = {}
        request.headers["X-XSRF-TOKEN"] = request.session.get_csrf_token()
        response = applications.application_create(request)
        assert response.code == 422

    def test_proper_data(self):
        import appenlight.views.applications as applications
        from ziggurat_foundations.models.services.user import UserService

        request = pyramid.threadlocal.get_current_request()
        request.user = UserService.by_user_name("testuser")
        request.unsafe_json_body = {"resource_name": "app name", "domains": "foo"}
        request.headers["X-XSRF-TOKEN"] = request.session.get_csrf_token()
        app_dict = applications.application_create(request)
        assert app_dict["public_key"] is not None
        assert app_dict["api_key"] is not None
        assert app_dict["resource_name"] == "app name"
        assert app_dict["owner_group_id"] is None
        assert app_dict["resource_id"] is not None
        assert app_dict["default_grouping"] == "url_traceback"
        assert app_dict["possible_permissions"] == ("view", "update_reports")
        assert app_dict["slow_report_threshold"] == 10
        assert app_dict["owner_user_name"] == "testuser"
        assert app_dict["owner_user_id"] == request.user.id
        assert app_dict["domains"] is "foo"
        assert app_dict["postprocessing_rules"] == []
        assert app_dict["error_report_threshold"] == 10
        assert app_dict["allow_permanent_storage"] is False
        assert app_dict["resource_type"] == "application"
        assert app_dict["current_permissions"] == []


@pytest.mark.usefixtures("default_application")
@pytest.mark.usefixtures("base_app", "with_migrations", "clean_tables")
class TestAPISentryView(object):
    def test_no_payload(self, default_application):
        import colander
        from appenlight.models.services.application import ApplicationService
        from appenlight.views.api import sentry_compat
        from appenlight.lib.request import JSONException

        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        request = testing.DummyRequest(headers={"Content-Type": "application/json"})
        request.unsafe_json_body = ""
        request.context = context
        route = mock.Mock()
        route.name = "api_sentry"
        request.matched_route = route
        with pytest.raises(JSONException):
            sentry_compat(request)

    def test_java_client_payload(self):
        from appenlight.views.api import sentry_compat
        from appenlight.models.services.application import ApplicationService
        from appenlight.models.report_group import ReportGroup

        route = mock.Mock()
        route.name = "api_sentry"
        request = pyramid.threadlocal.get_current_request()
        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        context.resource.allow_permanent_storage = True
        request.context = context
        request.matched_route = route
        request.body = (
            b"eJy1UmFr2zAQ/S0T+7BCLOzYThp/C6xjG6SDLd/GCBf57Ki"
            b"RJSHJJiXkv+/UlC7p2kAZA33Ru6f33t1pz3BAHVayZhWr87"
            b"JMs+I6q3MsrifFep2vc1iXM1HMpgBTNmIdeg8tEvlmJ9AGa"
            b"fQ7goOkQoDOUmGcZpMkLZO0WGZFRadMiaHIR1EVnTMu3k3b"
            b"oiMgqJrXpgOpOVjLLTiPkWAVhMa4jih3MAAholfWyUDAksz"
            b"m1iopICbg8fWH52B8VWXZVYwHrWfV/jBipD2gW2no8CFMa5"
            b"JButCDSjoQG6mR6LgLDojPPn/7sbydL25ep34HGl+y3DiE+"
            b"lH0xXBXjMzFBsXW99SS7pWKYXRw91zqgK4BgZ4/DZVVP/cs"
            b"3NuzSZPfAKqP2Cdj4tw7U/cKH0fEFeiWQFqE2FIHAmMPjaN"
            b"Y/kHvbzY/JqdHUq9o/KxqQHkcsabX4piDuT4aK+pXG1ZNi/"
            b"IwOpEyruXC1LiB3vPO3BmOOxTUCIqv5LIg5H12oh9cf0l+P"
            b"MvP5P8kddgoFIEvMGzM5cRSD2aLJ6qTdHKm6nv9pPcRFba0"
            b"Kd0eleeCFuGN+9JZ9TaXIn/V5JYMBvxXg3L6PwzSE4dkfOb"
            b"w7CtfWmP85SdCs8OvA53fUV19cg=="
        )
        sentry_compat(request)
        query = DBSession.query(ReportGroup)
        report = query.first()
        assert query.count() == 1
        assert report.total_reports == 1

    def test_ruby_client_payload(self):
        from appenlight.views.api import sentry_compat
        from appenlight.models.services.application import ApplicationService
        from appenlight.models.report_group import ReportGroup
        from appenlight.tests.payload_examples import SENTRY_RUBY_ENCODED

        route = mock.Mock()
        route.name = "api_sentry"
        request = testing.DummyRequest(
            headers={
                "Content-Type": "application/octet-stream",
                "User-Agent": "sentry-ruby/1.0.0",
                "X-Sentry-Auth": "Sentry sentry_version=5, "
                "sentry_client=raven-ruby/1.0.0, "
                "sentry_timestamp=1462378483, "
                "sentry_key=xxx, sentry_secret=xxx",
            }
        )
        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        context.resource.allow_permanent_storage = True
        request.context = context
        request.matched_route = route
        request.body = SENTRY_RUBY_ENCODED
        sentry_compat(request)
        query = DBSession.query(ReportGroup)
        report = query.first()
        assert query.count() == 1
        assert report.total_reports == 1

    def test_python_client_decoded_payload(self):
        from appenlight.views.api import sentry_compat
        from appenlight.models.services.application import ApplicationService
        from appenlight.models.report_group import ReportGroup
        from appenlight.tests.payload_examples import SENTRY_PYTHON_PAYLOAD_7

        route = mock.Mock()
        route.name = "api_sentry"
        request = pyramid.threadlocal.get_current_request()
        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        context.resource.allow_permanent_storage = True
        request.context = context
        request.matched_route = route
        request.body = json.dumps(SENTRY_PYTHON_PAYLOAD_7).encode("utf8")
        sentry_compat(request)
        query = DBSession.query(ReportGroup)
        report = query.first()
        assert query.count() == 1
        assert report.total_reports == 1

    def test_python_client_encoded_payload(self):
        from appenlight.views.api import sentry_compat
        from appenlight.models.services.application import ApplicationService
        from appenlight.models.report_group import ReportGroup
        from appenlight.tests.payload_examples import SENTRY_PYTHON_ENCODED

        route = mock.Mock()
        route.name = "api_sentry"
        request = testing.DummyRequest(
            headers={
                "Content-Type": "application/octet-stream",
                "Content-Encoding": "deflate",
                "User-Agent": "sentry-ruby/1.0.0",
                "X-Sentry-Auth": "Sentry sentry_version=5, "
                "sentry_client=raven-ruby/1.0.0, "
                "sentry_timestamp=1462378483, "
                "sentry_key=xxx, sentry_secret=xxx",
            }
        )
        context = DummyContext()
        context.resource = ApplicationService.by_id(1)
        context.resource.allow_permanent_storage = True
        request.context = context
        request.matched_route = route
        request.body = SENTRY_PYTHON_ENCODED
        sentry_compat(request)
        query = DBSession.query(ReportGroup)
        report = query.first()
        assert query.count() == 1
        assert report.total_reports == 1