30 Commits

Author SHA1 Message Date
cdeeceeb55 Werkende upload
All checks were successful
build / build (push) Successful in 1m33s
2024-04-07 22:59:11 +02:00
64089ce72c Versie 0.9.7 2024-04-05 16:35:37 +02:00
7683ec29e9 Cleanup of module documentation
Some checks failed
build / build (push) Failing after 1m29s
2024-04-05 16:27:54 +02:00
285b3514a8 Added cmdWrite 2024-04-05 16:19:55 +02:00
f32c007b3f lmwParameter comment in class comment geplaats
All checks were successful
build / build (push) Successful in 1m45s
2024-04-04 21:52:34 +02:00
014dc507cc Commentaar toegevoegd voor lmwParamers 2024-04-04 21:39:47 +02:00
cec151e564 Overgang gitea action
All checks were successful
build / build (push) Successful in 1m44s
2024-02-18 22:08:14 +01:00
26f1caa60e Aanpassing pre-commit 2024-02-18 22:05:47 +01:00
7092e69e6a Nieuwe python versies toegevoegd in tox test 2024-01-21 20:42:51 +01:00
b32bce956f Extra parameters aan lijst toegevoegd
All checks were successful
continuous-integration/drone/push Build is passing
2022-11-02 10:47:47 +01:00
22bb4e6bd3 Updates .drone
Some checks failed
continuous-integration/drone/push Build is failing
2022-06-30 10:52:31 +02:00
49954d762b Updates .drone
Some checks failed
continuous-integration/drone/push Build is failing
2022-06-30 10:24:31 +02:00
30504239e1 Versie verhogen na cleartelnet fix
Some checks failed
continuous-integration/drone/push Build is failing
2022-06-29 18:20:27 +02:00
Marcel Nijenhof
42799296a9 Bug fix: Cleartelnet code werkte niet omdat deze in de if stond voor 0 bytes ontvangen.
Some checks failed
continuous-integration/drone/push Build is failing
2022-06-29 14:23:40 +00:00
Marcel Nijenhof
ac54b7cd0d Upload naar pypi org prod
Some checks failed
continuous-integration/drone/push Build is failing
continuous-integration/drone Build is passing
2022-01-03 18:56:28 +01:00
Marcel Nijenhof
1df2a42196 Correctie .drone
Some checks reported errors
continuous-integration/drone/push Build is passing
continuous-integration/drone Build was killed
2022-01-03 18:42:44 +01:00
Marcel Nijenhof
2909ea7ed4 PYTHONPATH=src zodat lmwsip gevonden word
Some checks failed
continuous-integration/drone/push Build is failing
2022-01-03 18:39:16 +01:00
Marcel Nijenhof
f55d6c2e19 PYTHONPATH=. zodat llmwsip gevonden word
Some checks failed
continuous-integration/drone/push Build is failing
2022-01-03 18:38:08 +01:00
Marcel Nijenhof
664b215f16 Verplaatsen test directory
Some checks failed
continuous-integration/drone/push Build is failing
2022-01-03 18:32:08 +01:00
Marcel Nijenhof
5fe756ce97 Small corrections setup.cfg 2022-01-03 18:14:40 +01:00
Marcel Nijenhof
bf28862f1f Update README: install via pypi 2022-01-03 18:03:03 +01:00
Marcel Nijenhof
fe29441d1d Versie update
Some checks failed
continuous-integration/drone/push Build is failing
2022-01-02 23:50:13 +01:00
Marcel Nijenhof
d0043f3f18 Upload pypi test
Some checks failed
continuous-integration/drone/push Build is failing
2022-01-02 23:45:44 +01:00
Marcel Nijenhof
583997d64c Versie update
All checks were successful
continuous-integration/drone/push Build is passing
2021-12-31 23:18:45 +01:00
Marcel Nijenhof
04daf49b0a Update drone voor verplaatsing files
Some checks failed
continuous-integration/drone/push Build is failing
2021-12-31 23:16:26 +01:00
Marcel Nijenhof
b19f5d9049 Merge branch 'master' of https://git.marceln.org/Werk/lmwsip
Some checks failed
continuous-integration/drone/push Build is failing
2021-12-31 22:56:23 +01:00
Marcel Nijenhof
2052bfd2cc Verplaatse files. 2021-12-31 22:56:01 +01:00
Marcel Nijenhof
418d3e6041 Correctie pre-commit hoek 2021-12-31 22:54:49 +01:00
Marcel Nijenhof
08cd160713 Lokale test omgeschreven naar tox
Some checks failed
continuous-integration/drone/push Build is failing
2021-12-31 22:42:04 +01:00
Marcel Nijenhof
da80bd93d9 Files verwijderd uit oorspronkelijke dir. 2021-12-31 22:25:43 +01:00
13 changed files with 100 additions and 555 deletions

