- 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!
Особенность библиотеки hljs, отсюда и разница с ру.
1. тебя не напрягает?
2. прямые sql-скрипты без юзания ORM и параметризации? похуй, что это тесты, тесты - тоже адекватно писать надо
3. огромное кол-во параметров в методах (хороший тон не более: 3-4... остальное выделить в доп приватные методы)
4. хуле все в ф-циях и вне классов с использованием ABC (abstract base classes)? чо когда тестов станет 100500, как ты будешь жить?
https://msdn.microsoft.com/ru-ru/windows/desktop/aa374843
2. Один константный запрос. ORM нинужен.
4. Как классы спасут тебя от написания 100500 тестов?
1. Да, говно, по идее, под каждый if должен быть отдельный тест. Вот только, скорее всего, сделано это потому, что подготовка данных теста выполняется по времени, как обычно, два миллиона лет, и чтобы не ждать завершения прогонки интеграционных тестов до скончания времен или ухода продукта с рынка, несколько схожих проверок объединили в одну. Так можно делать.
2. ORM, [естественно], не нужен. Это однозначно. Если это единственный тест, то можно загнать SQL-запрос прямо в тело метода; если так руки чешутся порефакторить, можете вынести в отдельный метод/класс-хелпер. Если все тесты такие, да, было бы неплохо вынести все это в отдельный класс сервиса. Повторюсь, просто в класс сервиса, без использования полноценного ORM. Если что-то в мире разработки есть, это не значит, что это говно надо обязетельно тащить в тесты. Вы ещё в микросервис обращение к базе вынесите.
3. Какие приватные методы, что вы несете? Как вы передадите потом эти данные в тестовый метод? Вынести все входные данные можно в отдельную структуру. Выходные, если их много, тоже можно в отдельную структуру. Здесь решили не усложнять. Опять же, это не отлично, но приемлемо для такого теста.
4. Не понял суть претензии, наверное, это к тому, что часть логики можно вынести в класс-хелпер. Это можно. Это сделает чтение сценария тестового метода проще, но, естественно, не сделает сам тест проще.
Вообще, это далеко не самый худший интеграционный тест, который я встречал. Обычно такие тесты раз в пять больше, с кучей обработок исключений.