Test webserver en locatie test
All checks were successful
continuous-integration/drone/push Build is passing

This commit is contained in:
Marcel Nijenhof
2021-06-14 17:10:32 +02:00
parent 02666b5f7a
commit 0f50cc2a46
2 changed files with 165 additions and 3 deletions

View File

@@ -2,21 +2,44 @@
import unittest
import ddOperApi
from time import sleep
from ddOperApi.tests.testServer import forkTestServer, stopTestServer
class ddOperApiTest(unittest.TestCase):
@classmethod
def setUpClass(cls):
global forkTestServerPid
forkTestServerPid = forkTestServer()
print("forkTestServer pid: %i" % forkTestServerPid)
@classmethod
def tearDownClass(cls):
global forkTestServerPid
stopTestServer(forkTestServerPid)
print("signal pid: %i" % forkTestServerPid)
def setUp(self):
self.client = ddOperApi.ddOperApi(url="https://localhost:5000/dd-oper/2.0", checkssl=False)
def tearDown(self):
pass
def test_version(self):
self.assertTrue(hasattr(ddOperApi, '__version__'))
self.assertEqual(type(ddOperApi.__version__), str)
def setup():
pass
def test_setup(self):
self.assertEqual(type(self.client), ddOperApi.ddOperApi)
def test_locations(self):
locations = self.client.locations()
self.assertEqual(type(locations), ddOperApi.ddOperLocation)
self.assertTrue(len(locations.data) > 0)
self.assertEqual(type(locations.data[0]["properties"]["locationName"]), str)
def main():
setup()
unittest.main()
if __name__ == '__main__':

139
ddOperApi/tests/testServer.py Executable file
View File

@@ -0,0 +1,139 @@
#!/usr/bin/python
#https://blog.miguelgrinberg.com/post/running-your-flask-application-over-https
import json
from os import fork, kill
from time import sleep
from random import getrandbits
from tempfile import NamedTemporaryFile
from datetime import datetime
from OpenSSL import crypto, SSL
from flask import Flask
class testServer(Flask):
def __init__(self, name):
super().__init__(__name__)
self.createCert()
self.writeCert()
def run(self):
super().run(ssl_context=(self.certFile.name, self.keyFile.name))
def createCert(self):
CN="*"
k = crypto.PKey()
k.generate_key(crypto.TYPE_RSA, 2048)
serialnumber=getrandbits(64)
cert = crypto.X509()
cert.get_subject().C = "TS"
cert.get_subject().L = "Test"
cert.get_subject().ST = "Test"
cert.get_subject().O = "Test"
cert.get_subject().OU = "Test"
cert.get_subject().CN = CN
cert.set_serial_number(serialnumber)
cert.gmtime_adj_notBefore(0)
cert.gmtime_adj_notAfter(3600)
cert.set_issuer(cert.get_subject())
cert.add_extensions([crypto.X509Extension(b'subjectAltName', False, b'DNS:*')])
cert.set_pubkey(k)
cert.sign(k, 'sha512')
self.cert=crypto.dump_certificate(crypto.FILETYPE_PEM, cert)
self.key=crypto.dump_privatekey(crypto.FILETYPE_PEM, k)
def writeCert(self):
self.certFile = NamedTemporaryFile()
self.certFile.file.write(self.cert)
self.certFile.file.flush()
self.keyFile = NamedTemporaryFile()
self.keyFile.file.write(self.key)
self.keyFile.flush()
app = testServer(__name__)
def provider(responseType):
return({
"name": "testServerDDOper",
"supportUrl": "https://git.marceln.org/marceln/ddOperApi",
"responseType": responseType,
"responseTimestamp": "%sz" % datetime.now().isoformat()
})
def geometryPoint(x, y):
return({
"type": "Point",
"coordinates": [ x, y ]
})
def jsonReturn(data):
return(json.dumps(data))
@app.route("/")
def root():
return("Hello world!\n")
@app.route("/dd-oper/2.0/locations")
def locations():
ret = { "provider": provider("LocationListResponse"),
"results": [
{
"type": "Feature",
"geometry": geometryPoint(0, 0),
"properties": {
"locationName": "test1",
"locationNameSpace": "TS.DDOPER",
"displayNameGlobal": "Test location 1"
}
},
{
"type": "Feature",
"geometry": geometryPoint(1, 0),
"properties": {
"locationName": "test2",
"locationNameSpace": "TS.DDOPER",
"displayNameGlobal": "Test location 2"
}
},
{
"type": "Feature",
"geometry": geometryPoint(0, 1),
"properties": {
"locationName": "test3",
"locationNameSpace": "TS.DDOPER",
"displayNameGlobal": "Test location 3"
}
}]}
return(jsonReturn(ret))
def forkTestServer():
pid = None
try:
pid = fork()
if (pid == 0):
app.run()
else:
sleep(0.01)
print("Server started")
except Exception as e:
print("Failed to start test server")
raise(e)
return(pid)
def stopTestServer(pid):
try:
kill(pid, 15)
sleep(0.01)
except Exception as e:
print("ERROR kill: %s" % e)
return()
def main():
app.run()
if __name__ == '__main__':
main()