|
|
# -*- 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": "test@test.foo",
|
|
|
"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
|
|
|
|