b453c821c7
- Writes tests for aircox streamer application; - Add test utilities in aircox Co-authored-by: bkfox <thomas bkfox net> Reviewed-on: #110
71 lines
2.3 KiB
Python
71 lines
2.3 KiB
Python
import json
|
|
import os
|
|
import socket
|
|
|
|
from .conftest import working_dir
|
|
|
|
|
|
class TestConnector:
|
|
payload = "non_value_info\n" 'a="value_1"\n' 'b="value_b"\n' "END"
|
|
"""Test payload."""
|
|
payload_data = {"a": "value_1", "b": "value_b"}
|
|
"""Resulting data of payload."""
|
|
|
|
def test_open(self, connector):
|
|
assert connector.open() == 0
|
|
assert connector.is_open
|
|
assert connector.socket.family == socket.AF_UNIX
|
|
assert connector.socket.type == socket.SOCK_STREAM
|
|
assert connector.socket.address == os.path.join(
|
|
working_dir, "test.sock"
|
|
)
|
|
connector.close()
|
|
|
|
def test_open_af_inet(self, connector):
|
|
address = ("test", 30)
|
|
connector.address = address
|
|
assert connector.open() == 0
|
|
assert connector.is_open
|
|
assert connector.socket.family == socket.AF_INET
|
|
assert connector.socket.type == socket.SOCK_STREAM
|
|
assert connector.socket.address == address
|
|
|
|
def test_open_is_already_open(self, connector):
|
|
connector.open()
|
|
assert connector.open() == 1
|
|
|
|
def test_open_failure(self, fail_connector):
|
|
assert fail_connector.open() == -1
|
|
assert fail_connector.socket is None # close() called
|
|
|
|
def test_close(self, connector):
|
|
connector.open()
|
|
assert connector.socket is not None
|
|
connector.close()
|
|
assert connector.socket is None
|
|
|
|
def test_send(self, connector):
|
|
connector.open()
|
|
connector.socket.recv_data = self.payload
|
|
result = connector.send("fake_action", parse=True)
|
|
assert result == self.payload_data
|
|
|
|
def test_send_open_failure(self, fail_connector):
|
|
assert fail_connector.send("fake_action", parse=True) is None
|
|
|
|
def test_parse(self, connector):
|
|
result = connector.parse(self.payload)
|
|
assert result == self.payload_data
|
|
|
|
def test_parse_json(self, connector):
|
|
# include case where json string is surrounded by '"'
|
|
dumps = '"' + json.dumps(self.payload_data) + '"'
|
|
result = connector.parse_json(dumps)
|
|
assert result == self.payload_data
|
|
|
|
def test_parse_json_empty_value(self, connector):
|
|
assert connector.parse_json('""') is None
|
|
|
|
def test_parse_json_failure(self, connector):
|
|
assert connector.parse_json("-- invalid json string --") is None
|