Vito Sansevero
680afb5e20
Duplicate prompts, unused components, missing links, workflow storage and rotated logs now record one protected row per kept item, in the review's wording, and the job report lists every one of them. Kept links and log files keep their path. The duplicate file cleanup records each copy the review protected in its Kept group.
141 lines
5.9 KiB
Python
141 lines
5.9 KiB
Python
"""Protected maintenance items are listed in Activity, not only counted (#586).
|
|
|
|
Each record a maintenance job keeps gets one ``protected`` row with its
|
|
reason, in the wording of that operation's review, and the job's
|
|
downloadable report lists every one of them (never cut at 200 rows). A kept
|
|
image link names its file, so Activity can still open and locate it.
|
|
"""
|
|
|
|
import os
|
|
import shutil
|
|
import sys
|
|
from pathlib import Path
|
|
from unittest.mock import MagicMock
|
|
|
|
_mock_server = MagicMock()
|
|
_mock_server.PromptServer.instance.routes = MagicMock()
|
|
sys.modules.setdefault("server", _mock_server)
|
|
|
|
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
|
|
from py.api.jobs import MAX_OUTCOME_ROWS # noqa: E402
|
|
from py.config import GalleryConfig, PromptManagerConfig # noqa: E402
|
|
|
|
try:
|
|
from tests.test_api_maintenance import ApplyTestCase # noqa: E402
|
|
except ImportError: # discovered with tests/ as the top-level directory
|
|
from test_api_maintenance import ApplyTestCase # noqa: E402
|
|
|
|
|
|
def _record(detail):
|
|
return {"path": "", "root": "", "root_index": None, "detail": detail}
|
|
|
|
|
|
class TestProtectedRows(ApplyTestCase):
|
|
def duplicate(self, text, created_at="2999-01-01"):
|
|
with self.api.db.model.get_connection() as conn:
|
|
return conn.execute(
|
|
"INSERT INTO prompts(text, created_at) VALUES(?, ?)",
|
|
(text.upper(), created_at),
|
|
).lastrowid
|
|
|
|
def assert_complete(self, job, count):
|
|
self.assertEqual(job.counters["protected"], count)
|
|
report = job.report()["outcomes"]["protected"]
|
|
self.assertEqual(
|
|
(report["count"], len(report["rows"]), report["truncated"]),
|
|
(count, count, False),
|
|
)
|
|
|
|
async def test_duplicate_prompts_lists_each_kept_copy(self):
|
|
paired = self.save("A dog")
|
|
self.api.db.save_prompt_pair("A dog", "")
|
|
paired_copy = self.duplicate("A dog")
|
|
titled = self.save("A bird")
|
|
titled_copy = self.duplicate("A bird")
|
|
with self.api.db.model.get_connection() as conn:
|
|
conn.execute(
|
|
"INSERT INTO prompt_library_state(prompt_id,title) VALUES(?,?)",
|
|
(titled_copy, "Bird recipe"),
|
|
)
|
|
plan, _summary, job = await self.run_apply("duplicate_prompts")
|
|
self.assertEqual(job.state, "finished", job.log)
|
|
self.assertEqual(
|
|
sorted(row["detail"] for row in job.outcomes["protected"]),
|
|
sorted(
|
|
[
|
|
f"Prompt {paired}: kept, belongs to a saved pair",
|
|
f"Prompt {paired_copy}: kept, belongs to a saved pair",
|
|
f"Prompt {titled}: kept, a copy has its own Library title",
|
|
f"Prompt {titled_copy}: kept, a copy has its own Library title",
|
|
]
|
|
),
|
|
)
|
|
self.assertEqual(plan["protected"]["count"], 4)
|
|
self.assert_complete(job, 4)
|
|
|
|
async def test_unused_components_lists_each_kept_prompt_with_its_reasons(self):
|
|
self.save("Orphan")
|
|
tagged = self.save("Kept", tags=["__protected__"])
|
|
self.api.db.save_prompt_pair("Paired", "")
|
|
paired = self.api.db.search_prompts(text="Paired")[0]["id"]
|
|
with self.api.db.model.get_connection() as conn:
|
|
conn.execute(
|
|
"INSERT INTO prompt_tags(prompt_id, tag_id) "
|
|
"SELECT ?, id FROM tags WHERE name='__protected__'",
|
|
(paired,),
|
|
)
|
|
plan, _summary, job = await self.run_apply("unused_components")
|
|
self.assertEqual(job.state, "finished", job.log)
|
|
self.assertEqual(
|
|
sorted(job.outcomes["protected"], key=lambda row: row["detail"]),
|
|
sorted(
|
|
[
|
|
_record(f"Prompt {tagged}: kept, tagged __protected__"),
|
|
_record(
|
|
f"Prompt {paired}: kept, in a saved pair; "
|
|
"tagged __protected__"
|
|
),
|
|
],
|
|
key=lambda row: row["detail"],
|
|
),
|
|
)
|
|
self.assertEqual(plan["protected"]["count"], 2)
|
|
self.assert_complete(job, 2)
|
|
|
|
async def test_report_lists_every_kept_prompt(self):
|
|
PromptManagerConfig.SAFETY_BACKUP = False
|
|
count = MAX_OUTCOME_ROWS + 5
|
|
for n in range(count):
|
|
self.save(f"Kept {n:03d}", tags=["__protected__"])
|
|
_plan, _summary, job = await self.run_apply("unused_components")
|
|
self.assertEqual(len(job.outcomes["protected"]), MAX_OUTCOME_ROWS)
|
|
self.assert_complete(job, count)
|
|
|
|
async def test_missing_links_lists_each_kept_link_with_its_file(self):
|
|
unmounted = Path(self.tmpdir) / "unmounted"
|
|
unmounted.mkdir()
|
|
GalleryConfig.MONITORING_DIRECTORIES = [str(self.output_dir), str(unmounted)]
|
|
prompt = self.save("Links")
|
|
self.image(prompt, "gone.png", exists=False)
|
|
self.image(prompt, "sub/away.png", folder=unmounted, exists=False)
|
|
self.image(prompt, "x.png", folder=Path(self.tmpdir) / "elsewhere")
|
|
shutil.rmtree(unmounted)
|
|
plan, _summary, job = await self.run_apply("missing_links")
|
|
self.assertEqual(job.state, "finished", job.log)
|
|
self.assertEqual(plan["protected"]["count"], 2)
|
|
rows = sorted(job.outcomes["protected"], key=lambda row: row["path"])
|
|
self.assertEqual(
|
|
[(row["path"], row["detail"]) for row in rows],
|
|
[
|
|
("sub/away.png", "Kept: link under an unavailable source"),
|
|
("x.png", "Kept: link outside the image sources"),
|
|
],
|
|
)
|
|
# The gone source keeps the path under its label; outside every
|
|
# source the row names the file only.
|
|
self.assertEqual((rows[0]["root"], rows[0]["root_index"]), ("unmounted", None))
|
|
self.assertEqual((rows[1]["root"], rows[1]["root_index"]), ("", None))
|
|
self.assertNotIn(self.tmpdir, str(job.report()))
|
|
self.assert_complete(job, 2)
|