import tempfile import unittest from pathlib import Path from unittest.mock import patch from acceptance_service import ( XHS_URL_PATTERN, cancel_acceptance_task, collect_acceptance_now, collect_acceptance_crawler_now, create_acceptance_task, get_acceptance_task, run_due_acceptance_schedules, retry_failed_acceptance_devices, stop_acceptance_schedule, update_acceptance_schedule, ) from config import Config from database import connection, init_database, now_iso class _ThreadWithoutStart: def __init__(self, *args, **kwargs): pass def start(self): pass class AcceptanceCreationTest(unittest.TestCase): def test_xhs_short_link_cn_is_recognized(self): text = "笔记 https://xhslink.cn/o/example123 进入小红书" self.assertEqual(XHS_URL_PATTERN.findall(text), ["https://xhslink.cn/o/example123"]) def setUp(self): self.temp_dir = tempfile.TemporaryDirectory() self.original_database_path = Config.DATABASE_PATH self.original_file_storage_path = Config.FILE_STORAGE_PATH root = Path(self.temp_dir.name) Config.DATABASE_PATH = root / "holy.db" Config.FILE_STORAGE_PATH = root / "files" init_database() def tearDown(self): Config.DATABASE_PATH = self.original_database_path Config.FILE_STORAGE_PATH = self.original_file_storage_path self.temp_dir.cleanup() @patch("acceptance_service.threading.Thread", _ThreadWithoutStart) @patch("acceptance_service.get_wen_resources") def test_screenshot_task_requires_keywords_but_not_feishu_links(self, resources): resources.return_value = {"cookie_count": 0, "device_count": 2} task = create_acceptance_task({ "acceptance_method": "screenshot", "feishu_links": [], "keywords": ["山姆特价"], "device_count": 1, }) self.assertEqual(task["acceptance_method"], "screenshot") self.assertEqual(task["keywords"], ["山姆特价"]) self.assertEqual(task["feishu_links"], []) @patch("acceptance_service.threading.Thread", _ThreadWithoutStart) @patch("acceptance_service.get_wen_resources") def test_active_acceptance_task_can_be_cancelled(self, resources): resources.return_value = {"cookie_count": 0, "device_count": 2} task = create_acceptance_task({ "acceptance_method": "screenshot", "keywords": ["山姆特价"], "device_count": 1, }) update_acceptance_schedule(task["task_id"], [1], "00:00", "23:59") cancelled = cancel_acceptance_task(task["task_id"]) self.assertEqual(cancelled["status"], "cancelled") self.assertTrue(cancelled["schedule_enabled"]) self.assertIsNotNone(cancelled["next_run_at"]) self.assertEqual(cancel_acceptance_task(task["task_id"])["status"], "cancelled") @patch("acceptance_service.threading.Thread", _ThreadWithoutStart) @patch("acceptance_service.get_wen_resources") def test_custom_cookie_is_stored_as_secret_and_not_returned(self, resources): resources.return_value = {"cookie_count": 0, "device_count": 0, "credential_ids": []} custom_cookie = "web_session=secret-value; a1=another-secret" task = create_acceptance_task({ "acceptance_method": "data-analysis", "feishu_links": ["https://example.feishu.cn/docx/project"], "keywords": ["山姆特价"], "custom_cookie": custom_cookie, }) self.assertEqual(task["device_count"], 1) self.assertNotIn(custom_cookie, str(task)) self.assertTrue(task["resource_progress"][0]["resource_id"].startswith("custom-")) with connection() as conn: row = conn.execute( "SELECT secret_value FROM acceptance_task_secrets WHERE secret_type='custom_cookie'" ).fetchone() self.assertEqual(row["secret_value"], custom_cookie) @patch("acceptance_service.threading.Thread", _ThreadWithoutStart) @patch("acceptance_service.get_wen_resources") def test_weekly_schedule_and_immediate_collection(self, resources): resources.return_value = {"cookie_count": 0, "device_count": 2} source = create_acceptance_task({ "acceptance_method": "screenshot", "keywords": ["山姆特价"], "device_count": 1, }) scheduled = update_acceptance_schedule( source["task_id"], [1, 3, 5], "01:30", "04:30", ) self.assertTrue(scheduled["schedule_enabled"]) self.assertEqual(scheduled["schedule_weekday"], 1) self.assertEqual(scheduled["schedule_weekdays"], [1, 3, 5]) self.assertEqual(scheduled["schedule_start_time"], "01:30") self.assertEqual(scheduled["schedule_end_time"], "04:30") self.assertIsNotNone(scheduled["next_run_at"]) child = collect_acceptance_now(source["task_id"]) self.assertNotEqual(child["task_id"], source["task_id"]) self.assertEqual(child["keywords"], ["山姆特价"]) self.assertFalse(child["schedule_enabled"]) stopped = stop_acceptance_schedule(source["task_id"]) self.assertFalse(stopped["schedule_enabled"]) self.assertIsNone(stopped["next_run_at"]) @patch("acceptance_service.threading.Thread", _ThreadWithoutStart) @patch("acceptance_service.get_wen_resources") def test_screenshot_project_can_start_cookie_crawler_batch(self, resources): resources.return_value = {"cookie_count": 2, "device_count": 3, "credential_ids": ["cookie-1", "cookie-2"]} source = create_acceptance_task({ "acceptance_method": "screenshot", "keywords": ["山姆会员卡"], "match_keywords": ["民生"], "device_count": 3, }) child = collect_acceptance_crawler_now(source["task_id"]) self.assertEqual(child["acceptance_method"], "data-analysis") self.assertEqual(child["keywords"], ["山姆会员卡"]) self.assertEqual(child["match_keywords"], ["民生"]) self.assertEqual(child["cookie_count"], 2) @patch("acceptance_service.threading.Thread", _ThreadWithoutStart) @patch("acceptance_service._copy_acceptance_task") @patch("acceptance_service.get_wen_resources") def test_due_schedule_reads_persisted_multi_day_range(self, resources, copy_task): resources.return_value = {"cookie_count": 0, "device_count": 2} source = create_acceptance_task({ "acceptance_method": "screenshot", "keywords": ["山姆特价"], "device_count": 1, }) update_acceptance_schedule(source["task_id"], [1, 4, 7], "00:00", "23:59") with connection() as conn: conn.execute( "UPDATE acceptance_tasks SET next_run_at=? WHERE public_task_id=?", (now_iso(), source["task_id"]), ) self.assertEqual(run_due_acceptance_schedules(), 1) copy_task.assert_called_once_with(source["task_id"], trigger_source="acceptance-weekly") refreshed = get_acceptance_task(source["task_id"]) self.assertEqual(refreshed["schedule_weekdays"], [1, 4, 7]) self.assertIsNotNone(refreshed["last_run_at"]) self.assertIsNotNone(refreshed["next_run_at"]) @patch("acceptance_service.threading.Thread", _ThreadWithoutStart) @patch("acceptance_service.schedule_xapi_screenshots") @patch("acceptance_service.get_wen_resources") def test_retry_failed_devices_dispatches_only_failed_accounts(self, resources, schedule): resources.return_value = {"cookie_count": 0, "device_count": 2} schedule.return_value = { "execution_mode": "immediate", "batches": [{ "batch_id": "retry-batch-1", "items": [{"xhs_account_id": "failed-account"}], }], } source = create_acceptance_task({ "acceptance_method": "screenshot", "keywords": ["山姆特价"], "device_count": 2, }) with connection() as conn: task = conn.execute( "SELECT id FROM acceptance_tasks WHERE public_task_id=?", (source["task_id"],), ).fetchone() now = now_iso() run = conn.execute( """INSERT INTO acceptance_keyword_runs (task_id,keyword,run_type,run_order,status,created_at,updated_at) VALUES (?,'山姆特价','screenshot',0,'failed',?,?)""", (task["id"], now, now), ) conn.executemany( """INSERT INTO acceptance_task_resources (task_id,resource_type,resource_id,resource_order,created_at) VALUES (?,'device',?,?,?)""", [ (task["id"], "failed-account", 0, now), (task["id"], "successful-account", 1, now), ], ) conn.executemany( """INSERT INTO acceptance_xapi_runs (task_id,keyword_run_id,batch_id,xapi_task_id,keyword,xhs_account_id, status,artifacts_json,feishu_sync_status,raw_json,created_at,updated_at) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)""", [ (task["id"], run.lastrowid, "old-1", "task-failed", "山姆特价", "failed-account", "failed", "[]", "failed", "{}", now, now), (task["id"], run.lastrowid, "old-2", "task-success", "山姆特价", "successful-account", "completed", '[{"image_url":"https://example.test/ok.png"}]', "succeeded", "{}", now, now), ], ) retried = retry_failed_acceptance_devices(source["task_id"]) self.assertEqual(retried["retry_dispatched_device_count"], 1) self.assertEqual(retried["retry_dispatched_pair_count"], 1) schedule.assert_called_once_with( ["山姆特价"], trigger_source="acceptance", xhs_account_ids=["failed-account"], ) with connection() as conn: keyword_run = conn.execute( "SELECT status,crawler_task_id FROM acceptance_keyword_runs WHERE id=?", (run.lastrowid,), ).fetchone() self.assertEqual(keyword_run["status"], "waiting") self.assertEqual(keyword_run["crawler_task_id"], "retry-batch-1") @patch("acceptance_service.threading.Thread", _ThreadWithoutStart) @patch("acceptance_service.get_wen_resources") def test_combined_task_uses_all_resources_and_keeps_match_keywords(self, resources): resources.return_value = { "cookie_count": 3, "device_count": 5, "credential_ids": ["cookie-1", "cookie-2", "cookie-3"], } task = create_acceptance_task({ "acceptance_method": "combined", "feishu_links": ["https://example.feishu.cn/docx/project"], "keywords": ["山姆特价"], "match_keywords": ["民生", "联名卡"], }) self.assertEqual(task["device_count"], 5) self.assertEqual(task["cookie_count"], 3) self.assertEqual(task["match_keywords"], ["民生", "联名卡"]) self.assertEqual( len([item for item in task["resource_progress"] if item["resource_type"] == "cookie"]), 3, ) if __name__ == "__main__": unittest.main() cancel_acceptance_task,