1. Список говнокодов пользователя gvkdgvkd

    Всего: 1

  2. 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)