gistfile1.py

@pytest.fixture
def scheduler(request):
    sched = DummyScheduler()
    if 'start_scheduler' in request.keywords:
        sched.start()
        request.addfinalizer(lambda: sched.shutdown() if sched.running else None)

    return sched


@pytest.fixture
def logstream(request, scheduler):
    stream = BytesIO()
    loghandler = StreamHandler(stream)
    loghandler.setLevel(ERROR)
    scheduler.logger.addHandler(loghandler)
    request.addfinalizer(lambda: scheduler.logger.removeHandler(loghandler))
    return stream


class TestOfflineScheduler(object):
    def test_jobstore_twice(self, scheduler):
        with pytest.raises(KeyError):
            scheduler.add_jobstore(MemoryJobStore(), 'dummy')
            scheduler.add_jobstore(MemoryJobStore(), 'dummy')