mirror of
https://github.com/mvt-project/mvt.git
synced 2026-09-03 00:21:07 +02:00
Donncha Ó Cearbhaill
da33c2fdb3
mvt.plugin re-exports the names a plugin needs from MVT under one import path. It holds the module base classes and Command, the alert and result types, the database errors a module raises, the timestamp converters, the plugin settings API, MVT's settings, get_plugin_logger() and MVT_VERSION. The names it exports are kept working on a best-effort basis. Changes to them are announced in the release notes. Anything else in mvt can still be imported, and may change between releases without notice. get_plugin_logger(__name__) returns a logger under mvt.ext for plugin code outside a module class. Its records then reach the console and the command.log file of a run. A file loaded with --load-module or --load-command is named after the file.
217 lines
6.6 KiB
Python
217 lines
6.6 KiB
Python
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from mvt.common.cli_plugins import _module_name_for_path as _command_module_name
|
|
from mvt.common.module import MVTModule
|
|
from mvt.common.module_loader import (
|
|
CustomModuleLoadError,
|
|
_module_name_for_path,
|
|
get_module_logger,
|
|
get_plugin_logger,
|
|
load_custom_modules,
|
|
load_custom_modules_from_path,
|
|
module_supports_command,
|
|
)
|
|
from mvt.ios.modules.mixed.whatsapp import Whatsapp
|
|
|
|
|
|
MODULE_TEMPLATE = """
|
|
from mvt.common.module import MVTModule
|
|
|
|
|
|
class {name}(MVTModule):
|
|
supported_commands = {supported_commands!r}
|
|
|
|
def run(self):
|
|
pass
|
|
|
|
def check_indicators(self):
|
|
pass
|
|
|
|
def serialize(self, result):
|
|
return None
|
|
"""
|
|
|
|
|
|
def _write_module(path, name, supported_commands=()):
|
|
path.write_text(
|
|
MODULE_TEMPLATE.format(
|
|
name=name,
|
|
supported_commands=supported_commands,
|
|
),
|
|
encoding="utf-8",
|
|
)
|
|
return path
|
|
|
|
|
|
def test_load_custom_modules_from_python_file(tmp_path):
|
|
module_path = _write_module(tmp_path / "custom.py", "FileModule")
|
|
|
|
modules = load_custom_modules_from_path(str(module_path))
|
|
|
|
assert [module.__name__ for module in modules] == ["FileModule"]
|
|
assert issubclass(modules[0], MVTModule)
|
|
|
|
|
|
def test_load_custom_modules_from_folder_in_sorted_order(tmp_path):
|
|
_write_module(tmp_path / "b_module.py", "BModule")
|
|
_write_module(tmp_path / "a_module.py", "AModule")
|
|
_write_module(tmp_path / ".hidden.py", "HiddenModule")
|
|
_write_module(tmp_path / "__init__.py", "InitModule")
|
|
nested = tmp_path / "nested"
|
|
nested.mkdir()
|
|
_write_module(nested / "nested_module.py", "NestedModule")
|
|
|
|
modules = load_custom_modules_from_path(str(tmp_path))
|
|
|
|
assert [module.__name__ for module in modules] == ["AModule", "BModule"]
|
|
|
|
|
|
def test_discovery_ignores_imported_base_and_unrelated_classes(tmp_path):
|
|
module_path = tmp_path / "custom.py"
|
|
module_path.write_text(
|
|
"""
|
|
from mvt.common.module import MVTModule
|
|
|
|
|
|
class Unrelated:
|
|
pass
|
|
|
|
|
|
class DiscoveredModule(MVTModule):
|
|
def run(self):
|
|
pass
|
|
|
|
def check_indicators(self):
|
|
pass
|
|
|
|
def serialize(self, result):
|
|
return None
|
|
""",
|
|
encoding="utf-8",
|
|
)
|
|
|
|
modules = load_custom_modules_from_path(str(module_path))
|
|
|
|
assert [module.__name__ for module in modules] == ["DiscoveredModule"]
|
|
|
|
|
|
def test_load_custom_modules_deduplicates_same_class(tmp_path):
|
|
module_path = _write_module(tmp_path / "custom.py", "DuplicateModule")
|
|
|
|
modules = load_custom_modules([str(module_path), str(module_path)])
|
|
|
|
assert [module.__name__ for module in modules] == ["DuplicateModule"]
|
|
|
|
|
|
def test_load_custom_modules_raises_for_missing_path(tmp_path):
|
|
with pytest.raises(CustomModuleLoadError, match="does not exist"):
|
|
load_custom_modules_from_path(str(tmp_path / "missing.py"))
|
|
|
|
|
|
def test_load_custom_modules_raises_for_import_error(tmp_path):
|
|
module_path = tmp_path / "broken.py"
|
|
module_path.write_text("raise RuntimeError('broken import')", encoding="utf-8")
|
|
|
|
with pytest.raises(CustomModuleLoadError, match="broken import"):
|
|
load_custom_modules_from_path(str(module_path))
|
|
|
|
|
|
def test_load_custom_modules_loads_env_folder_first(tmp_path, monkeypatch):
|
|
env_folder = tmp_path / "env"
|
|
env_folder.mkdir()
|
|
cli_folder = tmp_path / "cli"
|
|
cli_folder.mkdir()
|
|
_write_module(env_folder / "env_module.py", "EnvModule")
|
|
_write_module(cli_folder / "cli_module.py", "CliModule")
|
|
monkeypatch.setenv("MVT_CUSTOM_MODULES", str(env_folder))
|
|
|
|
modules = load_custom_modules([str(cli_folder)])
|
|
|
|
assert [module.__name__ for module in modules] == ["EnvModule", "CliModule"]
|
|
|
|
|
|
def test_module_supports_command_requires_explicit_declaration(tmp_path, caplog):
|
|
module_path = _write_module(tmp_path / "custom.py", "DefaultModule")
|
|
module = load_custom_modules_from_path(str(module_path))[0]
|
|
|
|
assert not module_supports_command(module, "ios", "check-backup")
|
|
assert not module_supports_command(module, "android", "check-bugreport")
|
|
assert "DefaultModule has no supported_commands" in caplog.text
|
|
|
|
|
|
def test_module_supports_command_honors_supported_commands(tmp_path):
|
|
module_path = _write_module(
|
|
tmp_path / "custom.py",
|
|
"SpecificModule",
|
|
(("ios", "check-backup"),),
|
|
)
|
|
module = load_custom_modules_from_path(str(module_path))[0]
|
|
|
|
assert module_supports_command(module, "ios", "check-backup")
|
|
assert not module_supports_command(module, "ios", "check-fs")
|
|
|
|
|
|
def test_get_module_logger_keeps_builtin_names():
|
|
assert get_module_logger(Whatsapp).name == "mvt.ios.modules.mixed.whatsapp"
|
|
|
|
|
|
def test_get_module_logger_parents_package_modules_under_mvt_ext():
|
|
class PackageModule(MVTModule):
|
|
pass
|
|
|
|
PackageModule.__module__ = "some_plugin_package.ios.custom"
|
|
|
|
assert (
|
|
get_module_logger(PackageModule).name
|
|
== "mvt.ext.some_plugin_package.ios.custom"
|
|
)
|
|
|
|
|
|
def test_get_module_logger_strips_the_plugin_package_prefix():
|
|
class PluginModule(MVTModule):
|
|
pass
|
|
|
|
PluginModule.__module__ = "mvt_plugin_example_org.ios.custom"
|
|
|
|
assert get_module_logger(PluginModule).name == "mvt.ext.example_org.ios.custom"
|
|
|
|
|
|
def test_get_module_logger_only_strips_the_prefix_from_the_top_level():
|
|
class NestedModule(MVTModule):
|
|
pass
|
|
|
|
NestedModule.__module__ = "other_package.mvt_plugin_sub"
|
|
|
|
assert get_module_logger(NestedModule).name == "mvt.ext.other_package.mvt_plugin_sub"
|
|
|
|
|
|
def test_get_module_logger_names_path_modules_after_their_file(tmp_path):
|
|
module_path = _write_module(tmp_path / "my_custom_module.py", "PathModule")
|
|
module = load_custom_modules_from_path(str(module_path))[0]
|
|
|
|
assert get_module_logger(module).name == "mvt.ext.my_custom_module"
|
|
|
|
|
|
def test_get_plugin_logger_uses_the_same_namespace_as_modules():
|
|
assert (
|
|
get_plugin_logger("mvt_plugin_example_org.commands.summarize").name
|
|
== "mvt.ext.example_org.commands.summarize"
|
|
)
|
|
assert get_plugin_logger("example_plugin.cli").name == "mvt.ext.example_plugin.cli"
|
|
|
|
|
|
def test_get_plugin_logger_keeps_builtin_names():
|
|
assert get_plugin_logger("mvt.ios.cli").name == "mvt.ios.cli"
|
|
|
|
|
|
def test_get_plugin_logger_names_loaded_files_after_the_file():
|
|
# A file loaded with --load-command or --load-module is imported under a
|
|
# mangled name. The log names the file instead.
|
|
command_name = _command_module_name(Path("/tmp/case_summary.py"))
|
|
module_name = _module_name_for_path(Path("/tmp/my_custom_module.py"))
|
|
|
|
assert get_plugin_logger(command_name).name == "mvt.ext.case_summary"
|
|
assert get_plugin_logger(module_name).name == "mvt.ext.my_custom_module"
|