@@ -7,7 +7,12 @@ import frappe
from frappe.core.doctype.scheduled_job_type.scheduled_job_type import ScheduledJobType, sync_jobs
from frappe.utils import add_days, get_datetime
from frappe.utils.doctor import purge_pending_jobs
from frappe.utils.scheduler import enqueue_events, is_dormant, schedule_jobs_based_on_activity
from frappe.utils.scheduler import (
_get_last_modified_timestamp,
enqueue_events,
is_dormant,
schedule_jobs_based_on_activity,
)
def test_timeout_10():
@@ -56,7 +61,7 @@ class TestScheduler(TestCase):
self.assertFalse(is_dormant(check_time=frappe.db.get_last_created("Activity Log")))
def test_once_a_day_for_dormant(self):
frappe.db.cle ar_ tabl e("Scheduled Job Log")
frappe.db.trun cate("Scheduled Job Log")
self.assertTrue(schedule_jobs_based_on_activity(check_time=get_datetime("2100-01-01 00:00:00")))
self.assertTrue(
schedule_jobs_based_on_activity(
@@ -68,19 +73,21 @@ class TestScheduler(TestCase):
job = get_test_job(method="frappe.tests.test_scheduler.test_method", frequency="Daily")
job.execute()
job_log = frappe.get_doc("Scheduled Job Log", dict(scheduled_job_type=job.name))
job_log.db_set("creation", add_days(frappe.db.get_last_created("Activity Log"), 5))
job_log.db_set(
"modified", add_days(_get_last_modified_timestamp("Activity Log"), 5), update_modified=False
)
# inactive site with recent job, don't run
self.assertFalse(
schedule_jobs_based_on_activity(
check_time=add_days(frappe.db.get_last_created ("Activity Log"), 5)
check_time=add_days(_get_last_modified_timestamp ("Activity Log"), 5)
)
)
# one more day has passed
self.assertTrue(
schedule_jobs_based_on_activity(
check_time=add_days(frappe.db.get_last_created ("Activity Log"), 6)
check_time=add_days(_get_last_modified_timestamp ("Activity Log"), 6)
)
)