cmshoppe/tests/test_excel.py

import os
import unittest

from _helpers import TempDirMixin

try:
    from openpyxl import Workbook
except ModuleNotFoundError:
    raise unittest.SkipTest("openpyxl 未安装")

from app import db, excel


class ExcelImportTests(TempDirMixin, unittest.TestCase):
    def save_workbook(self, path, sheets):
        workbook = Workbook()
        default = workbook.active
        workbook.remove(default)
        for title, rows in sheets:
            sheet = workbook.create_sheet(title=title)
            for row in rows:
                sheet.append(row)
        workbook.save(path)
        workbook.close()

    def test_import_tasks_parses_rows_and_writes_batch_tasks(self):
        with self.make_temp_dir() as temp_dir:
            db_path = os.path.join(temp_dir, "cmshopee.db")
            excel_path = os.path.join(temp_dir, "input.xlsx")
            self.save_workbook(
                excel_path,
                [
                    (
                        "商品",
                        [
                            ["账号名", "别名", "商品id", "旧标题"],
                            ["主店", "alias-a", 51100639510, ""],
                            ["副店", "alias-b", "52999999", ""],
                            ["缺别名", "", "123456", ""],
                            ["坏商品", "alias-c", "abc123", ""],
                            [None, None, None, None],
                        ],
                    )
                ],
            )

            result = excel.import_tasks([excel_path], path=db_path, note="导入测试")

            self.assertIsNotNone(result["batch_id"])
            self.assertEqual(2, len(result["rows"]))
            self.assertEqual(1, result["stats"]["files"])
            self.assertEqual(4, result["stats"]["total"])
            self.assertEqual(2, result["stats"]["valid"])
            self.assertEqual(2, result["stats"]["invalid"])
            self.assertEqual(2, result["stats"]["inserted"])
            self.assertEqual([], result["stats"]["file_errors"])
            self.assertEqual(2, len(result["stats"]["row_errors"]))
            self.assertEqual("商品", result["rows"][0]["source_sheet"])
            self.assertEqual(2, result["rows"][0]["source_row"])
            self.assertEqual("51100639510", result["rows"][0]["item_id"])
            self.assertTrue(result["rows"][0]["row_key"].startswith(result["batch_id"] + ":"))

            batch = db.get_batch(result["batch_id"], path=db_path)
            tasks = db.list_tasks(batch_id=result["batch_id"], path=db_path)
            self.assertEqual("导入测试", batch.note)
            self.assertEqual(2, len(tasks))
            self.assertEqual("alias-a", tasks[0].alias)
            self.assertEqual(os.path.abspath(excel_path), tasks[0].source_file_abs)
            self.assertEqual(2, tasks[0].source_row)

        self.assert_removed(temp_dir)

    def test_missing_required_column_rejects_whole_file(self):
        with self.make_temp_dir() as temp_dir:
            db_path = os.path.join(temp_dir, "cmshopee.db")
            excel_path = os.path.join(temp_dir, "missing_alias.xlsx")
            self.save_workbook(
                excel_path,
                [
                    (
                        "商品",
                        [
                            ["账号名", "商品id"],
                            ["主店", "51100639510"],
                        ],
                    )
                ],
            )

            result = excel.import_tasks([excel_path], path=db_path)

            self.assertIsNone(result["batch_id"])
            self.assertEqual([], result["rows"])
            self.assertEqual(1, len(result["stats"]["file_errors"]))
            self.assertEqual(["别名"], result["stats"]["file_errors"][0]["missing_columns"])
            self.assertFalse(os.path.exists(db_path))

        self.assert_removed(temp_dir)

    def test_import_tasks_continues_when_another_file_has_column_error(self):
        with self.make_temp_dir() as temp_dir:
            db_path = os.path.join(temp_dir, "cmshopee.db")
            good_path = os.path.join(temp_dir, "good.xlsx")
            bad_path = os.path.join(temp_dir, "missing_item.xlsx")
            self.save_workbook(
                good_path,
                [
                    (
                        "商品",
                        [
                            ["账号名", "别名", "商品id"],
                            ["主店", "alias-a", "51100639510"],
                        ],
                    )
                ],
            )
            self.save_workbook(
                bad_path,
                [
                    (
                        "商品",
                        [
                            ["账号名", "别名"],
                            ["主店", "alias-a"],
                        ],
                    )
                ],
            )

            result = excel.import_tasks([good_path, bad_path], path=db_path)

            self.assertIsNotNone(result["batch_id"])
            self.assertEqual(2, result["stats"]["files"])
            self.assertEqual(1, result["stats"]["valid"])
            self.assertEqual(1, result["stats"]["inserted"])
            self.assertEqual(1, len(result["stats"]["file_errors"]))
            self.assertEqual(["商品id"], result["stats"]["file_errors"][0]["missing_columns"])
            self.assertEqual(os.path.abspath(good_path), result["rows"][0]["source_file_abs"])
            self.assertEqual(1, len(db.list_tasks(batch_id=result["batch_id"], path=db_path)))

        self.assert_removed(temp_dir)

    def test_match_summary_counts_known_and_unknown_aliases(self):
        rows = [
            {"alias": "a"},
            {"alias": "b"},
            {"alias": "a"},
            {"alias": "missing"},
        ]
        accounts = [
            {"alias": "a"},
            {"alias": "b"},
        ]

        self.assertEqual(
            {
                "matched": 3,
                "unmatched": 1,
                "by_account": {"a": 2, "b": 1},
                "unmatched_aliases": ["missing"],
            },
            excel.match_summary(rows, accounts),
        )


if __name__ == "__main__":
    unittest.main()