|
| 1 | +import pytest |
| 2 | +from django.core.exceptions import ImproperlyConfigured |
| 3 | +from django.test import override_settings |
| 4 | + |
| 5 | +from django_celery_beat.helpers import (clockedschedule_model, |
| 6 | + crontabschedule_model, |
| 7 | + intervalschedule_model, |
| 8 | + periodictask_model, |
| 9 | + periodictasks_model, |
| 10 | + solarschedule_model) |
| 11 | +from django_celery_beat.models.abstract import (AbstractClockedSchedule, |
| 12 | + AbstractCrontabSchedule, |
| 13 | + AbstractIntervalSchedule, |
| 14 | + AbstractPeriodicTask, |
| 15 | + AbstractPeriodicTasks, |
| 16 | + AbstractSolarSchedule) |
| 17 | + |
| 18 | +fetch_models = [ |
| 19 | + (AbstractCrontabSchedule, crontabschedule_model), |
| 20 | + (AbstractIntervalSchedule, intervalschedule_model), |
| 21 | + (AbstractPeriodicTask, periodictask_model), |
| 22 | + (AbstractSolarSchedule, solarschedule_model), |
| 23 | + (AbstractPeriodicTasks, periodictasks_model), |
| 24 | + (AbstractClockedSchedule, clockedschedule_model), |
| 25 | +] |
| 26 | + |
| 27 | + |
| 28 | +@pytest.mark.django_db |
| 29 | +@pytest.mark.parametrize(("abstract_model", "fetch_func"), fetch_models) |
| 30 | +def test_fetching_model_works_with_no_setting(abstract_model, fetch_func): |
| 31 | + """ |
| 32 | + Test fetching models when you're using the generic Django Celery Beat models. |
| 33 | + """ |
| 34 | + result = fetch_func() |
| 35 | + assert issubclass( |
| 36 | + result, abstract_model |
| 37 | + ), f"Expected {abstract_model}, got {type(result)}" |
| 38 | + |
| 39 | + |
| 40 | +@pytest.mark.django_db |
| 41 | +@pytest.mark.parametrize(("abstract_model", "fetch_func"), fetch_models) |
| 42 | +def test_fetching_model_works_with_correct_setting(abstract_model, fetch_func): |
| 43 | + """ |
| 44 | + Test fetching models when you're using custom Django Celery Beat models. |
| 45 | + """ |
| 46 | + model_name = abstract_model.__name__.replace("Abstract", "") |
| 47 | + model_setting_name = f"CELERY_BEAT_{model_name.upper()}_MODEL" |
| 48 | + app_label_and_generic_name = f"django_celery_beat.{model_name.upper()}" |
| 49 | + |
| 50 | + with override_settings(**{model_setting_name: app_label_and_generic_name}): |
| 51 | + result = fetch_func() |
| 52 | + assert issubclass( |
| 53 | + result, abstract_model |
| 54 | + ), f"Expected {abstract_model}, got {type(result)}" |
| 55 | + |
| 56 | + |
| 57 | +@pytest.mark.django_db |
| 58 | +@pytest.mark.parametrize(("abstract_model", "fetch_func"), fetch_models) |
| 59 | +def test_fetching_model_works_with_invalid_setting(abstract_model, fetch_func): |
| 60 | + """ |
| 61 | + Test fetching models when you're using custom Django Celery Beat models, |
| 62 | + but your model name is not of the form `app_label.model_name`. |
| 63 | + """ |
| 64 | + model_name = abstract_model.__name__.replace("Abstract", "") |
| 65 | + model_setting_name = f"CELERY_BEAT_{model_name.upper()}_MODEL" |
| 66 | + |
| 67 | + with override_settings(**{model_setting_name: f"{model_name}"}), pytest.raises( |
| 68 | + ImproperlyConfigured, |
| 69 | + match=f"{model_setting_name} must be of the form 'app_label.model_name'", |
| 70 | + ): |
| 71 | + fetch_func() |
| 72 | + |
| 73 | + |
| 74 | +@pytest.mark.django_db |
| 75 | +@pytest.mark.parametrize(("abstract_model", "fetch_func"), fetch_models) |
| 76 | +def test_fetching_model_works_with_improper_setting(abstract_model, fetch_func): |
| 77 | + """ |
| 78 | + Test fetching models when you're using custom Django Celery Beat models, |
| 79 | + but your model cannot be found. |
| 80 | + """ |
| 81 | + model_name = abstract_model.__name__.replace("Abstract", "") |
| 82 | + model_setting_name = f"CELERY_BEAT_{model_name.upper()}_MODEL" |
| 83 | + |
| 84 | + with override_settings( |
| 85 | + **{model_setting_name: f"improper.{model_name}"} |
| 86 | + ), pytest.raises( |
| 87 | + ImproperlyConfigured, |
| 88 | + match=f"{model_setting_name} refers to model 'improper.{model_name}' " |
| 89 | + "that has not been installed", |
| 90 | + ): |
| 91 | + fetch_func() |
0 commit comments