Test connection_index property returns correct index or raises appropriate exceptions.
(self)
| 71 | scheduler.register(say_hello, "nonexistent_queue", interval=300) |
| 72 | |
| 73 | def test_connection_index_property(self): |
| 74 | """Test connection_index property returns correct index or raises appropriate exceptions.""" |
| 75 | |
| 76 | scheduler = DjangoCronScheduler() |
| 77 | |
| 78 | # Before any registration, connection_index should raise ValueError |
| 79 | with self.assertRaises(ValueError): |
| 80 | _ = scheduler.connection_index |
| 81 | |
| 82 | # Register a job with 'test3' queue (secondary Redis DB configured for tests) |
| 83 | scheduler.register(say_hello, "test3", interval=60) |
| 84 | |
| 85 | # Now connection_index should return a valid index |
| 86 | connection_index = scheduler.connection_index |
| 87 | self.assertGreaterEqual(connection_index, 0) |
| 88 | |
| 89 | # Test with a queue using a different connection |
| 90 | scheduler2 = DjangoCronScheduler() |
| 91 | scheduler2.register(say_hello, "default", interval=60) # Uses DB=0 |
| 92 | |
| 93 | # Should have a different connection_index |
| 94 | self.assertNotEqual(scheduler2.connection_index, scheduler.connection_index) |
| 95 | |
| 96 | |
| 97 | class CronCommandTest(TestCase): |
nothing calls this directly
no test coverage detected