Eerste test op values

ddOperApi/tests/__init__.py

@@ -36,13 +36,16 @@ class ddOperApiTest(unittest.TestCase):
         self.assertEqual(type(locations), ddOperApi.ddOperLocation)
         self.assertTrue(len(locations.data) > 0)
         self.assertEqual(type(locations.data[0]["properties"]["locationName"]), str)
-
     def test_quantities(self):
         quantities = self.client.quantities()
         self.assertEqual(type(quantities), ddOperApi.ddOperQuantitie)
         self.assertTrue(len(quantities.data) > 0)
         self.assertEqual(type(quantities.data[0]), str)
 
+    def test_values(self):
+        values = self.client.values("test1", "null")
+        self.assertEqual(type(values), ddOperApi.ddOperValues)
+
 def main():
     unittest.main()
 

ddOperApi/tests/testServer.py

@@ -6,9 +6,11 @@ from os import fork, kill
 from time import sleep
 from random import getrandbits
 from tempfile import NamedTemporaryFile
-from datetime import datetime
+from datetime import datetime, timedelta
+from dateutil.parser import parse
+from dateutil.tz import gettz
 from OpenSSL import crypto, SSL
-from flask import Flask
+from flask import Flask, request
 
 #
 # TODO: createCert verplaatsen zodat die ook beschikbaar is in de module.
@@ -61,12 +63,15 @@ class testServer(Flask):
 
 app = testServer(__name__)
 
+def timeStr(dt):
+    return("%sz" % dt.isoformat())
+
 def provider(responseType):
     return({
         "name": "testServerDDOper",
         "supportUrl": "https://git.marceln.org/marceln/ddOperApi",
         "responseType": responseType,
-        "responseTimestamp": "%sz" % datetime.now().isoformat()
+        "responseTimestamp": timeStr(datetime.now())
     })
 
 def geometryPoint(x, y):
@@ -75,6 +80,37 @@ def geometryPoint(x, y):
         "coordinates": [ x, y ]
     })
 
+def roundTime(startTime, intervalSec):
+    t = startTime.timestamp()
+    m = intervalSec
+    if (t%m > 0):
+        s = int(t/m+1)*m
+        return(datetime.fromtimestamp(s, tz=gettz("UTC")))
+    else:
+        return(startTime)
+
+def singleEvent(eventTime, intervalSec, location, quantity, aspectSet):
+    return({
+        "timeStamp": timeStr(eventTime),
+        "startTime": timeStr(eventTime-timedelta(seconds=intervalSec/2)),
+        "endTime": timeStr(eventTime+timedelta(seconds=intervalSec/2)),
+        "quality": 10,
+        "value": 0,
+        "additionalInfo": 0
+    })
+            
+def events(location, quantity, aspectSet, intervalLength, startTime, endTime):
+    ret = []
+    if intervalLength == "10min":
+        intervalSec=600
+    elif intervalLength == "1min":
+        intervalSec=60
+    eventTime = roundTime(startTime, intervalSec)
+    while (eventTime < endTime):
+        ret.append(singleEvent(eventTime, intervalSec, location, quantity, aspectSet))
+        eventTime = eventTime + timedelta(seconds=intervalSec)
+    return(ret)
+
 def jsonReturn(data):
     return(json.dumps(data))
 
@@ -129,6 +165,66 @@ def quantities():
     }
     return(jsonReturn(ret))
 
+@app.route("/dd-oper/2.0/locations/<location>/quantities")
+def locationsQuantities(location):
+    return("TODO: locations: %s\n" % location, 404)
+
+@app.route("/dd-oper/2.0/quantities/<quantity>")
+def quantitiesLocations(quantity):
+    return("TODO: quantities: %s\n" % quantity, 404)
+
+@app.route("/dd-oper/2.0/locations/<location>/quantities/<quantity>/timeseries")
+def values(location, quantity):
+    startTime       = request.args.get('startTime', type = str)
+    endTime         = request.args.get('endTime', type = str)
+    process         = request.args.get('process', default='measurement', type = str)
+    intervalLength  = request.args.get('intervalLength', default='10min', type = str)
+    aspectSet       = request.args.get('aspectSet', default="minimum", type = str)
+    try:
+        startTime = parse(startTime)
+        endTime   = parse(endTime)
+    except Exception as e:
+        return("parseerror time", 404)
+    ret = { "provider": provider("Timeseries"),
+            "results": [{
+                "source": {
+                    "process": process,
+                    "institution": {
+                        "name": "testServerDDOper"
+                    },
+                    "location": {
+                        "geometry": {
+                            "type": "Point",
+                            "coordinates": [ 0, 0]
+                        }
+                    },
+                    "type": "Feature",
+                    "properties": {
+                        "locationName": location,
+                        "locationNameSpace": "TS.DDOPER",
+                        "crsName": "WGS84",
+                        "displayNameGlobal": "test 1"
+                    }
+                },
+                "startTime": timeStr(startTime),
+                "endTime":   timeStr(endTime),
+                "events":    events(location, quantity, aspectSet, intervalLength, startTime, endTime),
+                "observationType": {
+                    "quantityName": quantity,
+                    "aspectSet": {
+                        "name": "minimum",
+                        "aspects": [
+                            {
+                                "name": "Average",
+                                "unit": ""
+                            }
+                        ]
+                    }
+                }
+            }]
+          }
+    return(jsonReturn(ret))
+
 def forkTestServer():
     pid = None
     try:
@@ -137,9 +233,7 @@ def forkTestServer():
             app.run()
         else:
             sleep(0.01)
-            print("Server started")
     except Exception as e:
-        print("Failed to start test server")
         raise(e)
     return(pid)