Test NGW history

test_history.py

# -*- coding: utf-8 -*-

import sys
from osgeo import gdal
from osgeo import ogr
import requests
import json
import datetime
from time import sleep

gdal.SetConfigOption('CPL_CURL_VERBOSE', 'ON')
gdal.SetConfigOption('CPL_DEBUG', 'ON')

def fill_fields(f):
    f.SetField(0, "1")
    f.SetField(1, "2")
    f.SetField(2, "3")

ds_resource_id = '831'
base_url = 'http://dev.nextgis.com/practice2'

url = 'NGW:' + base_url + '/resource/' + ds_resource_id

# begin = str(datetime.datetime.utcnow().isoformat())
begin = str(datetime.datetime.now().isoformat())
for i in range(0,10):
    print('{} ...'.format(i)),
    sys.stdout.flush()
    sleep(1.15)

ngw_ds = gdal.OpenEx(url, gdal.OF_UPDATE, open_options=['USERPWD=administrator:admin'])
if ngw_ds is None:
    exit('Dataset open filed')

lyr = ngw_ds.GetLayer(0)
if lyr is None:
    exit('Get layer failed')

# end = str(datetime.datetime.utcnow().isoformat())
end = str(datetime.datetime.now().isoformat())
r = requests.get('http://dev.nextgis.com/practice2/api/resource/831/activity/?date_from={}&date_to={}'.format(begin, end), auth=('administrator', 'admin'))

json_data = r.json()
if len(json_data) > 0:
    exit('Get unexcepted changes count. Expected count is 3, got {}'.format(len(json_data)))

f = ogr.Feature(lyr.GetLayerDefn())
f.SetGeometry(ogr.CreateGeometryFromWkt('MULTILINESTRING((10 10, 20 20, 10 40),(40 40, 30 30, 40 20, 30 10))'))
fill_fields(f)
ret = lyr.CreateFeature(f)
if ret != 0 or f.GetFID() < 0:
    exit('Add feature failed')

f.SetGeometry(ogr.CreateGeometryFromWkt('MULTILINESTRING((0 0,1 1,1 2),(2 3,3 2,5 4))'))
ret = lyr.SetFeature(f)
if ret != 0:
    exit('Change feature failed')

lyr.DeleteFeature(f.GetFID())
#lyr.SyncToDisk()

for i in range(0,5):
    print('{} ...'.format(i)),
    sys.stdout.flush()
    sleep(1.15)

# end = str(datetime.datetime.utcnow().isoformat())
end = str(datetime.datetime.now().isoformat())

activity_url = '{}/api/resource/{}/activity/?date_from={}&date_to={}'.format(base_url, ds_resource_id, begin, end)
print activity_url
r = requests.get(activity_url, auth=('administrator', 'admin'))
json_data = r.json()
if len(json_data) != 3:
    exit('Get unexcepted changes count. Expected count is 3, got {}'.format(len(json_data)))

r = requests.get('{}/api/resource/{}/activity/?date_from={}&date_to={}&limit=1'.format(base_url, ds_resource_id, begin, end), auth=('administrator', 'admin'))
json_data = r.json()
if len(json_data) != 1:
    exit('Get unexcepted changes count. Expected count is 3, got {}'.format(len(json_data)))

r = requests.get('{}/api/resource/{}/activity/?date_from={}&date_to={}&limit=1&offset=1'.format(base_url, ds_resource_id, begin, end), auth=('administrator', 'admin'))
json_data = r.json()
if len(json_data) != 1:
    exit('Get unexcepted changes count. Expected count is 3, got {}'.format(len(json_data)))

print('All tests passed')