91 lines
3.3 KiB
Python
91 lines
3.3 KiB
Python
import sys
|
|
import unittest
|
|
from datetime import datetime, timedelta
|
|
|
|
|
|
sys.path.insert(0, "/www/wwwroot/getDomain/domainCheck")
|
|
|
|
from app.utils.database import _detect_job_item_step_priority
|
|
from app.utils.database import _ordered_step_claim_codes
|
|
from app.utils.database import _resolve_step_claim_quota
|
|
from app.utils.database import _select_preferred_claim_job_ids
|
|
|
|
|
|
class DetectJobItemClaimPriorityTestCase(unittest.TestCase):
|
|
def test_ordered_step_claim_codes_prioritize_deeper_steps_first(self):
|
|
ordered = _ordered_step_claim_codes()
|
|
self.assertLess(ordered.index("detect_360_site"), ordered.index("detect_baidu_site"))
|
|
self.assertLess(ordered.index("detect_chinaz"), ordered.index("detect_360_site"))
|
|
self.assertGreater(ordered.index("detect_register"), ordered.index("detect_baidu_site"))
|
|
|
|
def test_later_pipeline_steps_are_prioritized_ahead_of_earlier_steps(self):
|
|
self.assertLess(
|
|
_detect_job_item_step_priority("detect_360_site"),
|
|
_detect_job_item_step_priority("detect_baidu_site"),
|
|
)
|
|
self.assertLess(
|
|
_detect_job_item_step_priority("detect_chinaz"),
|
|
_detect_job_item_step_priority("detect_360_site"),
|
|
)
|
|
self.assertLess(
|
|
_detect_job_item_step_priority("detect_aizhan"),
|
|
_detect_job_item_step_priority("detect_chinaz"),
|
|
)
|
|
|
|
def test_non_register_steps_are_prioritized_ahead_of_register(self):
|
|
self.assertLess(
|
|
_detect_job_item_step_priority("detect_baidu_site"),
|
|
_detect_job_item_step_priority("detect_register"),
|
|
)
|
|
self.assertLess(
|
|
_detect_job_item_step_priority("detect_wayback"),
|
|
_detect_job_item_step_priority("detect_register"),
|
|
)
|
|
|
|
def test_blank_step_code_has_lowest_priority(self):
|
|
self.assertGreater(
|
|
_detect_job_item_step_priority(""),
|
|
_detect_job_item_step_priority("detect_register"),
|
|
)
|
|
|
|
def test_step_claim_quota_defaults_to_full_window_for_large_pools(self):
|
|
self.assertEqual(64, _resolve_step_claim_quota(120))
|
|
self.assertEqual(1200, _resolve_step_claim_quota(1200))
|
|
|
|
def test_select_preferred_claim_job_ids_prioritizes_running_jobs_even_if_older(self):
|
|
now = datetime(2026, 4, 24, 18, 30, 0)
|
|
|
|
selected = _select_preferred_claim_job_ids(
|
|
[
|
|
(19, "pending", now - timedelta(hours=60)),
|
|
(57, "running", now - timedelta(hours=72)),
|
|
(58, "pending", now - timedelta(hours=2)),
|
|
],
|
|
limit=3,
|
|
recent_hours=24,
|
|
now=now,
|
|
)
|
|
|
|
self.assertEqual([57, 58], selected)
|
|
|
|
def test_select_preferred_claim_job_ids_filters_out_stale_pending_jobs(self):
|
|
now = datetime(2026, 4, 24, 18, 30, 0)
|
|
|
|
selected = _select_preferred_claim_job_ids(
|
|
[
|
|
(19, "pending", now - timedelta(hours=60)),
|
|
(20, "pending", now - timedelta(hours=40)),
|
|
(58, "pending", now - timedelta(hours=2)),
|
|
(59, "pending", now - timedelta(minutes=30)),
|
|
],
|
|
limit=4,
|
|
recent_hours=24,
|
|
now=now,
|
|
)
|
|
|
|
self.assertEqual([59, 58], selected)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|