def test_task_timeout_set_in_session(tmpdir, execution, session):
    """Test task termination due to the task ran too long"""
    with tmpdir.as_cwd():

        func_run_slow = get_slow_func(execution)

        task = FuncTask(func_run_slow, name="slow task", start_cond=AlwaysTrue(), execution=execution, session=session)

        session.config.shut_cond = (TaskStarted(task="slow task") >= 2) | ~SchedulerStarted(period=TimeDelta("20 seconds"))
        session.config.timeout = 0.1
        assert session.config.timeout == datetime.timedelta(milliseconds=100)
        session.start()

        logger = task.logger
        assert 2 == logger.filter_by(action="run").count()
        assert 2 == logger.filter_by(action="terminate").count()
        assert 0 == logger.filter_by(action="success").count()
        assert 0 == logger.filter_by(action="fail").count()

        assert not os.path.exists("work.txt")
Test task termination due to the task ran too long
rocketry/test/schedule/test_terminate.py