Rietveld Code Review Tool
Help | Bug tracker | Discussion group | Source code | Sign in
(1672)

Unified Diff: tests/storage/fake_storage.py

Issue 334410043: [plaso] Moved storage implementations (Closed)
Patch Set: Changes after rebase Created 6 years, 2 months ago
Use n/p to move between diff chunks; N/P to move between comments. Please Sign in to add in-line comments.
Jump to:
View side-by-side diff with in-line comments
Download patch
Index: tests/storage/fake_storage.py
diff --git a/tests/storage/fake_storage.py b/tests/storage/fake_storage.py
deleted file mode 100644
index 65a4c23313a853ce5082a684f512d0d3129e679e..0000000000000000000000000000000000000000
--- a/tests/storage/fake_storage.py
+++ /dev/null
@@ -1,237 +0,0 @@
-#!/usr/bin/python
-# -*- coding: utf-8 -*-
-"""Tests for the fake storage."""
-
-from __future__ import unicode_literals
-
-import unittest
-
-from plaso.containers import errors
-from plaso.containers import event_sources
-from plaso.containers import reports
-from plaso.containers import sessions
-from plaso.containers import tasks
-from plaso.lib import definitions
-from plaso.storage import fake_storage
-
-from tests.storage import test_lib
-
-
-class FakeStorageWriterTest(test_lib.StorageTestCase):
- """Tests for the fake storage writer object."""
-
- def testAddAnalysisReport(self):
- """Tests the AddAnalysisReport function."""
- session = sessions.Session()
- analysis_report = reports.AnalysisReport(
- plugin_name='test', text='test report')
-
- storage_writer = fake_storage.FakeStorageWriter(session)
- storage_writer.Open()
-
- storage_writer.AddAnalysisReport(analysis_report)
-
- storage_writer.Close()
-
- with self.assertRaises(IOError):
- storage_writer.AddAnalysisReport(analysis_report)
-
- def testAddError(self):
- """Tests the AddError function."""
- session = sessions.Session()
- extraction_error = errors.ExtractionError(
- message='Test extraction error')
-
- storage_writer = fake_storage.FakeStorageWriter(session)
- storage_writer.Open()
-
- storage_writer.AddError(extraction_error)
-
- storage_writer.Close()
-
- with self.assertRaises(IOError):
- storage_writer.AddError(extraction_error)
-
- def testAddEvent(self):
- """Tests the AddEvent function."""
- session = sessions.Session()
- test_events = self._CreateTestEvents()
-
- storage_writer = fake_storage.FakeStorageWriter(session)
- storage_writer.Open()
-
- event = None
- for event in test_events:
- storage_writer.AddEvent(event)
-
- storage_writer.Close()
-
- with self.assertRaises(IOError):
- storage_writer.AddEvent(event)
-
- def testAddEventSource(self):
- """Tests the AddEventSource function."""
- session = sessions.Session()
- event_source = event_sources.EventSource()
-
- storage_writer = fake_storage.FakeStorageWriter(session)
- storage_writer.Open()
-
- storage_writer.AddEventSource(event_source)
-
- storage_writer.Close()
-
- with self.assertRaises(IOError):
- storage_writer.AddEventSource(event_source)
-
- def testAddEventTag(self):
- """Tests the AddEventTag function."""
- session = sessions.Session()
-
- storage_writer = fake_storage.FakeStorageWriter(session)
- storage_writer.Open()
-
- test_events = self._CreateTestEvents()
- for event in test_events:
- storage_writer.AddEvent(event)
-
- event_tag = None
- test_event_tags = self._CreateTestEventTags(test_events)
- for event_tag in test_event_tags:
- storage_writer.AddEventTag(event_tag)
-
- storage_writer.Close()
-
- with self.assertRaises(IOError):
- storage_writer.AddEventTag(event_tag)
-
- def testOpenClose(self):
- """Tests the Open and Close functions."""
- session = sessions.Session()
- storage_writer = fake_storage.FakeStorageWriter(session)
- storage_writer.Open()
- storage_writer.Close()
-
- storage_writer.Open()
- storage_writer.Close()
-
- storage_writer = fake_storage.FakeStorageWriter(
- session, storage_type=definitions.STORAGE_TYPE_TASK)
- storage_writer.Open()
- storage_writer.Close()
-
- storage_writer.Open()
-
- with self.assertRaises(IOError):
- storage_writer.Open()
-
- storage_writer.Close()
-
- with self.assertRaises(IOError):
- storage_writer.Close()
-
- def testGetEvents(self):
- """Tests the GetEvents function."""
- session = sessions.Session()
- test_events = self._CreateTestEvents()
-
- storage_writer = fake_storage.FakeStorageWriter(session)
- storage_writer.Open()
-
- event = None
- for event in test_events:
- storage_writer.AddEvent(event)
-
- events = list(storage_writer.GetEvents())
- self.assertEqual(len(events), len(test_events))
-
- storage_writer.Close()
-
- # TODO: add tests for GetEventSources.
- # TODO: add tests for GetEventTags.
- # TODO: add tests for GetFirstWrittenEventSource and
- # GetNextWrittenEventSource.
-
- def testGetSortedEvents(self):
- """Tests the GetSortedEvents function."""
- session = sessions.Session()
- test_events = self._CreateTestEvents()
-
- storage_writer = fake_storage.FakeStorageWriter(session)
- storage_writer.Open()
-
- event = None
- for event in test_events:
- storage_writer.AddEvent(event)
-
- events = list(storage_writer.GetSortedEvents())
- self.assertEqual(len(events), len(test_events))
-
- storage_writer.Close()
-
- # TODO: add test with time range.
-
- def testWriteSessionStartAndCompletion(self):
- """Tests the WriteSessionStart and WriteSessionCompletion functions."""
- session = sessions.Session()
-
- storage_writer = fake_storage.FakeStorageWriter(session)
- storage_writer.Open()
-
- storage_writer.WriteSessionStart()
- storage_writer.WriteSessionCompletion()
-
- storage_writer.Close()
-
- with self.assertRaises(IOError):
- storage_writer.WriteSessionStart()
-
- with self.assertRaises(IOError):
- storage_writer.WriteSessionCompletion()
-
- storage_writer = fake_storage.FakeStorageWriter(
- session, storage_type=definitions.STORAGE_TYPE_TASK)
- storage_writer.Open()
-
- with self.assertRaises(IOError):
- storage_writer.WriteSessionStart()
-
- with self.assertRaises(IOError):
- storage_writer.WriteSessionCompletion()
-
- storage_writer.Close()
-
- def testWriteTaskStartAndCompletion(self):
- """Tests the WriteTaskStart and WriteTaskCompletion functions."""
- session = sessions.Session()
- task = tasks.Task(session_identifier=session.identifier)
-
- storage_writer = fake_storage.FakeStorageWriter(
- session, storage_type=definitions.STORAGE_TYPE_TASK, task=task)
- storage_writer.Open()
-
- storage_writer.WriteTaskStart()
- storage_writer.WriteTaskCompletion()
-
- storage_writer.Close()
-
- with self.assertRaises(IOError):
- storage_writer.WriteTaskStart()
-
- with self.assertRaises(IOError):
- storage_writer.WriteTaskCompletion()
-
- storage_writer = fake_storage.FakeStorageWriter(session)
- storage_writer.Open()
-
- with self.assertRaises(IOError):
- storage_writer.WriteTaskStart()
-
- with self.assertRaises(IOError):
- storage_writer.WriteTaskCompletion()
-
- storage_writer.Close()
-
-
-if __name__ == '__main__':
- unittest.main()

Powered by Google App Engine
RSS Feeds Recent Issues | This issue
This is Rietveld f62528b