133 lines
4.2 KiB
Python
133 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from support import * # noqa: F403
|
|
|
|
from netbox_store_agent.config import load_config
|
|
from netbox_store_agent.errors import ValidationError
|
|
|
|
|
|
class ConfigTests(unittest.TestCase):
|
|
def setUp(self) -> None:
|
|
self.temporary = tempfile.TemporaryDirectory()
|
|
self.root = Path(self.temporary.name).resolve()
|
|
|
|
def tearDown(self) -> None:
|
|
self.temporary.cleanup()
|
|
|
|
@staticmethod
|
|
def q(path: Path) -> str:
|
|
return json.dumps(path.as_posix())
|
|
|
|
def config_text(self, **changes: str) -> str:
|
|
root = self.root
|
|
values = {
|
|
"agent_extra": "",
|
|
"store_extra": "",
|
|
"paths_extra": "",
|
|
"commands_extra": "",
|
|
"policy_extra": "",
|
|
"base_url": '"https://store.test"',
|
|
"include": self.q(root / "netbox" / "store_plugins.py"),
|
|
}
|
|
values.update(changes)
|
|
return f"""
|
|
[agent]
|
|
socket_path = {self.q(root / 'agent.sock')}
|
|
journal_path = {self.q(root / 'state' / 'journal.sqlite3')}
|
|
lock_path = {self.q(root / 'state' / 'lock')}
|
|
backup_dir = {self.q(root / 'state' / 'backups')}
|
|
require_root = false
|
|
require_peer_credentials = false
|
|
{values['agent_extra']}
|
|
|
|
[store]
|
|
base_url = {values['base_url']}
|
|
allowed_hosts = ["store.test"]
|
|
{values['store_extra']}
|
|
|
|
[paths]
|
|
allowed_root = {self.q(root / 'netbox')}
|
|
include_path = {values['include']}
|
|
requirements_path = {self.q(root / 'netbox' / 'requirements.txt')}
|
|
temp_dir = {self.q(root / 'state' / 'tmp')}
|
|
{values['paths_extra']}
|
|
|
|
[commands]
|
|
python_path = {self.q(root / 'bin' / 'python')}
|
|
manage_path = {self.q(root / 'netbox' / 'manage.py')}
|
|
systemctl_path = {self.q(root / 'bin' / 'systemctl')}
|
|
{values['commands_extra']}
|
|
|
|
[policy]
|
|
{values['policy_extra']}
|
|
"""
|
|
|
|
def write(self, text: str) -> Path:
|
|
path = self.root / "agent.toml"
|
|
path.write_text(text, encoding="utf-8")
|
|
path.chmod(0o600)
|
|
return path
|
|
|
|
def test_defaults_are_dry_run_and_block_all_self_slugs(self) -> None:
|
|
config = load_config(self.write(self.config_text()), allow_insecure_owner=True)
|
|
self.assertTrue(config.agent.dry_run)
|
|
self.assertIn("netbox-plugin-store", config.policy.self_plugin_slugs)
|
|
self.assertIn("netbox_plugin_store", config.policy.self_plugin_slugs)
|
|
|
|
def test_unknown_setting_rejected(self) -> None:
|
|
with self.assertRaises(ValidationError):
|
|
load_config(
|
|
self.write(self.config_text(agent_extra="surprise = true")),
|
|
allow_insecure_owner=True,
|
|
)
|
|
|
|
def test_http_requires_explicit_testing_switch(self) -> None:
|
|
with self.assertRaises(ValidationError):
|
|
load_config(
|
|
self.write(self.config_text(base_url='"http://store.test"')),
|
|
allow_insecure_owner=True,
|
|
)
|
|
config = load_config(
|
|
self.write(
|
|
self.config_text(
|
|
base_url='"http://store.test"', store_extra="allow_http_for_testing = true"
|
|
)
|
|
),
|
|
allow_insecure_owner=True,
|
|
)
|
|
self.assertTrue(config.store.allow_http_for_testing)
|
|
|
|
def test_managed_path_escape_rejected(self) -> None:
|
|
with self.assertRaises(ValidationError):
|
|
load_config(
|
|
self.write(self.config_text(include=self.q(self.root / "outside.py"))),
|
|
allow_insecure_owner=True,
|
|
)
|
|
|
|
def test_protocol_size_cannot_exceed_64k(self) -> None:
|
|
with self.assertRaises(ValidationError):
|
|
load_config(
|
|
self.write(self.config_text(agent_extra="max_request_bytes = 65537")),
|
|
allow_insecure_owner=True,
|
|
)
|
|
|
|
def test_endpoint_placeholders_are_exact(self) -> None:
|
|
with self.assertRaises(ValidationError):
|
|
load_config(
|
|
self.write(
|
|
self.config_text(
|
|
store_extra='release_endpoint_template = "/api/{plugin_slug}/{other}"'
|
|
)
|
|
),
|
|
allow_insecure_owner=True,
|
|
)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|