View File

@@ -1,166 +0,0 @@
---
kind: pipeline
type: exec
name: CentOS7_test
platform:
os: linux
arch: amd64
variant: CentOS7
steps:
- name: Run unit test
environment:
PYTHONPATH: .
commands:
- python3 setup.py test
---
kind: pipeline
type: exec
name: CentOS8_test
platform:
os: linux
arch: amd64
variant: CentOS8
steps:
- name: Run unit test
environment:
PYTHONPATH: .
commands:
- python3 setup.py test
---
kind: pipeline
type: exec
name: Ubuntu1804_test
platform:
os: linux
arch: amd64
variant: Ubuntu1804
steps:
- name: Run unit test
environment:
PYTHONPATH: .
commands:
- python3 setup.py test
---
kind: pipeline
type: exec
name: Fedora_test
platform:
os: linux
arch: amd64
variant: Fedora
steps:
- name: Run unit test
environment:
PYTHONPATH: .
commands:
- python3 setup.py test
---
kind: pipeline
type: docker
name: docker_python:3.6
steps:
- name: Run unit test
image: python:3.6
environment:
PYTHONPATH: .
commands:
- pip install python-dateutil
- python setup.py test
---
kind: pipeline
type: docker
name: docker_python:latest
steps:
- name: Run unit test
image: python:latest
environment:
PYTHONPATH: .
commands:
- pip install python-dateutil
- python --version
- python setup.py test
---
kind: pipeline
type: exec
name: Build
platform:
os: linux
arch: amd64
variant: Fedora
steps:
- name: Build package files
commands:
- python3 setup.py sdist bdist_wheel
- name: Versie toevoegen aan download
environment:
DOWNLOADDIR: /usr/share/nginx/html/download/python/lmwsip
commands:
- mkdir -p "$${DOWNLOADDIR}"
- cd dist
- |
for f in *
do
if [ -f "$${DOWNLOADDIR}/$${f}" ]
then
echo version error
exit 1
else
cp "$${f}" "$${DOWNLOADDIR}";
fi
done
- ls -l "$${DOWNLOADDIR}"
when:
branch:
- master
depends_on:
- CentOS7_test
- CentOS8_test
- Ubuntu1804_test
- Fedora_test
- docker_python:latest
- docker_python:3.6
---
kind: pipeline
type: docker
name: installCheck
steps:
- name: install Check
image: python:3.6
# Make sure we run the pip installed version
commands:
- rm -rf lmwsip
- pip install --extra-index-url https://marceln.org/download/python lmwsip
- python -c "import lmwsip"
- python -c "import lmwsip; print(lmwsip.__version__)"
- >
[ $(python -c "import lmwsip; print(lmwsip.__version__)") =
$(python setup.py --version) ]
- python -m unittest -v lmwsip.tests
when:
branch:
- master
depends_on:
- Build

View File

