test_unit.py
1592 lines
| 64.9 KiB
| text/x-python
|
PythonLexer
r0 | # -*- coding: utf-8 -*- | |||
r112 | # Copyright 2010 - 2017 RhodeCode GmbH and the AppEnlight project authors | |||
r0 | # | |||
r112 | # 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 | ||||
r0 | # | |||
r112 | # http://www.apache.org/licenses/LICENSE-2.0 | |||
r0 | # | |||
r112 | # 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. | ||||
r0 | ||||
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): | ||||
r50 | dummy_json = [{'tags': [['counter_a', 15.5], ['counter_b', 63]]}] | |||
r0 | deserialized = general_metrics_schema.deserialize(dummy_json)[0] | |||
expected = {'namespace': '', | ||||
'server_name': 'unknown', | ||||
r50 | 'tags': [('counter_a', 15.5), ('counter_b', 63)], | |||
r0 | '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 | ||||
r49 | from appenlight.models.metric import Metric | |||
r0 | 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 | ||||
r49 | from appenlight.models.metric import Metric | |||
r0 | 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(): | ||||
r41 | ||||
@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 | ||||
r0 | @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' | ||||
r135 | assert len(user.user_password) >= 60 | |||
r0 | ||||
@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) | ||||
r95 | context.resource.allow_permanent_storage = True | |||
r0 | 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) | ||||
r95 | context.resource.allow_permanent_storage = True | |||
r0 | 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) | ||||
r95 | context.resource.allow_permanent_storage = True | |||
r0 | 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) | ||||
r95 | context.resource.allow_permanent_storage = True | |||
r0 | 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 | ||||