nodes/ComfyAssets-ComfyUI_PromptManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Merge branch 'work/332-vocab' into feat/332-image-categories-pr2

Browse Source
This commit is contained in:
Vito Sansevero
2026-10-27 18:29:54 -07:00
parent 883409f5c6 6a026d143f
commit f1be8db44c
4 changed files with 434 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
TESTING.md
+14
View File
@@ -5474,6 +5474,20 @@ rotates the epoch, so a pre-restore `expected_epoch` is refused with `storage-ch
Owner rows and their timestamps never change on a tag command; a category apply only
touches rows carrying the exact source.
Image categories (#332 §7, MC9, MC10, DC7): each `GET categories` row also carries
`image_count`, the stored `image_categories` rows holding that exact value (files
missing on disk included). A value prompts and images share appears once, ordered by
its oldest positive; a value only images hold appears once after every prompt-backed
value (`first_pid` 9223372036854775807 in the keyset cursor), ordered by value, with
`prompt_count` 0. `/library/options` keeps its own query and never lists image-only
values. Category preview and apply report `impact.images` (rows holding the source),
the target exists when an image row holds it, an image-only value can be renamed, the
apply renames the image rows in the same transaction (file keys and rowids kept, so a
merge cannot collide), and the impact token hashes the sorted image rowids, so an image
row gaining or losing the value between preview and apply is 409
`vocabulary-conflict` with nothing written. Every image-category statement reads
`idx_image_categories_category` (`test_category_reads_use_the_image_category_index`).
Two legacy routes change additively: `GET /tags` now lists unused registry tags, and
`/tags/suggest` without context appends unused names at count 0. Unused tags become
deletable only through the native Delete tag (vocabulary UI PR), not on the legacy
database/vocabulary.py
+55 -15
View File
@@ -13,7 +13,9 @@ same connection before commit. Owner rows and their timestamps never change.
Category consolidation is previewed in one ``library_snapshot`` and applied in
one ``_pair_transaction`` that recomputes the owner projection and requires the
preview's ``impact_token``; only rows carrying the exact source change.
preview's ``impact_token``; only rows carrying the exact source change. Image
rows holding the source (``image_categories``, #332) are owners too: they are
counted, hashed into the token and renamed with the prompts and pairs.
"""
import sqlite3
@@ -97,16 +99,30 @@ def _validate_tag_id(tag_id: object) -> None:
raise ValueError("Invalid tag ID")
# The /library/options universe: nonblank values on eligible parents or on
# active pairs, grouped exactly (whitespace and case variants stay distinct).
_CATEGORY_GROUPS = (
# Category values in Tags & categories (#332 §7): the prompt-backed values
# (nonblank values on eligible parents or on active pairs, grouped exactly, so
# whitespace and case variants stay distinct; /library/options lists the same
# prompt-backed set with its own query) plus the values only image rows hold.
# A prompt-backed value is ordered by its oldest owning positive; an image-only
# value sorts after all of them (first_pid is SQLite's INTEGER maximum, above
# any prompt id), then by value. ``image_count`` counts stored rows, including
# files missing on disk. No FULL OUTER JOIN, so it runs on SQLite 3.31+.
_IMAGE_ONLY_FIRST_PID = 2**63 - 1
_CATEGORY_ROWS = (
"WITH owners(value,pid) AS ("
"SELECT p.category,p.id FROM prompts p WHERE p.category IS NOT NULL "
"AND TRIM(p.category)!='' AND " + _ELIGIBLE + " "
"UNION SELECT pp.category,pp.positive_prompt_id FROM prompt_pairs pp "
"WHERE pp.archived=0 AND pp.category IS NOT NULL AND TRIM(pp.category)!=''),"
"grouped AS (SELECT value,MIN(pid) AS first_pid,"
"COUNT(DISTINCT pid) AS prompt_count FROM owners GROUP BY value) "
"image_values(value,image_count) AS ("
"SELECT category,COUNT(*) FROM image_categories GROUP BY category),"
"grouped(value,first_pid,prompt_count) AS ("
"SELECT value,MIN(pid),COUNT(DISTINCT pid) FROM owners GROUP BY value "
"UNION ALL SELECT value," + str(_IMAGE_ONLY_FIRST_PID) + ",0 FROM image_values "
"WHERE value NOT IN (SELECT value FROM owners)),"
"counted AS (SELECT g.value,g.first_pid,g.prompt_count,"
"COALESCE(i.image_count,0) AS image_count FROM grouped g "
"LEFT JOIN image_values i ON i.value=g.value) "
)
@@ -117,15 +133,23 @@ _CATEGORY_COMPONENTS = (
"WHERE p.category=? ORDER BY p.id"
)
_CATEGORY_PAIRS = "SELECT id,archived FROM prompt_pairs WHERE category=? ORDER BY id"
# Image rows holding one exact value (#332 DC7), an index search.
_IMAGE_CATEGORY_ROWIDS = (
"SELECT rowid FROM image_categories WHERE category=? ORDER BY rowid"
)
_CATEGORY_EXISTS = (
"SELECT EXISTS(SELECT 1 FROM prompts WHERE category=:v)"
" OR EXISTS(SELECT 1 FROM prompt_pairs WHERE category=:v)"
" OR EXISTS(SELECT 1 FROM image_categories WHERE category=:v)"
)
# Pairs after components, so a failing pair update rolls both back.
# Pairs after components and image rows last, so a failing update rolls every
# earlier one back. An image row is keyed by its file, never by its value, so
# merging into a value other files already hold cannot conflict.
_CATEGORY_UPDATES = (
"UPDATE prompts SET category=?,updated_at=? WHERE category=?",
"UPDATE prompt_pairs SET category=?,updated_at=? WHERE category=?",
"UPDATE image_categories SET category=?,updated_at=? WHERE category=?",
)
@@ -136,8 +160,15 @@ def category_impact_token(
target_exists: bool,
components: list,
pairs: list,
*,
images: list = (),
) -> str:
"""Hash a category projection like ``library_edit_token`` (no secret)."""
"""Hash a category projection like ``library_edit_token`` (no secret).
``images`` are the rowids of the image rows holding ``source`` (#332 DC7).
With none the preimage has no ``images`` key, so a prompt-only token keeps
the value it had before image categories existed.
"""
preimage = {
"epoch": epoch,
"source": source,
@@ -146,6 +177,8 @@ def category_impact_token(
"components": sorted([int(i), bool(flag)] for i, flag in components),
"pairs": sorted([int(i), bool(flag)] for i, flag in pairs),
}
if images:
preimage["images"] = sorted(int(i) for i in images)
return canonical_token(_CATEGORY_DOMAIN, preimage)
@@ -325,7 +358,8 @@ class VocabularyOperations:
pairs = [
[row[0], bool(row[1])] for row in conn.execute(_CATEGORY_PAIRS, (source,))
]
if not components and not pairs:
images = [row[0] for row in conn.execute(_IMAGE_CATEGORY_ROWIDS, (source,))]
if not components and not pairs and not images:
raise LibraryCommandError("vocabulary-conflict", epoch)
target_exists = bool(
conn.execute(_CATEGORY_EXISTS, {"v": target}).fetchone()[0]
@@ -339,9 +373,10 @@ class VocabularyOperations:
"other_prompts": len(components) - eligible,
"negative_variants": len(pairs) - archived,
"archived_negative_variants": archived,
"images": len(images),
},
"impact_token": category_impact_token(
epoch, source, target, target_exists, components, pairs
epoch, source, target, target_exists, components, pairs, images=images
),
}
@@ -365,7 +400,8 @@ class VocabularyOperations:
def apply_vocabulary_category(
self, *, expected_epoch: str, source: str, target: str, impact_token: str
) -> dict:
"""Set the exact ``source`` to ``target`` on every component and pair.
"""Set the exact ``source`` to ``target`` on every component, pair
and image row.
The recomputed projection must still match the preview's token; a
mismatch is ``vocabulary-conflict`` with nothing written.
@@ -422,15 +458,15 @@ class VocabularyOperations:
_limit(limit)
key = _category_after(after)
total = conn.execute(
_CATEGORY_GROUPS + "SELECT COUNT(*) FROM grouped"
_CATEGORY_ROWS + "SELECT COUNT(*) FROM counted"
).fetchone()[0]
where, params = "", []
if key is not None:
where = " WHERE first_pid>? OR (first_pid=? AND value>?)"
params = [key[0], key[0], key[1]]
rows = conn.execute(
_CATEGORY_GROUPS
+ "SELECT value,first_pid,prompt_count FROM grouped"
_CATEGORY_ROWS
+ "SELECT value,first_pid,prompt_count,image_count FROM counted"
+ where
+ " ORDER BY first_pid,value LIMIT ?",
[*params, limit + 1],
@@ -440,7 +476,11 @@ class VocabularyOperations:
return {
"library_epoch": epoch,
"categories": [
{"value": row["value"], "prompt_count": row["prompt_count"]}
{
"value": row["value"],
"prompt_count": row["prompt_count"],
"image_count": row["image_count"],
}
for row in rows
],
"total": total,
tests/test_api_vocabulary.py
+74 -2
View File
@@ -62,12 +62,16 @@ class TestVocabularyAPI(ImageAPITestCase):
},
)
self.assertEqual(
categories["categories"], [{"value": "Animal", "prompt_count": 1}]
categories["categories"],
[{"value": "Animal", "prompt_count": 1, "image_count": 0}],
)
_, rest = await self.get(
"categories", limit=1, cursor=categories["next_cursor"]
)
self.assertEqual(rest["categories"], [{"value": "Pet", "prompt_count": 1}])
self.assertEqual(
rest["categories"],
[{"value": "Pet", "prompt_count": 1, "image_count": 0}],
)
self.assertFalse(rest["has_more"])
async def test_vocabulary_cursor_is_signed_epoch_scoped(self):
@@ -488,6 +492,7 @@ class TestVocabularyAPI(ImageAPITestCase):
"other_prompts": 0,
"negative_variants": 1,
"archived_negative_variants": 0,
"images": 0,
}
self.assertEqual(
preview,
@@ -528,6 +533,73 @@ class TestVocabularyAPI(ImageAPITestCase):
self.assertEqual(response.status, 409)
self.assertEqual(replay["code"], "vocabulary-conflict")
async def test_category_rows_report_images_and_image_only_values(self):
self._save_prompt("Owner", category="Prompt value")
_, options_before = await self.get_options()
self.api.db.set_image_categories(
[("root", "a.png"), ("root", "b.png")], "Prompt value"
)
self.api.db.set_image_categories([("root", "c.png")], "Image value")
response, first = await self.get("categories", limit=1)
self.assertEqual(response.status, 200)
self.assertEqual(
first["categories"],
[{"value": "Prompt value", "prompt_count": 1, "image_count": 2}],
)
self.assertEqual(first["total"], 2)
_, rest = await self.get("categories", limit=1, cursor=first["next_cursor"])
self.assertEqual(
rest["categories"],
[{"value": "Image value", "prompt_count": 0, "image_count": 1}],
)
self.assertFalse(rest["has_more"])
# The Prompt library's category list does not gain image values.
_, options_after = await self.get_options()
self.assertEqual(options_after, options_before)
self.assertEqual(options_after["categories"], ["Prompt value"])
async def get_options(self):
response = await self.client.get("/prompt_manager/library/options")
return response, await response.json()
async def test_category_routes_rename_image_only_value(self):
epoch = self.api.db.get_execution_epoch()
self._save_prompt("Bystander", category="Kept")
self.api.db.set_image_categories(
[("root", "a.png"), ("second", "b.png")], "Image value"
)
prompts_before = self.category_rows()
body = {"expected_epoch": epoch, "source": "Image value", "target": "Moved"}
response, preview = await self.send("POST", "categories/preview", body)
self.assertEqual(response.status, 200, preview)
impact = {
"positive_prompts": 0,
"other_prompts": 0,
"negative_variants": 0,
"archived_negative_variants": 0,
"images": 2,
}
self.assertEqual(preview["impact"], impact)
stale = preview["impact_token"]
self.api.db.set_image_categories([("root", "late.png")], "Image value")
response, refused = await self.send(
"POST", "categories/apply", {**body, "impact_token": stale}
)
self.assertEqual(response.status, 409)
self.assertEqual(refused["code"], "vocabulary-conflict")
self.assertEqual(self.api.db.image_category("root", "a.png"), "Image value")
_, preview = await self.send("POST", "categories/preview", body)
response, applied = await self.send(
"POST",
"categories/apply",
{**body, "impact_token": preview["impact_token"]},
)
self.assertEqual(response.status, 200, applied)
self.assertEqual(applied["impact"], {**impact, "images": 3})
for key in (("root", "a.png"), ("second", "b.png"), ("root", "late.png")):
self.assertEqual(self.api.db.image_category(*key), "Moved")
self.assertEqual(self.category_rows(), prompts_before)
async def test_preview_validation(self):
epoch = self.api.db.get_execution_epoch()
self._save_prompt("Owner", category="Scenery")
tests/test_vocabulary.py
+291 -1
View File
@@ -13,6 +13,11 @@ from database.library_commands import LibraryCommandError
from database.operations import PromptDatabase
from utils.hashing import generate_prompt_hash
try:
from test_image_categories import category_checksum
except ImportError:
from tests.test_image_categories import category_checksum
class TestVocabularyReads(unittest.TestCase):
def setUp(self):
@@ -168,7 +173,8 @@ class TestVocabularyReads(unittest.TestCase):
self.assertEqual(page["total"], 4)
self.assertFalse(page["has_more"])
for row in page["categories"]:
self.assertEqual(set(row), {"value", "prompt_count"})
self.assertEqual(set(row), {"value", "prompt_count", "image_count"})
self.assertEqual(row["image_count"], 0)
self.assertEqual(
row["prompt_count"],
self.db.list_library(category=row["value"])["total"],
@@ -238,6 +244,111 @@ class TestVocabularyReads(unittest.TestCase):
with self.assertRaises(ValueError):
self.read("list_vocabulary_tags", **options)
def own(self, value, *paths, root="root"):
"""Give each relative path under ``root`` the own category ``value``."""
self.db.set_image_categories([(root, path) for path in paths], value)
def all_categories(self, limit=200):
rows, pages, after = [], [], None
while True:
page = self.read("list_vocabulary_categories", limit=limit, after=after)
pages.append(page)
rows.extend(page["categories"])
if not page["has_more"]:
return rows, pages
after = page["next_key"]
def test_category_list_includes_image_only_values_last(self):
self.save("Oldest", category="Zeta")
self.save("Newer", category="Alpha")
self.own("Zeta", "a.png", "b.png")
self.own("Only images B", "c.png")
self.own("Only images A", "d.png", "e.png", "gone.png")
rows, _pages = self.all_categories()
self.assertEqual(
rows,
[
{"value": "Zeta", "prompt_count": 1, "image_count": 2},
{"value": "Alpha", "prompt_count": 1, "image_count": 0},
{"value": "Only images A", "prompt_count": 0, "image_count": 3},
{"value": "Only images B", "prompt_count": 0, "image_count": 1},
],
)
page = self.read("list_vocabulary_categories")
self.assertEqual(page["total"], 4)
def test_category_cursor_crosses_into_image_only_values(self):
for index in range(3):
self.save(f"Owner {index}", category=f"P{index}")
self.own("P1", "shared.png")
self.own("I-b", "x.png")
self.own("I-a", "y.png")
self.own("I-c", "z.png", root="other")
for limit in (1, 2, 3, 4):
with self.subTest(limit=limit):
rows, pages = self.all_categories(limit=limit)
self.assertEqual(
[row["value"] for row in rows],
["P0", "P1", "P2", "I-a", "I-b", "I-c"],
)
self.assertEqual({page["total"] for page in pages}, {6})
self.assertEqual(
[row["image_count"] for row in rows], [0, 1, 0, 1, 1, 1]
)
first = self.read("list_vocabulary_categories", limit=3)
self.assertEqual(first["next_key"], [3, "P2"])
second = self.read("list_vocabulary_categories", limit=1, after=[3, "P2"])
self.assertEqual(second["next_key"], [2**63 - 1, "I-a"])
def test_image_rows_on_a_prompt_value_count_once(self):
self.save("Owner", category="Shared")
self.db.save_prompt_pair("Owner", "Noise", category="Shared")
self.own("Shared", *[f"{n}.png" for n in range(5)])
self.own("Shared", "elsewhere.png", root="second")
rows, _pages = self.all_categories()
self.assertEqual(
rows, [{"value": "Shared", "prompt_count": 1, "image_count": 6}]
)
def test_library_options_ignore_image_categories(self):
self.save("Owner", category="Prompt value")
before = self.db.get_library_options()
self.own("Image value", "a.png")
self.own("Prompt value", "b.png")
self.assertEqual(self.db.get_library_options(), before)
self.assertEqual(before["categories"], ["Prompt value"])
def test_category_reads_use_the_image_category_index(self):
from database import vocabulary
self.own("Value", "a.png")
statements = [
vocabulary._CATEGORY_ROWS + "SELECT COUNT(*) FROM counted",
vocabulary._CATEGORY_ROWS
+ "SELECT value,first_pid,prompt_count,image_count FROM counted "
"WHERE first_pid>? OR (first_pid=? AND value>?) "
"ORDER BY first_pid,value LIMIT ?",
]
plans = []
with self.db.model.get_connection() as conn:
for sql in statements:
params = [1] * sql.count("?")
plans.extend(
row[-1] for row in conn.execute("EXPLAIN QUERY PLAN " + sql, params)
)
for sql in (
vocabulary._IMAGE_CATEGORY_ROWIDS,
*vocabulary._CATEGORY_UPDATES,
):
params = [1] * sql.count("?")
plans.extend(
row[-1] for row in conn.execute("EXPLAIN QUERY PLAN " + sql, params)
)
image_steps = [step for step in plans if "image_categories" in step]
self.assertTrue(image_steps)
for step in image_steps:
self.assertIn("idx_image_categories_category", step, step)
def test_reads_share_one_snapshot_and_report_epoch(self):
self.save("Cat", category="Animal", tags=["cat"])
epoch = self.db.get_execution_epoch()
@@ -741,6 +852,7 @@ class TestVocabularyCategoryCommands(unittest.TestCase):
"other_prompts": 1,
"negative_variants": 2,
"archived_negative_variants": 1,
"images": 0,
},
)
self.assertFalse(result["target_exists"])
@@ -810,6 +922,7 @@ class TestVocabularyCategoryCommands(unittest.TestCase):
"other_prompts": 0,
"negative_variants": 0,
"archived_negative_variants": 0,
"images": 0,
},
)
self.assertEqual(self.rows("prompts")[variant]["category"], "Other")
@@ -902,6 +1015,171 @@ class TestVocabularyCategoryCommands(unittest.TestCase):
self.apply(bad)
self.assertEqual(self.snapshot(), before)
def own(self, value, *paths, root="root"):
self.db.set_image_categories([(root, path) for path in paths], value)
def image_rows(self):
with self.db.model.get_connection() as conn:
return {
(row[0], row[1]): (row[2], row[3], row[4])
for row in conn.execute(
"SELECT root_key,relative_path,category,updated_at,rowid "
"FROM image_categories ORDER BY rowid"
)
}
def checksum(self):
with self.db.model.get_connection() as conn:
return category_checksum(conn)
def test_preview_counts_image_rows(self):
self.owners()
self.own("Scenery", "a.png", "b.png", "missing.png")
self.own("Scenery", "c.png", root="second")
self.own("Other", "d.png")
before = self.image_rows()
result = self.preview()
self.assertEqual(result["impact"]["images"], 4)
self.assertEqual(self.image_rows(), before)
self.assertTrue(self.preview(target="Other")["target_exists"])
def test_apply_renames_image_categories(self):
self.owners()
bystander = self.save("Bystander", category="Other")
self.own("Scenery", "a.png", "b.png")
self.own("Scenery", "c.png", root="second")
self.own(" Scenery ", "padded.png")
self.own("Other", "d.png")
self.age_rows()
with self.db.model.get_connection() as conn:
conn.execute("UPDATE image_categories SET updated_at=?", (_OLD_STAMP,))
images_before = self.image_rows()
prompts_before, pairs_before = self.snapshot()
preview = self.preview()
receipt = self.apply(preview["impact_token"])
self.assertEqual(receipt["impact"]["images"], 3)
images_after = self.image_rows()
self.assertEqual(set(images_after), set(images_before))
renamed = {("root", "a.png"), ("root", "b.png"), ("second", "c.png")}
for key, (value, stamp, rowid) in images_after.items():
old_value, old_stamp, old_rowid = images_before[key]
self.assertEqual(rowid, old_rowid, key)
if key in renamed:
self.assertEqual(value, "Landscape")
self.assertNotEqual(stamp, _OLD_STAMP)
else:
self.assertEqual((value, stamp), (old_value, old_stamp))
prompts_after, pairs_after = self.snapshot()
self.assertEqual(prompts_after[bystander], prompts_before[bystander])
for before, after in (
(prompts_before, prompts_after),
(pairs_before, pairs_after),
):
for row_id, row in after.items():
if before[row_id]["category"] != "Scenery":
self.assertEqual(row, before[row_id])
# Merging into a value images already hold keeps every file key.
merge = self.preview(source="Landscape", target="Other")
self.assertTrue(merge["target_exists"])
self.assertEqual(merge["impact"]["images"], 3)
receipt = self.apply(merge["impact_token"], source="Landscape", target="Other")
self.assertTrue(receipt["consolidated"])
merged = self.image_rows()
self.assertEqual(set(merged), set(images_before))
self.assertEqual(
sorted(key for key, row in merged.items() if row[0] == "Other"),
sorted([*renamed, ("root", "d.png")]),
)
def test_image_only_value_can_be_renamed(self):
self.save("Bystander", category="Kept")
self.db.save_prompt_pair("Pair", "Noise", category="Kept")
self.own("Image only", "a.png", "b.png")
self.age_rows()
before = self.checksum()
preview = self.preview(source="Image only", target="Renamed")
self.assertEqual(
preview["impact"],
{
"positive_prompts": 0,
"other_prompts": 0,
"negative_variants": 0,
"archived_negative_variants": 0,
"images": 2,
},
)
self.assertFalse(preview["target_exists"])
receipt = self.apply(
preview["impact_token"], source="Image only", target="Renamed"
)
self.assertFalse(receipt["consolidated"])
self.assertEqual({row[0] for row in self.image_rows().values()}, {"Renamed"})
self.assertEqual(self.checksum(), before)
# An image-only target counts as existing.
self.own("Image target", "c.png")
self.assertTrue(
self.preview(source="Renamed", target="Image target")["target_exists"]
)
# Nothing holds the value any more: refused.
with self.assertRaises(LibraryCommandError) as caught:
self.preview(source="Image only", target="Again")
self.assertEqual(caught.exception.code, "vocabulary-conflict")
def test_token_refused_after_an_image_row_gains_or_loses_the_value(self):
self.owners()
self.own("Scenery", "a.png", "b.png")
def gain():
self.own("Scenery", "late.png")
def lose():
self.db.clear_image_categories([("root", "a.png")])
def move_away():
self.own("Elsewhere", "b.png")
def target_on_image():
self.own("Landscape", "target.png")
for mutate in (gain, lose, move_away, target_on_image):
with self.subTest(mutate.__name__):
token = self.preview()["impact_token"]
mutate()
before = self.snapshot(), self.image_rows()
with self.assertRaises(LibraryCommandError) as caught:
self.apply(token)
self.assertEqual(caught.exception.code, "vocabulary-conflict")
self.assertEqual((self.snapshot(), self.image_rows()), before)
def test_image_category_failure_rolls_back_prompts(self):
self.owners()
self.own("Scenery", "a.png")
token = self.preview()["impact_token"]
with self.db.model.get_connection() as conn:
conn.execute(
"CREATE TRIGGER abort_images BEFORE UPDATE ON image_categories "
"BEGIN SELECT RAISE(ABORT, 'boom'); END"
)
before = self.snapshot(), self.image_rows()
with self.assertRaises(sqlite3.Error):
self.apply(token)
self.assertEqual((self.snapshot(), self.image_rows()), before)
def test_category_exists_reads_the_image_index(self):
from database import vocabulary
with self.db.model.get_connection() as conn:
plan = [
row[-1]
for row in conn.execute(
"EXPLAIN QUERY PLAN " + vocabulary._CATEGORY_EXISTS, {"v": "x"}
)
]
steps = [step for step in plan if "image_categories" in step]
self.assertTrue(steps, plan)
for step in steps:
self.assertIn("idx_image_categories_category", step)
def test_category_apply_failure_rolls_back(self):
self.owners()
token = self.preview()["impact_token"]
@@ -942,6 +1220,18 @@ class TestSharedTokenHashing(unittest.TestCase):
),
"cefe7f6a5544b7f6a203f5527810899c6af568aa7abc1b75adffbfb5ae6251ae",
)
with_images = category_impact_token(
"e", "S", "T", True, [(1, True)], [], images=[9, 4]
)
self.assertRegex(with_images, r"\A[0-9a-f]{64}\Z")
self.assertEqual(
with_images,
category_impact_token("e", "S", "T", True, [(1, True)], [], images=[4, 9]),
)
self.assertNotEqual(
with_images,
category_impact_token("e", "S", "T", True, [(1, True)], [], images=[4]),
)
if __name__ == "__main__":