@@ -0,0 +1,21 @@
---
name: build
on:
- push
jobs:
build:
runs-on: fedora-builder
steps:
- name: Check out repository code
uses: actions/checkout@v3
- name: Build package
run: python3 -m build
- name: run tox
run: tox
- name: Upload na pypi
env:
TWINE_USERNAME: ${{ secrets.PYPI_USER }}
TWINE_PASSWORD: ${{ secrets.PYPI_PASSWORD }}
if: github.ref_type == 'tag'
run: python3 -m twine upload dist/*

1
.gitignore vendored
View File

@@ -7,3 +7,4 @@ lmwsip_marceln.egg-info
test/__pycache__ test/__pycache__
lmwsip.egg-info lmwsip.egg-info
*.swp *.swp
.tox

View File

@@ -12,12 +12,10 @@ how to use it.
## Installing ## Installing
The package is not available on [PyPI](https://pypi.org/). Just install the package with 'pip':
At the moment the package is hosted at https://marceln.org/download/python.
You can install the package with pip:
``` ```
pip install --extra-index-url https://marceln.org/download/python lmwsip pip install lmwsip
``` ```
## Examples ## Examples

View File

@@ -2,9 +2,10 @@
set -e set -e
yamllint .gitea/workflows/*yml
VERSION=$(grep version setup.cfg | sed 's/.*= *//') VERSION=$(grep version setup.cfg | sed 's/.*= *//')
sed -i "s/^__version__ = .*/__version__ = '${VERSION}'/" src/lmwsip/__init__.py sed -i "s/^__version__ = .*/__version__ = '${VERSION}'/" src/lmwsip/__init__.py
git add lmwsip/__init__.py git add src/lmwsip/__init__.py
python setup.py test tox
yamllint .drone.yml

View File

@@ -1,64 +0,0 @@
#!/usr/bin/env python3
import sys
import getopt
import argparse
import logging
from lmwsip import LmwSip
def run(args):
logging.basicConfig(level=args.debug)
logging.debug("lmwsip.run %s" % args)
try:
lmwsip = LmwSip(host=args.host, port=args.port,
ssl=not args.unencrypted,
check_ssl=not args.acceptssl,
cleartelnet=args.cleartelnet)
except Exception as e:
print("Connect to lmw failed: %s" % e)
exit(1)
for f in args.files:
for cmd in f:
cmd = cmd.replace('{DATE}', args.date)
cmd = cmd.replace('{TIME}', args.time)
cmd = cmd.replace('\n', '\r')
print("> [%s]" % (cmd.strip('\r')))
try:
lmwsip.send(cmd)
print("< [%s]" % (lmwsip.recv().strip('\r')))
except:
pass
try:
lmwsip.closesocket()
except:
pass
def main():
lastTime=LmwSip(host=None).lasttime("H10")
parser = argparse.ArgumentParser(description="Run a sip file.")
parser.add_argument("-u", "--unencrypted", action="store_true",
help="Run a sip connection without ssl")
parser.add_argument("-a", "--acceptssl", action="store_true",
help="Accept ssl certificate")
parser.add_argument("-c", "--cleartelnet", action="store_true",
help="Clear telnet protocol in tcp session")
parser.add_argument("-H", "--host", action='store',
default="sip-lmw.rws.nl",
help="Host to connect to")
parser.add_argument("-p", "--port", action='store', type=int, default=443,
help="Port to connect to")
parser.add_argument("-d", "--date", action='store',
default=lastTime["day"],
help="Date replacement string [DD-MM-YYYY]")
parser.add_argument("-t", "--time", action='store',
default=lastTime["time_of_day"],
help="Time replacement string [HH:MM]")
parser.add_argument("-D", "--debug", action='store',
default="WARN",
help="Debug level")
parser.add_argument("files", type=argparse.FileType('r'), nargs="+",
help="Sip files to run")
run(parser.parse_args())
if __name__ == "__main__":
main()

View File

