phares3757/barcode-server
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Init

Browse Source
This commit is contained in:
Mike Phares
2022-06-24 12:14:10 -07:00
commit 27fc1f5e16
31 changed files with 3769 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files

34
tests/__init__.py Normal file
View File

@ -0,0 +1,34 @@
# Copyright (c) 2019 Markus Ressel
# .
# Permission is hereby granted, free of charge, to any person obtaining a copy
# of this software and associated documentation files (the "Software"), to deal
# in the Software without restriction, including without limitation the rights
# to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
# copies of the Software, and to permit persons to whom the Software is
# furnished to do so, subject to the following conditions:
# .
# The above copyright notice and this permission notice shall be included in all
# copies or substantial portions of the Software.
# .
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
# SOFTWARE.
from unittest import IsolatedAsyncioTestCase
class TestBase(IsolatedAsyncioTestCase):
from barcode_server.config import AppConfig
from container_app_conf.source.yaml_source import YamlSource
# load config from test folder
config = AppConfig(
singleton=True,
data_sources=[
YamlSource("barcode_server", "./tests/")
]
)

32
tests/api_test.py Normal file
View File

@ -0,0 +1,32 @@
from datetime import datetime
from unittest.mock import Mock
from barcode_server.barcode import BarcodeEvent
from barcode_server.util import barcode_event_to_json
from tests import TestBase
class ApiTest(TestBase):
def test_json(self):
date_str = "2020-08-03T10:00:00+00:00"
input_device = Mock()
input_device.name = "Barcode Scanner"
input_device.path = "/dev/input/event2"
input_device.info.vendor = 1
input_device.info.product = 2
date = datetime.fromisoformat(str(date_str))
barcode = "4006824000970"
server_id = "server-id"
event = BarcodeEvent(input_device, barcode, date)
event_json = str(barcode_event_to_json(server_id, event))
self.assertIn(event.id, event_json)
self.assertIn(server_id, event_json)
self.assertIn(date_str, event_json)
self.assertIn(input_device.path, event_json)
self.assertIn(barcode, event_json)
self.assertIsNotNone(event_json)

11
tests/barcode_reader_test.py Normal file
View File

@ -0,0 +1,11 @@
from barcode_server.barcode import BarcodeReader
from barcode_server.config import AppConfig
from tests import TestBase
class BarcodeReaderTest(TestBase):
async def test_initialization(self):
config = AppConfig()
reader = BarcodeReader(config)
self.assertIsNotNone(reader)

9
tests/barcode_server.yaml Normal file
View File

@ -0,0 +1,9 @@
barcode_server:
server:
host: "127.0.0.1"
port: 9654
api_token: "EmUSqjXGfnQwn5wn6CpzJRZgoazMTRbMNgH7CXwkQG7Ph7stex"
devices:
- "Barcode/i"
stats:
port:

167
tests/key_event_reader_test.py Normal file
View File

