File: manager.py

package info (click to toggle)
plaso 20201007-2
  • links: PTS, VCS
  • area: main
  • in suites: bullseye
  • size: 519,924 kB
  • sloc: python: 79,002; sh: 629; xml: 72; sql: 14; vhdl: 11; makefile: 10
file content (82 lines) | stat: -rw-r--r-- 2,447 bytes parent folder | download
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""Tests for the analysis plugin manager."""

from __future__ import unicode_literals

import unittest

from plaso.analysis import interface
from plaso.analysis import manager

from tests import test_lib as shared_test_lib


class TestAnalysisPlugin(interface.AnalysisPlugin):
  """Test analysis plugin."""

  NAME = 'test_plugin'

  def CompileReport(self, mediator):
    """Compiles a report of the analysis.

    After the plugin has received every copy of an event to
    analyze this function will be called so that the report
    can be assembled.

    Args:
      mediator (AnalysisMediator): mediates interactions between analysis
          plugins and other components, such as storage and dfvfs.
    """
    return

  # pylint: disable=arguments-differ,unused-argument
  def ExamineEvent(self, mediator, event, **unused_kwargs):
    """Analyzes an event object.

    Args:
      mediator (AnalysisMediator): mediates interactions between analysis
          plugins and other components, such as storage and dfvfs.
      event (EventObject): event.
    """
    return


class AnalysisPluginManagerTest(shared_test_lib.BaseTestCase):
  """Tests for the analysis plugin manager."""

  # pylint: disable=protected-access

  def testPluginRegistration(self):
    """Tests the RegisterPlugin and DeregisterPlugin functions."""
    number_of_plugins = len(manager.AnalysisPluginManager._plugin_classes)

    manager.AnalysisPluginManager.RegisterPlugin(TestAnalysisPlugin)
    self.assertEqual(
        len(manager.AnalysisPluginManager._plugin_classes),
        number_of_plugins + 1)

    with self.assertRaises(KeyError):
      manager.AnalysisPluginManager.RegisterPlugin(TestAnalysisPlugin)

    manager.AnalysisPluginManager.DeregisterPlugin(TestAnalysisPlugin)
    self.assertEqual(
        len(manager.AnalysisPluginManager._plugin_classes),
        number_of_plugins)

  def testGetPlugins(self):
    """Tests the GetPlugins function."""
    manager.AnalysisPluginManager.RegisterPlugin(TestAnalysisPlugin)

    # Use set-comprehension to create a set of the analysis plugin names.
    plugin_set = {name for name, _ in list(
        manager.AnalysisPluginManager.GetPlugins())}
    self.assertTrue('test_plugin' in plugin_set)

    manager.AnalysisPluginManager.DeregisterPlugin(TestAnalysisPlugin)

  # TODO: add tests for GetPluginNames.


if __name__ == '__main__':
  unittest.main()