feat: complete T-534 generated reset enhancements

This commit is contained in:
chengma
2026-07-06 20:35:47 +08:00
parent 46995b301d
commit 71f8beae57
9 changed files with 488 additions and 63 deletions
+76 -1
View File
@@ -168,7 +168,7 @@ class DbTests(TempDirMixin, unittest.TestCase):
self.assertEqual(new_cover, reset_generated["new_cover_path"])
self.assertIsNone(reset_generated["deleted_file"])
self.assertTrue(os.path.exists(new_cover))
self.assertEqual("collected", after_generated.stage)
self.assertEqual("generated", after_generated.stage)
self.assertEqual("success", after_generated.status)
self.assertIsNone(after_generated.new_title)
self.assertIsNone(after_generated.new_cover_path)
@@ -177,6 +177,81 @@ class DbTests(TempDirMixin, unittest.TestCase):
self.assert_removed(temp_dir)
def test_reset_generated_can_reset_title_or_cover_components(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")
db.init_db(db_path)
batch_id = db.create_batch(["input.xlsx"], path=db_path)
db.insert_tasks(
batch_id,
[
{
"source_file_abs": os.path.join(temp_dir, "input.xlsx"),
"source_sheet": "Sheet1",
"source_row": 2,
"account_name": "shop",
"alias": "alias",
"item_id": "51100639510",
},
{
"source_file_abs": os.path.join(temp_dir, "input.xlsx"),
"source_sheet": "Sheet1",
"source_row": 3,
"account_name": "shop",
"alias": "alias",
"item_id": "51100639511",
},
],
path=db_path,
)
first, second = db.list_tasks(batch_id=batch_id, path=db_path)
first_cover = os.path.join(temp_dir, "first_new.jpg")
second_cover = os.path.join(temp_dir, "second_new.jpg")
for cover in [first_cover, second_cover]:
with open(cover, "wb") as fh:
fh.write(b"jpeg")
db.set_collected(first.id, "旧标题A", "old-a.jpg", path=db_path)
db.set_generated(first.id, "手动标题A", first_cover, path=db_path)
db.mark_failed(first.id, "generate", "图片不满意", path=db_path)
db.set_collected(second.id, "旧标题B", "old-b.jpg", path=db_path)
db.set_generated(second.id, "手动标题B", second_cover, path=db_path)
title_reset = db.reset_generated(
first.id,
reset_title=True,
reset_cover=False,
path=db_path,
)
after_title = title_reset["after"]
self.assertEqual("generated", after_title.stage)
self.assertEqual("success", after_title.status)
self.assertIsNone(after_title.new_title)
self.assertEqual(first_cover, after_title.new_cover_path)
self.assertIsNone(after_title.last_error)
self.assertTrue(os.path.exists(first_cover))
cover_reset = db.reset_generated(
second.id,
reset_title=False,
reset_cover=True,
delete_file=True,
path=db_path,
)
after_cover = cover_reset["after"]
self.assertEqual("generated", after_cover.stage)
self.assertEqual("success", after_cover.status)
self.assertEqual("手动标题B", after_cover.new_title)
self.assertIsNone(after_cover.new_cover_path)
self.assertEqual(second_cover, cover_reset["deleted_file"])
self.assertFalse(os.path.exists(second_cover))
with self.assertRaises(db.DbError):
db.reset_generated(first.id, reset_title=False, reset_cover=False, path=db_path)
with self.assertRaises(db.DbError):
db.reset_generated(first.id, reset_title=True, reset_cover=False, delete_file=True, path=db_path)
self.assert_removed(temp_dir)
def test_delete_batch_soft_hides_batch_and_tasks(self):
with self.make_temp_dir() as temp_dir: