diff --git a/ddOperApi/tests/__init__.py b/ddOperApi/tests/__init__.py index c65c586..2cb33be 100644 --- a/ddOperApi/tests/__init__.py +++ b/ddOperApi/tests/__init__.py @@ -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__': diff --git a/ddOperApi/tests/testServer.py b/ddOperApi/tests/testServer.py new file mode 100755 index 0000000..164d655 --- /dev/null +++ b/ddOperApi/tests/testServer.py @@ -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()