@@ -1,20 +1,23 @@
[metadata] [metadata]
name = lmwsip name = lmwsip
version = 0.9.1 version = 0.9.7
author = Marcel Nijenhof author = Marcel Nijenhof
author_email = pypi@marceln.org author_email = pypi@marceln.org
description = Interface for the lmw sip protocol description = Interface for the lmw sip protocol
long_description = file: README.md long_description = file: README.md
long_description_content_type = text/markdown long_description_content_type = text/markdown
url = https://marceln.org/git/Werk/lmwsip url = https://git.marceln.org/Werk/lmwsip
classifiers = classifiers =
Programming Language :: Python :: 3 Programming Language :: Python :: 3
License :: OSI Approved :: MIT License License :: OSI Approved :: MIT License
Operating System :: OS Independent Operating System :: OS Independent
Development Status :: 4 - Beta
[options] [options]
package_dir = package_dir =
= src = src
install_requires =
python-dateutil
packages = find: packages = find:
python_requires = >= 3.6 python_requires = >= 3.6

View File

@@ -12,7 +12,7 @@ from datetime import datetime, timedelta
from dateutil import tz from dateutil import tz
""" Version info changed by git hook """ """ Version info changed by git hook """
__version__ = '0.9.1' __version__ = '0.9.7'
class LmwSip: class LmwSip:
"""Class to connect to the LMW Standard Interface prototcol (sip) """Class to connect to the LMW Standard Interface prototcol (sip)
@@ -27,7 +27,12 @@ Support for:
ti ti
cmd(wn, vw, as) cmd(wn, vw, as)
lmwParameters:
- Type: WN, VW, AS
- Array size: [1-201]
- Periode: 1, 10
""" """
lmwParameters = { lmwParameters = {
'Tm02_MV': ('VW', 1, 0), 'Tm02_MV': ('VW', 1, 0),
'xH1': ('WN', 1, 1), 'xH1': ('WN', 1, 1),
@@ -489,7 +494,12 @@ Support for:
'xH60': ('WN', 1, 60), 'xH60': ('WN', 1, 60),
'xNI60': ('WN', 1, 60), 'xNI60': ('WN', 1, 60),
'xQ60': ('WN', 1, 60), 'xQ60': ('WN', 1, 60),
'xH60R': ('WN', 1, 60) 'xH60R': ('WN', 1, 60),
'QfQStt10': ('WN', 1, 10),
'QfQTr10': ('WN', 1, 10),
'QfQSt10': ('WN', 1, 10),
'QfHYS10': ('WN', 1, 10),
'QfQSy10': ('WN', 1, 10)
} }
def __init__(self, user=None, password=None, def __init__(self, user=None, password=None,
@@ -588,7 +598,7 @@ Opens the connection and logs in.
"time_of_day": time.strftime("%H:%M", time.gmtime(now)) } "time_of_day": time.strftime("%H:%M", time.gmtime(now)) }
def connect(self): def connect(self):
"""connect() """Setup the network connection
connects to lmw with tcp using the values of the object creation. connects to lmw with tcp using the values of the object creation.
""" """
@@ -624,7 +634,7 @@ connects to lmw with tcp using the values of the object creation.
self._socket = None self._socket = None
def send(self, sipcmd): def send(self, sipcmd):
"""send(sipcmd) """Send a sip command to the server
send a sip command to the server send a sip command to the server
""" """
@@ -654,7 +664,7 @@ send a sip command to the server
raise LmwSipConnectError("LmwSip.telnetheader: Socket connection lost") raise LmwSipConnectError("LmwSip.telnetheader: Socket connection lost")
def recv(self): def recv(self):
"""recv() """Recieve the results
recieve a answer from the sip server recieve a answer from the sip server
""" """
@@ -668,9 +678,9 @@ recieve a answer from the sip server
if (len(bytebuf) == 0): if (len(bytebuf) == 0):
c+=1 c+=1
self.log.debug("recv: bytebuf: %s" % bytebuf) self.log.debug("recv: bytebuf: %s" % bytebuf)
if self.cleartelnet: if self.cleartelnet:
if bytebuf[0] == 255: if bytebuf[0] == 255:
bytebuf = b'' bytebuf = b''
except Exception as e: except Exception as e:
self.log.error("SipLmw.recv: socket timeout: %s", e) self.log.error("SipLmw.recv: socket timeout: %s", e)
self.closesocket() self.closesocket()
@@ -697,7 +707,7 @@ recieve a answer from the sip server
return(stringbuf) return(stringbuf)
def login(self): def login(self):
"""login() """Login the sip server
Login lmw using the object creation user, password. Login lmw using the object creation user, password.
Raises a LmwLoginFailure exception on failure Raises a LmwLoginFailure exception on failure
@@ -721,7 +731,7 @@ Raises a LmwLoginFailure exception on failure
self.login() self.login()
def reconnectcheck(self): def reconnectcheck(self):
"""reconnectcheck() """Check if we need to reconnect.
Checks if a reconnect is nessecery. Checks if a reconnect is nessecery.
@@ -739,7 +749,7 @@ There are two timeouts:
self.reconnect() self.reconnect()
def sendrecv(self, cmd): def sendrecv(self, cmd):
"""sendrecv(cmd) """Send a a command and recieve the result.
send the command and recieve the answer. send the command and recieve the answer.
retry on socket failure. retry on socket failure.
@@ -768,7 +778,7 @@ retry on socket failure.
return(ret) return(ret)
def ti(self): def ti(self):
"""ti() """Recieve the time from the sipserver.
Request the time from lmw and returns the string. Request the time from lmw and returns the string.
@@ -780,7 +790,7 @@ Raises a LmwCmdWarn of failure
def cmd(self, process, location, parameter, time_delta, day, def cmd(self, process, location, parameter, time_delta, day,
time_of_day, cmd_type="DATA"): time_of_day, cmd_type="DATA"):
"""cmd(process, location, parameter, time_delta, day, time_of_day) """Create a sip command from the paramters
Send a cmd to LMW and returns the lmw string Send a cmd to LMW and returns the lmw string
@@ -812,9 +822,35 @@ Returns:
raise LmwCmdWarn(cmdstr, d) raise LmwCmdWarn(cmdstr, d)
return (d[2:-1]) return (d[2:-1])
def cmdWrite(self, process, location, parameter, time_delta, day,
time_of_day, values):
"""Write data to LMW
process: <WNT|VWT|AST>
location: <lmw location (e.g. HOEK)>
parameter: <lmw parameter (e.g. H10)>
time_delta: <Time windows (max 23:59, e.g. +01:00>
day: <Date>
time_of_day: <Time>
data: Values to be writen (e.g. 33/10;35/10).
Example:
lmw.cmd("WNT", "HOEK", "H10", "+00:20", "13-08-2018", "16:00", "33/10;35/10")
Returns:
The LMW answer string
"""
cmdstr=process + " " + self.meetnet + "," + location + "," + \
parameter + "," + time_delta + "," + day + "," + \
time_of_day + values + "\r"
d = self.sendrecv(cmdstr)
if (d[0] != '!'):
raise LmwCmdWarn(cmdstr, d)
return (d[2:-1])
def valueStr(self, process, location, parameter, day = None, def valueStr(self, process, location, parameter, day = None,
time_of_day = None): time_of_day = None):
"""value(process, location, parameter, [day], [time_of_day]): """Get string of values from sip
Parameters: Parameters:
process: <WN|VW|AS> process: <WN|VW|AS>
@@ -841,7 +877,7 @@ Returns a single string value with quality
def value(self, process, location, parameter, day = None, def value(self, process, location, parameter, day = None,
time_of_day = None): time_of_day = None):
"""value(process, location, parameter, [day], [time_of_day]): """Get one value from sip
Parameters: Parameters:
process: <WN|VW|AS> process: <WN|VW|AS>
@@ -885,7 +921,7 @@ Returns a single string value or None
def timeSerie(self, process, location, parameter, def timeSerie(self, process, location, parameter,
startTime, endTime, cmd_type="DATB"): startTime, endTime, cmd_type="DATB"):
"""timeSerie(process, location, parameter, startTime, endTime, cmd_type="DATA") """Get a python data structure with the results.
Parameters: Parameters:
process: <WN|VW|AS> process: <WN|VW|AS>
@@ -954,7 +990,7 @@ Errors:
return(res) return(res)
def logout(self): def logout(self):
"""logout() """Logut of the sip server,
Logs of Logs of
""" """
@@ -978,7 +1014,7 @@ Note:
""" """
def __init__(self, start, delta, values=""): def __init__(self, start, delta, values=""):
"""lmwTimeSerie(start, delta, values) """lmwTimeSerie init
Create a lmwTimeSerie object with: Create a lmwTimeSerie object with:
start: Start time start: Start time
@@ -991,7 +1027,7 @@ Create a lmwTimeSerie object with:
self.addvalues(start, values) self.addvalues(start, values)
def addvalues(self, start, values): def addvalues(self, start, values):
"""addvalues(start, delta, values) """Add values
Add values to a timeserie Add values to a timeserie
start: Start time start: Start time

