144 lines
4.9 KiB
Python
144 lines
4.9 KiB
Python
import hashlib
|
|
import io
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import patch
|
|
|
|
import _bootstrap # noqa: F401
|
|
|
|
from netbox_plugin_store.client import CatalogPlugin, StoreClient, StoreClientError, URLPolicy
|
|
|
|
|
|
def plugin_payload():
|
|
artifact = b"verified artifact"
|
|
return {
|
|
"slug": "example-plugin",
|
|
"name": "Example",
|
|
"summary": "Example plugin",
|
|
"description": "README",
|
|
"repository_url": "https://git.mrblake.cc/team/example",
|
|
"latest_version": "1.2.0",
|
|
"package_name": "netbox-example",
|
|
"import_name": "netbox_example",
|
|
"min_netbox_version": "4.6.5",
|
|
"max_netbox_version": "4.6.8",
|
|
"approved": True,
|
|
"releases": [
|
|
{
|
|
"version": "1.2.0",
|
|
"download_url": "https://store.example/artifacts/example.whl",
|
|
"sha256": hashlib.sha256(artifact).hexdigest(),
|
|
"approved_payload_sha256": "a" * 64,
|
|
"approved": True,
|
|
"immutable": True,
|
|
"min_netbox_version": "4.6.5",
|
|
"max_netbox_version": "4.6.8",
|
|
}
|
|
],
|
|
}, artifact
|
|
|
|
|
|
class URLPolicyTests(unittest.TestCase):
|
|
def test_exact_origin_and_path_prefix(self):
|
|
policy = URLPolicy(["https://store.example/internal"])
|
|
self.assertEqual(
|
|
policy.check("https://store.example/internal/api/v1/plugins/"),
|
|
"https://store.example/internal/api/v1/plugins/",
|
|
)
|
|
with self.assertRaises(StoreClientError):
|
|
policy.check("https://store.example.evil/internal")
|
|
with self.assertRaises(StoreClientError):
|
|
policy.check("https://store.example/other")
|
|
|
|
def test_release_selection_checks_netbox_version(self):
|
|
payload, _ = plugin_payload()
|
|
plugin = CatalogPlugin.from_mapping(payload)
|
|
self.assertEqual(plugin.select_release("4.6.8").version, "1.2.0")
|
|
with self.assertRaises(StoreClientError):
|
|
plugin.select_release("4.6.9")
|
|
|
|
def test_only_verified_supported_artifacts_are_installable(self):
|
|
payload, _ = plugin_payload()
|
|
payload["releases"].append(
|
|
{
|
|
"version": "1.3.0",
|
|
"download_url": "https://store.example/source.tar.gz",
|
|
"sha256": "b" * 64,
|
|
"approved": False,
|
|
"immutable": True,
|
|
"artifact_kind": "source_archive",
|
|
"min_netbox_version": "4.6.5",
|
|
"max_netbox_version": "4.6.8",
|
|
}
|
|
)
|
|
plugin = CatalogPlugin.from_mapping(payload)
|
|
|
|
self.assertEqual(
|
|
tuple(release.version for release in plugin.installable_releases("4.6.8")),
|
|
("1.2.0",),
|
|
)
|
|
|
|
|
|
class _Response(io.BytesIO):
|
|
def __init__(self, body):
|
|
super().__init__(body)
|
|
self.headers = {"Content-Length": str(len(body))}
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *args):
|
|
self.close()
|
|
|
|
|
|
class _Opener:
|
|
def __init__(self, body):
|
|
self.body = body
|
|
|
|
def open(self, request, timeout):
|
|
return _Response(self.body)
|
|
|
|
|
|
class DownloadTests(unittest.TestCase):
|
|
def test_download_is_hashed_before_use(self):
|
|
_, artifact = plugin_payload()
|
|
client = StoreClient(
|
|
"https://store.example", ("https://store.example",), ("https://store.example",)
|
|
)
|
|
with tempfile.TemporaryDirectory() as temp_name:
|
|
target = Path(temp_name) / "plugin.whl"
|
|
with patch("netbox_plugin_store.client.build_opener", return_value=_Opener(artifact)):
|
|
actual, size = client.download_artifact(
|
|
"https://store.example/artifacts/example.whl",
|
|
target,
|
|
expected_sha256=hashlib.sha256(artifact).hexdigest(),
|
|
timeout=10,
|
|
max_bytes=1_000,
|
|
)
|
|
self.assertEqual(actual, hashlib.sha256(artifact).hexdigest())
|
|
self.assertEqual(size, len(artifact))
|
|
self.assertEqual(target.read_bytes(), artifact)
|
|
|
|
def test_mismatched_hash_removes_download(self):
|
|
_, artifact = plugin_payload()
|
|
client = StoreClient(
|
|
"https://store.example", ("https://store.example",), ("https://store.example",)
|
|
)
|
|
with tempfile.TemporaryDirectory() as temp_name:
|
|
target = Path(temp_name) / "plugin.whl"
|
|
with patch("netbox_plugin_store.client.build_opener", return_value=_Opener(artifact)):
|
|
with self.assertRaises(StoreClientError):
|
|
client.download_artifact(
|
|
"https://store.example/artifacts/example.whl",
|
|
target,
|
|
expected_sha256="0" * 64,
|
|
timeout=10,
|
|
max_bytes=1_000,
|
|
)
|
|
self.assertFalse(target.exists())
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|