test_tasks.py 文件源码

def test_receivers_exception(app, db, sample_record_string):
    """Test receivers."""
    responses.add(
        responses.POST, "http://localhost:6800/schedule.json",
        body=json.dumps({"jobid": None, "status": "ok"}),
        status=200
    )

    mock_record = MagicMock()
    prop_mock = PropertyMock(return_value=sample_record_string)
    type(mock_record).raw = prop_mock

    with app.app_context():
        with pytest.raises(CrawlerScheduleError):
            receive_oaiharvest_job(
                request=None,
                records=[mock_record],
                name="",
                spider="Test",
                workflow="test"
            )