@ -0,0 +1,167 @@
from typing import List
from unittest.mock import Mock
from evdev import InputEvent, ecodes, KeyEvent
from barcode_server.keyevent_reader import KeyEventReader
from tests import TestBase
class KeyEventReaderTest(TestBase):
@staticmethod
def fake_input_loop(input: List[InputEvent]):
input_events = input
def read_loop():
for event in input_events:
yield event
return read_loop
@staticmethod
def mock_input_event(keycode, keystate) -> InputEvent:
input_event = Mock(spec=InputEvent)
input_event.type = 1
input_event.keystate = keystate # 0: UP, 1: Down, 2: Hold
input_event.keycode = keycode
# inverse lookup of the event code in the target structure
code = next(key for key, value in ecodes.keys.items() if value == keycode)
input_event.code = code
return input_event
def generate_input_event_sequence(self, expected: str, finish_line: bool = True) -> List[InputEvent]:
events = []
keycodes = list(map(lambda x: self.character_to_keycode(x), expected))
for item in keycodes:
for keystate in [KeyEvent.key_down, KeyEvent.key_up]:
event = self.mock_input_event(keycode=item, keystate=keystate)
events.append(event)
if finish_line:
for keystate in [KeyEvent.key_down, KeyEvent.key_up]:
event = self.mock_input_event(keycode="KEY_ENTER", keystate=keystate)
events.append(event)
return events
@staticmethod
def character_to_keycode(character: str) -> str:
char_to_keycode_map = {
'0': "KEY_KP0",
'1': "KEY_KP1",
'2': "KEY_KP2",
'3': "KEY_KP3",
'4': "KEY_KP4",
'5': "KEY_KP5",
'6': "KEY_KP6",
'7': "KEY_KP7",
'8': "KEY_KP8",
'9': "KEY_KP9",
'*': "KEY_KPASTERISK",
'/': "KEY_SLASH",
'-': "KEY_KPMINUS",
'+': "KEY_KPPLUS",
'.': "KEY_DOT",
',': "KEY_COMMA",
'?': "KEY_QUESTION",
'\n': "KEY_ENTER",
}
return char_to_keycode_map[character]
def test_mock_gen(self):
# GIVEN
expected = [
self.mock_input_event(keycode="KEY_KPMINUS", keystate=KeyEvent.key_down),
self.mock_input_event(keycode="KEY_KPMINUS", keystate=KeyEvent.key_up),
self.mock_input_event(keycode="KEY_DOT", keystate=KeyEvent.key_down),
self.mock_input_event(keycode="KEY_DOT", keystate=KeyEvent.key_up),
self.mock_input_event(keycode="KEY_KPMINUS", keystate=KeyEvent.key_down),
self.mock_input_event(keycode="KEY_KPMINUS", keystate=KeyEvent.key_up),
self.mock_input_event(keycode="KEY_ENTER", keystate=KeyEvent.key_down),
self.mock_input_event(keycode="KEY_ENTER", keystate=KeyEvent.key_up),
]
text = "-.-"
# WHEN
input_events = self.generate_input_event_sequence(text)
# THEN
self.assertEqual(len(expected), len(input_events))
for i in range(0, len(expected)):
self.assertEqual(expected[i].keycode, input_events[i].keycode)
async def test_numbers(self):
# GIVEN
under_test = KeyEventReader()
expected = "0123456789"
input_events = self.generate_input_event_sequence(expected)
input_device = Mock()
input_device.read_loop = self.fake_input_loop(input_events)
# WHEN
line = under_test.read_line(input_device)
# THEN
self.assertEqual(expected, line)
async def test_special_characters(self):
# GIVEN
under_test = KeyEventReader()
expected = "+.,*/-?"
input_events = self.generate_input_event_sequence(expected)
input_device = Mock()
input_device.read_loop = self.fake_input_loop(input_events)
# WHEN
line = under_test.read_line(input_device)
# THEN
self.assertEqual(expected, line)
async def test_none_event(self):
# GIVEN
under_test = KeyEventReader()
unexpected = "0"
expected = "1"
input_events = self.generate_input_event_sequence(unexpected + expected)
input_events[0] = None
input_events[1] = None
input_device = Mock()
input_device.read_loop = self.fake_input_loop(input_events)
# WHEN
line = under_test.read_line(input_device)
# THEN
self.assertEqual(expected, line)
async def test_event_without_keystate_attribute(self):
# GIVEN
under_test = KeyEventReader()
unexpected = "0"
input_events_without_keystate = self.generate_input_event_sequence(unexpected)
for event in input_events_without_keystate:
delattr(event, "keystate")
expected = "1"
input_events = self.generate_input_event_sequence(expected)
input_events = input_events + input_events_without_keystate
input_device = Mock()
input_device.read_loop = self.fake_input_loop(input_events)
# WHEN
line = under_test.read_line(input_device)
# THEN
self.assertEqual(expected, line)

13
tests/notifier_test.py Normal file
View File

@ -0,0 +1,13 @@
from barcode_server.notifier.http import HttpNotifier
from tests import TestBase
class NotifierTest(TestBase):
def test_http(self):
method = "POST"
url = "test.de"
headers = []
reader = HttpNotifier(method, url, headers)
self.assertIsNotNone(reader)

199
tests/websocket_notifier_test.py Normal file
View File

