tests/test_feeds.py
Rama master · réplica de lectura del repositorio original.
"""番号发现源：解析、两层去重、已读与忽略、空壳与真实资产的边界（ADR-0042）。"""
from __future__ import annotations
import sys
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
from peach import feeds # noqa: E402
from peach.migrations import upgrade # noqa: E402
from peach.repository import LedgerDatabase # noqa: E402
from peach.sources.dmm import ActressWorks, ListedWork # noqa: E402
MIGRATIONS = Path(__file__).resolve().parents[1] / "migrations"
JAVDB_ACTOR = """
<div class="movie-list h cols-4 vcols-8">
<div class="item">
<a href="/v/5nr8mp" class="box" title="ignored">
<div class="cover"><img src="x.jpg" /></div>
<div class="video-title"><strong>PBD-528</strong> 気高きお姉さん達</div>
<div class="score"><span class="value">4.0分</span></div>
<div class="meta">
2026-10-20
</div>
</a>
</div>
<div class="item">
<a href="/v/RkPb5z" class="box" title="ignored">
<div class="cover"><img src="y.jpg" /></div>
<div class="video-title"><strong>BBSS-106</strong> 別の作品</div>
<div class="meta">
2026-10-13
</div>
</a>
</div>
<div class="item">
<a href="/v/Zx9Qa1" class="box" title="ignored">
<div class="cover"><img src="z.jpg" /></div>
<div class="video-title"><strong></strong> 合集 30 部打包</div>
<div class="meta">
2026-10-01
</div>
</a>
</div>
</div>
"""
#: 旧艺名那一页：`/actors/MmbDR`（白石アイリ）2026-09-28 抓回来的标题栏与空列表。
JAVDB_EMPTY_ACTOR = """
<div class="columns is-desktop section-columns">
<div class="column section-title">
<h2 class="title is-4 has-text-justified">
<span class="actor-section-name">白石アイリ</span>
<br>
<span class="section-meta">0 部影片</span>
</h2>
</div>
</div>
<div class="empty-message">暂无内容</div>
"""
#: 本名那一页（`/actors/bkxd`）的标题栏：别名一行在前，作品数一行在后。
JAVDB_ACTOR_HEADER = """
<h2 class="title is-4 has-text-justified">
<span class="actor-section-name">森日向子</span>
<br>
<span class="section-meta">朝比奈えみり, ひー, ひなこちゃん, 雛子ちゃん, 日向結衣</span>
<br>
<span class="section-meta">1060 部影片</span>
</h2>
"""
def _database(root: Path) -> LedgerDatabase:
path = root / "ledger.db"
upgrade(path, MIGRATIONS)
return LedgerDatabase(path)
class FeedParsingTest(unittest.TestCase):
def test_code_is_found_inside_a_title_full_of_prose(self):
# 整段丢给 `release_code_from_text` 认不出来，词元扫描才认得出。
title = "HMN-071 新人 帶來超稀有妹子 戶川步[有碼高清中文字幕]"
self.assertEqual(feeds.scan_code(title), "HMN-071")
self.assertEqual(feeds.scan_code("[H265 1080p] DSOD-114 花宮京子"), "DSOD-114")
self.assertIsNone(feeds.scan_code("合集 30 部打包"))
def test_javdb_actor_page_reads_code_and_release_date(self):
parsed = feeds.parse_javdb_actor(JAVDB_ACTOR, "https://javdb.com/actors/pRMq")
self.assertEqual([entry.item_key for entry in parsed.entries],
["/v/5nr8mp", "/v/RkPb5z", "/v/Zx9Qa1"])
self.assertEqual(parsed.entries[0].code, "PBD-528")
self.assertEqual(parsed.entries[0].link, "https://javdb.com/v/5nr8mp")
self.assertEqual(parsed.entries[0].published_at, "2026-10-20T00:00:00.000Z")
self.assertIsNone(parsed.entries[2].code)
def test_an_unknown_kind_is_a_parse_failure(self):
# 账本里若留着别的类型，拉取把原因写在那一行上，不拿演员页的解析器硬读。
self.assertIsNone(feeds.parse("rss", JAVDB_ACTOR.encode("utf-8"),
"https://example.test/rss"))
def test_a_page_without_any_work_is_a_failure_not_an_empty_poll(self):
# 带查询串的演员页会回一份不含作品的页面。当成「这次没有新作」会让一个坏掉的
# 订阅永远不报错。
self.assertIsNone(feeds.parse_javdb_actor("<html>nothing</html>",
"https://javdb.com/actors/pRMq"))
def test_a_page_that_says_it_has_no_work_is_a_quiet_source(self):
# 旧艺名页常年是空的；页面自报 0 部，它就是一个没有作品的真实源。
parsed = feeds.parse_javdb_actor(JAVDB_EMPTY_ACTOR, "https://javdb.com/actors/MmbDR")
self.assertEqual((parsed.title, parsed.entries), ("白石アイリ", ()))
def test_a_page_that_claims_works_but_lists_none_is_still_a_failure(self):
self.assertIsNone(feeds.parse_javdb_actor(
JAVDB_ACTOR_HEADER + '<div class="empty-message">暂无内容</div>',
"https://javdb.com/actors/bkxd"))
def test_the_page_name_is_read_from_the_title_bar(self):
parsed = feeds.parse_javdb_actor(JAVDB_ACTOR_HEADER + JAVDB_ACTOR,
"https://javdb.com/actors/bkxd")
self.assertEqual(parsed.title, "森日向子")
self.assertEqual(len(parsed.entries), 3)
class CompilationTest(unittest.TestCase):
"""标题与名单取自一份订阅实际抓回的壳。"""
def test_marked_titles_and_long_cast_lists_are_group_compilations(self):
cast = "、".join(f"女優{index}" for index in range(8))
for title, performers in (
("ひくひくアナル丸見え！デカ尻バック激ピストンBEST", "奏音かのん、丘えりな"),
("乱交ベスト", ""),
("主観手コキ50連発", ""),
("絶対忠実秘書 BEST 8時間 Vol.01", "八掛うみ、野々浦暖"),
("季刊性年KMP 花火 ノーカット 1397分", ""),
("レス界隈奥さん他人棒で淫乱覚醒する巨乳人妻", cast)):
with self.subTest(title=title):
self.assertEqual(feeds.compilation_kind(title, performers), feeds.GROUP_COMPILATION)
def test_one_listed_performer_with_a_marked_title_is_a_solo_compilation(self):
for title in ("涼森れむ 8時間 BEST PRESTIGE PREMIUM RESTRICTED vol.12",
"涼森れむ 480分 総集編"):
with self.subTest(title=title):
self.assertEqual(feeds.compilation_kind(title, "涼森れむ"), feeds.SOLO_COMPILATION)
def test_a_few_co_stars_and_ordinary_titles_are_not_compilations(self):
for title, performers in (
("初めてのセックスが大優勝 ～神女優で童貞卒業～", "鈴村あいり、涼森れむ、八掛うみ"),
("月刊ハメ撮り 本能剥き出し3本番 涼森れむ", "涼森れむ"),
("BESTIE と過ごす2時間", ""),
(None, None)):
with self.subTest(title=title):
self.assertIsNone(feeds.compilation_kind(title, performers))
def test_the_highlight_studio_is_an_excerpt_whatever_the_title_says(self):
self.assertEqual(feeds.compilation_kind("愛する妻が…", "篠田ゆう", "ハイライト"),
feeds.EXCERPT)
self.assertEqual(feeds.compilation_kind("涼森れむ 480分 総集編", "涼森れむ", "ハイライト"),
feeds.EXCERPT)
self.assertIsNone(feeds.compilation_kind("愛する妻が…", "篠田ゆう", "マドンナ"))
class FeedPhaseTest(unittest.TestCase):
"""下次拉取按源的相位排（ADR-0134）：同时加的源错开，跳与跳之间的延迟不累积。"""
MOMENT = datetime(2026, 11, 3, 3, 28, 57, tzinfo=timezone.utc)
PERIOD = timedelta(minutes=feeds.DEFAULT_INTERVAL_MINUTES)
def test_sources_fetched_together_spread_over_the_window(self):
upcoming = sorted(feeds.next_fetch(source_id, self.MOMENT, feeds.DEFAULT_INTERVAL_MINUTES)
for source_id in range(1, 12))
for moment in upcoming:
self.assertGreaterEqual(moment, self.MOMENT + self.PERIOD / 2)
self.assertLessEqual(moment, self.MOMENT + self.PERIOD * 1.5)
slots = {int((moment.timestamp() % self.PERIOD.total_seconds())
// timedelta(minutes=feeds.TICK_MINUTES).total_seconds()) for moment in upcoming}
# 11 个源落进 12 个半小时档里的至少 9 个：一跳最多拉两三个，而不是 11 个一起。
self.assertGreaterEqual(len(slots), 9)
def test_tick_lag_does_not_drift_the_schedule(self):
"""到期之后要等到下一跳才拉，拉完排的下次仍回到同一个相位，周期保持一个间隔。"""
moment = self.MOMENT
due = feeds.next_fetch(7, moment, feeds.DEFAULT_INTERVAL_MINUTES)
for lag in (timedelta(minutes=29), timedelta(minutes=3), timedelta(minutes=17)):
fetched = due + lag
following = feeds.next_fetch(7, fetched, feeds.DEFAULT_INTERVAL_MINUTES)
self.assertAlmostEqual((following - due).total_seconds(),
self.PERIOD.total_seconds(), places=2)
due = following
class FeedStoreTest(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.root = Path(self._tmp.name)
self.database = _database(self.root)
self.addCleanup(self._tmp.cleanup)
def _source(self, **kwargs) -> int:
with self.database.write_transaction() as connection:
return feeds.add_source(connection,
url=kwargs.pop("url", "https://javdb.com/actors/pRMq"),
**kwargs)
def _poll(self, source_id: int) -> dict:
parsed = feeds.parse_javdb_actor(JAVDB_ACTOR, "https://javdb.com/actors/pRMq")
with self.database.write_transaction() as connection:
source = connection.execute("SELECT * FROM feed_source WHERE id=?",
(source_id,)).fetchone()
return feeds.poll(connection, source, parsed)
def test_https_only(self):
with self.assertRaises(ValueError):
feeds.normalize_url("http://example.test/rss")
def test_same_url_twice_is_refused(self):
self._source()
with self.assertRaises(ValueError):
self._source()
def test_first_poll_records_every_entry_and_shells_the_coded_ones(self):
source_id = self._source()
outcome = self._poll(source_id)
self.assertEqual(outcome["seen"], 3)
self.assertEqual(outcome["fresh"], 3)
# 解不出番号的条目照样记一行，只是不建壳。
self.assertEqual(outcome["without_code"], 1)
self.assertEqual(sorted(code for _id, code in outcome["created"]),
["BBSS-106", "PBD-528"])
with self.database.read_connection() as connection:
self.assertEqual(connection.execute(
"SELECT count(*) FROM feed_item").fetchone()[0], 3)
def test_second_poll_of_the_same_document_creates_nothing(self):
source_id = self._source()
self._poll(source_id)
outcome = self._poll(source_id)
self.assertEqual(outcome["fresh"], 0)
self.assertEqual(outcome["created"], [])
def test_two_sources_reporting_one_code_share_a_single_shell(self):
first = self._source(url="https://javdb.com/actors/pRMq")
second = self._source(url="https://javdb.com/actors/d45k9")
self._poll(first)
outcome = self._poll(second)
# 源内那一层是各自的，所以第二个源仍然把三条都记成新条目……
self.assertEqual(outcome["fresh"], 3)
# ……但全局那一层只认番号，壳一个都不再建。
self.assertEqual(outcome["created"], [])
with self.database.read_connection() as connection:
self.assertEqual(connection.execute(
"SELECT count(*) FROM feed_discovery").fetchone()[0], 2)
self.assertEqual(connection.execute(
"SELECT count(*) FROM feed_item").fetchone()[0], 6)
def test_a_code_already_in_the_library_gets_no_shell(self):
with self.database.write_transaction() as connection:
connection.execute(
"INSERT INTO asset(location,path,name,medium,code) "
"VALUES('R','R:\\\\media\\\\pbd528.mp4','pbd528.mp4','video','pbd00528')")
source_id = self._source()
outcome = self._poll(source_id)
# 归一化之后 `pbd00528` 与 `PBD-528` 是同一部片。
self.assertEqual([code for _id, code in outcome["created"]], ["BBSS-106"])
def test_read_and_ignore_are_independent_and_do_not_touch_dedupe(self):
source_id = self._source()
created = self._poll(source_id)["created"]
discovery_id = created[0][0]
with self.database.write_transaction() as connection:
connection.execute("UPDATE feed_discovery SET read_at=? WHERE id=?",
(feeds.stamp(), discovery_id))
connection.execute("UPDATE feed_discovery SET ignored_at=? WHERE id=?",
(feeds.stamp(), discovery_id))
row = connection.execute(
"SELECT read_at,ignored_at FROM feed_discovery WHERE id=?",
(discovery_id,)).fetchone()
self.assertIsNotNone(row["read_at"])
self.assertIsNotNone(row["ignored_at"])
# 忽略之后再拉一轮，这条既不重建也不变回未忽略。
outcome = self._poll(source_id)
self.assertEqual(outcome["created"], [])
with self.database.read_connection() as connection:
row = connection.execute(
"SELECT ignored_at FROM feed_discovery WHERE id=?",
(discovery_id,)).fetchone()
self.assertIsNotNone(row["ignored_at"])
def test_settle_writes_the_next_time_on_every_path(self):
source_id = self._source()
with self.database.write_transaction() as connection:
feeds.settle(connection, source_id, error="来源回了 HTTP 503",
interval_minutes=60)
with self.database.read_connection() as connection:
row = connection.execute("SELECT * FROM feed_source WHERE id=?",
(source_id,)).fetchone()
self.assertEqual(row["last_error"], "来源回了 HTTP 503")
self.assertIsNotNone(row["next_fetch_at"])
self.assertGreater(row["next_fetch_at"], row["last_fetched_at"])
def test_disabled_sources_are_never_due(self):
source_id = self._source()
with self.database.write_transaction() as connection:
feeds.set_enabled(connection, source_id, False)
with self.database.read_connection() as connection:
self.assertEqual(feeds.due_sources(connection), [])
def test_the_longest_waiting_source_is_due_first(self):
first = self._source(url="https://javdb.com/actors/a1")
second = self._source(url="https://javdb.com/actors/a2")
fresh = self._source(url="https://javdb.com/actors/a3")
with self.database.write_transaction() as connection:
feeds.defer(connection, first, datetime(2026, 11, 3, 2, tzinfo=timezone.utc))
feeds.defer(connection, second, datetime(2026, 11, 3, 1, tzinfo=timezone.utc))
with self.database.read_connection() as connection:
due = [int(row["id"]) for row in feeds.due_sources(connection, "2026-11-03T03:00:00Z")]
self.assertEqual(due, [fresh, second, first])
def test_settle_after_a_cooldown_retries_at_its_end(self):
source_id = self._source()
until = datetime(2026, 11, 3, 3, 15, tzinfo=timezone.utc)
with self.database.write_transaction() as connection:
feeds.settle(connection, source_id, error="来源拒绝访问", interval_minutes=360,
now=datetime(2026, 11, 3, 3, tzinfo=timezone.utc), retry_at=until)
row = connection.execute("SELECT next_fetch_at FROM feed_source WHERE id=?",
(source_id,)).fetchone()
self.assertEqual(row[0], feeds.stamp(until))
def test_performer_names_match_existing_entities_only(self):
with self.database.write_transaction() as connection:
connection.execute(
"INSERT INTO entity(kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES('performer','深田えいみ',peach_normalize('深田えいみ'),?,?)",
(feeds.stamp(), feeds.stamp()))
with self.database.read_connection() as connection:
found = feeds.match_performers(connection, ["深田えいみ", "查无此人"])
self.assertEqual(len(found), 1)
def dmm_works(*rows) -> ActressWorks:
"""DMM 片单：每行（cid, 官方番号, 发售日, 预购, 出演）。"""
return ActressWorks(name="森日向子", total=len(rows), works=tuple(
ListedWork(cid=cid, title="作品タイトル", maker_code=code, release_date=date,
pre_order=pre_order, performers=tuple(cast), studio="ムーディーズ")
for cid, code, date, pre_order, cast in rows))
#: 2026-11-04 实测片单里的几种：预购的单体作品、和 JavDB 同一部、分卷、几十人的合集。
DMM_WORKS = dmm_works(
("mikr00133", "MIKR-133", "2026-11-13", True, ["森日向子"]),
("pbd00528", "PBD-528", "2026-10-20", False, ["森日向子", "共演者"]),
("h_068mxdlp00343", "MXDLP-0343", "2026-10-01", False, ["森日向子"]),
("h_1454jksr74802", "JKSR-748-02", "2026-09-01", False, ["森日向子"]),
("mizd00570", "MIZD-570", "2026-11-27", True, [f"女優{index}" for index in range(37)]),
)
class DmmActressFeedTest(unittest.TestCase):
"""DMM 按女优编号的片单当第二个发现源：番号写法、跨源去重、合集与补订（ADR-0135）。"""
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.addCleanup(self._tmp.cleanup)
self.database = _database(Path(self._tmp.name))
with self.database.write_transaction() as connection:
self.performer = self._performer(connection, "森日向子", [("r18dev", "1061586")])
@staticmethod
def _performer(connection, name, refs) -> int:
entity_id = int(connection.execute(
"INSERT INTO entity(kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES('performer',?,peach_normalize(?),?,?)",
(name, name, feeds.stamp(), feeds.stamp())).lastrowid)
for provider, external_id in refs:
connection.execute(
"INSERT INTO entity_external_ref(entity_id,provider,external_kind,external_id)"
" VALUES(?,?,'performer',?)", (entity_id, provider, external_id))
return entity_id
def _poll(self, source_id, parsed, hidden=feeds.DEFAULT_HIDDEN_COMPILATIONS) -> dict:
with self.database.write_transaction() as connection:
source = connection.execute("SELECT * FROM feed_source WHERE id=?", (source_id,)).fetchone()
return feeds.poll(connection, source, parsed, hidden)
def _add(self, url) -> int:
with self.database.write_transaction() as connection:
return feeds.add_source(connection, url=url, entity_id=self.performer)
def test_entries_carry_the_official_code_in_the_same_spelling_as_javdb(self):
parsed = feeds.parse_dmm_actress(DMM_WORKS)
self.assertEqual(parsed.title, "森日向子")
self.assertEqual([entry.code for entry in parsed.entries],
["MIKR-133", "PBD-528", "MXDLP-343", None, "MIZD-570"])
head = parsed.entries[0]
self.assertEqual((head.item_key, head.link, head.published_at[:10], head.title),
("mikr00133", "https://video.dmm.co.jp/av/content/?id=mikr00133",
"2026-11-13", "MIKR-133 作品タイトル"))
# 官方番号没问到时不回头从标题里找。
blank = feeds.parse_dmm_actress(dmm_works(("x1", "", "", False, [])))
self.assertIsNone(blank.entries[0].code)
def test_dmm_and_javdb_reporting_one_work_share_a_single_shell(self):
javdb = self._add("https://javdb.com/actors/pRMq")
self._poll(javdb, feeds.parse_javdb_actor(JAVDB_ACTOR, "https://javdb.com/actors/pRMq"))
dmm = self._add(feeds.dmm_actress_url("1061586"))
outcome = self._poll(dmm, feeds.parse_dmm_actress(DMM_WORKS))
# 预购的 MIKR-133 与 MXDLP-343 建壳；PBD-528 已有 JavDB 的壳；分卷认不出番号；合集收起。
self.assertEqual([code for _id, code in outcome["created"]], ["MIKR-133", "MXDLP-343"])
self.assertEqual((outcome["fresh"], outcome["without_code"], outcome["compilations"]), (5, 1, 1))
with self.database.read_connection() as connection:
self.assertEqual(connection.execute(
"SELECT count(*) FROM feed_discovery WHERE code='PBD-528'").fetchone()[0], 1)
self.assertEqual(connection.execute(
"SELECT count(*) FROM feed_item WHERE code='PBD-528'").fetchone()[0], 2)
shell = connection.execute(
"SELECT release_date,performers,studio FROM feed_discovery WHERE code='MIKR-133'").fetchone()
self.assertEqual(tuple(shell), ("2026-11-13", "森日向子", "ムーディーズ"))
# 合集开关打开时那一部照常建壳。
other = self._add(feeds.dmm_actress_url("1047128"))
shown = self._poll(other, feeds.parse_dmm_actress(DMM_WORKS), hidden=frozenset())
self.assertEqual([code for _id, code in shown["created"]], ["MIZD-570"])
def test_actress_ids_come_from_dmm_and_r18dev_evidence(self):
with self.database.write_transaction() as connection:
for provider, value in (("dmm", "1061586"), ("dmm", "1047128"), ("javdb", "bkxd"),
("avwikidb", "5555"), ("dmm", "not-a-number")):
connection.execute(
"INSERT OR IGNORE INTO entity_external_ref(entity_id,provider,external_kind,external_id)"
" VALUES(?,?,'performer',?)", (self.performer, provider, value))
ids = feeds.dmm_actress_ids(connection, self.performer)
self.assertEqual(ids, ["1047128", "1061586"])
self.assertEqual(feeds.dmm_actress_id(feeds.dmm_actress_url("1061586")), "1061586")
self.assertEqual(feeds.dmm_actress_id("https://javdb.com/actors/pRMq"), "")
self.assertEqual(feeds.kind_of(feeds.dmm_actress_url("1061586")), feeds.KIND_DMM_ACTRESS)
with self.assertRaises(ValueError):
feeds.dmm_actress_url("1061586&x=1")
def test_sync_adds_dmm_sources_only_for_followed_people_with_an_id(self):
with self.database.write_transaction() as connection:
nameless = self._performer(connection, "編号なし", [("javdb", "Zz9")])
# 有 DMM 编号但没订阅的人不补。
self._performer(connection, "未订阅", [("r18dev", "1000001")])
for entity_id, page in ((self.performer, "pRMq"), (nameless, "Zz9")):
feeds.add_source(connection, url=f"https://javdb.com/actors/{page}", entity_id=entity_id)
created = feeds.sync_dmm_sources(connection)
rows = connection.execute(
"SELECT kind,url,entity_id FROM feed_source WHERE id IN (%s)" % ",".join("?" * len(created)),
created).fetchall()
self.assertEqual([tuple(row) for row in rows],
[(feeds.KIND_DMM_ACTRESS, "https://video.dmm.co.jp/av/list/?actress=1061586",
self.performer)])
# 用户单独停掉的 DMM 那一条不会被重新打开，也不另建一条。
with self.database.write_transaction() as connection:
feeds.set_enabled(connection, created[0], False)
self.assertEqual(feeds.sync_dmm_sources(connection), [])
self.assertTrue(feeds.entity_following(connection, self.performer))
feeds.unfollow_entity(connection, self.performer)
self.assertFalse(feeds.entity_following(connection, self.performer))
self.assertEqual(connection.execute(
"SELECT count(*) FROM feed_source WHERE enabled=1").fetchone()[0], 1)
class ShellBoundaryTest(unittest.TestCase):
"""壳不进 asset，所以整理、抽帧与馆藏统计看不到它（ADR-0042 第二条）。"""
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.database = _database(Path(self._tmp.name))
self.addCleanup(self._tmp.cleanup)
def _poll_once(self, connection) -> int:
source_id = feeds.add_source(connection, url="https://javdb.com/actors/pRMq")
feeds.poll(connection, connection.execute(
"SELECT * FROM feed_source WHERE id=?", (source_id,)).fetchone(),
feeds.parse_javdb_actor(JAVDB_ACTOR, "https://javdb.com/actors/pRMq"))
return source_id
def test_a_shell_adds_no_asset_row(self):
with self.database.write_transaction() as connection:
self._poll_once(connection)
with self.database.read_connection() as connection:
self.assertEqual(connection.execute(
"SELECT count(*) FROM asset").fetchone()[0], 0)
self.assertGreater(connection.execute(
"SELECT count(*) FROM feed_discovery").fetchone()[0], 0)
def test_the_shell_table_is_not_an_asset_reference(self):
# 物理删除资产时要清的引用名单里没有它：壳不引用 asset，两者只按番号对上。
from peach.web_batch import ASSET_REFERENCE_TABLES
self.assertNotIn("feed_discovery", ASSET_REFERENCE_TABLES)
self.assertNotIn("feed_item", ASSET_REFERENCE_TABLES)
def test_deleting_a_source_takes_its_items_but_leaves_the_shell(self):
# 壳是「这个番号还没入库」，它不该随着某个订阅被删而消失。连接和服务一样不开
# 外键：子表靠 `remove_source` 自己处理，不靠建表时的 CASCADE 声明。
with self.database.write_transaction() as connection:
feeds.remove_source(connection, self._poll_once(connection))
with self.database.read_connection() as connection:
self.assertEqual(connection.execute(
"SELECT count(*) FROM feed_item").fetchone()[0], 0)
self.assertEqual(tuple(connection.execute(
"SELECT count(*), count(source_id) FROM feed_discovery").fetchone()), (2, 0))
self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), [])
if __name__ == "__main__": # pragma: no cover
unittest.main()