от
Я работаю над сервисом Python, который отслеживает каталог в файловой системе. Когда он видит, что файл был создан или перемещен туда, он отправляет путь к файлу в очередь Kafka. У меня служба работает точно так, как мне нужно, но моя проблема в том, что у меня должно быть как минимум 90% покрытия модульными тестами. Я относительно новичок в Python, и я никогда раньше не использовал модульное тестирование ни на одном языке, поэтому я чувствую себя действительно не в своей тарелке. Я просто не могу обернуться, как бы я начал тестировать эти классы. Это класс, который контролирует файловую систему, я использую сторожевую библиотеку. Я добавил параметр
handler=FileHandler
в init, потому что решил, что смогу использовать его для передачи классу фальшивого обработчика, который я мог бы использовать для тестов, но кажется, что он излишне сложен.
class FileSystemMonitor:

    def __init__(self, target_path, kafka_queue, handler=FileHandler):
        self.path = target_path
        self.queue = kafka_queue
        self.handler = handler(self.queue)

    def start(self):
        observer = Observer()
        observer.schedule(self.handler, self.path, recursive=True)
        observer.start()
        try:
            while True:
                time.sleep(1)
        except KeyboardInterrupt:
            observer.stop()
        observer.join()

def parse_args():
    path = sys.argv[1] if len(sys.argv) > 1 else '.'
    queue = sys.argv[2] if len(sys.argv) > 2 else 'default'
    return path, queue

if __name__ == "__main__":
    path, queue = parse_args()
    monitor = FileSystemMonitor(path, queue)
    monitor.start()
Это класс, который я создал, который обрабатывает события, генерируемые монитором, и передает путь в очередь Кафки.
class FileHandler(PatternMatchingEventHandler):

    def __init__(self, queue):
        super(FileHandler, self).__init__(ignore_patterns=["*/.DS_Store"], ignore_directories=True)
        self.queue = queue

    def on_any_event(self, event):
        super(FileHandler, self).on_any_event(event)
        #print(event, self.queue)
        result = kafkaProducer.send_msg(self.queue, event.src_path, event.event_type)
        print("Handler:", result)
        return result
Я написал несколько тестов для класса kafkaProducer, и у меня не было таких трудностей, потому что он на самом деле возвращает значение, которое я мог протестировать. FileSystemMonitor работает бесконечно и просто ждет прерывания клавиатуры, а когда оно заканчивается, ничего не возвращается, так как мне написать для него модульные тесты? Что касается класса FileHandler, то он зависит от событий, запускаемых классом монитора, так как же мне выделить класс Handler для его проверки?              

Ваш ответ

Отображаемое имя (по желанию):
Конфиденциальность: Ваш электронный адрес будет использоваться только для отправки уведомлений.
Анти-спам проверка:
Чтобы избежать проверки в будущем, пожалуйста войдите или зарегистрируйтесь.
...