tests/test_follow_creators.py
Rama master · réplica de lectura del repositorio original.
"""关注作者建档（ADR-0096）：建谁、等谁、并进谁，链接从哪来，认人与按批撤回。"""
import contextlib
import io
import json
import sqlite3
import tempfile
import unittest
from datetime import datetime, timezone
from pathlib import Path
from peach import follow_creators, web_follow
from peach.follow_sources import FollowCandidate, SourceFetch
from peach.follow_store import FollowStore
from peach.web_contract import WebContract, dispatch_api_get, dispatch_api_post
from peach.web_entity import q_entity
from peach.follow_identity import author_key
from scripts import revert_auto_landing
from support.ledger import fresh_ledger
MOMENT = datetime(2026, 9, 1, tzinfo=timezone.utc)
class FollowCreatorTests(unittest.TestCase):
def setUp(self):
self.temporary = tempfile.TemporaryDirectory()
self.addCleanup(self.temporary.cleanup)
self.root = Path(self.temporary.name).resolve()
self.db = fresh_ledger(self.root)
self.contract = WebContract(
self.db, follow_sources_root=self.root / "sources",
follow_secrets_root=self.root / "secrets", follow_shared_root=self.root / "shared",
candidate_root=self.root / "generated", follow_state_root=self.root / "state")
def source(self, provider, ref, label, metadata=None, items=1):
with self.contract.database.write_transaction() as connection:
store = FollowStore(lambda: connection, sources_root=self.contract.follow_sources_root)
source_id = store.register(provider=provider, ref=ref, label=label,
url=f"https://{provider}.test/{ref}", metadata=metadata,
moment=MOMENT)
candidates = tuple(FollowCandidate(
provider=provider, external_id=f"{source_id}-{at}", title=f"{label} {source_id}-{at}",
url=f"https://{provider}.test/{ref}/{at}") for at in range(items))
store.record(source_id, SourceFetch(
provider=provider, ref=ref, request_url=f"https://{provider}.test/{ref}",
semantics="work", candidates=candidates, raw_body=b"<html/>"), moment=MOMENT)
return source_id
def creator(self, name):
with self.contract.database.write_transaction() as connection:
return int(connection.execute(
"INSERT INTO entity(kind,canonical_name,normalized_name,created_at,updated_at)"
" VALUES('creator',?,?,?,?)", (name, name.casefold(), "2026", "2026")).lastrowid)
def plan(self):
with self.contract.database.read_connection() as connection:
return {landing.key: landing for landing in follow_creators.plan(connection)}
def run_followup(self):
return follow_creators.run(self.contract, "follow-creator:all", None)
def read(self, sql, *args):
with self.contract.database.read_connection() as connection:
return [tuple(row) for row in connection.execute(sql, args)]
def jul3d(self):
self.source("kemono", "fanbox/12345", "Jul3D", items=2)
self.source("kemono", "patreon/777", "Jul3D")
self.source("f95zone", "123", "Jul3D Collection", metadata={
"author_key": "jul3d",
"official_links": [
{"service": "patreon", "handle": "jul3d", "url": "https://www.patreon.com/jul3d/posts"},
{"service": "twitter", "handle": "Jul3D_", "url": "http://twitter.com/Jul3D_?s=20"},
{"service": "discord", "handle": "abc", "url": "https://discord.gg/abc"},
]})
def test_a_new_author_becomes_a_creator_with_its_channels_and_socials(self):
self.jul3d()
landing = self.plan()["name:jul3d"]
self.assertEqual((landing.action, landing.name), ("create", "Jul3D"))
self.assertEqual(sorted(link["url"] for link in landing.links), [
"https://www.patreon.com/jul3d", "https://www.pixiv.net/users/12345", "https://x.com/Jul3D_"])
result = self.run_followup()
self.assertEqual((result["created"], result["bound"], result["links"]), (1, 3, 3))
entity_id, metadata = self.read(
"SELECT id,metadata_json FROM entity WHERE canonical_name='Jul3D'")[0]
self.assertEqual(json.loads(metadata)["source"], follow_creators.SOURCE)
self.assertEqual(sorted(self.read("SELECT link_kind,label FROM entity_link WHERE entity_id=?", entity_id)),
[("official", "Patreon"), ("social", "X"), ("social", "pixiv")])
self.assertEqual({row[0] for row in self.read("SELECT entity_id FROM follow_source")}, {entity_id})
# 重跑是一张空计划。
self.assertEqual(self.run_followup()["created"], 0)
authors = dispatch_api_get(self.contract, "/api/follow/authors", {})["items"]
self.assertEqual([(row["k"], row["key"], row["entity_id"]) for row in authors],
[("Jul3D", f"entity:{entity_id}", entity_id)])
# 资料页的读数与名册那一格数的是同一批更新。
page = q_entity(self.contract, {"kind": "creator", "name": "Jul3D"})
self.assertEqual((page["follow"]["key"], page["follow"]["n"], page["follow"]["held"]),
(f"entity:{entity_id}", authors[0]["n"], []))
def test_a_later_source_under_the_same_name_key_joins_the_existing_creator(self):
self.jul3d()
self.run_followup()
later = self.source("kemono", "fanbox/99999", "Jul3D")
entity_id = self.read("SELECT id FROM entity WHERE canonical_name='Jul3D'")[0][0]
landing = self.plan()["name:jul3d"]
self.assertEqual((landing.action, landing.entity_id), ("join", entity_id))
with self.contract.database.read_connection() as connection:
self.assertEqual(follow_creators.declare(connection)[0]["task_key"], "follow-creator")
self.run_followup()
self.assertEqual(self.read("SELECT entity_id FROM follow_source WHERE id=?", later), [(entity_id,)])
self.assertIn(("https://www.pixiv.net/users/99999",),
self.read("SELECT url FROM entity_link WHERE entity_id=?", entity_id))
def test_a_same_named_creator_is_only_offered_until_the_user_confirms(self):
held_id = self.creator("InitialA")
source_id = self.source("kemono", "fanbox/4242", "InitialA", items=3)
landing = self.plan()["name:initiala"]
self.assertEqual((landing.action, landing.entity_id), ("hold", held_id))
with self.contract.database.read_connection() as connection:
self.assertEqual(follow_creators.declare(connection), [])
self.run_followup()
self.assertEqual(self.read("SELECT entity_id FROM follow_source"), [(None,)])
authors = dispatch_api_get(self.contract, "/api/follow/authors", {})["items"]
self.assertEqual([(row["key"], row["held_by"]) for row in authors], [("name:initiala", "InitialA")])
follow = q_entity(self.contract, {"kind": "creator", "name": "InitialA"})["follow"]
self.assertEqual(follow["key"], "")
self.assertEqual([(group["key"], group["n"]) for group in follow["held"]], [("name:initiala", 3)])
written = dispatch_api_post(self.contract, "/api/follow/creator",
{"entity_id": held_id, "key": "name:initiala"})
self.assertEqual((written["bound"], written["links"]), (1, 1))
self.assertEqual(self.read("SELECT entity_id FROM follow_source WHERE id=?", source_id), [(held_id,)])
link_metadata = json.loads(self.read("SELECT metadata_json FROM entity_link")[0][0])
self.assertEqual(link_metadata["source"], follow_creators.CONFIRM_SOURCE)
follow = q_entity(self.contract, {"kind": "creator", "name": "InitialA"})["follow"]
self.assertEqual((follow["key"], follow["held"]), (f"entity:{held_id}", []))
with self.assertRaises(ValueError):
dispatch_api_post(self.contract, "/api/follow/creator", {"entity_id": held_id, "key": "name:initiala"})
def test_revert_unbinds_the_batch_and_keeps_creators_people_rely_on(self):
self.jul3d()
held_id = self.creator("InitialA")
self.source("kemono", "fanbox/4242", "InitialA")
self.run_followup()
dispatch_api_post(self.contract, "/api/follow/creator", {"entity_id": held_id, "key": "name:initiala"})
with contextlib.redirect_stdout(io.StringIO()):
revert_auto_landing.main(["--db", str(self.db), "--apply", "--backup", str(self.root / "backup.db"),
"--source", follow_creators.SOURCE])
self.assertEqual(self.read("SELECT canonical_name FROM entity WHERE kind='creator'"), [("InitialA",)])
# 人认过的那一组不归自动撤回。
self.assertEqual(sorted(row[0] or 0 for row in self.read("SELECT entity_id FROM follow_source")),
[0, 0, 0, held_id])
self.assertEqual(len(self.read("SELECT id FROM entity_link")), 1)
with contextlib.closing(sqlite3.connect(self.db)) as connection:
self.assertEqual(connection.execute("PRAGMA foreign_key_check").fetchall(), [])
def test_the_follow_check_declares_the_followup_only_while_someone_is_waiting(self):
self.jul3d()
result = web_follow._run_follow_check(self.contract, {"sources": []})
self.assertEqual([row["key"] for row in result["followups"]], ["follow-creator:all"])
self.run_followup()
self.assertEqual(web_follow._run_follow_check(self.contract, {"sources": []})["followups"], [])
def test_author_key_follows_the_binding(self):
self.jul3d()
self.run_followup()
with self.contract.database.read_connection() as connection:
rows = FollowStore(lambda: connection).sources()
self.assertEqual({author_key(row) for row in rows},
{f"entity:{rows[0]['entity_id']}"})
class SourceLinkTests(unittest.TestCase):
def row(self, provider, ref, metadata=None, url=""):
return {"provider": provider, "ref": ref, "url": url, "label": ref,
"metadata_json": json.dumps(metadata or {})}
def urls(self, row):
return [link["url"] for link in follow_creators.source_links(row)]
def test_archives_restore_only_the_account_shapes_that_were_checked(self):
self.assertEqual(self.urls(self.row("kemono", "fanbox/123")), ["https://www.pixiv.net/users/123"])
self.assertEqual(self.urls(self.row("coomer", "onlyfans/some.one_")), ["https://onlyfans.com/some.one_"])
self.assertEqual(self.urls(self.row("kemono", "fanbox/name")), [])
self.assertEqual(self.urls(self.row("kemono", "gumroad/123")), [])
def test_an_official_source_is_its_own_channel(self):
self.assertEqual(self.urls(self.row("subscribestar", "artist", url="https://subscribestar.adult/artist")),
["https://subscribestar.adult/artist"])
def test_booru_profile_links_count_only_when_the_handle_is_the_author(self):
row = self.row("rule34video", "artist", {"official_links": [
{"service": "twitter", "handle": "artist", "url": "https://x.com/artist"},
{"service": "twitter", "handle": "collaborator", "url": "https://x.com/collaborator"}]})
self.assertEqual(self.urls(row), ["https://x.com/artist"])
def test_subscribestar_keeps_the_host_from_the_profile(self):
row = self.row("f95zone", "1", {"official_links": [
{"service": "subscribestar", "handle": "artist", "url": "https://www.subscribestar.com/artist?x=1"}]})
self.assertEqual(self.urls(row), ["https://subscribestar.com/artist"])
if __name__ == "__main__":
unittest.main()