test: added tests to improve coverage

Author: jstucke

src/test/unit/analysis/test_analysis_plugin.py

@@ -0,0 +1,60 @@
+from __future__ import annotations
+
+from pydantic import BaseModel
+from semver import Version
+
+from analysis.plugin import AnalysisPluginV0, Tag
+from helperFunctions.tag import TagColor
+
+
+class DummyPlugin(AnalysisPluginV0):
+    class Schema(BaseModel):
+        foo: str
+
+    def __init__(self):
+        metadata = self.MetaData(name='DummyPlugin', description='', Schema=self.Schema, version=Version(0, 1, 0))
+        super().__init__(metadata)
+
+    def analyze(self, file_handle, virtual_file_path, analyses):
+        return self.Schema(foo='foo')
+
+
+class ExtendedDummyPlugin(DummyPlugin):
+    def summarize(self, result):
+        return [result.foo]
+
+    def get_tags(self, result, summary):
+        return [Tag(name=result.foo, value=result.foo, color=TagColor.GREEN)]
+
+
+def test_get_analysis():
+    plugin = DummyPlugin()
+
+    result = plugin.get_analysis(None, {}, {})
+    expected_keys = ['analysis_date', 'plugin_version', 'result', 'summary', 'system_version', 'tags']
+    assert all(k in result for k in expected_keys)
+    assert 'foo' in result['result']
+    assert result['summary'] == []
+    assert result['tags'] == {}
+
+
+def test_get_analysis_extended():
+    plugin = ExtendedDummyPlugin()
+
+    result = plugin.get_analysis(None, {}, {})
+    expected_keys = ['analysis_date', 'plugin_version', 'result', 'summary', 'system_version', 'tags']
+    assert all(k in result for k in expected_keys)
+    assert 'foo' in result['result']
+    assert result['summary'] == ['foo']
+    assert len(result['tags']) == 1
+    assert result['tags']['foo'] == {'color': TagColor.GREEN, 'propagate': False, 'value': 'foo'}
+
+
+def test_summarize():
+    plugin = DummyPlugin()
+    extended_plugin = ExtendedDummyPlugin()
+
+    assert (
+        plugin.summarize(plugin.Schema(foo='foo')) == []
+    ), 'if the plugin does not implement summarize, it should return an empty list'
+    assert extended_plugin.summarize(plugin.Schema(foo='foo')) == ['foo']

src/test/unit/scheduler/test_analysis_status.py

@@ -10,9 +10,18 @@
 ROOT_UID = 'root_uid'
 
 
+class MockRedis:
+    def __init__(self):
+        self.analysis_status = None
+
+    def set_analysis_status(self, status):
+        self.analysis_status = status
+
+
 class TestAnalysisStatus:
     def setup_method(self):
         self.status = AnalysisStatus()
+        self.status._worker.redis = MockRedis()
 
     def test_add_firmware_to_current_analyses(self):
         fw = Firmware(binary=b'foo')
@@ -207,3 +216,73 @@ def test_cancel_unknown_uid(self):
         self.status._worker._update_status()
 
         assert ROOT_UID in self.status._worker.currently_running
+
+    def test_add_update(self):
+        fw = Firmware(binary=b'foo')
+        fw.files_included = ['foo', 'bar']
+        self.status.add_update(fw, fw.files_included)
+        # _update_status is called twice, because add_object is called first and then add_update
+        self.status._worker._update_status()
+        self.status._worker._update_status()
+
+        assert fw.uid in self.status._worker.currently_running
+        result = self.status._worker.currently_running[fw.uid]
+        assert result.files_to_unpack == set()
+        assert result.completed_files == set()
+        assert result.unpacked_files_count == result.total_files_count
+        assert result.analyzed_files_count == 0
+        assert result.files_to_analyze == {fw.uid, *fw.files_included}
+        assert result.total_files_count == len(fw.files_included) + 1
+
+    def test_add_analysis(self):
+        self.status._worker.currently_running = {
+            ROOT_UID: FwAnalysisStatus(
+                files_to_unpack=set(),
+                files_to_analyze={'foo', 'bar'},
+                analysis_plugins={},
+                hid='',
+                total_files_count=3,
+            )
+        }
+        self.status._currently_analyzed[ROOT_UID] = True
+        fo = FileObject(binary=b'foo')
+        fo.root_uid = ROOT_UID
+        fo.uid = 'foo'
+
+        result = self.status._worker.currently_running[ROOT_UID]
+        assert result.analysis_plugins == {}
+
+        self.status.add_analysis(fo, 'some_plugin')
+        self.status._worker._update_status()
+
+        assert result.analysis_plugins == {'some_plugin': 1}
+
+    def test_store_status(self):
+        self.status._worker.currently_running = {
+            ROOT_UID: FwAnalysisStatus(
+                files_to_unpack={'bar'},
+                files_to_analyze={'foo', 'bar'},
+                analysis_plugins={},
+                hid='',
+                unpacked_files_count=2,
+                total_files_count=3,
+            )
+        }
+        recently_finished = {
+            'other_UID': {
+                'duration': 13.37,
+                'total_files_count': 1337,
+                'time_finished': 1337,
+                'hid': 'analysis_status.hid',
+            }
+        }
+        self.status._worker.recently_finished = recently_finished
+        self.status._worker._store_status()
+
+        status = self.status._worker.redis.analysis_status
+        assert ROOT_UID in status['current_analyses']
+        assert status['current_analyses'][ROOT_UID]['unpacked_count'] == 2
+        assert status['current_analyses'][ROOT_UID]['analyzed_count'] == 0
+        assert status['current_analyses'][ROOT_UID]['total_count'] == 3
+        assert status['recently_canceled_analyses'] == {}
+        assert status['recently_finished_analyses'] == recently_finished

src/test/unit/statistic/test_analysis_stats.py

@@ -3,6 +3,7 @@
 from dataclasses import dataclass
 
 import pytest
+from pydantic import BaseModel
 
 from scheduler.analysis import plugin
 from statistic.analysis_stats import get_plugin_stats
@@ -15,6 +16,14 @@ class MockMetadata:
     dependencies: list[str]
 
 
+class MockDependency:
+    metadata = MockMetadata('dependency', [])
+
+    class Schema(BaseModel):
+        a: int
+        b: str
+
+
 @dataclass
 class MockPlugin:
     metadata: MockMetadata
@@ -31,8 +40,12 @@ def generate_path(self, fw):
 @pytest.fixture
 def mock_runner():
     runner_config = plugin.PluginRunner.Config(process_count=1, timeout=5)
-    metadata = MockMetadata(name='test', dependencies=[])
-    runner = plugin.PluginRunner(MockPlugin(metadata), runner_config, {})
+    metadata = MockMetadata(name='test', dependencies=[MockDependency.metadata.name])
+    runner = plugin.PluginRunner(
+        MockPlugin(metadata),
+        runner_config,
+        {MockDependency.metadata.name: MockDependency.Schema},
+    )
     runner._fsorganizer = MockFSOrganizer()
     yield runner
     runner.shutdown()
@@ -61,6 +74,7 @@ def test_update_duration_stats(mock_runner):
     mock_runner.start()
     assert mock_runner.stats_count.value == mock_runner._stats_idx.value == 0
     fw = create_test_firmware()
+    fw.processed_analysis[MockDependency.metadata.name] = {'result': {'a': 1, 'b': '2'}}
     for _ in range(4):
         mock_runner.queue_analysis(fw)
         mock_runner.out_queue.get(timeout=5)