1. Python / Говнокод #26052

    0

    1. 01
    2. 02
    3. 03
    4. 04
    5. 05
    6. 06
    7. 07
    8. 08
    9. 09
    10. 10
    11. 11
    12. 12
    13. 13
    14. 14
    15. 15
    16. 16
    17. 17
    18. 18
    19. 19
    20. 20
    21. 21
    22. 22
    23. 23
    24. 24
    25. 25
    26. 26
    27. 27
    28. 28
    29. 29
    30. 30
    31. 31
    32. 32
    33. 33
    34. 34
    35. 35
    36. 36
    37. 37
    38. 38
    39. 39
    40. 40
    41. 41
    42. 42
    43. 43
    44. 44
    45. 45
    46. 46
    47. 47
    48. 48
    49. 49
    50. 50
    51. 51
    52. 52
    53. 53
    54. 54
    55. 55
    56. 56
    57. 57
    58. 58
    59. 59
    60. 60
    61. 61
    62. 62
    63. 63
    64. 64
    65. 65
    66. 66
    67. 67
    68. 68
    69. 69
    70. 70
    71. 71
    72. 72
    73. 73
    74. 74
    75. 75
    76. 76
    77. 77
    78. 78
    79. 79
    80. 80
    81. 81
    82. 82
    83. 83
    84. 84
    85. 85
    86. 86
    87. 87
    88. 88
    89. 89
    90. 90
    91. 91
    92. 92
    93. 93
    94. 94
    95. 95
    96. 96
    97. 97
    98. 98
    99. 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))

    интеграционный тест одной тупой педовки

    Запостил: gvkdgvkd, 27 Ноября 2019

    Комментарии (17) RSS

    • Где говно?
      Ответить
      • показать все, что скрытоvanished
        Ответить
      • @Rooster, а чо? вложенные циклы, и ересь в виде вложенных:

        if
          if
            if
              if
            else
          else


        1. тебя не напрягает?
        2. прямые sql-скрипты без юзания ORM и параметризации? похуй, что это тесты, тесты - тоже адекватно писать надо
        3. огромное кол-во параметров в методах (хороший тон не более: 3-4... остальное выделить в доп приватные методы)
        4. хуле все в ф-циях и вне классов с использованием ABC (abstract base classes)? чо когда тестов станет 100500, как ты будешь жить?
        Ответить
        • 3. Сколько здесь параметров? Максимум восемь? Смотрите, как надо:
          https://msdn.microsoft.com/ru-ru/windows/desktop/aa374843
          Ответить
        • 1. Нет, потому что мне не надо с этим разбираться) Да и в портянках из ифов ничего особо интересного нет.
          2. Один константный запрос. ORM нинужен.
          4. Как классы спасут тебя от написания 100500 тестов?
          Ответить
        • Я вас умоляю, практически все проекты, за редким исключением, имеют именно такие интеграционные тесты.

          1. Да, говно, по идее, под каждый if должен быть отдельный тест. Вот только, скорее всего, сделано это потому, что подготовка данных теста выполняется по времени, как обычно, два миллиона лет, и чтобы не ждать завершения прогонки интеграционных тестов до скончания времен или ухода продукта с рынка, несколько схожих проверок объединили в одну. Так можно делать.
          2. ORM, [естественно], не нужен. Это однозначно. Если это единственный тест, то можно загнать SQL-запрос прямо в тело метода; если так руки чешутся порефакторить, можете вынести в отдельный метод/класс-хелпер. Если все тесты такие, да, было бы неплохо вынести все это в отдельный класс сервиса. Повторюсь, просто в класс сервиса, без использования полноценного ORM. Если что-то в мире разработки есть, это не значит, что это говно надо обязетельно тащить в тесты. Вы ещё в микросервис обращение к базе вынесите.
          3. Какие приватные методы, что вы несете? Как вы передадите потом эти данные в тестовый метод? Вынести все входные данные можно в отдельную структуру. Выходные, если их много, тоже можно в отдельную структуру. Здесь решили не усложнять. Опять же, это не отлично, но приемлемо для такого теста.
          4. Не понял суть претензии, наверное, это к тому, что часть логики можно вынести в класс-хелпер. Это можно. Это сделает чтение сценария тестового метода проще, но, естественно, не сделает сам тест проще.

          Вообще, это далеко не самый худший интеграционный тест, который я встречал. Обычно такие тесты раз в пять больше, с кучей обработок исключений.
          Ответить

    Добавить комментарий