Correctie locatie tests directory
Some checks failed
continuous-integration/drone/push Build is failing
Some checks failed
continuous-integration/drone/push Build is failing
This commit is contained in:
Marcel Nijenhof
@@ -1,85 +0,0 @@
|
||||
#!/usr/bin/python
|
||||
|
||||
import sys
|
||||
import io
|
||||
import unittest
|
||||
import lmwsip
|
||||
import stubSipServer
|
||||
|
||||
from lmwsip.run import run
|
||||
from datetime import datetime, timedelta
|
||||
from dateutil import tz
|
||||
|
||||
class myTestArgs():
|
||||
pass
|
||||
|
||||
class lmwsipTest(unittest.TestCase):
|
||||
|
||||
def setUp(self):
|
||||
self.sipserver = stubSipServer.sipServer()
|
||||
self.sip = None
|
||||
self.sipserver.run()
|
||||
|
||||
def login(self):
|
||||
self.sip = lmwsip.LmwSip("USER", "PASS", "localhost",
|
||||
self.sipserver.port, ssl=False)
|
||||
|
||||
def tearDown(self):
|
||||
if self.sip:
|
||||
self.sip.closesocket()
|
||||
self.sipserver.kill()
|
||||
|
||||
def test_sipobj(self):
|
||||
self.login()
|
||||
self.assertEqual(type(self.sip), lmwsip.LmwSip)
|
||||
|
||||
def test_ti(self):
|
||||
self.login()
|
||||
self.assertEqual(type(self.sip.ti()), str)
|
||||
|
||||
def test_cmd(self):
|
||||
self.login()
|
||||
self.assertEqual(type(self.sip.cmd("WN", "DUMMY", "D10", "+00:59", "2020-01-01", "00:00")), str)
|
||||
|
||||
def test_logout(self):
|
||||
self.login()
|
||||
self.assertEqual(self.sip.logout(), None)
|
||||
|
||||
def test_lmwTimeSerie(self):
|
||||
self.login()
|
||||
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):
|
||||
self.login()
|
||||
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)
|
||||
|
||||
def test_run(self):
|
||||
capturedOutput = io.StringIO()
|
||||
sys.stdout = capturedOutput
|
||||
args = myTestArgs()
|
||||
args.debug = "WARN"
|
||||
args.host = "localhost"
|
||||
args.port = self.sipserver.port
|
||||
args.unencrypted = True
|
||||
args.cleartelnet = False
|
||||
args.time = "+00:59"
|
||||
args.date = "2020-01-01"
|
||||
args.files = [open("tests/test.sip")]
|
||||
run(args)
|
||||
args.files[0].close()
|
||||
self.assertEqual(capturedOutput.getvalue().find("!")>= 0, True)
|
||||
self.assertEqual(capturedOutput.getvalue().find("?"), -1)
|
||||
sys.stdout = sys.__stdout__
|
||||
|
||||
if __name__ == '__main__':
|
||||
unittest.main()
|
||||
@@ -1,116 +0,0 @@
|
||||
#!/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
|
||||
@@ -1,3 +0,0 @@
|
||||
LI USER,PASS
|
||||
TI LMW
|
||||
LO
|
||||
Reference in New Issue
Block a user