# -*- 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', ''), ('request', ''), ('conn', '')]) 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 #<#: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 '' 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((, 'Connection to domain.gr timed out. (connect timeout=10)')) caused by: ConnectTimeoutError((, '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 ''.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