@ -0,0 +1,199 @@
import asyncio
import random
from unittest.mock import MagicMock
import aiohttp
from aiohttp.test_utils import AioHTTPTestCase, unittest_run_loop
from barcode_server import const
from barcode_server.barcode import BarcodeEvent
from barcode_server.util import barcode_event_to_json
from barcode_server.webserver import Webserver
def create_barcode_event_mock(barcode: str = None):
device = lambda: None
device.info = lambda: None
device.name = "BARCODE SCANNER BARCODE SCANNER"
device.path = "/dev/input/event3"
device.info.vendor = 1
device.info.product = 1
event = BarcodeEvent(
device,
barcode if barcode is not None else f"{random.getrandbits(24)}"
)
return event
class WebsocketNotifierTest(AioHTTPTestCase):
from barcode_server.config import AppConfig
from container_app_conf.source.yaml_source import YamlSource
# load config from test folder
config = AppConfig(
singleton=True,
data_sources=[
YamlSource("barcode_server", "./tests/")
]
)
webserver = None
async def get_application(self):
"""
Override the get_app method to return your application.
"""
barcode_reader = MagicMock()
self.webserver = Webserver(self.config, barcode_reader)
app = self.webserver.create_app()
runner = aiohttp.web.AppRunner(app)
await runner.setup()
site = aiohttp.web.TCPSite(
runner,
host=self.config.SERVER_HOST.value,
port=self.config.SERVER_PORT.value
)
await site.start()
return app
# the unittest_run_loop decorator can be used in tandem with
# the AioHTTPTestCase to simplify running
# tests that are asynchronous
@unittest_run_loop
async def test_ws_connect_and_event(self):
sample_event = create_barcode_event_mock("abcdefg")
server_id = self.config.INSTANCE_ID.value
expected_json = barcode_event_to_json(server_id, sample_event)
import uuid
client_id = str(uuid.uuid4())
async with aiohttp.ClientSession() as session:
async with session.ws_connect(
'http://127.0.0.1:9654/',
headers={
const.Client_Id: client_id,
const.X_Auth_Token: self.config.SERVER_API_TOKEN.value or ""
}) as ws:
asyncio.create_task(self.webserver.on_barcode(sample_event))
async for msg in ws:
if msg.type == aiohttp.WSMsgType.BINARY:
self.assertEqual(expected_json, msg.data)
await ws.close()
return
else:
self.fail("No event received")
assert False
@unittest_run_loop
async def test_ws_reconnect_event_catchup(self):
server_id = self.config.INSTANCE_ID.value
missed_event = create_barcode_event_mock("abcdefg")
second_event = create_barcode_event_mock("123456")
missed_event_json = barcode_event_to_json(server_id, missed_event)
second_event_json = barcode_event_to_json(server_id, second_event)
import uuid
client_id = str(uuid.uuid4())
# connect to the server once
async with aiohttp.ClientSession() as session:
async with session.ws_connect(
'http://127.0.0.1:9654/',
headers={
const.Client_Id: client_id,
const.X_Auth_Token: self.config.SERVER_API_TOKEN.value or ""
}) as ws:
await ws.close()
# then emulate a barcode scan event
asyncio.create_task(self.webserver.on_barcode(missed_event))
await asyncio.sleep(0.1)
# and then reconnect again, expecting the event in between
async with aiohttp.ClientSession() as session:
async with session.ws_connect(
'http://127.0.0.1:9654/',
headers={
const.Client_Id: client_id,
const.X_Auth_Token: self.config.SERVER_API_TOKEN.value or ""
}) as ws:
# emulate another event, while connected
asyncio.create_task(self.webserver.on_barcode(second_event))
missed_event_received = False
async for msg in ws:
if msg.type == aiohttp.WSMsgType.BINARY:
if missed_event_json == msg.data:
if missed_event_received:
assert False
missed_event_received = True
elif second_event_json == msg.data:
if not missed_event_received:
assert False
await ws.close()
return
else:
assert False
else:
self.fail("No event received")
assert False
@unittest_run_loop
async def test_ws_reconnect_drop_queue(self):
server_id = self.config.INSTANCE_ID.value
missed_event = create_barcode_event_mock("abcdefg")
second_event = create_barcode_event_mock("123456")
missed_event_json = barcode_event_to_json(server_id, missed_event)
second_event_json = barcode_event_to_json(server_id, second_event)
import uuid
client_id = str(uuid.uuid4())
# connect to the server once
async with aiohttp.ClientSession() as session:
async with session.ws_connect(
'http://127.0.0.1:9654/',
headers={
const.Client_Id: client_id,
const.X_Auth_Token: self.config.SERVER_API_TOKEN.value or ""
}) as ws:
await ws.close()
# then emulate a barcode scan event while not connected
asyncio.create_task(self.webserver.on_barcode(missed_event))
await asyncio.sleep(0.1)
# and then reconnect again, passing the "drop cache" header, expecting only
# the new live event
async with aiohttp.ClientSession() as session:
async with session.ws_connect(
'http://127.0.0.1:9654/',
headers={
const.Client_Id: client_id,
const.Drop_Event_Queue: "",
const.X_Auth_Token: self.config.SERVER_API_TOKEN.value or ""
}) as ws:
# emulate another event, while connected
asyncio.create_task(self.webserver.on_barcode(second_event))
async for msg in ws:
if msg.type == aiohttp.WSMsgType.BINARY:
if missed_event_json == msg.data:
self.fail("Received missed event despite queue drop")
elif second_event_json == msg.data:
await ws.close()
assert True
return
else:
self.fail("Received unexpected event")
else:
self.fail("No event received")
assert False