feat(ai-studio): separate image pool from task states

This commit is contained in:
chengma
2026-07-13 09:54:08 +08:00
parent 42fe233ed9
commit b1c050f633
9 changed files with 500 additions and 59 deletions
+127 -3
View File
@@ -928,19 +928,143 @@ class GuiTests(TempDirMixin, unittest.TestCase):
tab._select_project(project.id)
self.assertEqual("继续查询任务", tab.resume_button.text())
self.assertEqual(1, tab.pool_grid.count())
self.assertTrue(
all(
tab.pool_grid.item(row).data(gui.Qt.UserRole)["type"] == "asset"
for row in range(tab.pool_grid.count())
)
)
self.assertFalse(tab.job_section.isHidden())
statuses = [
tab.pool_grid.item(row).text()
for row in range(tab.pool_grid.count())
if tab.pool_grid.item(row).data(gui.Qt.UserRole)["type"] == "job"
tab.job_grid.item(row).text()
for row in range(tab.job_grid.count())
]
self.assertEqual(1, len(statuses))
self.assertIn("已提交", statuses[0])
self.assertIn("可继续查询", statuses[0])
self.assertIn("扣点2", statuses[0])
self.assertIn("余额88", statuses[0])
self.assertIn("call_id=call-1", statuses[0])
self.assert_removed(temp_dir)
def test_image_studio_tab_separates_usable_assets_from_job_states(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)
db.init_db(cfg["db_path"])
project = image_studio.create_or_get_project(
account_alias="alias-a",
account_slug="alias_a",
item_id="51100639510",
path=cfg["db_path"],
)
other_project = image_studio.create_or_get_project(
account_alias="alias-b",
account_slug="alias_b",
item_id="51100639511",
path=cfg["db_path"],
)
source = image_studio.add_asset(
project.id,
"original",
local_path=self.write_test_image(os.path.join(temp_dir, "source.jpg")),
path=cfg["db_path"],
)
resumable = image_studio.create_job(
project.id,
source_asset_id=source.id,
task_key="resume-job",
path=cfg["db_path"],
)
image_studio.set_job_submitted(resumable.id, "remote-resume", path=cfg["db_path"])
image_studio.update_job_status(
resumable.id,
"failed",
error="下载失败:https://secret.example.com/api/v1/image?token=private",
path=cfg["db_path"],
)
terminal = image_studio.create_job(
project.id,
source_asset_id=source.id,
task_key="terminal-job",
path=cfg["db_path"],
)
image_studio.set_job_submitted(terminal.id, "remote-terminal", path=cfg["db_path"])
image_studio.update_job_status(
terminal.id,
"expired",
error="上游任务已过期",
recovery_action=image_studio.JOB_RECOVERY_REGENERATE,
path=cfg["db_path"],
)
stopped = image_studio.create_job(
project.id,
source_asset_id=source.id,
task_key="stopped-job",
path=cfg["db_path"],
)
image_studio.update_job_status(
stopped.id,
"cancelled",
error="用户停止",
recovery_action=image_studio.JOB_RECOVERY_REGENERATE,
path=cfg["db_path"],
)
tab = ImageStudioTab(config=cfg, db_path=cfg["db_path"])
self.addCleanup(tab.close)
tab._select_project(project.id)
self.assertEqual(1, tab.pool_grid.count())
self.assertTrue(
all(
tab.pool_grid.item(row).data(gui.Qt.UserRole)["type"] == "asset"
for row in range(tab.pool_grid.count())
)
)
self.assertFalse(tab.job_section.isHidden())
self.assertEqual(3, tab.job_grid.count())
job_items = [tab.job_grid.item(row) for row in range(tab.job_grid.count())]
self.assertTrue(
all(item.data(gui.Qt.UserRole)["draggable"] is False for item in job_items)
)
details = "\n".join(item.text() for item in job_items)
tooltips = "\n".join(item.toolTip() for item in job_items)
self.assertIn("生成失败", details)
self.assertIn("任务过期", details)
self.assertIn("已停止", details)
self.assertIn("可继续查询", details)
self.assertIn("需要重新生成,可能再次扣点", details)
self.assertNotIn("https://", details)
self.assertNotIn("secret.example", tooltips)
failed_item = next(item for item in job_items if "生成失败" in item.text())
self.assertEqual("#ffebe9", failed_item.background().color().name())
tab._select_project(other_project.id)
self.assertEqual(0, tab.job_grid.count())
self.assertTrue(tab.job_section.isHidden())
tab._select_project(project.id)
output = image_studio.add_asset(
project.id,
"generated_main",
parent_asset_id=source.id,
local_path=self.write_test_image(os.path.join(temp_dir, "generated.jpg")),
path=cfg["db_path"],
)
image_studio.update_job_status(
resumable.id,
"succeeded",
output_asset_id=output.id,
path=cfg["db_path"],
)
tab.refresh_project_assets()
self.assertEqual(2, tab.pool_grid.count())
self.assertEqual(2, tab.job_grid.count())
self.assert_removed(temp_dir)
def test_image_studio_generation_log_uses_cmhub_tier_summary(self):
with self.make_temp_dir() as temp_dir:
cfg = self.make_config(temp_dir)