from io import StringIO from unittest.mock import patch from allauth.mfa.models import Authenticator from celery import shared_task from django.contrib.auth import get_user_model from django.core.cache import cache from django.core.management import call_command from django.core.management.base import CommandError from django.test import RequestFactory, TestCase, override_settings from waffle import flag_is_active, get_waffle_flag_model from club.models import Club from .jobs import JOB_REGISTRY from .models import JobRun, JobToggle, Maintenance Flag = get_waffle_flag_model() User = get_user_model() class ClubScopedFlagTests(TestCase): @classmethod def setUpTestData(cls): cls.club = Club.objects.create(name="Ajax United", slug="ajax-united") cls.other = Club.objects.create(name="Rival FC", slug="rival-fc") cls.flag = Flag.objects.create(name="shop") def setUp(self): # waffle caches flags by name, and its cache is NOT rolled back with the # test transaction -- a flag row whose targeting changed in the previous # test would otherwise be shadowed by that test's cached object. Has to # stay per-test: it is the cache, not the rows, that leaks. cache.clear() self.addCleanup(cache.clear) def request_for(self, club): request = RequestFactory().get("/") request.club = club request.user = None return request def active_for(self, club): return flag_is_active(self.request_for(club), "shop") def test_off_for_every_club_by_default(self): self.assertFalse(self.active_for(self.club)) self.assertFalse(self.active_for(self.other)) def test_on_only_for_the_targeted_club(self): self.flag.clubs.add(self.club) self.assertTrue(self.active_for(self.club)) self.assertFalse(self.active_for(self.other)) def test_removing_a_club_turns_it_off_again(self): self.flag.clubs.add(self.club) self.flag.clubs.remove(self.club) self.assertFalse(self.active_for(self.club)) def test_everyone_true_overrides_club_targeting(self): self.flag.everyone = True self.flag.save() self.assertTrue(self.active_for(self.other)) # not targeted, still on def test_everyone_false_beats_club_targeting(self): # waffle's contract: `everyone` overrides ALL other settings, so a flag # switched off for everyone must stay off even for a targeted club. self.flag.clubs.add(self.club) self.flag.everyone = False self.flag.save() self.assertFalse(self.active_for(self.club)) def test_no_club_on_the_request_is_not_active(self): # e.g. the base domain / control panel, where there is no tenant. self.flag.clubs.add(self.club) self.assertFalse(flag_is_active(self.request_for(None), "shop")) def test_a_flag_name_with_no_matching_row_is_off_not_an_error(self): # Regression: waffle's own BaseModel.get() falls back to a transient, # unsaved Flag(name=...) instance when nothing in the DB matches the # name -- an M2M lookup (self.clubs) against that unsaved instance used # to raise ValueError instead of just resolving to "off". This is the # normal state for any flag nobody has created in the control panel yet. self.assertFalse(flag_is_active(self.request_for(self.club), "no-such-flag")) def test_is_active_for_club_without_a_request(self): self.flag.clubs.add(self.club) self.assertTrue(self.flag.is_active_for_club(self.club)) self.assertFalse(self.flag.is_active_for_club(self.other)) def test_is_active_for_club_respects_everyone(self): self.flag.everyone = False self.flag.save() self.assertFalse(self.flag.is_active_for_club(self.club)) self.flag.everyone = True self.flag.save() self.assertTrue(self.flag.is_active_for_club(self.club)) def test_cache_is_flushed_when_club_targeting_changes(self): # The M2M does not call save(), so without the flush signal waffle would # keep answering from a stale cached set. self.assertFalse(self.active_for(self.club)) # primes the cache self.flag.clubs.add(self.club) self.assertTrue(self.active_for(self.club)) def test_cache_is_flushed_on_reverse_edit(self): self.assertFalse(self.active_for(self.club)) self.club.flags.add(self.flag) self.assertTrue(self.active_for(self.club)) @override_settings( ROSTERCHIEF_BASE_DOMAIN="rosterchief.app", ALLOWED_HOSTS=["rosterchief.app", "ajax-united.rosterchief.app", "testserver"], ) class MaintenanceModeTests(TestCase): @classmethod def setUpTestData(cls): cls.club = Club.objects.create(name="Ajax United", slug="ajax-united") cls.user = User.objects.create_user(email="root@example.com", password="pw-secret-123", is_staff=True) Authenticator.objects.create(user=cls.user, type=Authenticator.Type.TOTP, data={"secret": "JBSWY3DPEHPK3PXP"}) def setUp(self): # Maintenance state lives in the shared cache, which no transaction rolls # back -- each test has to start from a reopened platform. cache.clear() self.addCleanup(cache.clear) def club_get(self, path="/"): return self.client.get(path, HTTP_HOST="ajax-united.rosterchief.app") def platform_get(self, path): return self.client.get(path, HTTP_HOST="rosterchief.app") def test_a_club_subdomain_is_closed(self): Maintenance.start(message="Upgrading the database.") response = self.club_get("/accounts/login/") self.assertEqual(response.status_code, 503) self.assertContains(response, "Upgrading the database", status_code=503) self.assertEqual(response["Retry-After"], "3600") def test_a_club_is_open_again_when_maintenance_ends(self): Maintenance.start() Maintenance.stop() self.assertEqual(self.club_get("/accounts/login/").status_code, 200) def test_the_control_panel_stays_reachable(self): Maintenance.start() self.client.force_login(self.user) self.assertEqual(self.platform_get("/controlpanel/").status_code, 200) def test_signing_in_stays_possible(self): # Close /accounts/ as well and you cannot sign in to turn maintenance off: a # lock-down with no key, fixable only from a shell. Maintenance.start() self.assertEqual(self.platform_get("/accounts/login/").status_code, 200) def test_the_club_logo_still_loads_on_the_maintenance_page(self): # The maintenance page is rendered through the club's own skin specifically to show # its logo -- closing /media/ on the club subdomain too would serve the maintenance # page itself in place of that logo, making it look like the logo vanished. Maintenance.start() self.assertEqual(self.club_get("/media/clubs/ajax-united/crest.png").status_code, 404) def test_the_health_check_still_answers(self): # Close it and the load balancer decides the node is dead and stops routing to it — # taking the control panel down with everything else. Maintenance.start() self.assertEqual(self.platform_get("/healthz").status_code, 200) self.assertEqual(self.client.get("/healthz", HTTP_HOST="ajax-united.rosterchief.app").status_code, 200) def test_the_rest_of_the_base_domain_is_closed(self): Maintenance.start() self.assertEqual(self.platform_get("/").status_code, 503) def test_a_json_caller_gets_json(self): Maintenance.start(message="Back soon.") response = self.client.get("/", HTTP_HOST="ajax-united.rosterchief.app", headers={"accept": "application/json"}) self.assertEqual(response.status_code, 503) self.assertEqual(response.json()["status"], "maintenance") def test_the_message_is_optional(self): Maintenance.start() self.assertContains(self.club_get("/"), "down for maintenance", status_code=503) def test_it_says_which_state_it_is_in(self): self.assertEqual(str(Maintenance.start()), "Maintenance on") self.assertEqual(str(Maintenance.stop()), "Maintenance off") def test_it_records_who_closed_the_platform_and_when(self): maintenance = Maintenance.start(message="db upgrade", user=self.user) self.assertTrue(maintenance.is_active) self.assertEqual(maintenance.started_by, self.user) self.assertIsNotNone(maintenance.started_at) def test_the_state_is_shared_rather_than_per_process(self): # The cache is the shared one the flags use, so closing the platform reaches every # worker and every server. A per-process cache would leave some workers serving clubs. Maintenance.start() self.assertTrue(cache.get(Maintenance.CACHE_KEY).is_active) self.assertTrue(Maintenance.is_on()) Maintenance.stop() self.assertFalse(cache.get(Maintenance.CACHE_KEY).is_active) class MaintenanceCommandTests(TestCase): def setUp(self): cache.clear() self.addCleanup(cache.clear) def run_command(self, name, *args): out = StringIO() call_command(name, *args, stdout=out, stderr=out) return out.getvalue() def test_a_scheduled_job_stands_down(self): Maintenance.start() with self.assertRaises(CommandError): self.run_command("archive_overdue_clubs") def test_it_runs_again_once_the_platform_reopens(self): Maintenance.start() Maintenance.stop() self.assertIn("Nothing overdue", self.run_command("archive_overdue_clubs")) def test_an_override_exists_for_when_you_mean_it(self): Maintenance.start() self.assertIn("Nothing overdue", self.run_command("archive_overdue_clubs", "--ignore-maintenance")) def test_migrate_is_not_blocked(self): # Maintenance is usually declared IN ORDER to migrate. A guard on every command would # mean turning the mode off to do the work you turned it on for. Maintenance.start() self.run_command("migrate", "--check") # raises SystemExit only if migrations are pending @shared_task(name="features.tests.succeed") def _succeed_task(): return "did the thing" @shared_task(name="features.tests.fail") def _fail_task(): raise RuntimeError("boom") @shared_task(name="features.tests.untracked") def _untracked_task(): return "quiet" class JobRunTests(TestCase): """The Celery signal wiring in features/signals.py, exercised against throwaway tasks (module-level, like any real task -- Celery's registry gets confused if the same task name is redefined per-test) rather than the real billing/club/events ones: what matters here is that a JobRun row appears for anything in JOB_REGISTRY and only for that, not the domain logic of any one scheduled job (each of those has its own tests alongside its management command).""" def setUp(self): self.succeed_task, self.fail_task, self.untracked_task = _succeed_task, _fail_task, _untracked_task patcher = patch.dict(JOB_REGISTRY, {"features.tests.succeed": {}, "features.tests.fail": {}}) patcher.start() self.addCleanup(patcher.stop) def test_a_successful_run_is_recorded(self): # .apply() runs the task synchronously, in-process, regardless of # CELERY_TASK_ALWAYS_EAGER -- exactly what the task_prerun/task_postrun signal # handlers in features/signals.py are wired to react to either way. self.succeed_task.apply() run = JobRun.objects.get(name="features.tests.succeed") self.assertEqual(run.status, JobRun.Status.SUCCESS) self.assertEqual(run.detail, "did the thing") self.assertEqual(run.error, "") self.assertIsNotNone(run.started_at) self.assertIsNotNone(run.finished_at) def test_a_failed_run_is_recorded(self): # A real worker never raises a task's exception back into whoever called .delay() -- # it's async, the caller is long gone by the time the task runs -- so eager mode # doesn't either (CELERY_TASK_EAGER_PROPAGATES is left at its default False; see # rosterchief/settings.py). The result carries FAILURE instead, same as production. result = self.fail_task.apply() self.assertEqual(result.state, "FAILURE") run = JobRun.objects.get(name="features.tests.fail") self.assertEqual(run.status, JobRun.Status.FAILURE) self.assertIn("boom", run.error) def test_a_task_outside_the_registry_is_not_tracked(self): self.untracked_task.apply() self.assertFalse(JobRun.objects.filter(name="features.tests.untracked").exists()) class JobToggleTests(TestCase): """Per-job pause/resume switch -- see features.models.JobToggle and its use in events/billing/club's scheduled tasks (each checks JobToggle.is_enabled(name) right alongside the existing Maintenance.is_on() gate).""" def setUp(self): # Same reasoning as MaintenanceModeTests -- the toggle's cache isn't # rolled back with the test transaction. cache.clear() self.addCleanup(cache.clear) def test_a_job_with_no_row_is_enabled_by_default(self): self.assertTrue(JobToggle.is_enabled("events.tasks.extend_event_series")) def test_disabling_a_job(self): JobToggle.set_enabled("events.tasks.extend_event_series", False) self.assertFalse(JobToggle.is_enabled("events.tasks.extend_event_series")) def test_re_enabling_a_job(self): JobToggle.set_enabled("events.tasks.extend_event_series", False) JobToggle.set_enabled("events.tasks.extend_event_series", True) self.assertTrue(JobToggle.is_enabled("events.tasks.extend_event_series")) def test_toggling_one_job_does_not_affect_another(self): JobToggle.set_enabled("events.tasks.extend_event_series", False) self.assertTrue(JobToggle.is_enabled("events.tasks.send_deadline_reminders"))