View File

@@ -1,164 +0,0 @@
#!/usr/bin/python
import sys
import io
import unittest
import lmwsip
import lmwsip.tests.stubSipServer
import logging
from lmwsip.tests.stubSipServer import sipServer
from lmwsip.run import run
from datetime import datetime, timedelta
from dateutil import tz
from time import sleep
class myTestArgs():
pass
class lmwsipTest(unittest.TestCase):
def setUp(self):
self.sipserver = sipServer()
self.sip = None
self.sipserver.run()
def login(self, **args):
log = logging.basicConfig(level=logging.DEBUG)
self.sip = lmwsip.LmwSip("USER", "PASS", "localhost",
self.sipserver.port, ssl=False,
log=log, **args)
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_H1(self):
self.sip = lmwsip.LmwSip(host=None)
self.assertEqual(self.sip.period('H1'), 1)
def test_H10(self):
self.sip = lmwsip.LmwSip(host=None)
self.assertEqual(self.sip.period('H10'), 10)
def test_xHm0(self):
self.sip = lmwsip.LmwSip(host=None)
self.assertEqual(self.sip.period('xHm0'), 10)
def test_Noparm(self):
self.sip = lmwsip.LmwSip(host=None)
with self.assertRaises(lmwsip.LmwParmWarn):
self.assertEqual(self.sip.period('Noparm'), None)
def test_loginfail(self):
with self.assertRaises(lmwsip.LmwLoginFailure):
self.sip = lmwsip.LmwSip("FAIL", "FAIL", "localhost",
self.sipserver.port, ssl=False)
def test_ti(self):
self.login()
self.assertEqual(type(self.sip.ti()), str)
def test_telnetti(self):
self.login(cleartelnet=True)
self.assertEqual(type(self.sip.ti()), str)
def test_cmd(self):
self.login()
self.assertEqual(type(self.sip.cmd("WN", "DUMMY", "H10", "+00:59", "2020-01-01", "00:00")), str)
def test_cmderr(self):
self.login()
with self.assertRaises(lmwsip.LmwCmdWarn):
self.assertEqual(type(self.sip.cmd("NOP", "DUMMY", "H10", "+00:59", "2020-01-01", "00:00")), str)
def test_value(self):
self.login()
self.assertEqual(type(self.sip.value("WN", "DUMMY", "H10")), str)
def test_value1min(self):
self.login()
self.assertEqual(type(self.sip.value("WN", "DUMMY", "H1")), str)
def test_valueStr(self):
self.login()
self.assertEqual(type(self.sip.valueStr("WN", "DUMMY", "H10")), 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", "H10",
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, timedelta(minutes=10)), t1)
self.assertEqual(self.sip._roundtime_(t2, timedelta(minutes=10)), t1)
def test_closerecv(self):
self.login()
self.sip.send("CLOSE")
with self.assertRaises(lmwsip.LmwSipConnectError):
self.sip.recv()
def test_closeti(self):
self.login()
self.sip.send("CLOSE")
self.assertEqual(type(self.sip.ti()), str)
def test_closecmd(self):
self.login()
self.sip.send("CLOSE")
self.assertEqual(type(self.sip.cmd("WN", "DUMMY", "H10", "+00:59", "2020-01-01", "00:00")), str)
def test_reconnect(self):
self.login(reconnecttime=1)
sleep(2)
self.assertEqual(self.sip.sendrecv("LOGOUTCOUNT"), "1\r")
def test_idlereconnect(self):
self.login(idlereconnect=1)
sleep(2)
self.assertEqual(self.sip.sendrecv("LOGOUTCOUNT"), "1\r")
def test_versionstr(self):
self.assertEqual(type(lmwsip.__version__), str)
def test_run(self):
capturedOutput = io.StringIO()
sys.stdout = capturedOutput
testSipFile = io.StringIO("LI USER,PASS\rTI LMW\rLO")
testSipFile.seek(0)
args = myTestArgs()
args.debug = "DEBUG"
args.host = "localhost"
args.port = self.sipserver.port
args.unencrypted = True
args.acceptssl = True
args.cleartelnet = False
args.time = "+00:59"
args.date = "2020-01-01"
args.files = [testSipFile]
run(args)
args.files[0].close()
self.assertEqual(capturedOutput.getvalue().find("!")>= 0, True)
self.assertEqual(capturedOutput.getvalue().find("?"), -1)
if __name__ == '__main__':
unittest.main()

