(self)
| 846 | assert loads(dumps(s)) |
| 847 | |
| 848 | def test_start(self): |
| 849 | s, sh = self.get_service() |
| 850 | schedule = s.scheduler.schedule |
| 851 | assert isinstance(schedule, dict) |
| 852 | assert isinstance(s.scheduler, beat.Scheduler) |
| 853 | scheduled = list(schedule.keys()) |
| 854 | for task_name in sh['entries'].keys(): |
| 855 | assert task_name in scheduled |
| 856 | |
| 857 | s.sync() |
| 858 | assert sh.closed |
| 859 | assert sh.synced |
| 860 | assert s._is_stopped.is_set() |
| 861 | s.sync() |
| 862 | s.stop(wait=False) |
| 863 | assert s._is_shutdown.is_set() |
| 864 | s.stop(wait=True) |
| 865 | assert s._is_shutdown.is_set() |
| 866 | |
| 867 | p = s.scheduler._store |
| 868 | s.scheduler._store = None |
| 869 | try: |
| 870 | s.scheduler.sync() |
| 871 | finally: |
| 872 | s.scheduler._store = p |
| 873 | |
| 874 | def test_start_embedded_process(self): |
| 875 | s, sh = self.get_service() |
nothing calls this directly
no test coverage detected