93 lines
3.5 KiB
Python
93 lines
3.5 KiB
Python
from __future__ import annotations
|
|
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from support import make_config, plan
|
|
|
|
from netbox_store_agent.errors import PolicyError
|
|
from netbox_store_agent.journal import ManagedPlugin
|
|
from netbox_store_agent.managed_files import ManagedFiles
|
|
|
|
|
|
class ManagedFilesTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temporary.name)
|
|
self.config = make_config(self.root, dry_run=False)
|
|
self.files = ManagedFiles(self.config)
|
|
release = plan()
|
|
self.plugin = ManagedPlugin(
|
|
"demo-plugin",
|
|
"demo-plugin",
|
|
"demo_plugin",
|
|
"1.2.3",
|
|
True,
|
|
(release.requirement(),),
|
|
)
|
|
|
|
def tearDown(self) -> None:
|
|
self.temporary.cleanup()
|
|
|
|
def test_include_exports_constant_without_referencing_plugins(self) -> None:
|
|
text = self.files.render_include([self.plugin]).decode()
|
|
self.assertIn('STORE_PLUGINS = [\n "demo_plugin"\n]', text)
|
|
self.assertNotIn("PLUGINS = list", text)
|
|
|
|
def test_atomic_write_backup_and_restore(self) -> None:
|
|
self.config.paths.include_path.write_text("old include", encoding="utf-8")
|
|
self.config.paths.requirements_path.write_text("old requirements", encoding="utf-8")
|
|
snapshot = self.files.write(
|
|
"20f4274f-d4e5-42bf-9164-967b1a774481", [self.plugin]
|
|
)
|
|
self.assertIn("STORE_PLUGINS", self.config.paths.include_path.read_text())
|
|
self.assertIn("--hash=sha256:", self.config.paths.requirements_path.read_text())
|
|
self.files.restore(snapshot)
|
|
self.assertEqual(self.config.paths.include_path.read_text(), "old include")
|
|
self.assertEqual(self.config.paths.requirements_path.read_text(), "old requirements")
|
|
|
|
def test_new_managed_files_are_readable_by_netbox(self) -> None:
|
|
previous = self.files._read_previous(self.config.paths.include_path)
|
|
|
|
self.assertEqual(previous.mode, 0o644)
|
|
|
|
def test_conflicting_locked_distribution_rejected(self) -> None:
|
|
other_requirement = {**self.plugin.requirements[0], "version": "2.0.0"}
|
|
other = ManagedPlugin(
|
|
"other", "demo-plugin", "other_plugin", "2.0.0", False, (other_requirement,)
|
|
)
|
|
with self.assertRaises(PolicyError):
|
|
self.files.render_requirements([self.plugin, other])
|
|
|
|
def test_requirement_host_must_be_allowlisted(self) -> None:
|
|
requirement = {**self.plugin.requirements[0], "download_url": "http://evil.test/x.whl"}
|
|
plugin = ManagedPlugin(
|
|
"demo-plugin", "demo-plugin", "demo_plugin", "1.2.3", False, (requirement,)
|
|
)
|
|
with self.assertRaises(Exception):
|
|
self.files.render_requirements([plugin])
|
|
|
|
def test_path_escape_rejected(self) -> None:
|
|
escaped = self.root / "outside.py"
|
|
changed = self.config.__class__(
|
|
self.config.agent,
|
|
self.config.store,
|
|
self.config.paths.__class__(
|
|
self.config.paths.allowed_root,
|
|
escaped,
|
|
self.config.paths.requirements_path,
|
|
self.config.paths.temp_dir,
|
|
),
|
|
self.config.commands,
|
|
self.config.policy,
|
|
)
|
|
with self.assertRaises(PolicyError):
|
|
ManagedFiles(changed).write(
|
|
"20f4274f-d4e5-42bf-9164-967b1a774481", [self.plugin]
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|