84 lines
3.1 KiB
Python
84 lines
3.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import unittest
|
|
|
|
from support import * # noqa: F403
|
|
|
|
from netbox_store_agent.errors import ValidationError
|
|
from netbox_store_agent.protocol import decode_request, encode_response, parse_request, response
|
|
|
|
|
|
class ProtocolTests(unittest.TestCase):
|
|
def submit(self, **body_overrides: object) -> dict[str, object]:
|
|
body: dict[str, object] = {
|
|
"request_id": "eea17d87-8944-4ee2-a076-363338ab746d",
|
|
"action": "install",
|
|
"plugin_slug": "demo-plugin",
|
|
"version": "1.2.3",
|
|
"approved_payload_sha256": "c" * 64,
|
|
"requested_by": "netbox:alice",
|
|
}
|
|
body.update(body_overrides)
|
|
return {
|
|
"protocol_version": 1,
|
|
"method": "POST",
|
|
"path": "/v1/operations",
|
|
"idempotency_key": "20f4274f-d4e5-42bf-9164-967b1a774481",
|
|
"body": body,
|
|
}
|
|
|
|
def test_exact_submit_contract_and_opaque_token(self) -> None:
|
|
request = parse_request(self.submit())
|
|
self.assertEqual(request.operation.approved_payload_sha256, "c" * 64)
|
|
|
|
def test_unknown_top_level_field_rejected(self) -> None:
|
|
value = self.submit(extra=True)
|
|
value["unexpected"] = True
|
|
with self.assertRaises(ValidationError):
|
|
parse_request(value)
|
|
|
|
def test_unknown_body_field_rejected(self) -> None:
|
|
with self.assertRaises(ValidationError):
|
|
parse_request(self.submit(extra=True))
|
|
|
|
def test_non_lifecycle_fields_must_be_null(self) -> None:
|
|
with self.assertRaises(ValidationError):
|
|
parse_request(self.submit(action="disable"))
|
|
request = parse_request(
|
|
self.submit(action="disable", version=None, approved_payload_sha256=None)
|
|
)
|
|
self.assertEqual(request.operation.action, "disable")
|
|
|
|
def test_noncanonical_uuid_rejected(self) -> None:
|
|
with self.assertRaises(ValidationError):
|
|
parse_request(self.submit(request_id="EEA17D87-8944-4EE2-A076-363338AB746D"))
|
|
|
|
def test_multiple_lines_and_oversize_rejected(self) -> None:
|
|
raw = json.dumps(self.submit()).encode()
|
|
with self.assertRaises(ValidationError):
|
|
decode_request(raw + b"\n{}", 65536)
|
|
with self.assertRaises(ValidationError):
|
|
decode_request(b"x" * 10, 9)
|
|
|
|
def test_capabilities_and_status_paths(self) -> None:
|
|
cap = parse_request({"protocol_version": 1, "method": "GET", "path": "/v1/capabilities"})
|
|
self.assertEqual(cap.path, "/v1/capabilities")
|
|
status = parse_request(
|
|
{
|
|
"protocol_version": 1,
|
|
"method": "GET",
|
|
"path": "/v1/operations/20f4274f-d4e5-42bf-9164-967b1a774481",
|
|
}
|
|
)
|
|
self.assertEqual(status.idempotency_key, "20f4274f-d4e5-42bf-9164-967b1a774481")
|
|
|
|
def test_response_shape_and_single_line(self) -> None:
|
|
encoded = encode_response(response(200, {"ok": True}))
|
|
self.assertEqual(encoded.count(b"\n"), 1)
|
|
self.assertEqual(set(json.loads(encoded)), {"protocol_version", "status", "body"})
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|