Skip to content
Snippets Groups Projects
Commit a29c4ae3 authored by Jan Zerdik's avatar Jan Zerdik
Browse files

Modification of tests. 

(Redmine issue: #4489)
parent b50a3fb0
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
lib/mentat/reports/test_event.py
+ 27
− 52
View file @ a29c4ae3
......@@ -32,6 +32,7 @@ import mentat.reports.utils
import mentat.reports.event
from mentat.datatype.sqldb import GroupModel, FilterModel, NetworkModel, \
SettingsReportingModel, EventReportModel
from pynspect.jpath import jpath_values
#-------------------------------------------------------------------------------
# NOTE: Sorry for the long lines in this file. They are deliberate, because the
......@@ -95,7 +96,7 @@ class TestMentatReportsEvent(unittest.TestCase):
'Description': 'Synthetic example 02',
'Source': [
{
'IP4': ['192.168.1.2-192.168.1.5', '192.169.0.0/25', '10.0.0.1'],
'IP4': ['10.0.1.2-10.0.1.5', '10.0.0.0/25', '10.0.0.0/22', '10.0.2.1'],
'IP6': ['2002:db8::ff00:42:0/112']
}
],
......@@ -158,7 +159,8 @@ class TestMentatReportsEvent(unittest.TestCase):
group = GroupModel(name = 'abuse@cesnet.cz', source = 'manual', description = 'CESNET, z.s.p.o.')
FilterModel(group = group, name = 'FLT1', type = 'basic', filter = 'Node.Name == "org.example.kippo_honey"', description = 'DESC1', enabled = True)
FilterModel(group = group, name = 'FLT2', type = 'basic', filter = 'Category == "Recon.Scanning"', description = 'DESC2', enabled = True)
FilterModel(group = group, name = 'FLT2', type = 'basic', filter = 'Source.IP4 IN [10.0.0.0/24]', description = 'DESC2', enabled = True)
FilterModel(group = group, name = 'FLT3', type = 'basic', filter = 'Source.IP4 IN [10.0.1.0/28]', description = 'DESC3', enabled = True)
NetworkModel(group = group, netname = 'UNET1', source = 'manual', network = '10.0.0.0/8')
SettingsReportingModel(group = group)
......@@ -329,24 +331,7 @@ class TestMentatReportsEvent(unittest.TestCase):
os.unlink(report_path)
os.unlink("{}.zip".format(report_path))
def test_05_aggr_events_by_source(self):
"""
Test :py:func:`mentat.reports.event.EventReporter.aggregate_events_by_source` function.
"""
self.maxDiff = None
abuse_group = self.sqlstorage.session.query(GroupModel).filter(GroupModel.name == 'abuse@cesnet.cz').one()
self.sqlstorage.session.commit()
reporting_settings = mentat.reports.utils.ReportingSettings(
abuse_group
)
events_aggr = self.reporter.aggregate_events_by_source(self.ideas_obj, reporting_settings)
self.assertEqual(list(sorted(events_aggr.keys())), ['10.0.0.1'])
self.assertEqual(list(sorted(map(lambda x: x['ID'], events_aggr['10.0.0.1']))), ['msg01','msg02'])
def test_06_filter_events(self):
def test_05_filter_events(self):
"""
Test :py:class:`mentat.reports.event.EventReporter.filter_events` function.
"""
......@@ -358,28 +343,32 @@ class TestMentatReportsEvent(unittest.TestCase):
reporting_settings = mentat.reports.utils.ReportingSettings(
abuse_group
)
events, fltlog = self.reporter.filter_events(self.ideas_obj, abuse_group, reporting_settings)
self.assertEqual(events, [])
self.assertEqual(fltlog, {'FLT1': 1, 'FLT2': 1})
events, aggr, fltlog = self.reporter.filter_events(self.ideas_obj, abuse_group, reporting_settings)
self.assertEqual(fltlog, {'FLT1': 1})
self.assertEqual(len(aggr), 2)
self.reporter.logger.assert_has_calls([
call.debug("Event matched filtering rule '%s'", 'FLT1'),
call.debug("Event matched filtering rule '%s'", 'FLT2'),
call.info('%s: Filters blocked all %d events, nothing to report.', 'abuse@cesnet.cz', 2)
call.debug("Event matched filtering rule '%s', all sources filtered", 'FLT1'),
call.debug("Event matched filtering rule '%s', some sources allowed through", 'FLT2'),
call.info('%s: Filters let %d events through, %d blocked.', 'abuse@cesnet.cz', 1, 1)
])
self.sqlstorage.session.commit()
events, fltlog = self.reporter.filter_events(self.ideas_obj, abuse_group, reporting_settings)
events, aggr, fltlog = self.reporter.filter_events(self.ideas_obj, abuse_group, reporting_settings)
self.sqlstorage.session.commit()
flt1 = self.sqlstorage.session.query(FilterModel).filter(FilterModel.name == 'FLT1').one()
self.assertEqual(flt1.hits, 2)
events, fltlog = self.reporter.filter_events(self.ideas_obj, abuse_group, reporting_settings)
events, fltlog = self.reporter.filter_events(self.ideas_obj, abuse_group, reporting_settings)
events, aggr, fltlog = self.reporter.filter_events(self.ideas_obj, abuse_group, reporting_settings)
events, aggr, fltlog = self.reporter.filter_events(self.ideas_obj, abuse_group, reporting_settings)
self.sqlstorage.session.commit()
flt1 = self.sqlstorage.session.query(FilterModel).filter(FilterModel.name == 'FLT1').one()
self.assertEqual(flt1.hits, 4)
def test_07_fetch_severity_events(self):
aggr = self.reporter.aggregate_events(aggr)
self.assertEqual(list(sorted(aggr.keys())), ['anomaly-traffic'])
self.assertEqual(list(aggr['anomaly-traffic'].keys()), ['10.0.2.1', '10.0.0.0/22'])
def test_06_fetch_severity_events(self):
"""
Test :py:class:`mentat.reports.event.EventReporter.fetch_severity_events` function.
"""
......@@ -412,7 +401,7 @@ class TestMentatReportsEvent(unittest.TestCase):
)
self.assertEqual(list(map(lambda x: x['ID'], events)), [])
def test_08_j2t_idea_path_valueset(self):
def test_07_j2t_idea_path_valueset(self):
"""
Test :py:class:`mentat.reports.event.EventReporter.j2t_idea_path_valueset` function.
"""
......@@ -445,25 +434,7 @@ class TestMentatReportsEvent(unittest.TestCase):
['https', 'ssh']
)
def test_09_aggregate_events(self):
"""
Test :py:class:`mentat.reports.event.EventReporter.aggregate_events` function.
"""
self.maxDiff = None
abuse_group = self.sqlstorage.session.query(GroupModel).filter(GroupModel.name == 'abuse@cesnet.cz').one()
self.sqlstorage.session.commit()
reporting_settings = mentat.reports.utils.ReportingSettings(
abuse_group
)
events_aggr = self.reporter.aggregate_events_by_source(self.ideas_obj, reporting_settings)
aggr = self.reporter.aggregate_events(events_aggr)
self.assertEqual(list(sorted(aggr.keys())), ['anomaly-traffic', 'class01'])
self.assertEqual(list(sorted([list(v.keys()) for v in aggr.values()])), [['10.0.0.1'], ['10.0.0.1']])
def test_10_render_report_summary(self):
def test_08_render_report_summary(self):
"""
Test :py:class:`mentat.reports.event.EventReporter.render_report_summary` function.
"""
......@@ -504,7 +475,7 @@ class TestMentatReportsEvent(unittest.TestCase):
self.assertTrue(report_txt)
self.assertEqual(report_txt.split('\n')[0], 'Vážení kolegové.')
def test_11_render_report_extra(self):
def test_09_render_report_extra(self):
"""
Test :py:class:`mentat.reports.event.EventReporter.render_report_extra` function.
"""
......@@ -596,7 +567,11 @@ class TestMentatReportsEvent(unittest.TestCase):
report.statistics = mentat.stats.idea.truncate_evaluations(
mentat.stats.idea.evaluate_events(self.ideas_obj)
)
events_aggr = self.reporter.aggregate_events_by_source(self.ideas_obj, self.reporting_settings)
events_aggr = {}
for obj in self.ideas_obj:
for src in (jpath_values(obj, 'Source.IP4') + jpath_values(obj, 'Source.IP6')):
events_aggr[src] = [obj]
report.structured_data = self.reporter.prepare_structured_data(events_aggr, events_aggr, self.reporting_settings)
return report
......
lib/mentat/reports/test_utils.py
+ 26
− 33
View file @ a29c4ae3
......@@ -25,6 +25,7 @@ import pprint
# Custom libraries
#
from pynspect.gparser import PynspectFilterParser
from pynspect.jpath import jpath_values
import mentat.const
import mentat.services.sqlstorage
......@@ -175,23 +176,15 @@ class TestMentatReportsUtils(unittest.TestCase):
"""
self.maxDiff = None
for ip in (jpath_values(self.ideas_raw[0], "Source.IP4") + jpath_values(self.ideas_raw[0], "Source.IP6")):
key = self.stcache._generate_cache_key(self.ideas_raw[0], ip)
self.assertEqual(self.stcache.get_source_from_cache_key(key), ip)
self.assertEqual(
self.stcache._generate_cache_keys(self.ideas_raw[0]), # pylint: disable=locally-disabled,protected-access
[
'class01+++192.168.0.2-192.168.0.5',
'class01+++192.168.0.0/25',
'class01+++10.0.0.1',
'class01+++2001:db8::ff00:42:0/112'
]
)
self.assertEqual(
self.stcache._generate_cache_keys({ # pylint: disable=locally-disabled,protected-access
self.stcache._generate_cache_key({ # pylint: disable=locally-disabled,protected-access
'Category': ['Test', 'Value'],
'Source': [{'IP4': ['195.113.144.194']}]
}),
[
'Test/Value+++195.113.144.194'
]
}, '195.113.144.194'),
'Test/Value+++195.113.144.194'
)
def test_02_no_thr_cache(self):
......@@ -205,13 +198,16 @@ class TestMentatReportsUtils(unittest.TestCase):
thresholdtime = relapsetime - datetime.timedelta(seconds = 600)
self.assertFalse(
self.ntcache.event_is_thresholded(self.ideas_obj[0], ttltime)
self.ntcache.event_is_thresholded(self.ideas_obj[0], '192.168.1.1', ttltime)
)
self.ntcache.set_threshold(self.ideas_obj[0], '192.168.1.1', thresholdtime, relapsetime, ttltime)
self.assertFalse(
self.ntcache.event_is_thresholded(self.ideas_obj[0], ttltime)
self.ntcache.event_is_thresholded(self.ideas_obj[0], '192.168.1.1', ttltime)
)
self.ntcache.threshold_event(self.ideas_obj[0], '192.168.1.1', 'Test', 'low', datetime.datetime.utcnow())
self.assertFalse(
self.ntcache.event_is_thresholded(self.ideas_obj[0], '192.168.1.1', ttltime)
)
self.ntcache.threshold_event(self.ideas_obj[0], 'Test', 'low', datetime.datetime.utcnow())
def test_03_storage_thr_cache(self):
"""
......@@ -224,36 +220,33 @@ class TestMentatReportsUtils(unittest.TestCase):
thrtime = reltime - datetime.timedelta(seconds = 300)
self.assertFalse(
self.stcache.event_is_thresholded(self.ideas_obj[0], ttltime)
self.stcache.event_is_thresholded(self.ideas_obj[0], '192.168.0.2-192.168.0.5', ttltime)
)
self.stcache.set_threshold(self.ideas_obj[0], '192.168.0.2-192.168.0.5', thrtime, reltime, ttltime)
self.assertFalse(
self.stcache.event_is_thresholded(self.ideas_obj[0], ttltime - datetime.timedelta(seconds = 50))
)
self.stcache.set_threshold(self.ideas_obj[0], '192.168.0.0/25', thrtime, reltime, ttltime)
self.assertFalse(
self.stcache.event_is_thresholded(self.ideas_obj[0], ttltime - datetime.timedelta(seconds = 50))
self.assertTrue(
self.stcache.event_is_thresholded(self.ideas_obj[0], '192.168.0.2-192.168.0.5', ttltime - datetime.timedelta(seconds = 50))
)
self.stcache.set_threshold(self.ideas_obj[0], '10.0.0.1', thrtime, reltime, ttltime)
self.assertFalse(
self.stcache.event_is_thresholded(self.ideas_obj[0], ttltime - datetime.timedelta(seconds = 50))
self.stcache.event_is_thresholded(self.ideas_obj[0], '192.168.0.0/25', ttltime - datetime.timedelta(seconds = 50))
)
self.stcache.set_threshold(self.ideas_obj[0], '2001:db8::ff00:42:0/112', thrtime, reltime, ttltime)
self.stcache.set_threshold(self.ideas_obj[0], '192.168.0.0/25', thrtime, reltime, ttltime)
self.assertTrue(
self.stcache.event_is_thresholded(self.ideas_obj[0], ttltime - datetime.timedelta(seconds = 50))
self.stcache.event_is_thresholded(self.ideas_obj[0], '192.168.0.0/25', ttltime - datetime.timedelta(seconds = 50))
)
self.stcache.threshold_event(self.ideas_obj[0], 'test@domain.org', 'low', ttltime - datetime.timedelta(seconds = 50))
self.stcache.threshold_event(self.ideas_obj[0], '192.168.0.2-192.168.0.5', 'test@domain.org', 'low', ttltime - datetime.timedelta(seconds = 50))
self.stcache.threshold_event(self.ideas_obj[0], '192.168.0.0/25', 'test@domain.org', 'low', ttltime - datetime.timedelta(seconds = 50))
relapses = self.stcache.relapses('test@domain.org', 'low', ttltime + datetime.timedelta(seconds = 50))
self.assertEqual(len(relapses), 1)
self.assertEqual(relapses[0]['ID'], 'msg01')
self.assertEqual(relapses[0][0], 'msg01')
self.assertEqual(len(relapses[0][2]), 2)
self.assertEqual(self.stcache.eventservice.thresholds_count(), 4)
self.assertEqual(self.stcache.eventservice.thresholded_events_count(), 4)
self.assertEqual(self.stcache.eventservice.thresholds_count(), 2)
self.assertEqual(self.stcache.eventservice.thresholded_events_count(), 2)
self.assertEqual(
self.stcache.cleanup(ttltime + datetime.timedelta(seconds = 50)),
{'thresholds': 4, 'events': 4}
{'thresholds': 2, 'events': 2}
)
def test_04_reporting_settings(self):
......
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment