import hashlib import json from pathlib import Path import shutil import sys import tempfile import unittest ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT)) from articles import inspect_article, short_title from store import Store class QueueTests(unittest.TestCase): def setUp(self): self.tmp = tempfile.TemporaryDirectory() self.root = Path(self.tmp.name) self.article = self.root / 'article' self.article.mkdir() self.md = self.article / 'a.md' self.md.write_text('# 测试新闻\n\n正文内容。\n\n来源：测试来源，如有侵权，请联系删除。\n') self.store = Store(self.root / 'queue.sqlite3') self.store.configure(dict(fid=125, minimum=1, maximum=1, interval=0, history_checked=True)) self.owner = dict(owner='tab-a', account=dict(uid='123', name='测试账号'), extension={'idle': True}, url='https://www.dww11.com/') def tearDown(self): self.tmp.cleanup() def begin(self): self.store.enqueue(self.article) self.store.control('start') return self.store.next(self.owner)['job'] def transition(self, job, state, **extra): return self.store.event(dict(owner='tab-a', id=job['id'], state=state, epoch=self.store.run['epoch'], **extra)) def submit(self, job): for state in ('importing', 'formatting', 'awaiting_user_review'): self.transition(job, state) self.transition(job, 'user_submitting', snapshot={'title': job['title']}) def accept(self, job): return self.store.accept(dict(owner='tab-a', id=job['id'], post_url='https://www.dww11.com/thread/12345.html', thread_id='123', first_post_pid='456', author_uid='123', fid=125, title=job['title'], state='accepted_pending_moderation', evidence={'thread_id': '123', 'test': 'fixture only'})) def prompt_data(self, job): current = self.store.state()['jobs'][0] message = '新主题需要审核，您的帖子通过审核后才能显示' return dict(owner='tab-a', id=job['id'], attempt=current.get('attempt'), state='accepted_pending_moderation', evidence=dict(adapter='moderation_prompt', message=message, observed_text=message + '，现在将转入首页', observed_url='https://www.dww11.com/forum.php?topicsubmit=yes')) def test_prompt_accept_counts_once_and_persists_dedup(self): job = self.begin() self.submit(job) data = self.prompt_data(job) self.store.accept(data) self.assertTrue(self.store.accept(data)['duplicate']) state = self.store.state() self.assertEqual(state['run']['batch']['count'], 1) self.assertIsNone(state['run']['active']) self.assertTrue(state['run']['return_home']) self.assertTrue(state['run']['running']) saved = state['jobs'][0] self.assertEqual(saved['state'], 'accepted_pending_moderation') self.assertEqual(saved.get('post_url'), '') self.assertNotIn('thread_id', saved) restarted = Store(self.root / 'queue.sqlite3') self.assertIn('已受理', restarted.enqueue(self.article)) def test_prompt_requires_manual_submission_and_current_attempt(self): job = self.begin() with self.assertRaisesRegex(ValueError, '人工提交'): self.store.accept(self.prompt_data(job)) self.submit(job) data = self.prompt_data(job) for changed in ({'attempt': 'old'}, {'evidence': {**data['evidence'], 'observed_text': '(审核中)'}}, {'evidence': {**data['evidence'], 'observed_url': 'https://example.com/'}}): with self.assertRaises(ValueError): self.store.accept({**data, **changed}) self.assertEqual(self.store.run['batch']['count'], 0) def test_prompt_after_stop_records_without_resuming(self): job = self.begin() self.submit(job) data = self.prompt_data(job) self.store.control('stop') self.store.accept(data) self.assertFalse(self.store.run['running']) self.assertEqual(self.store.run['batch']['count'], 1) def test_stable_identity_and_original_preserved(self): before = self.md.read_bytes() item = inspect_article(self.article) copy = self.root / 'renamed' shutil.copytree(self.article, copy) (copy / 'a.md').write_text(self.md.read_text().replace('测试新闻', '改标题')) self.assertEqual(inspect_article(copy)['article_key'], item['article_key']) self.assertEqual(self.md.read_bytes(), before) self.assertEqual(item['source'], '测试来源') def test_title_does_not_break_emoji(self): short = short_title('😀' * 50) self.assertTrue(short.endswith('...')) self.assertLessEqual(len(short.encode('utf-8')), 78) self.assertEqual(short_title('短标题'), '短标题') def test_existing_queued_title_updates_before_import_only(self): original = '中' * 30 self.md.write_text('# ' + original + '\n\n正文。\n\n来源：测试，如有侵权，请联系删除。') self.store.enqueue(self.article) with self.store.db() as db: job = self.store.state()['jobs'][0] job['title'] = original # Simulate a queue created by the old version. self.store._save(db, job) self.store.control('start') job = self.store.next(self.owner)['job'] self.assertEqual(job['title'], '中' * 25 + '...') self.assertEqual(self.store.package(dict(owner='tab-a', id=job['id']))['title'], job['title']) self.submit(job) restarted = Store(self.root / 'queue.sqlite3') self.assertEqual(restarted.state()['jobs'][0]['snapshot']['title'], job['title']) self.assertEqual(restarted.state()['jobs'][0]['original_title'], original) def test_unsupported_images_stop_before_queue(self): (self.article / 'a.webp').write_bytes(b'webp') with self.assertRaisesRegex(ValueError, '暂不支持'): self.store.enqueue(self.article) self.assertEqual(self.store.state()['jobs'], []) def test_missing_and_outside_image_rejected(self): for image in ('missing.png', '../outside.png', 'https://example.com/a.png'): self.md.write_text('# 标题\n\n正文 ![](' + image + ')\n\n来源：测试，如有侵权，请联系删除。') with self.assertRaises(ValueError): self.store.enqueue(self.article) def test_no_duplicate_queue_or_cross_tab_claim(self): job = self.begin() self.assertIn('已在队列', self.store.enqueue(self.article)) self.assertEqual(self.store.next({**self.owner, 'owner': 'tab-b'})['action'], 'other_tab') self.assertEqual(self.store.next(self.owner)['job']['id'], job['id']) def test_stop_invalidates_old_continuation(self): job = self.begin() epoch = self.store.run['epoch'] self.store.control('stop') self.store.control('start') with self.assertRaisesRegex(ValueError, '过期'): self.store.event(dict(owner='tab-a', id=job['id'], state='importing', epoch=epoch)) def test_restart_pauses_and_preserves_claim(self): job = self.begin() restarted = Store(self.root / 'queue.sqlite3') self.assertFalse(restarted.run['running']) self.assertEqual(restarted.run['active'], job['id']) self.assertEqual(restarted.run['batch']['target'], 1) def test_accept_requires_manual_submission(self): job = self.begin() with self.assertRaisesRegex(ValueError, '人工提交'): self.accept(job) def test_accept_is_atomic_and_idempotent(self): job = self.begin() self.submit(job) self.accept(job) self.accept(job) self.assertEqual(self.store.run['batch']['count'], 1) self.assertIn('已受理', self.store.enqueue(self.article)) with self.assertRaisesRegex(ValueError, '已受理'): self.store.resolve(dict(id=job['id'], confirmed_not_accepted=True, retry=True)) def test_unknown_never_requeues_automatically(self): job = self.begin() self.submit(job) self.transition(job, 'submit_unknown') self.assertEqual(self.store.next(self.owner)['job']['state'], 'submit_unknown') self.assertEqual(self.store.run['batch']['count'], 0) self.assertIsNone(self.store.run['switch']) def test_migrate_real_ledger_shape(self): record = dict(markdown_sha256=hashlib.sha256(self.md.read_bytes()).hexdigest(), block_repost=True, markdown_path=str(self.md), status='accepted_pending_moderation', post_id='12345') self.store.import_history([record]) self.assertIn('已受理', self.store.enqueue(self.article)) def test_switch_only_after_target_and_with_more_work(self): job = self.begin() self.submit(job) self.accept(job) other = self.root / 'other';shutil.copytree(self.article, other) (other / 'a.md').write_text(self.md.read_text().replace('正文内容', '另一篇内容')) self.store.enqueue(other) self.assertEqual(self.store.next(self.owner)['action'], 'switch') self.assertEqual(self.store.next(self.owner)['action'], 'switch') self.store.switch_begin({'owner': 'tab-a'}) with self.assertRaisesRegex(ValueError, '禁止重复'): self.store.switch_begin({'owner': 'tab-a'}) self.assertEqual(self.store.next(self.owner)['action'], 'switch_wait') next_job = self.store.next({**self.owner, 'account': {'uid': '456', 'name': '新测试账号'}}) self.assertEqual(next_job['action'], 'prepare') self.assertEqual(self.store.run['batch']['count'], 0) def test_no_switch_when_queue_done(self): job = self.begin();self.submit(job);self.accept(job) self.assertEqual(self.store.next(self.owner)['action'], 'idle') self.assertIsNone(self.store.run['switch']) def test_skip_preserves_history_and_continues_next_without_counting(self): job = self.begin() other = self.root / 'other' shutil.copytree(self.article, other) (other / 'a.md').write_text(self.md.read_text().replace('正文内容', '下一篇内容')) self.store.enqueue(other) self.transition(job, 'importing') self.transition(job, 'formatting') self.transition(job, 'awaiting_user_review') result = self.store.skip(dict(owner='tab-a', id=job['id'], epoch=self.store.run['epoch'])) self.assertEqual(result['fid'], 125) self.assertEqual(self.store.run['batch']['count'], 0) self.assertEqual(self.store.state()['history_count'], 0) self.assertEqual(next(j for j in self.store.state()['jobs'] if j['id'] == job['id'])['state'], 'skipped') self.assertIn('已在队列', self.store.enqueue(self.article)) self.assertEqual(self.store.next(self.owner)['action'], 'prepare') self.assertNotEqual(self.store.run['active'], job['id']) restarted = Store(self.root / 'queue.sqlite3') self.assertEqual(next(j for j in restarted.state()['jobs'] if j['id'] == job['id'])['state'], 'skipped') def test_skip_last_article_finishes_queue(self): job = self.begin() self.store.skip(dict(owner='tab-a', id=job['id'], epoch=self.store.run['epoch'])) self.assertEqual(self.store.next(self.owner)['action'], 'idle') self.assertFalse(self.store.run['running']) self.assertIsNone(self.store.run['switch']) def test_skip_rejects_other_tab_stale_request_and_duplicate(self): job = self.begin() data = dict(owner='tab-a', id=job['id'], epoch=self.store.run['epoch']) with self.assertRaisesRegex(ValueError, '另一个'): self.store.skip({**data, 'owner': 'tab-b'}) self.store.control('stop') with self.assertRaisesRegex(ValueError, '状态已改变'): self.store.skip(data) data['epoch'] = self.store.run['epoch'] self.store.skip(data) with self.assertRaises(ValueError): self.store.skip(data) def test_skip_resumes_unsubmitted_error_but_rejects_submitted_errors(self): job = self.begin() self.transition(job, 'needs_attention', error='上传失败') self.store.skip(dict(owner='tab-a', id=job['id'], epoch=self.store.run['epoch'])) self.assertTrue(self.store.run['running']) self.assertEqual(self.store.next(self.owner)['action'], 'idle') def test_submitted_article_cannot_be_skipped(self): job = self.begin() self.submit(job) for state in ('user_submitting', 'submit_unknown', 'needs_attention'): if state != 'user_submitting': self.transition(job, state) with self.assertRaisesRegex(ValueError, '已提交'): self.store.skip(dict(owner='tab-a', id=job['id'], epoch=self.store.run['epoch'])) self.assertEqual(self.store.run['active'], job['id']) def test_real_tid_and_return_home_gate(self): job = self.begin(); self.submit(job); self.accept(job) saved = self.store.state()['jobs'][0] self.assertEqual(saved['thread_id'], '123') self.assertEqual(saved['first_post_pid'], '456') self.assertEqual(self.store.next({**self.owner, 'url': saved['post_url']})['action'], 'home') self.assertEqual(self.store.next(self.owner)['action'], 'idle') def test_integrity_issue_records_acceptance_but_stops(self): job = self.begin(); self.submit(job) self.store.accept(dict(owner='tab-a', id=job['id'], post_url='https://www.dww11.com/thread/12311.html', thread_id='123', author_uid='123', fid=125, title=job['title'], state='accepted_pending_moderation', evidence={'thread_id':'123','integrity_issues':['正文图片位置改变']})) self.assertFalse(self.store.run['running']) self.assertEqual(self.store.run['batch']['count'], 1) self.assertIn('已受理', self.store.enqueue(self.article)) def test_historical_reconciliation_does_not_count_or_resume(self): job = self.begin(); self.submit(job); self.store.control('stop') data=dict(id=job['id'], article_key=job['article_key'], confirmed=True, post_url='https://www.dww11.com/thread/12311.html',thread_id='123',author_uid='123',fid=125, title=job['title'],state='accepted_pending_moderation',evidence={'thread_id':'123'}) self.store.reconcile(data); self.store.reconcile(data) self.assertFalse(self.store.run['running']) self.assertEqual(self.store.run['batch']['count'], 0) self.assertIn('已受理', self.store.enqueue(self.article)) def test_transaction_rolls_back_in_memory_run(self): import sqlite3 self.begin() before = json.loads(json.dumps(self.store.run)) with self.assertRaises(sqlite3.IntegrityError): with self.store.db() as db: self.store.run['batch']['count'] = 999 db.execute("INSERT INTO posts VALUES(NULL,NULL)") self.assertEqual(self.store.run, before) def test_accept_after_stop_only_records_no_resumption(self): job = self.begin(); self.submit(job); self.store.control('stop'); self.accept(job) self.assertFalse(self.store.run['running']) self.assertEqual(self.store.run['batch']['count'], 1) self.assertEqual(self.store.next(self.owner)['action'], 'idle') def test_switch_same_uid_and_idle_stops_without_reclick(self): import time job = self.begin(); self.submit(job); self.accept(job) other = self.root / 'other'; shutil.copytree(self.article, other) (other / 'a.md').write_text(self.md.read_text().replace('正文内容', '其他内容')) self.store.enqueue(other) self.assertEqual(self.store.next(self.owner)['action'], 'switch') self.store.switch_begin({'owner':'tab-a'}) self.store.run['switch']['clicked_at'] = time.time() - 11 self.assertEqual(self.store.next(self.owner)['action'], 'idle') self.assertFalse(self.store.run['running']) self.assertEqual(self.store.run['switch']['state'], 'clicked') def test_rejected_submission_requires_manual_resolution(self): job = self.begin(); self.submit(job) self.transition(job, 'rejected', error='标题超限') self.assertFalse(self.store.run['running']) self.assertEqual(self.store.run['batch']['count'], 0) self.store.resolve(dict(id=job['id'],confirmed_not_accepted=True,retry=True)) saved = self.store.state()['jobs'][0] self.assertNotIn('submitted_at', saved) self.assertEqual(saved['state'], 'queued') def test_legacy_identity_is_preserved_without_guessing_tid(self): job = self.begin(); self.submit(job); self.accept(job) with self.store.db() as db: saved = self.store._job(db,job['id']); saved.pop('thread_id') self.store._save(db,saved) restarted = Store(self.root/'queue.sqlite3') saved = restarted.state()['jobs'][0] self.assertEqual(saved['legacy_url_id'],'123') self.assertTrue(saved['identity_needs_reconciliation']) self.assertFalse(restarted.run['running']) def test_remove_selected_preserves_files_and_survives_restart(self): self.store.enqueue(self.article) other = self.root / 'other' shutil.copytree(self.article, other) (other / 'a.md').write_text(self.md.read_text().replace('正文内容', '另一篇内容')) self.store.enqueue(other) jobs = self.store.state()['jobs'] result = self.store.remove({'ids': [jobs[0]['id'], jobs[0]['id']]}) self.assertEqual(result['removed'], 1) self.assertTrue(self.md.is_file()) restarted = Store(self.root / 'queue.sqlite3') self.assertEqual([j['id'] for j in restarted.state()['jobs']], [jobs[1]['id']]) self.assertEqual(restarted.enqueue(self.article), '已入队') self.assertEqual(restarted.remove({'ids': [j['id'] for j in restarted.state()['jobs']]})['removed'], 2) self.assertEqual(restarted.state()['jobs'], []) def test_remove_is_atomic_and_protects_current_task(self): job = self.begin() with self.assertRaisesRegex(ValueError, '暂停'): self.store.remove({'ids': [job['id']]}) self.store.control('pause') with self.assertRaisesRegex(ValueError, '当前任务'): self.store.remove({'ids': [job['id']]}) self.store.resolve(dict(id=job['id'], confirmed_not_accepted=True, retry=False)) for ids in ([], 'invalid', [None], [job['id'], 'missing']): with self.assertRaises(ValueError): self.store.remove({'ids': ids}) self.assertEqual(len(self.store.state()['jobs']), 1) self.store.remove({'ids': [job['id']]}) self.assertEqual(self.store.state()['jobs'], []) def test_remove_accepted_retains_deduplication_and_counts(self): job = self.begin(); self.submit(job); self.accept(job) self.store.control('pause') before = self.store.state()['history_count'] self.store.remove({'ids': [job['id']]}) self.assertEqual(self.store.state()['jobs'], []) self.assertEqual(self.store.state()['history_count'], before) self.assertEqual(self.store.run['batch']['count'], 1) self.assertIn('已受理', self.store.enqueue(self.article)) with self.store.db() as db: self.assertEqual(db.execute('SELECT count(*) FROM posts').fetchone()[0], 1) def test_remove_unresolved_submission_is_blocked_even_without_active_pointer(self): job = self.begin(); self.submit(job); self.store.control('pause') self.store.run['active'] = None with self.assertRaisesRegex(ValueError, '尚未核对'): self.store.remove({'ids': [job['id']]}) self.assertEqual(len(self.store.state()['jobs']), 1) def test_bad_result_does_not_increment(self): job = self.begin();self.submit(job) with self.assertRaises(ValueError): self.store.accept(dict(owner='tab-a', id=job['id'], post_url='https://example.com/thread/12345.html')) self.assertEqual(self.store.run['batch']['count'], 0) if __name__ == '__main__': unittest.main()