tests/test_jav_discovery_metadata.py
Rama master · réplica de lectura del repositorio original.
"""发现资料边车、固定批次、预算和失败状态的行为契约。"""
import tempfile
import json
import unittest
import threading
from pathlib import Path
from unittest.mock import patch
from peach.javdb_public import PublicAPI, PublicSourceUnavailable
from peach.jav_discovery_metadata import DiscoveryMetadata
from peach.web_jav_discovery import _metadata_response
class DiscoveryMetadataTests(unittest.TestCase):
def test_legacy_sidecar_identity_is_not_trusted_and_new_original_title_survives_restart(self):
self.path.write_text(json.dumps({'version': 1, 'records': {'MIKR-023': {
'movie': {'id': 'WrongFirst', 'number': 'MIKR-023'}, 'checked_at': self.now[0]}}}), encoding='utf8')
original = self.path.read_bytes()
old_job = self.path.with_suffix('.job.json')
old_job.write_text(json.dumps({'status': 'running', 'codes': ['MIKR-023'],
'completed_codes': ['MIKR-023']}), encoding='utf8')
old = DiscoveryMetadata(self.path, api=self.api, clock=lambda: self.now[0])
self.addCleanup(old.stop)
self.assertEqual(old.peek('MIKR-023')[0], {})
self.assertIsNone(old.job.snapshot())
old.remember('MIKR-023', {'id': 'XeYkdb', 'number': 'MIKR-023',
'title': '显示题', 'origin_title': 'Original 日本語', 'javdb_rule': 2})
restarted = DiscoveryMetadata(self.path, api=self.api, clock=lambda: self.now[0])
self.addCleanup(restarted.stop)
self.assertEqual(restarted.peek('MIKR-023')[0]['origin_title'], 'Original 日本語')
self.assertEqual(self.path.read_bytes(), original)
self.assertTrue(old_job.exists())
def test_restart_retains_batch_identity_budget_and_completed_codes(self):
self.cache.remember('EXM-001', {'number': 'EXM-001', 'tags': [], 'score': 4, 'duration': 120})
self.cache.job_path.write_text(json.dumps(dict(
javdb_rule=2, job_id='original', status='running', codes=['EXM-001', 'EXM-002'], refresh=True,
completed=1, completed_codes=['EXM-001'], attempted_codes=['EXM-001'], requests=3, bytes=120,
active_seconds=10)), encoding='utf-8')
with patch.object(self.api, 'movie', return_value={'number': 'EXM-002', 'tags': [], 'score': 4, 'duration': 120}) as movie:
restarted = DiscoveryMetadata(self.path, api=self.api, clock=lambda: self.now[0])
self.addCleanup(restarted.stop)
restarted.job.thread.join(5)
final = restarted.job.snapshot()
movie.assert_called_once_with('EXM-002', refresh=True)
self.assertEqual((final['job_id'], final['completed'], final['requests'], final['bytes']), ('original', 2, 3, 120))
def setUp(self):
temporary = tempfile.TemporaryDirectory()
self.addCleanup(temporary.cleanup)
self.path = Path(temporary.name).resolve() / "metadata.json"
self.now = [1000.0]
self.api = PublicAPI(sleeper=lambda _: None)
self.cache = DiscoveryMetadata(self.path, api=self.api, clock=lambda: self.now[0])
self.addCleanup(self.cache.stop)
def test_repeated_restart_keeps_checkpoint_before_the_next_item_finishes(self):
self.cache.job_path.write_text(json.dumps(dict(
javdb_rule=2, job_id='original', status='running', codes=['EXM-001', 'EXM-002'], refresh=True,
started_at=100, completed=1, completed_codes=['EXM-001'], attempted_codes=['EXM-001'],
failed_codes=[], requests=3, bytes=120, active_seconds=10)), encoding='utf-8')
with patch.object(DiscoveryMetadata, '_run_batch'):
first = DiscoveryMetadata(self.path, api=self.api, clock=lambda: self.now[0])
first.job.thread.join(5)
first.stop()
saved = json.loads(self.cache.job_path.read_text(encoding='utf-8'))
self.assertEqual(saved['completed_codes'], ['EXM-001'])
self.assertEqual(saved['started_at'], 100)
with patch.object(self.api, 'movie', return_value={'number': 'EXM-002', 'tags': [], 'score': 4, 'duration': 120}) as movie:
second = DiscoveryMetadata(self.path, api=self.api, clock=lambda: self.now[0])
self.addCleanup(second.stop)
second.job.thread.join(5)
movie.assert_called_once_with('EXM-002', refresh=True)
self.assertEqual(second.job.snapshot()['completed'], 2)
def test_sidecar_survives_restart_and_distinguishes_empty_unknown_and_stale(self):
movie = {"number": "EXM-001", "tags": [], "score": "4.23", "duration": 180,
"preview_video_url": "https://example.invalid/secret",
"actors": [{"name": "甲", "gender": 0, "avatar_url": "secret"}]}
self.cache.remember("EXM-001", movie)
restarted = DiscoveryMetadata(self.path, clock=lambda: self.now[0])
source, state = restarted.peek("EXM001")
self.assertEqual(state["metadata_state"], "ready")
self.assertEqual(state["metadata_fields"]["tags"], "empty")
self.assertNotIn("preview_video_url", source)
self.assertNotIn("avatar_url", source["actors"][0])
self.assertEqual(restarted.peek("EXM-002")[1]["metadata_fields"]["tags"], "unavailable")
self.now[0] += 86400
self.assertEqual(restarted.peek("EXM-001")[1]["metadata_state"], "stale")
self.cache.remember("EXM-002", error="资料未取得")
self.assertEqual(self.cache.peek("EXM-002")[1]["metadata_state"], "unavailable")
def test_fixed_batch_finishes_without_detail_open_or_polling_new_batch(self):
def movie_for(code, **_):
return {"number": code, "tags": [{"name": "制服"}], "score": 4, "duration": 120}
with patch.object(self.api, "movie", side_effect=movie_for) as movie:
self.cache.ensure([f"EXM-{index:03d}" for index in range(1, 105)])
self.cache.job.thread.join(5)
state = self.cache.job.snapshot()
self.assertEqual((state["status"], state["completed"], state["total"]), ("complete", 100, 100))
self.cache.ensure(["EXM-999"])
self.assertEqual(movie.call_count, 100)
self.assertEqual(self.cache.peek("EXM-001")[1]["metadata_state"], "ready")
self.cache.ensure(["EXM-001", "EXM-999"], refresh=True, job_id=state["job_id"])
self.cache.job.thread.join(5)
self.assertEqual(movie.call_count, 102)
def test_static_portraits_survive_restart_without_signed_addresses(self):
self.cache.remember('EXM-001', {'actors': [
{'id': 'Static', 'name': '甲', 'avatar_url': 'https://tp.spfcas.com/actor.jpg'},
{'id': 'Signed', 'name': '乙', 'avatar_url': 'https://tp.spfcas.com/actor.jpg?signature=example'},
{'id': 'Private', 'name': '丙', 'avatar_url': 'http://127.0.0.1/private'}]})
source, _ = DiscoveryMetadata(self.path).peek('EXM-001')
actors = source['actors']
self.assertEqual(actors[0]['avatar_image'], 'https://tp.spfcas.com/actor.jpg')
self.assertNotIn('avatar_image', actors[1])
self.assertTrue(actors[1]['has_avatar'])
self.assertFalse(actors[2]['has_avatar'])
self.assertNotIn('signature', self.cache.path.read_text(encoding='utf-8'))
def test_source_cooldown_stops_batch_and_preserves_failure(self):
def unavailable(code):
self.api.paused_until = self.api.clock() + 900
raise PublicSourceUnavailable("资料未取得")
with patch.object(self.api, "movie", side_effect=unavailable) as movie:
self.cache.ensure(["EXM-001", "EXM-002"])
self.cache.job.thread.join(5)
state = self.cache.job.snapshot()
self.assertEqual(state["status"], "stopped")
self.assertEqual(state["stop_reason"], "source_cooldown")
self.assertEqual(movie.call_count, 1)
self.assertEqual(self.cache.peek("EXM-001")[1]["metadata_state"], "unavailable")
def test_cancel_preserves_completed_records_and_requires_explicit_resume(self):
entered, release = threading.Event(), threading.Event()
def movie_for(code, **_):
entered.set()
release.wait(5)
return {"number": code, "tags": [], "score": 4, "duration": 120}
with patch.object(self.api, "movie", side_effect=movie_for) as movie:
self.cache.ensure(["EXM-001", "EXM-002"])
self.assertTrue(entered.wait(5))
thread = self.cache.job.thread
stopped = self.cache.cancel()
release.set()
thread.join(5)
self.assertEqual(stopped["stop_reason"], "cancelled")
self.assertEqual(self.cache.peek("EXM-001")[1]["metadata_state"], "ready")
self.cache.ensure(["EXM-001", "EXM-002"])
self.assertEqual(movie.call_count, 1)
self.cache.ensure(["EXM-001", "EXM-002"], refresh=True)
self.cache.job.thread.join(5)
self.assertEqual(movie.call_count, 3)
def test_finished_collection_rechecks_expired_new_work_and_keeps_mature_work(self):
self.now[0] = 1791892800.0
def movie_for(code):
return {'number': code, 'release_date': '2026-10-01' if code == 'EXM-001' else '2020-01-01',
'tags': [], 'score': 4, 'duration': 120}
with patch.object(self.api, 'movie', side_effect=movie_for) as movie:
self.cache.ensure(['EXM-001', 'EXM-002'])
self.cache.job.thread.join(5)
saved = self.cache.peek('EXM-002')[1]['metadata_checked_at']
self.cache.ensure(['EXM-999'])
self.assertEqual(movie.call_count, 2)
self.now[0] += 86400
self.cache.ensure(['EXM-999'])
self.cache.job.thread.join(5)
self.assertEqual(movie.call_count, 3)
self.assertEqual(movie.call_args.args, ('EXM-001',))
self.assertEqual(self.cache.peek('EXM-002')[1]['metadata_checked_at'], saved)
def test_source_cache_read_and_failed_refresh_do_not_renew_source_timestamp(self):
source = {'number': 'EXM-001', 'release_date': '2020-01-01', 'tags': [], 'score': 4, 'duration': 120}
self.cache.remember('EXM-001', source, checked_at=100)
self.now[0] = 2000
self.cache.remember('EXM-001', source, checked_at=100)
self.cache.remember('EXM-001', error='来源未取得')
self.assertEqual(self.cache.peek('EXM-001')[1]['metadata_checked_at'], 100)
def test_public_job_status_excludes_internal_errors_and_private_codes(self):
failure = {"job_id": "sample", "status": "failed", "error": "RuntimeError: https://example.invalid/?private=sample",
"codes": ["EXM-001"], "completed": 0, "total": 1, "stop_reason": ""}
with patch("peach.web_jav_discovery.discovery_metadata") as metadata:
metadata.return_value.current.return_value = failure
response = _metadata_response(None, {}, [], [], 60, "external")
self.assertEqual(response["metadata_job"]["message"], "作品资料未取得")
self.assertEqual(response["metadata_job"]["error"], "作品资料未取得")
self.assertNotIn("codes", response["metadata_job"])
self.assertNotIn("poll_after_ms", response)