tests/test_field_owners.py
Rama master · réplica de lectura del repositorio original.
"""真相字段的字段级归属与乐观并发。
这一组用例盯的是覆盖规则本身：谁能改谁写下的值，以及一次写入里有字段写不得时
剩下的字段还写不写得进去。判据必须在 SQL 那一层成立——调用方自觉检查的版本，
换一个调用点就失效了。
"""
from __future__ import annotations
import json
import sqlite3
import tempfile
import unittest
from pathlib import Path
from peach.field_owners import (
OWNED_FIELDS,
USER_MANUAL,
RevisionConflict,
auto_owner,
check_revision,
owner_label,
owner_of,
parse_owners,
review_owner,
script_owner,
write_owned_fields,
)
from peach.migrations import upgrade
from support.ledger import fresh_ledger
MIGRATIONS = Path(__file__).resolve().parents[1] / "migrations"
class FieldOwnerWriteTests(unittest.TestCase):
def setUp(self):
self.tmp = tempfile.TemporaryDirectory()
self.addCleanup(self.tmp.cleanup)
database = fresh_ledger(self.tmp.name)
self.connection = sqlite3.connect(database)
self.connection.row_factory = sqlite3.Row
self.addCleanup(self.connection.close)
for asset_id in (1, 2):
self.connection.execute(
"INSERT INTO asset(id,location,path,name,medium) "
"VALUES(?,'local',?,?,'video')",
(asset_id, f"R:\\media\\{asset_id}.mp4", f"{asset_id}.mp4"))
def _row(self, asset_id: int) -> sqlite3.Row:
return self.connection.execute(
"SELECT * FROM asset WHERE id=?", (asset_id,)).fetchone()
def test_a_write_records_the_owner_and_bumps_the_revision(self):
result = write_owned_fields(
self.connection, [1], {"release_date": "2026-09-11"}, auto_owner("r18dev"))
self.assertEqual((result.assets, result.written, result.refused),
(1, ("release_date",), ()))
row = self._row(1)
self.assertEqual(row["release_date"], "2026-09-11")
self.assertEqual(parse_owners(row["field_owners"]),
{"release_date": "auto:r18dev"})
self.assertEqual(row["mutation_revision"], 1)
def test_automatic_writers_cannot_touch_what_the_user_decided(self):
write_owned_fields(self.connection, [1], {"studio": "用户写的"}, USER_MANUAL)
result = write_owned_fields(
self.connection, [1], {"studio": "刮削给的"}, auto_owner("javbus"))
self.assertEqual(result.refused, ("studio",))
self.assertEqual(result.written, ())
row = self._row(1)
self.assertEqual(row["studio"], "用户写的")
self.assertEqual(owner_of(row["field_owners"], "studio"), USER_MANUAL)
def test_a_blocked_field_does_not_take_the_rest_of_the_write_with_it(self):
"""按字段判而不是按行判：一个字段被接管不该让同批其它无主字段也写不进去。"""
write_owned_fields(self.connection, [1], {"studio": "用户写的"}, USER_MANUAL)
result = write_owned_fields(
self.connection, [1],
{"studio": "刮削给的", "series": "某系列", "release_date": "2026-01-02"},
auto_owner("javbus"))
self.assertEqual(result.refused, ("studio",))
self.assertEqual(sorted(result.written), ["release_date", "series"])
row = self._row(1)
self.assertEqual((row["studio"], row["series"], row["release_date"]),
("用户写的", "某系列", "2026-01-02"))
self.assertEqual(parse_owners(row["field_owners"]), {
"studio": USER_MANUAL, "series": "auto:javbus",
"release_date": "auto:javbus"})
def test_the_user_overrides_any_automatic_owner(self):
write_owned_fields(self.connection, [1], {"code": "ABC-001"},
script_owner("apply_metadata_tags"))
result = write_owned_fields(self.connection, [1], {"code": "ABC-002"}, USER_MANUAL)
self.assertEqual(result.refused, ())
self.assertEqual(self._row(1)["code"], "ABC-002")
self.assertEqual(owner_of(self._row(1)["field_owners"], "code"), USER_MANUAL)
def test_approval_overrides_what_an_earlier_approval_wrote(self):
write_owned_fields(self.connection, [1], {"series": "旧系列"},
review_owner("r18dev"))
write_owned_fields(self.connection, [1], {"series": "新系列"},
review_owner("javbus"))
row = self._row(1)
self.assertEqual(row["series"], "新系列")
self.assertEqual(owner_of(row["field_owners"], "series"), "review:javbus")
self.assertEqual(row["mutation_revision"], 2)
def test_require_empty_refuses_a_field_that_already_has_a_value(self):
write_owned_fields(self.connection, [1], {"code": "ABC-001"},
auto_owner("r18dev"))
result = write_owned_fields(self.connection, [1], {"code": "ZZZ-999"},
"scan:filename", require_empty=True)
self.assertEqual(result.refused, ("code",))
self.assertEqual(self._row(1)["code"], "ABC-001")
def test_the_revision_only_counts_truth_field_changes(self):
write_owned_fields(self.connection, [1], {"studio": "同一个值"},
auto_owner("r18dev"))
write_owned_fields(self.connection, [1], {"studio": "同一个值"},
auto_owner("javbus"))
row = self._row(1)
self.assertEqual(row["mutation_revision"], 1)
# 取值没变也要把归属换成最后一个写入者：回答的是「现在这个值算谁的」。
self.assertEqual(owner_of(row["field_owners"], "studio"), "auto:javbus")
def test_a_stale_revision_is_refused_and_reports_the_current_one(self):
write_owned_fields(self.connection, [1], {"studio": "甲"}, USER_MANUAL)
with self.assertRaises(RevisionConflict) as caught:
write_owned_fields(self.connection, [1], {"studio": "乙"}, USER_MANUAL,
expected_revision=0)
self.assertEqual(caught.exception.expected, 0)
self.assertEqual(caught.exception.revisions, {1: 1})
self.assertEqual(self._row(1)["studio"], "甲")
def test_a_matching_revision_lets_the_write_through(self):
write_owned_fields(self.connection, [1], {"studio": "甲"}, USER_MANUAL,
expected_revision=0)
self.assertEqual(self._row(1)["studio"], "甲")
write_owned_fields(self.connection, [1], {"studio": "乙"}, USER_MANUAL,
expected_revision=1)
self.assertEqual(self._row(1)["studio"], "乙")
def test_check_revision_covers_paths_that_write_other_tables(self):
write_owned_fields(self.connection, [1], {"studio": "甲"}, USER_MANUAL)
check_revision(self.connection, [2], 0)
with self.assertRaises(RevisionConflict):
check_revision(self.connection, [1, 2], 0)
def test_one_statement_covers_every_targeted_asset(self):
result = write_owned_fields(
self.connection, [1, 2], {"catalog_title": "同番号两卷"}, review_owner("mgstage"))
self.assertEqual(result.assets, 2)
self.assertEqual(result.revisions, {1: 1, 2: 1})
for asset_id in (1, 2):
self.assertEqual(self._row(asset_id)["catalog_title"], "同番号两卷")
def test_a_null_owner_map_never_wipes_the_existing_owners(self):
"""`json_patch` 碰上 NULL 会把整个结果变成 NULL，基值与跳过分支都得给 `'{}'`。"""
write_owned_fields(self.connection, [1], {"studio": "用户写的"}, USER_MANUAL)
write_owned_fields(self.connection, [1],
{"studio": "刮削给的", "series": "某系列"},
auto_owner("javbus"))
stored = self._row(1)["field_owners"]
self.assertIsNotNone(stored)
self.assertEqual(json.loads(stored),
{"studio": USER_MANUAL, "series": "auto:javbus"})
class FieldOwnerVocabularyTests(unittest.TestCase):
def test_only_the_declared_truth_fields_can_be_written(self):
connection = sqlite3.connect(":memory:")
self.addCleanup(connection.close)
with self.assertRaises(ValueError):
write_owned_fields(connection, [1], {"play_count": 3}, USER_MANUAL)
def test_an_unknown_owner_category_is_refused(self):
connection = sqlite3.connect(":memory:")
self.addCleanup(connection.close)
with self.assertRaises(ValueError):
write_owned_fields(connection, [1], {"studio": "甲"}, "plugin:whatever")
with self.assertRaises(ValueError):
write_owned_fields(connection, [1], {"studio": "甲"}, "user")
def test_owner_identifiers_are_restricted_to_a_safe_shape(self):
self.assertEqual(review_owner("r18dev"), "review:r18dev")
self.assertEqual(auto_owner("aventertainment"), "auto:aventertainment")
self.assertEqual(script_owner("apply_metadata_tags"),
"script:apply_metadata_tags")
for bad in ("", "R18Dev", "has space", 'quote"'):
with self.assertRaises(ValueError):
review_owner(bad)
def test_a_broken_owner_map_reads_as_unowned(self):
self.assertEqual(parse_owners("not json"), {})
self.assertEqual(parse_owners(None), {})
self.assertEqual(parse_owners('{"play_count":"user:manual"}'), {})
self.assertEqual(owner_of('{"studio":"user:manual"}', "studio"), USER_MANUAL)
self.assertEqual(owner_of('{"studio":"user:manual"}', None), "")
def test_every_owned_field_is_a_real_asset_column(self):
database = Path(tempfile.mkdtemp()) / "ledger.db"
sqlite3.connect(database).close()
upgrade(database, MIGRATIONS)
connection = sqlite3.connect(database)
self.addCleanup(connection.close)
columns = {row[1] for row in connection.execute("PRAGMA table_info(asset)")}
self.assertTrue(set(OWNED_FIELDS) <= columns)
def test_owner_labels_say_who_wrote_it(self):
self.assertEqual(owner_label(USER_MANUAL), "手动填写")
self.assertEqual(owner_label("review:javbus"), "复核批准 javbus")
self.assertEqual(owner_label("auto:r18dev"), "r18dev 免复核落库")
self.assertEqual(owner_label("scan:filename"), "文件名推导")
self.assertEqual(owner_label("script:clean_names"), "维护脚本 clean_names")
self.assertEqual(owner_label(""), "")
if __name__ == "__main__":
unittest.main()