diff --git a/.gitignore b/.gitignore index 8bfc6cc..2033ffe 100644 --- a/.gitignore +++ b/.gitignore @@ -1,3 +1,8 @@ *.pyc *.sip +tmp __pycache__ +build +dist +lmwsip_marceln.egg-info +test/__pycache__ diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..1333ed7 --- /dev/null +++ b/LICENSE @@ -0,0 +1 @@ +TODO diff --git a/lmwsip/__init__.py b/lmwsip/__init__.py index 73ed59d..93c18c3 100644 --- a/lmwsip/__init__.py +++ b/lmwsip/__init__.py @@ -151,7 +151,7 @@ send a sip command to the server self.closesocket() raise LmwSipConnectError("LmwSip.send: Socket connection lost") else: - self.log.warn("LmwSip.send: No connection") + self.log.warning("LmwSip.send: No connection") def recv(self): """recv() diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..fb026dc --- /dev/null +++ b/setup.py @@ -0,0 +1,22 @@ +import setuptools + +with open("README.md", "r") as fh: + long_description = fh.read() + +setuptools.setup( + name="lmwsip-marceln", # Replace with your own username + version="0.0.1", + author="Marcel Nijenhof", + author_email="pip@pion.xs4all.nl", + description="Interface for the lmw sip protocol", + long_description=long_description, + long_description_content_type="text/markdown", + url="https://marceln.org/git/Werk/lmwsip", + packages=setuptools.find_packages(), + classifiers=[ + "Programming Language :: Python :: 3", + "License :: TODO", + "Operating System :: OS Independent", + ], + python_requires='>=3.6', +) diff --git a/test/__init__.py b/test/__init__.py new file mode 100755 index 0000000..e27e7c8 --- /dev/null +++ b/test/__init__.py @@ -0,0 +1,51 @@ +#!/usr/bin/python + +import unittest +import lmwsip +import stubSipServer + +from datetime import datetime, timedelta +from dateutil import tz + +class lmwsipTest(unittest.TestCase): + + def setUp(self): + self.sipserver = stubSipServer.sipServer() + self.sipserver.run() + self.sip = lmwsip.LmwSip("USER", "PASS", "localhost", + self.sipserver.port, ssl=False) + + def tearDown(self): + self.sipserver.kill() + self.sip.closesocket() + + def test_sipobj(self): + self.assertEqual(type(self.sip), lmwsip.LmwSip) + + def test_ti(self): + self.assertEqual(type(self.sip.ti()), str) + + def test_cmd(self): + self.assertEqual(type(self.sip.cmd("WN", "DUMMY", "D10", "+00:59", "2020-01-01", "00:00")), str) + + def test_logout(self): + self.assertEqual(self.sip.logout(), None) + + def test_lmwTimeSerie(self): + timezone = tz.gettz('GMT+1') + res = self.sip.timeSerie("WN", "DUMMY", "D10", + datetime.now(timezone)-timedelta(minutes=60), + datetime.now(timezone)) + self.assertEqual(type(res.ts), list) + self.assertEqual(len(res.ts), 6) + self.assertEqual(res.ts[1][1][0], '1') + + def test_roundtime(self): + timezone = tz.gettz('GMT+1') + t1 = datetime(2020, 1, 1, 0, 10, 0, 0, timezone) + t2 = datetime(2020, 1, 1, 0, 0, 0, 1, timezone) + self.assertEqual(self.sip._roundtime_(t1, "D10"), t1) + self.assertEqual(self.sip._roundtime_(t2, "D10"), t1) + +if __name__ == '__main__': + unittest.main() diff --git a/test/stubSipServer.py b/test/stubSipServer.py new file mode 100755 index 0000000..1760f94 --- /dev/null +++ b/test/stubSipServer.py @@ -0,0 +1,116 @@ +#!/usr/bin/python + +"""A stub sipserver for testing lmwsip + +This is a stub sipserver that implements a small subset of the sip +protocol to perform unit tests. + +Implements the following commands: + + CMD> LI USER,PASS + ANS< ! + + CMD> TI LMW + ANS< ! 20-JAN-01 00:00:00 + + CMD> WN LMW,DUMMY,D10,+HH:MM,yyyy-mm-dd,HH:MM,DATA + ANS< ! 1/10,;2/10;.... + + CMD> WN LMW,DUMMY,D10,+HH:MM,yyyy-mm-dd,HH:MM,DATB + ANS< ! 1/10/0;2/10/0;.... + + CMD> LO + ANS< ! + +All other commands result in a "?" + + CMD> * + ANS< ? ERROR + +Note: + for a WN command the time and date are ignored. + The duration is used to calculare the number of results to send. + + The sip syntax for time is much flexibler. + The stub only support this format! +""" + + +import os +import time +import random +import socketserver + +class sipProtocol(socketserver.BaseRequestHandler): + + def match(self, m): + return(self.data.find(m.encode()) == 0) + + def send(self, a): + a = "%s\r" % a + self.request.sendall(a.encode()) + + def read(self): + try: + self.data = self.request.recv(1024).strip() + except: + self.data = None + + def meting(self): + res = "" + sep = "! " + if self.data[18:19] == b'0': + h = int(self.data[19:20].decode()) + else: + h = int(self.data[18:20].decode()) + if self.data[21:22] == b'0': + m = int(self.data[22:23].decode()) + else: + m = int(self.data[21:23].decode()) + aantal = 1+(60*h+m)//10 + if self.data[-1:] == b'A': + data = "%i/10" + else: + data = "%i/10/0" + for i in range(aantal): + res += sep+data % i + sep=";" + self.send(res) + + def handle(self): + self.read() + while self.data: + if self.match("LI USER,PASS"): + self.send("!") + elif self.match("TI LMW"): + self.send("! 20-JAN-01 00:00:00") + elif self.match("WN LMW,DUMMY,D10"): + self.meting() + elif self.match("LO"): + self.send("!") + else: + self.send("? ERROR") + self.read() + +class sipServer(socketserver.TCPServer): + def __init__(self): + self.port = None + while self.port == None: + self.port = random.randint(20000, 50000) + try: + super(sipServer, self).__init__(("localhost", self.port), sipProtocol) + except: + self.port = None + + def run(self): + self.pid = os.fork() + if self.pid == 0: + self.serve_forever() + + def kill(self): + if self.pid != 0: + os.kill(self.pid, 15) + self.server_close() + +if __name__ == '__main__': + pass