else
AC_SUBST(PYTHONEGGINSTALLDIR, "--install-dir $pythoneggdir")
fi
+ AC_PATH_PROG([TOX], [tox], [], [$PATH:/bin:/usr/bin:/usr/local/bin])
AC_PATH_PROG([PY_TEST], [py.test], [], [$PATH:/bin:/usr/bin:/usr/local/bin])
fi
AM_CONDITIONAL(PYTHON_EGGS_INSTALL, [test "x$python_eggs_install" = xtrue])
AM_CONDITIONAL(USE_RUBY_GEMS, test x$ruby_gems = xtrue)
AM_CONDITIONAL(USE_PYTHON_EGGS, test x$python_eggs = xtrue)
AM_CONDITIONAL(USE_PERL_CPAN, test x$perl_cpan = xtrue)
-AM_CONDITIONAL(USE_PY_TEST, test "x$PY_TEST" != x)
+AM_CONDITIONAL(USE_TOX, test "x$TOX" != x)
+AM_CONDITIONAL(USE_PY_TEST, test "x$PY_TEST" != x -a "x$TOX" = x)
# ========================
# set global definitions
libmysqlclient-dev libsqlite3-dev clearsilver-dev libfcgi-dev
libpcsclite-dev libpam0g-dev binutils-dev libunwind8-dev libnm-dev
libjson-c-dev iptables-dev python-pip libtspi-dev libsystemd-dev"
- PYDEPS="pytest"
+ PYDEPS="tox"
if test "$1" = "deps"; then
build_botan
build_wolfssl
include LICENSE
include README.rst
+include tox.ini
+recursive-include test *.py
EXTRA_DIST = LICENSE README.rst MANIFEST.in \
setup.py.in \
- vici/test/__init__.py \
- vici/test/test_protocol.py \
+ tox.ini \
+ tox.sh \
+ test/__init__.py \
+ test/test_protocol.py \
vici/__init__.py \
vici/command_wrappers.py \
vici/compat.py \
dist/vici-$(PYTHON_PACKAGE_VERSION)-py$(PYTHON_VERSION).egg
endif
+if USE_TOX
+ TESTS_ENVIRONMENT = export TOX=$(TOX);
+ AM_TESTS_FD_REDIRECT = 1>&2
+ TESTS = tox.sh
+endif
+
if USE_PY_TEST
TESTS = $(PY_TEST)
endif
"Programming Language :: Python :: 3.2",
"Programming Language :: Python :: 3.3",
"Programming Language :: Python :: 3.4",
+ "Programming Language :: Python :: 3.5",
+ "Programming Language :: Python :: 3.6",
+ "Programming Language :: Python :: 3.7",
+ "Programming Language :: Python :: 3.8",
"Topic :: Security",
"Topic :: Software Development :: Libraries",
)
--- /dev/null
+import pytest
+
+from vici.protocol import Packet, Message, FiniteStream
+from vici.exception import DeserializationException
+
+
+class TestPacket(object):
+ # test data definitions for outgoing packet types
+ cmd_request = b"\x00\x0c" b"command_type"
+ cmd_request_msg = b"\x00\x07" b"command" b"payload"
+ event_register = b"\x03\x0a" b"event_type"
+ event_unregister = b"\x04\x0a" b"event_type"
+
+ # test data definitions for incoming packet types
+ cmd_response = b"\x01" b"reply"
+ cmd_unknown = b"\x02"
+ event_confirm = b"\x05"
+ event_unknown = b"\x06"
+ event = b"\x07\x03" b"log" b"message"
+
+ def test_request(self):
+ assert Packet.request("command_type") == self.cmd_request
+ assert Packet.request("command", b"payload") == self.cmd_request_msg
+
+ def test_register_event(self):
+ assert Packet.register_event("event_type") == self.event_register
+
+ def test_unregister_event(self):
+ assert Packet.unregister_event("event_type") == self.event_unregister
+
+ def test_parse(self):
+ parsed_cmd_response = Packet.parse(self.cmd_response)
+ assert parsed_cmd_response.response_type == Packet.CMD_RESPONSE
+ assert parsed_cmd_response.payload.getvalue() == self.cmd_response
+
+ parsed_cmd_unknown = Packet.parse(self.cmd_unknown)
+ assert parsed_cmd_unknown.response_type == Packet.CMD_UNKNOWN
+ assert parsed_cmd_unknown.payload.getvalue() == self.cmd_unknown
+
+ parsed_event_confirm = Packet.parse(self.event_confirm)
+ assert parsed_event_confirm.response_type == Packet.EVENT_CONFIRM
+ assert parsed_event_confirm.payload.getvalue() == self.event_confirm
+
+ parsed_event_unknown = Packet.parse(self.event_unknown)
+ assert parsed_event_unknown.response_type == Packet.EVENT_UNKNOWN
+ assert parsed_event_unknown.payload.getvalue() == self.event_unknown
+
+ parsed_event = Packet.parse(self.event)
+ assert parsed_event.response_type == Packet.EVENT
+ assert parsed_event.payload.getvalue() == self.event
+
+
+class TestMessage(object):
+ """Message (de)serialization test."""
+
+ # data definitions for test of de(serialization)
+ # serialized messages holding a section
+ ser_sec_unclosed = b"\x01\x08unclosed"
+ ser_sec_single = b"\x01\x07section\x02"
+ ser_sec_nested = b"\x01\x05outer\x01\x0asubsection\x02\x02"
+
+ # serialized messages holding a list
+ ser_list_invalid = b"\x04\x07invalid\x05\x00\x02e1\x02\x03sec\x06"
+ ser_list_0_item = b"\x04\x05empty\x06"
+ ser_list_1_item = b"\x04\x01l\x05\x00\x02e1\x06"
+ ser_list_2_item = b"\x04\x01l\x05\x00\x02e1\x05\x00\x02e2\x06"
+
+ # serialized messages with key value pairs
+ ser_kv_pair = b"\x03\x03key\x00\x05value"
+ ser_kv_zero = b"\x03\x0azerolength\x00\x00"
+
+ # deserialized messages holding a section
+ des_sec_single = {"section": {}}
+ des_sec_nested = {"outer": {"subsection": {}}}
+
+ # deserialized messages holding a list
+ des_list_0_item = {"empty": []}
+ des_list_1_item = {"l": [b"e1"]}
+ des_list_2_item = {"l": [b"e1", b"e2"]}
+
+ # deserialized messages with key value pairs
+ des_kv_pair = {"key": b"value"}
+ des_kv_zero = {"zerolength": b""}
+
+ def test_section_serialization(self):
+ assert Message.serialize(self.des_sec_single) == self.ser_sec_single
+ assert Message.serialize(self.des_sec_nested) == self.ser_sec_nested
+
+ def test_list_serialization(self):
+ assert Message.serialize(self.des_list_0_item) == self.ser_list_0_item
+ assert Message.serialize(self.des_list_1_item) == self.ser_list_1_item
+ assert Message.serialize(self.des_list_2_item) == self.ser_list_2_item
+
+ def test_key_serialization(self):
+ assert Message.serialize(self.des_kv_pair) == self.ser_kv_pair
+ assert Message.serialize(self.des_kv_zero) == self.ser_kv_zero
+
+ def test_section_deserialization(self):
+ single = Message.deserialize(FiniteStream(self.ser_sec_single))
+ nested = Message.deserialize(FiniteStream(self.ser_sec_nested))
+
+ assert single == self.des_sec_single
+ assert nested == self.des_sec_nested
+
+ with pytest.raises(DeserializationException):
+ Message.deserialize(FiniteStream(self.ser_sec_unclosed))
+
+ def test_list_deserialization(self):
+ l0 = Message.deserialize(FiniteStream(self.ser_list_0_item))
+ l1 = Message.deserialize(FiniteStream(self.ser_list_1_item))
+ l2 = Message.deserialize(FiniteStream(self.ser_list_2_item))
+
+ assert l0 == self.des_list_0_item
+ assert l1 == self.des_list_1_item
+ assert l2 == self.des_list_2_item
+
+ with pytest.raises(DeserializationException):
+ Message.deserialize(FiniteStream(self.ser_list_invalid))
+
+ def test_key_deserialization(self):
+ pair = Message.deserialize(FiniteStream(self.ser_kv_pair))
+ zerolength = Message.deserialize(FiniteStream(self.ser_kv_zero))
+
+ assert pair == self.des_kv_pair
+ assert zerolength == self.des_kv_zero
+
+ def test_roundtrip(self):
+ message = {
+ "key1": "value1",
+ "section1": {
+ "sub-section": {
+ "key2": b"value2",
+ },
+ "list1": ["item1", "item2"],
+ },
+ }
+ serialized_message = FiniteStream(Message.serialize(message))
+ deserialized_message = Message.deserialize(serialized_message)
+
+ # ensure that list items and key values remain as undecoded bytes
+ deserialized_section = deserialized_message["section1"]
+ assert deserialized_message["key1"] == b"value1"
+ assert deserialized_section["sub-section"]["key2"] == b"value2"
+ assert deserialized_section["list1"] == [b"item1", b"item2"]
--- /dev/null
+[tox]
+envlist = py27, py35, py36, py37, py38
+
+[testenv]
+deps =
+ pytest
+ pytest-pycodestyle
+commands = pytest --pycodestyle
+
+[testenv:py27]
+deps = pytest
+commands = pytest
+
+[pycodestyle]
+max-line-length = 80
+show-source = True
--- /dev/null
+#!/bin/sh
+
+${TOX} -c ${srcdir} --skip-missing-interpreters
"""
self.request("load-shared", secret)
-
def unload_shared(self, identifier):
"""Unload a previously loaded shared secret by its unique identifier.
+++ /dev/null
-import pytest
-
-from ..protocol import Packet, Message, FiniteStream
-from ..exception import DeserializationException
-
-
-class TestPacket(object):
- # test data definitions for outgoing packet types
- cmd_request = b"\x00\x0c" b"command_type"
- cmd_request_msg = b"\x00\x07" b"command" b"payload"
- event_register = b"\x03\x0a" b"event_type"
- event_unregister = b"\x04\x0a" b"event_type"
-
- # test data definitions for incoming packet types
- cmd_response = b"\x01" b"reply"
- cmd_unknown = b"\x02"
- event_confirm = b"\x05"
- event_unknown = b"\x06"
- event = b"\x07\x03" b"log" b"message"
-
- def test_request(self):
- assert Packet.request("command_type") == self.cmd_request
- assert Packet.request("command", b"payload") == self.cmd_request_msg
-
- def test_register_event(self):
- assert Packet.register_event("event_type") == self.event_register
-
- def test_unregister_event(self):
- assert Packet.unregister_event("event_type") == self.event_unregister
-
- def test_parse(self):
- parsed_cmd_response = Packet.parse(self.cmd_response)
- assert parsed_cmd_response.response_type == Packet.CMD_RESPONSE
- assert parsed_cmd_response.payload.getvalue() == self.cmd_response
-
- parsed_cmd_unknown = Packet.parse(self.cmd_unknown)
- assert parsed_cmd_unknown.response_type == Packet.CMD_UNKNOWN
- assert parsed_cmd_unknown.payload.getvalue() == self.cmd_unknown
-
- parsed_event_confirm = Packet.parse(self.event_confirm)
- assert parsed_event_confirm.response_type == Packet.EVENT_CONFIRM
- assert parsed_event_confirm.payload.getvalue() == self.event_confirm
-
- parsed_event_unknown = Packet.parse(self.event_unknown)
- assert parsed_event_unknown.response_type == Packet.EVENT_UNKNOWN
- assert parsed_event_unknown.payload.getvalue() == self.event_unknown
-
- parsed_event = Packet.parse(self.event)
- assert parsed_event.response_type == Packet.EVENT
- assert parsed_event.payload.getvalue() == self.event
-
-
-class TestMessage(object):
- """Message (de)serialization test."""
-
- # data definitions for test of de(serialization)
- # serialized messages holding a section
- ser_sec_unclosed = b"\x01\x08unclosed"
- ser_sec_single = b"\x01\x07section\x02"
- ser_sec_nested = b"\x01\x05outer\x01\x0asubsection\x02\x02"
-
- # serialized messages holding a list
- ser_list_invalid = b"\x04\x07invalid\x05\x00\x02e1\x02\x03sec\x06"
- ser_list_0_item = b"\x04\x05empty\x06"
- ser_list_1_item = b"\x04\x01l\x05\x00\x02e1\x06"
- ser_list_2_item = b"\x04\x01l\x05\x00\x02e1\x05\x00\x02e2\x06"
-
- # serialized messages with key value pairs
- ser_kv_pair = b"\x03\x03key\x00\x05value"
- ser_kv_zero = b"\x03\x0azerolength\x00\x00"
-
- # deserialized messages holding a section
- des_sec_single = { "section": {} }
- des_sec_nested = { "outer": { "subsection": {} } }
-
- # deserialized messages holding a list
- des_list_0_item = { "empty": [] }
- des_list_1_item = { "l": [ b"e1" ] }
- des_list_2_item = { "l": [ b"e1", b"e2" ] }
-
- # deserialized messages with key value pairs
- des_kv_pair = { "key": b"value" }
- des_kv_zero = { "zerolength": b"" }
-
- def test_section_serialization(self):
- assert Message.serialize(self.des_sec_single) == self.ser_sec_single
- assert Message.serialize(self.des_sec_nested) == self.ser_sec_nested
-
- def test_list_serialization(self):
- assert Message.serialize(self.des_list_0_item) == self.ser_list_0_item
- assert Message.serialize(self.des_list_1_item) == self.ser_list_1_item
- assert Message.serialize(self.des_list_2_item) == self.ser_list_2_item
-
- def test_key_serialization(self):
- assert Message.serialize(self.des_kv_pair) == self.ser_kv_pair
- assert Message.serialize(self.des_kv_zero) == self.ser_kv_zero
-
- def test_section_deserialization(self):
- single = Message.deserialize(FiniteStream(self.ser_sec_single))
- nested = Message.deserialize(FiniteStream(self.ser_sec_nested))
-
- assert single == self.des_sec_single
- assert nested == self.des_sec_nested
-
- with pytest.raises(DeserializationException):
- Message.deserialize(FiniteStream(self.ser_sec_unclosed))
-
- def test_list_deserialization(self):
- l0 = Message.deserialize(FiniteStream(self.ser_list_0_item))
- l1 = Message.deserialize(FiniteStream(self.ser_list_1_item))
- l2 = Message.deserialize(FiniteStream(self.ser_list_2_item))
-
- assert l0 == self.des_list_0_item
- assert l1 == self.des_list_1_item
- assert l2 == self.des_list_2_item
-
- with pytest.raises(DeserializationException):
- Message.deserialize(FiniteStream(self.ser_list_invalid))
-
- def test_key_deserialization(self):
- pair = Message.deserialize(FiniteStream(self.ser_kv_pair))
- zerolength = Message.deserialize(FiniteStream(self.ser_kv_zero))
-
- assert pair == self.des_kv_pair
- assert zerolength == self.des_kv_zero
-
- def test_roundtrip(self):
- message = {
- "key1": "value1",
- "section1": {
- "sub-section": {
- "key2": b"value2",
- },
- "list1": [ "item1", "item2" ],
- },
- }
- serialized_message = FiniteStream(Message.serialize(message))
- deserialized_message = Message.deserialize(serialized_message)
-
- # ensure that list items and key values remain as undecoded bytes
- deserialized_section = deserialized_message["section1"]
- assert deserialized_message["key1"] == b"value1"
- assert deserialized_section["sub-section"]["key2"] == b"value2"
- assert deserialized_section["list1"] == [ b"item1", b"item2" ]
projects / strongswan.git / commitdiff