catch-22:「生成标题」用 read_all_rows 取行,而它要求 标题(A)+原始图片路径(C) 都非空;但标题生成的目的就是填 A。印花导出表若 A 空(被清空/早期导出), 所有行被跳过 → 误报「该表没有可处理的行」。实测 output/20260623_094529.xlsx: A 全空、C 是印花子目录。 - excel_service.read_all_rows 增 require_title=True 开关:False 时只在 原始图片路径(C) 空时跳过、允许 标题(A) 空;默认 True 行为不变(预览/ 无待处理统计/图片生成 load_outfit_tasks 仍要求 A) - ai_outfit_panel._TitleWorker.run 改调 read_all_rows(..., require_title=False) 测试:A 空+C 有的行 require_title=False 收录、默认跳过;C 空两种都跳过。 全套 py37 通过(test_config_service 的 packaging 模板失败属并行 §19.13,无关)。 离屏冒烟:真表 20260623_094529.xlsx 默认 0 行→require_title=False 加载 6 行; worker 把 6 条标题按序回填 A、C(印花目录)不动。 Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
231 lines
9.5 KiB
Python
231 lines
9.5 KiB
Python
"""Tests for AI outfit Excel service.
|
||
|
||
Pure service tests: no GUI dependency.
|
||
"""
|
||
import shutil
|
||
import sys
|
||
import tempfile
|
||
import unittest
|
||
from pathlib import Path
|
||
from unittest.mock import patch
|
||
|
||
from openpyxl import Workbook, load_workbook
|
||
|
||
sys.path.insert(0, str(Path(__file__).parent.parent / "src"))
|
||
|
||
|
||
class TestOutfitExcelService(unittest.TestCase):
|
||
def setUp(self):
|
||
self.tmp = Path(tempfile.mkdtemp())
|
||
self.excel_path = self.tmp / "outfit.xlsx"
|
||
self._write_workbook()
|
||
|
||
def tearDown(self):
|
||
shutil.rmtree(str(self.tmp), ignore_errors=True)
|
||
|
||
def _write_workbook(self):
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.append(["标题", "货号", "衣服图", "结果图", "状态", "原因"])
|
||
ws.append(["T恤", "TY001", r"D:\img\ty001.png", "", "", ""])
|
||
ws.append(["已完成", "TY002", r"D:\img\ty002.png", r"D:\out\ty002.jpg", "完成", ""])
|
||
ws.append(["失败行", "TY003", r"D:\img\ty003.png", "", "失败", "旧错误"])
|
||
ws.append(["缺路径", "TY004", "", "", "", ""])
|
||
ws.append([" 保留空格 ", 1005, r"D:\img\ty005.png", "", "", ""])
|
||
wb.save(str(self.excel_path))
|
||
wb.close()
|
||
|
||
def _read_row(self, row):
|
||
wb = load_workbook(str(self.excel_path))
|
||
try:
|
||
ws = wb.active
|
||
return (
|
||
ws.cell(row, 4).value,
|
||
ws.cell(row, 5).value,
|
||
ws.cell(row, 6).value,
|
||
)
|
||
finally:
|
||
wb.close()
|
||
|
||
def test_load_tasks_skips_done_failed_and_incomplete_by_default(self):
|
||
from services.excel_service import load_outfit_tasks
|
||
|
||
tasks = load_outfit_tasks(self.excel_path)
|
||
|
||
self.assertEqual([task.row_index for task in tasks], [2, 6])
|
||
self.assertEqual(tasks[0].title, "T恤")
|
||
self.assertEqual(tasks[0].product_id, "TY001")
|
||
self.assertEqual(tasks[0].garment_path, r"D:\img\ty001.png")
|
||
|
||
def test_load_tasks_includes_failed_when_retry_enabled(self):
|
||
from services.excel_service import load_outfit_tasks
|
||
|
||
tasks = load_outfit_tasks(self.excel_path, retry_failed=True)
|
||
|
||
self.assertEqual([task.row_index for task in tasks], [2, 4, 6])
|
||
self.assertEqual(tasks[1].status, "失败")
|
||
|
||
def test_load_tasks_preserves_user_cell_text(self):
|
||
from services.excel_service import load_outfit_tasks
|
||
|
||
tasks = load_outfit_tasks(self.excel_path)
|
||
task = tasks[-1]
|
||
|
||
self.assertEqual(task.title, " 保留空格 ")
|
||
self.assertEqual(task.product_id, "1005")
|
||
|
||
def test_write_success_result_updates_d_e_f_and_saves(self):
|
||
from core.models import OutfitResult, OutfitTask
|
||
from services.excel_service import write_outfit_result
|
||
|
||
task = OutfitTask(row_index=2, title="T恤", product_id="TY001", garment_path="g.png")
|
||
result = OutfitResult(task=task, success=True, output_path=r"D:\out\TY001.jpg")
|
||
|
||
write_outfit_result(self.excel_path, result)
|
||
|
||
self.assertEqual(self._read_row(2), (r"D:\out\TY001.jpg", "完成", None))
|
||
|
||
def test_write_failure_result_updates_e_f_and_saves(self):
|
||
from core.models import OutfitResult, OutfitTask
|
||
from services.excel_service import write_outfit_result
|
||
|
||
task = OutfitTask(row_index=2, title="T恤", product_id="TY001", garment_path="g.png")
|
||
result = OutfitResult(task=task, success=False, error="衣服图打不开")
|
||
|
||
write_outfit_result(self.excel_path, result)
|
||
|
||
self.assertEqual(self._read_row(2), (None, "失败", "衣服图打不开"))
|
||
|
||
def test_read_all_rows_includes_done_failed_skips_incomplete(self):
|
||
from services.excel_service import read_all_rows
|
||
|
||
rows = read_all_rows(self.excel_path)
|
||
|
||
# rows 2(待处理) 3(完成) 4(失败) 6(待处理) — row 5 (空衣服图) skipped
|
||
self.assertEqual([r.row_index for r in rows], [2, 3, 4, 6])
|
||
by_idx = {r.row_index: r for r in rows}
|
||
self.assertEqual(by_idx[3].status, "完成")
|
||
self.assertEqual(by_idx[4].status, "失败")
|
||
self.assertEqual(by_idx[6].title, " 保留空格 ") # raw text preserved
|
||
|
||
def test_empty_product_id_rows_are_included(self):
|
||
"""货号(B)可空:只要求 标题 + 衣服图路径(docs/11 §4)。"""
|
||
from services.excel_service import load_outfit_tasks, read_all_rows
|
||
|
||
p = self.tmp / "nopid.xlsx"
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.append(["标题", "货号", "衣服图", "结果", "状态", "原因"])
|
||
ws.append(["有标题无货号", "", r"D:\img\a.png", "", "", ""]) # 收录
|
||
ws.append(["缺图也缺货号", "", "", "", "", ""]) # 缺图 → 仍跳过
|
||
wb.save(str(p))
|
||
wb.close()
|
||
|
||
tasks = load_outfit_tasks(p)
|
||
self.assertEqual([t.row_index for t in tasks], [2])
|
||
self.assertEqual(tasks[0].product_id, "")
|
||
self.assertEqual([r.row_index for r in read_all_rows(p)], [2])
|
||
|
||
def test_read_all_rows_require_title_false_keeps_empty_title_rows(self):
|
||
"""§19.27: 标题生成的行来源——A 空、C 非空的行 require_title=False 收录,
|
||
默认(True)跳过;C 空的行两种都跳过。"""
|
||
from services.excel_service import read_all_rows
|
||
|
||
p = self.tmp / "notitle.xlsx"
|
||
wb = Workbook()
|
||
ws = wb.active
|
||
ws.append(["标题", "商品id", "原始图片路径", "结果", "状态", "原因"])
|
||
ws.append([None, None, "D:\\out\\TY030\\", "", "", ""]) # A空、C有(印花表)
|
||
ws.append([None, None, "D:\\out\\TY037\\", "", "", ""]) # A空、C有
|
||
ws.append([None, None, "", "", "", ""]) # C空 → 两种都跳过
|
||
wb.save(str(p))
|
||
wb.close()
|
||
|
||
# 默认要求标题 → A 全空 → 跳过印花行(旧 catch-22 现象)
|
||
self.assertEqual([r.row_index for r in read_all_rows(p)], [])
|
||
# require_title=False → A 空也收录,只按 C 过滤
|
||
rows = read_all_rows(p, require_title=False)
|
||
self.assertEqual([r.row_index for r in rows], [2, 3])
|
||
self.assertEqual(rows[0].title, "") # A 留空(待生成)
|
||
self.assertTrue(rows[0].garment_path.endswith("TY030" + "\\"))
|
||
|
||
def test_write_title_result_updates_a_only_and_saves(self):
|
||
"""标题生成回填只改 A 列,不动 D/E/F(docs/11 §17)。"""
|
||
from services.excel_service import write_title_result
|
||
|
||
# Row 3 starts as 完成 with D/E filled; writing a title must not touch them.
|
||
write_title_result(self.excel_path, 3, "AI 生成的电商标题")
|
||
|
||
wb = load_workbook(str(self.excel_path))
|
||
try:
|
||
ws = wb.active
|
||
self.assertEqual(ws.cell(3, 1).value, "AI 生成的电商标题") # A 改写
|
||
self.assertEqual(ws.cell(3, 4).value, r"D:\out\ty002.jpg") # D 不动
|
||
self.assertEqual(ws.cell(3, 5).value, "完成") # E 不动
|
||
finally:
|
||
wb.close()
|
||
|
||
def test_check_excel_writable_true_for_existing_workbook(self):
|
||
from services.excel_service import check_excel_writable
|
||
|
||
self.assertTrue(check_excel_writable(self.excel_path))
|
||
|
||
def test_ensure_excel_writable_raises_when_probe_fails(self):
|
||
from services.excel_service import ExcelInUseError, ensure_excel_writable
|
||
|
||
with patch("builtins.open", side_effect=OSError("locked")):
|
||
with self.assertRaises(ExcelInUseError):
|
||
ensure_excel_writable(self.excel_path)
|
||
|
||
# -- write_outfit_source_excel (docs/02 §6.12) ----------------------------
|
||
|
||
def test_write_outfit_source_excel_header_and_rows(self):
|
||
"""One row per (print_name, subdir_path); header + A/B/C values correct."""
|
||
import os
|
||
from services.excel_service import write_outfit_source_excel
|
||
|
||
p = self.tmp / "batch.xlsx"
|
||
rows = [
|
||
("floral", r"D:\out\20260623\floral" + os.sep),
|
||
("stripe", r"D:\out\20260623\stripe" + os.sep),
|
||
]
|
||
write_outfit_source_excel(p, rows)
|
||
|
||
wb = load_workbook(str(p))
|
||
ws = wb.active
|
||
try:
|
||
self.assertEqual(
|
||
[ws.cell(1, c).value for c in range(1, 8)],
|
||
["标题", "商品id", "原始图片路径", "生成结果图片路径", "完成状态", "失败原因", "货憨憨上传状态"],
|
||
)
|
||
self.assertEqual(ws.max_row, 3) # 1 header + 2 data
|
||
self.assertEqual(ws.cell(2, 1).value, "floral")
|
||
self.assertIsNone(ws.cell(2, 2).value) # 货号留空
|
||
self.assertIn("floral", ws.cell(2, 3).value)
|
||
self.assertTrue(ws.cell(2, 3).value.endswith(os.sep))
|
||
self.assertEqual(ws.cell(3, 1).value, "stripe")
|
||
for col in (4, 5, 6): # D/E/F 留空
|
||
self.assertIsNone(ws.cell(2, col).value)
|
||
finally:
|
||
wb.close()
|
||
|
||
def test_write_outfit_source_excel_ai_outfit_can_read_it(self):
|
||
"""AI穿搭 read_all_rows picks up the generated Excel as a directory row."""
|
||
import os
|
||
from services.excel_service import read_all_rows, write_outfit_source_excel
|
||
|
||
subdir = self.tmp / "floral"
|
||
subdir.mkdir()
|
||
p = self.tmp / "batch.xlsx"
|
||
write_outfit_source_excel(p, [("floral", str(subdir) + os.sep)])
|
||
|
||
rows = read_all_rows(p)
|
||
self.assertEqual(len(rows), 1)
|
||
self.assertEqual(rows[0].title, "floral")
|
||
self.assertEqual(rows[0].product_id, "")
|
||
self.assertTrue(rows[0].garment_path.endswith(os.sep))
|
||
|
||
|
||
if __name__ == "__main__":
|
||
unittest.main()
|