From bb65bd7bf03da7ad5a9ae25d6caab57d06211b99 Mon Sep 17 00:00:00 2001 From: Roberto Hueso Gómez Date: Thu, 16 Jan 2020 11:33:26 +0100 Subject: Add unit testing basic structure --- tests/run-tests.py | 6 ++++++ tests/units/__init__.py | 1 + tests/units/client.py | 17 +++++++++++++++++ tests/units/server.py | 42 ++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 66 insertions(+) create mode 100644 tests/run-tests.py create mode 100644 tests/units/__init__.py create mode 100644 tests/units/client.py create mode 100644 tests/units/server.py diff --git a/tests/run-tests.py b/tests/run-tests.py new file mode 100644 index 0000000..5e6bfe7 --- /dev/null +++ b/tests/run-tests.py @@ -0,0 +1,6 @@ +#!/usr/bin/env python3 + +import subprocess + +# Run all tests in folder units. +subprocess.run('python3 -m unittest discover -s units -v', shell=True) diff --git a/tests/units/__init__.py b/tests/units/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/tests/units/__init__.py @@ -0,0 +1 @@ + diff --git a/tests/units/client.py b/tests/units/client.py new file mode 100644 index 0000000..4e780a7 --- /dev/null +++ b/tests/units/client.py @@ -0,0 +1,17 @@ +import subprocess +import os + +class Client(): + + def __init__(self): + self.null = open(os.devnull, 'wb') + self.proc = subprocess.Popen(['python3', 'main.py'], + cwd='../', + stdout=self.null, + stderr=self.null) + + def stop(self): + self.proc.terminate() + self.proc.kill() + self.proc.wait() + self.null.close() diff --git a/tests/units/server.py b/tests/units/server.py new file mode 100644 index 0000000..764ae6a --- /dev/null +++ b/tests/units/server.py @@ -0,0 +1,42 @@ +import socket +import sys + +HOST = '127.0.0.1' +PORT = '1234' + +class Server(): + + _probe_json = '{"id": 0, "name": "test_local", "center": 0, "room": 0}' + _probe_msg = 'POST /probe HTTP/1.0\r\nContent-Length:'+ \ + str(len(_probe_json)) + \ + '\r\nContent-Type:application/json\r\n\r\n' + _probe_json + + def __init__(self, host='127.0.0.1', port=1234): + self.host = host + self.port = port + self.sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM) + self.sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + + def connect(self, probe=True): + try: + self.sock.bind((self.host, self.port)) + except socket.error as msg: + print('Bind failed. Error Code : ' + str(msg[0]) + ' Message ' + + msg[1]) + sys.exit() + + self.sock.listen(10) + self.conn, self.addr = self.sock.accept() + if probe: + self.send(self._probe_msg) + return self.recv() + + def send(self, msg): + self.conn.send(msg.encode()) + + def recv(self): + return self.conn.recv(1024).decode('utf-8') + + def stop(self): + self.conn.close() + self.sock.close() -- cgit v1.2.3-18-g5258