Я переношу код конвейера Apache Beam с Python 2 на Python 3.
У меня есть тест конвейера Beam, написанный на Python2, и когда я пытаюсь запустить его с python3 каждый раз, когда я получаю разные результаты от своего Beamконвейер, приводящий к сбою тестовых случаев.
Это код, в котором, я думаю, проблема возникает при запуске конвейера и получении образцов данных.
result = []
with open_shards(temp_path + '.result-*-of-*') as result_file:
for line in result_file:
result.append(line)
self.assertEqual(sorted(result[0]), sorted(expected_data))
Каждый раз, когда я получаю случайное значение (разные) данные (первый элемент) в списке результатов и, таким образом, мой тестовый случай иногда дает сбой, потому что он не соответствует моим ожидаемым данным.Это работает без проблем в Python 2. Есть идеи, что является причиной проблемы и как я могу ее решить?