chengma/cmshoppe
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(status): default legacy batches to normal

Browse Source
This commit is contained in:
chengma
2026-07-20 09:41:55 +08:00
parent 00c695d035
commit d687c833e0
17 changed files with 205 additions and 991 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_db.py
+114 -1
View File
@@ -122,12 +122,125 @@ class DbTests(TempDirMixin, unittest.TestCase):
task = db.get_task(1, conn=conn)
self.assertEqual(0, task.cover_reset_count)
self.assertIsNone(task.cover_reset_at)
self.assertIsNone(task.product_status)
self.assertEqual("normal", task.product_status)
self.assertEqual(
db.LEGACY_PRODUCT_STATUS_DEFAULT_NOTE,
task.product_status_note,
)
self.assertIsNone(task.product_status_at)
finally:
conn.close()
self.assert_removed(temp_dir)
def test_init_db_defaults_only_pre_feature_active_missing_status_tasks(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")
db.init_db(db_path)
batch_ids = {}
for index, name in enumerate(
(
"legacy-null",
"legacy-blank",
"legacy-unknown",
"boundary",
"after",
"deleted",
),
start=1,
):
batch_id = db.create_batch([f"{name}.xlsx"], path=db_path)
batch_ids[name] = batch_id
db.insert_tasks(
batch_id,
[
{
"source_file_abs": os.path.join(temp_dir, f"{name}.xlsx"),
"source_sheet": "Sheet1",
"source_row": 2,
"account_name": "店铺",
"alias": "alias",
"item_id": str(51100639700 + index),
}
],
path=db_path,
)
conn = db.connect(db_path)
try:
with conn:
conn.execute(
"UPDATE batches SET created_at = ? WHERE id IN (?, ?, ?, ?)",
(
"2026-07-18T16:34:36",
batch_ids["legacy-null"],
batch_ids["legacy-blank"],
batch_ids["legacy-unknown"],
batch_ids["deleted"],
),
)
conn.execute(
"UPDATE batches SET created_at = ? WHERE id = ?",
(db.PRODUCT_STATUS_FEATURE_INTRODUCED_AT, batch_ids["boundary"]),
)
conn.execute(
"UPDATE batches SET created_at = ? WHERE id = ?",
("2026-07-18T16:34:38", batch_ids["after"]),
)
blank_task = db.list_tasks(
batch_id=batch_ids["legacy-blank"],
conn=conn,
)[0]
unknown_task = db.list_tasks(
batch_id=batch_ids["legacy-unknown"],
conn=conn,
)[0]
conn.execute(
"UPDATE tasks SET product_status = ' ' WHERE id = ?",
(blank_task.id,),
)
conn.execute(
"UPDATE tasks SET product_status = 'unknown', product_status_note = '已有未知状态' WHERE id = ?",
(unknown_task.id,),
)
conn.execute(
"UPDATE batches SET deleted_at = ? WHERE id = ?",
("2026-07-19T00:00:00", batch_ids["deleted"]),
)
finally:
conn.close()
db.init_db(db_path)
legacy_null = db.list_tasks(batch_id=batch_ids["legacy-null"], path=db_path)[0]
legacy_blank = db.list_tasks(batch_id=batch_ids["legacy-blank"], path=db_path)[0]
legacy_unknown = db.list_tasks(batch_id=batch_ids["legacy-unknown"], path=db_path)[0]
boundary = db.list_tasks(batch_id=batch_ids["boundary"], path=db_path)[0]
after = db.list_tasks(batch_id=batch_ids["after"], path=db_path)[0]
deleted = db.list_tasks(
batch_id=batch_ids["deleted"],
path=db_path,
include_deleted=True,
)[0]
for task in (legacy_null, legacy_blank):
self.assertEqual("normal", task.product_status)
self.assertEqual(db.LEGACY_PRODUCT_STATUS_DEFAULT_NOTE, task.product_status_note)
self.assertIsNone(task.product_status_at)
self.assertEqual("unknown", legacy_unknown.product_status)
self.assertEqual("已有未知状态", legacy_unknown.product_status_note)
self.assertIsNone(boundary.product_status)
self.assertIsNone(after.product_status)
self.assertIsNone(deleted.product_status)
first_updated_at = legacy_null.updated_at
db.init_db(db_path)
self.assertEqual(
first_updated_at,
db.get_task(legacy_null.id, path=db_path).updated_at,
)
self.assert_removed(temp_dir)
def test_account_batch_task_lifecycle(self):
with self.make_temp_dir() as temp_dir:
db_path = os.path.join(temp_dir, "cmshopee.db")