View File

@@ -1,131 +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,H10,+HH:MM,yyyy-mm-dd,HH:MM,DATA
ANS< ! 1/10,;2/10;....
CMD> WN LMW,DUMMY,H10,+HH:MM,yyyy-mm-dd,HH:MM,DATB
ANS< ! 1/10/0;2/10/0;....
CMD> WN LMW,DUMMY,H1,+HH:MM,yyyy-mm-dd,HH:MM,DATA
ANS< ! 1/10,;2/10;....
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
logoutcount=0
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 number(self, b):
if b[0] == b'0':
return(int(b[0:1]))
else:
return(int(b[0:2]))
def meting(self, delta=10):
res = ""
sep = "! "
elem = self.data.decode().split(",")
h = self.number(elem[3][1:3])
m = self.number(elem[3][4:6])
aantal = 1+(60*h+m)//delta
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):
global logoutcount
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,H10,"):
self.meting(10)
elif self.match("WN LMW,DUMMY,H1,"):
self.meting(1)
elif self.match("LOGOUTCOUNT"):
self.send(str(logoutcount))
elif self.match("LO"):
logoutcount+=1
self.send("!")
elif self.match("CLOSE"):
self.request.close()
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__':
s = sipServer()
s.run()
pass

View File

@@ -4,10 +4,10 @@ import sys
import io import io
import unittest import unittest
import lmwsip import lmwsip
import lmwsip.tests.stubSipServer import stubSipServer
import logging import logging
from lmwsip.tests.stubSipServer import sipServer from stubSipServer import sipServer
from lmwsip.run import run from lmwsip.run import run
from datetime import datetime, timedelta from datetime import datetime, timedelta
from dateutil import tz from dateutil import tz
@@ -77,6 +77,10 @@ class lmwsipTest(unittest.TestCase):
with self.assertRaises(lmwsip.LmwCmdWarn): with self.assertRaises(lmwsip.LmwCmdWarn):
self.assertEqual(type(self.sip.cmd("NOP", "DUMMY", "H10", "+00:59", "2020-01-01", "00:00")), str) self.assertEqual(type(self.sip.cmd("NOP", "DUMMY", "H10", "+00:59", "2020-01-01", "00:00")), str)
def test_cmdWrite(self):
self.login()
self.assertEqual(type(self.sip.cmdWrite("WN", "DUMMY", "H10", "+00:20", "2020-01-01", "00:00", "35/10;33/10")), str)
def test_value(self): def test_value(self):
self.login() self.login()
self.assertEqual(type(self.sip.value("WN", "DUMMY", "H10")), str) self.assertEqual(type(self.sip.value("WN", "DUMMY", "H10")), str)

6
tox.ini Normal file
View File

@@ -0,0 +1,6 @@
[tox]
envlist = py36,py37,py38,py310,py311,py312,py313
[testenv]
deps = python-dateutil
commands = python tests/main.py