- 01
- 02
- 03
- 04
- 05
- 06
- 07
- 08
- 09
- 10
- 11
- 12
- 13
- 14
- 15
- 16
- 17
- 18
- 19
- 20
- 21
- 22
- 23
- 24
- 25
- 26
- 27
- 28
- 29
- 30
- 31
- 32
- 33
- 34
- 35
- 36
- 37
- 38
- 39
- 40
- 41
- 42
- 43
- 44
- 45
- 46
- 47
- 48
- 49
- 50
- 51
- 52
- 53
- 54
- 55
- 56
- 57
- 58
- 59
- 60
- 61
- 62
- 63
- 64
- 65
- 66
- 67
- 68
- 69
- 70
- 71
- 72
- 73
- 74
- 75
- 76
- 77
- 78
- 79
- 80
- 81
- 82
- 83
- 84
- 85
- 86
- 87
- 88
- 89
- 90
- 91
- 92
- 93
- 94
- 95
- 96
- 97
- 98
- 99
from datetime import datetime, timedelta
from dateutil import parser
import os
import pytest
from tests.db_support import psg_db
intake_iot_mapper = [('sourceId', 'DEVICE_ID', str),
('altitude', 'ALTITUDE', int),
('odometer', 'ODOMETER', int),
('battery', 'BATTERY_LEVEL', int),
('speed', 'SPEED', int),
('satCount', 'SAT_COUNT', float),
('gpsQuality', 'GPSQUALITY', float),
('lat', 'LAT', float),
('lon', 'LON', float),
('radius', 'RADIUS', int),
('objectId', 'OBJECT_ID', str),
('direction', 'DIRECTION', int)]
@pytest.fixture(scope='module')
def device_ids():
sql_device = """SELECT
dvc.id,
dvc.source_id device_id,
dvc_m.object_id
FROM iot_platform.iot_device_mecomo dvc_m
JOIN iot_platform.iot_device dvc on dvc.id = dvc_m.id
WHERE dvc_m.object_id is not NULL ORDER BY dvc_m.object_id"""
ids = psg_db(sql=sql_device)
ids_dict = [(row['device_id'], row['id'], row['object_id']) for row in ids]
return ids_dict
@pytest.mark.parametrize('device_id, uuid, object_id', device_ids())
@pytest.mark.parametrize('check_date', [str((datetime.now() - timedelta(days=1)).date())])
def test_telemetry_all(device_id, uuid, object_id, get_intake_data, devices_list, get_iot_data, check_date, expect):
_from = parser.parse(check_date)
_to = _from + timedelta(hours=23, minutes=59, seconds=59)
intake_from = _from.strftime('%Y/%m/%d %H:%M:%S')
intake_to = _to.strftime('%Y/%m/%d %H:%M:%S')
# take wider period from Dymano (+24 hours)
dynamo_from = _from.timestamp()*1000
dynamo_to = (_to + timedelta(hours=24)).timestamp()*1000
xml_file_name = '%s/IntakeRaw/device_telemetry/telemetry_device_%s_%s.xml' % (os.path.dirname(__file__), device_id, _from.date())
# write response data to file if there is no file saved
if not os.path.isfile(xml_file_name):
params = {'objectId': object_id, 'startIndex': 1, 'startDate': intake_from, 'endDate': intake_to}
content = get_intake_data('positions_report', **params)
# create dir, get intake data and write it to the file
os.makedirs(os.path.dirname(xml_file_name), exist_ok=True)
with open(xml_file_name, 'w') as f_xml:
f_xml.write(content.decode('utf-8'))
telemetry_in = devices_list(xml_file_name, 'POSITION')
telemetry_out = get_iot_data(uuid, dynamo_from, dynamo_to, 'telemetry')
# check if IOT data is empty but there are entries in the intake, go no further if this fails
if telemetry_in:
assert telemetry_out, \
'Fail: empty data received for device %s, period %s - %s: Entries count: Intake %s != %s Dynamo DB' \
% (uuid, dynamo_from, dynamo_to, len(telemetry_in), len(telemetry_out))
for pos_id in telemetry_in:
# check if the position id was saved in Dynamo
if pos_id in telemetry_out:
for key_out, key_in, to_type in intake_iot_mapper:
# check if the parameter is in the intake and it is not null
if key_in in telemetry_in[pos_id] and telemetry_in[pos_id][key_in] is not None:
if key_out in telemetry_out[pos_id]:
if key_in in ('LAT', 'LON'):
expect(
to_type(telemetry_out[pos_id]['location'][key_out]) == to_type(float(telemetry_in[pos_id][key_in])),
'Fail: position id %s, %s: in %s != %s out' % (pos_id, key_out, telemetry_in[pos_id][key_in],
telemetry_out[pos_id]['location'][key_out]))
else:
expect(str(telemetry_out[pos_id][key_out]) == telemetry_in[pos_id][key_in],
'Fail: position id %s, %s: in %s != %s out' %
(pos_id, key_out, telemetry_in[pos_id][key_in], telemetry_out[pos_id][key_out]))
else:
expect(key_out in telemetry_out[pos_id],
'Fail: record time %s, device id: %s:%s: %s in %s != None %s out'
% (pos_id, device_id, uuid, key_in, telemetry_in[pos_id][key_in], key_out))
Follow us!