602 lines
26 KiB
Python
602 lines
26 KiB
Python
import os
|
|
import sys
|
|
import unittest
|
|
from types import SimpleNamespace
|
|
from unittest import mock
|
|
|
|
sys.path.insert(0, os.path.dirname(__file__))
|
|
|
|
from _helpers import TempDirMixin
|
|
|
|
from app import accounts, db, editor, image_studio
|
|
|
|
|
|
class ImageStudioTests(TempDirMixin, unittest.TestCase):
|
|
def _config(self, temp_dir):
|
|
return {
|
|
"data_dir": temp_dir,
|
|
"db_path": os.path.join(temp_dir, "cmshopee.db"),
|
|
"user_data_root": os.path.join(temp_dir, "chrome_user_data_dir"),
|
|
"chrome_path": "chrome.exe",
|
|
"default_debug_port": 9222,
|
|
"debug_port_range": [9222, 9230],
|
|
"cdp_ready_timeout": 1,
|
|
}
|
|
|
|
def test_init_db_adds_image_studio_tables_without_breaking_existing_tables(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
db_path = os.path.join(temp_dir, "cmshopee.db")
|
|
|
|
db.init_db(db_path)
|
|
db.init_db(db_path)
|
|
|
|
conn = db.connect(db_path)
|
|
try:
|
|
tables = {
|
|
row["name"]
|
|
for row in conn.execute(
|
|
"SELECT name FROM sqlite_master WHERE type = 'table'"
|
|
).fetchall()
|
|
}
|
|
self.assertTrue({"batches", "accounts", "tasks"}.issubset(tables))
|
|
self.assertTrue(
|
|
{
|
|
"image_studio_projects",
|
|
"image_studio_assets",
|
|
"image_studio_jobs",
|
|
"image_studio_selections",
|
|
}.issubset(tables)
|
|
)
|
|
|
|
projects_columns = {
|
|
row["name"]
|
|
for row in conn.execute("PRAGMA table_info(image_studio_projects)").fetchall()
|
|
}
|
|
self.assertTrue(
|
|
{
|
|
"account_alias",
|
|
"account_slug",
|
|
"item_id",
|
|
"target_main_count",
|
|
"target_detail_count",
|
|
"deleted_at",
|
|
}.issubset(projects_columns)
|
|
)
|
|
finally:
|
|
conn.close()
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_project_crud_unique_per_account_and_image_dirs(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
db_path = os.path.join(temp_dir, "cmshopee.db")
|
|
db.init_db(db_path)
|
|
|
|
account = SimpleNamespace(
|
|
alias="alias A",
|
|
account_name="店铺A",
|
|
slug="alias_a_slug",
|
|
)
|
|
project = image_studio.create_or_get_project(
|
|
account,
|
|
item_id="51100639510",
|
|
draft_prompt="初始提示词",
|
|
path=db_path,
|
|
)
|
|
same_project = image_studio.create_or_get_project(
|
|
account_alias="alias A",
|
|
account_slug="ignored_slug",
|
|
item_id="51100639510",
|
|
path=db_path,
|
|
)
|
|
other_account_project = image_studio.create_or_get_project(
|
|
account_alias="alias B",
|
|
account_slug="alias_b_slug",
|
|
item_id="51100639510",
|
|
path=db_path,
|
|
)
|
|
|
|
self.assertEqual(project.id, same_project.id)
|
|
self.assertNotEqual(project.id, other_account_project.id)
|
|
self.assertEqual("alias A", project.account_alias)
|
|
self.assertEqual("alias_a_slug", project.account_slug)
|
|
self.assertEqual("店铺A", project.account_name)
|
|
self.assertEqual("初始提示词", project.draft_prompt)
|
|
|
|
updated = image_studio.update_project_prompt(project.id, "二次提示词", path=db_path)
|
|
self.assertEqual("二次提示词", updated.draft_prompt)
|
|
|
|
dirs = image_studio.project_image_dirs(os.path.join(temp_dir, "images"), project)
|
|
self.assertEqual(
|
|
os.path.join(
|
|
temp_dir,
|
|
"images",
|
|
"pool",
|
|
"alias_a_slug",
|
|
"51100639510",
|
|
"originals",
|
|
),
|
|
dirs["originals"],
|
|
)
|
|
self.assertEqual(
|
|
os.path.join(
|
|
temp_dir,
|
|
"images",
|
|
"pool",
|
|
"alias_a_slug",
|
|
"51100639510",
|
|
"generated",
|
|
),
|
|
dirs["generated"],
|
|
)
|
|
self.assertEqual(
|
|
os.path.join(
|
|
temp_dir,
|
|
"images",
|
|
"pool",
|
|
"alias_a_slug",
|
|
"51100639510",
|
|
"exports",
|
|
),
|
|
dirs["exports"],
|
|
)
|
|
|
|
deleted = image_studio.soft_delete_project(project.id, "测试删除", path=db_path)
|
|
self.assertEqual("测试删除", deleted.deleted_reason)
|
|
self.assertEqual(1, len(image_studio.list_projects(path=db_path)))
|
|
restored = image_studio.create_or_get_project(
|
|
account,
|
|
item_id="51100639510",
|
|
path=db_path,
|
|
)
|
|
self.assertEqual(project.id, restored.id)
|
|
self.assertIsNone(restored.deleted_at)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_asset_crud_parent_status_and_sorting(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
db_path = os.path.join(temp_dir, "cmshopee.db")
|
|
db.init_db(db_path)
|
|
project = image_studio.create_or_get_project(
|
|
account_alias="alias",
|
|
account_slug="alias_slug",
|
|
item_id="51100639510",
|
|
path=db_path,
|
|
)
|
|
|
|
original_path = os.path.join(temp_dir, "old.png")
|
|
generated_path = os.path.join(temp_dir, "generated.png")
|
|
original = image_studio.add_asset(
|
|
project.id,
|
|
"original",
|
|
remote_url="https://example.test/old.png",
|
|
local_path=original_path,
|
|
aspect_ratio="1:1",
|
|
source_order=2,
|
|
path=db_path,
|
|
)
|
|
generated = image_studio.add_asset(
|
|
project.id,
|
|
"generated_main",
|
|
local_path=generated_path,
|
|
parent_asset_id=original.id,
|
|
prompt="生成提示词",
|
|
source_order=1,
|
|
path=db_path,
|
|
)
|
|
|
|
self.assertEqual(os.path.abspath(original_path), original.local_path)
|
|
self.assertEqual(original.id, generated.parent_asset_id)
|
|
self.assertEqual("生成提示词", generated.prompt)
|
|
|
|
assets = image_studio.list_assets(project.id, path=db_path)
|
|
self.assertEqual([generated.id, original.id], [asset.id for asset in assets])
|
|
|
|
missing = image_studio.mark_asset_status(
|
|
original.id,
|
|
image_studio.ASSET_STATUS_MISSING,
|
|
path=db_path,
|
|
)
|
|
self.assertEqual(image_studio.ASSET_STATUS_MISSING, missing.status)
|
|
with self.assertRaises(db.DbError):
|
|
image_studio.mark_asset_status(original.id, "deleted", path=db_path)
|
|
available_assets = image_studio.list_assets(
|
|
project.id,
|
|
include_missing=False,
|
|
path=db_path,
|
|
)
|
|
self.assertEqual([generated.id], [asset.id for asset in available_assets])
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_remove_asset_only_when_not_referenced(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
db_path = os.path.join(temp_dir, "cmshopee.db")
|
|
db.init_db(db_path)
|
|
project = image_studio.create_or_get_project(
|
|
account_alias="alias",
|
|
account_slug="alias_slug",
|
|
item_id="51100639510",
|
|
path=db_path,
|
|
)
|
|
free_asset = image_studio.add_asset(project.id, "original", path=db_path)
|
|
referenced_asset = image_studio.add_asset(project.id, "original", path=db_path)
|
|
image_studio.create_job(
|
|
project.id,
|
|
source_asset_id=referenced_asset.id,
|
|
path=db_path,
|
|
)
|
|
|
|
counts = image_studio.asset_reference_counts(referenced_asset.id, path=db_path)
|
|
self.assertEqual(1, counts["source_job"])
|
|
with self.assertRaisesRegex(db.DbError, "引用"):
|
|
image_studio.remove_asset_if_unused(referenced_asset.id, path=db_path)
|
|
|
|
removed = image_studio.remove_asset_if_unused(free_asset.id, path=db_path)
|
|
self.assertEqual(free_asset.id, removed.id)
|
|
self.assertIsNone(image_studio.get_asset(free_asset.id, path=db_path))
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_sync_original_asset_urls_is_idempotent_and_marks_missing(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
db_path = os.path.join(temp_dir, "cmshopee.db")
|
|
db.init_db(db_path)
|
|
project = image_studio.create_or_get_project(
|
|
account_alias="alias",
|
|
account_slug="alias_slug",
|
|
item_id="51100639510",
|
|
path=db_path,
|
|
)
|
|
|
|
first_sync = image_studio.sync_original_asset_urls(
|
|
project.id,
|
|
[
|
|
{"index": 1, "src": "https://susercontent.com/a.jpg"},
|
|
{"index": 2, "src": "https://susercontent.com/b.jpg"},
|
|
],
|
|
path=db_path,
|
|
)
|
|
second_sync = image_studio.sync_original_asset_urls(
|
|
project.id,
|
|
[
|
|
{"index": 1, "src": "https://susercontent.com/a.jpg"},
|
|
{"index": 2, "src": "https://susercontent.com/b.jpg"},
|
|
],
|
|
path=db_path,
|
|
)
|
|
third_sync = image_studio.sync_original_asset_urls(
|
|
project.id,
|
|
[{"index": 1, "src": "https://susercontent.com/b.jpg"}],
|
|
path=db_path,
|
|
)
|
|
|
|
self.assertEqual([asset.id for asset in first_sync], [asset.id for asset in second_sync])
|
|
self.assertEqual(2, len(third_sync))
|
|
by_url = {asset.remote_url: asset for asset in third_sync}
|
|
self.assertEqual(image_studio.ASSET_STATUS_MISSING, by_url["https://susercontent.com/a.jpg"].status)
|
|
self.assertEqual(image_studio.ASSET_STATUS_AVAILABLE, by_url["https://susercontent.com/b.jpg"].status)
|
|
self.assertEqual(1, by_url["https://susercontent.com/b.jpg"].source_order)
|
|
self.assertIsNone(by_url["https://susercontent.com/b.jpg"].local_path)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_job_lifecycle_and_resumable_query(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
db_path = os.path.join(temp_dir, "cmshopee.db")
|
|
db.init_db(db_path)
|
|
project = image_studio.create_or_get_project(
|
|
account_alias="alias",
|
|
account_slug="alias_slug",
|
|
item_id="51100639510",
|
|
path=db_path,
|
|
)
|
|
source_asset = image_studio.add_asset(project.id, "original", path=db_path)
|
|
|
|
job = image_studio.create_job(
|
|
project.id,
|
|
source_asset_id=source_asset.id,
|
|
job_type="main",
|
|
task_key="stable-task-key",
|
|
prompt="生成主图",
|
|
path=db_path,
|
|
)
|
|
self.assertEqual("stable-task-key", job.task_key)
|
|
self.assertEqual("pending", job.status)
|
|
self.assertEqual("cmhub", job.provider)
|
|
|
|
submitted = image_studio.set_job_submitted(
|
|
job.id,
|
|
"cmhub-task-1",
|
|
call_id="call-1",
|
|
points_cost=2,
|
|
points_balance=98,
|
|
path=db_path,
|
|
)
|
|
self.assertEqual("submitted", submitted.status)
|
|
self.assertEqual("cmhub-task-1", submitted.task_id)
|
|
self.assertEqual(2, submitted.points_cost)
|
|
self.assertEqual([job.id], [item.id for item in image_studio.list_resumable_jobs(path=db_path)])
|
|
|
|
running = image_studio.update_job_status(
|
|
job.id,
|
|
"running",
|
|
increment_attempts=True,
|
|
path=db_path,
|
|
)
|
|
self.assertEqual("running", running.status)
|
|
self.assertEqual(1, running.attempts)
|
|
with self.assertRaises(db.DbError):
|
|
image_studio.update_job_status(job.id, "unknown", path=db_path)
|
|
|
|
output_asset = image_studio.add_asset(
|
|
project.id,
|
|
"generated_main",
|
|
parent_asset_id=source_asset.id,
|
|
path=db_path,
|
|
)
|
|
succeeded = image_studio.update_job_status(
|
|
job.id,
|
|
"succeeded",
|
|
output_asset_id=output_asset.id,
|
|
points_balance=96,
|
|
path=db_path,
|
|
)
|
|
self.assertEqual("succeeded", succeeded.status)
|
|
self.assertEqual(output_asset.id, succeeded.output_asset_id)
|
|
self.assertEqual(96, succeeded.points_balance)
|
|
self.assertIsNotNone(succeeded.finished_at)
|
|
self.assertEqual([], image_studio.list_resumable_jobs(path=db_path))
|
|
|
|
with self.assertRaises(db.DbError):
|
|
image_studio.create_job(
|
|
project.id,
|
|
task_key="stable-task-key",
|
|
path=db_path,
|
|
)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_selections_are_consecutive_unique_and_replaceable(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
db_path = os.path.join(temp_dir, "cmshopee.db")
|
|
db.init_db(db_path)
|
|
project = image_studio.create_or_get_project(
|
|
account_alias="alias",
|
|
account_slug="alias_slug",
|
|
item_id="51100639510",
|
|
path=db_path,
|
|
)
|
|
other_project = image_studio.create_or_get_project(
|
|
account_alias="other",
|
|
account_slug="other_slug",
|
|
item_id="51100639510",
|
|
path=db_path,
|
|
)
|
|
first = image_studio.add_asset(project.id, "generated_main", source_order=1, path=db_path)
|
|
second = image_studio.add_asset(project.id, "generated_main", source_order=2, path=db_path)
|
|
third = image_studio.add_asset(project.id, "generated_detail", source_order=3, path=db_path)
|
|
other_asset = image_studio.add_asset(other_project.id, "generated_main", path=db_path)
|
|
|
|
main = image_studio.replace_selections(project.id, "main", [second.id, first.id], path=db_path)
|
|
self.assertEqual([1, 2], [selection.position for selection in main])
|
|
self.assertEqual([second.id, first.id], [selection.asset_id for selection in main])
|
|
|
|
detail = image_studio.replace_selections(project.id, "detail", [second.id, third.id], path=db_path)
|
|
self.assertEqual([second.id, third.id], [selection.asset_id for selection in detail])
|
|
|
|
replaced = image_studio.replace_selections(project.id, "main", [first.id], path=db_path)
|
|
self.assertEqual([1], [selection.position for selection in replaced])
|
|
self.assertEqual([first.id], [selection.asset_id for selection in replaced])
|
|
|
|
all_selections = image_studio.list_selections(project.id, path=db_path)
|
|
self.assertEqual(
|
|
[("detail", 1, second.id), ("detail", 2, third.id), ("main", 1, first.id)],
|
|
[
|
|
(selection.selection_type, selection.position, selection.asset_id)
|
|
for selection in all_selections
|
|
],
|
|
)
|
|
|
|
with self.assertRaises(db.DbError):
|
|
image_studio.replace_selections(project.id, "main", [first.id, first.id], path=db_path)
|
|
with self.assertRaises(db.DbError):
|
|
image_studio.replace_selections(project.id, "invalid", [first.id], path=db_path)
|
|
with self.assertRaises(db.DbError):
|
|
image_studio.replace_selections(project.id, "main", [other_asset.id], path=db_path)
|
|
with self.assertRaises(db.DbError):
|
|
image_studio.create_job(project.id, source_asset_id=other_asset.id, path=db_path)
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_pull_remote_main_image_urls_reuses_running_chrome_and_writes_snapshot(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self._config(temp_dir)
|
|
account = accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
|
|
images = [
|
|
{"index": index, "src": f"https://susercontent.com/main-{index}.jpg"}
|
|
for index in range(1, 10)
|
|
]
|
|
cdp = SimpleNamespace()
|
|
|
|
with mock.patch("app.image_studio.chrome.is_running", return_value=True) as is_running, \
|
|
mock.patch("app.image_studio.accounts.launch_for_login") as launch_for_login, \
|
|
mock.patch(
|
|
"app.image_studio.accounts.detect_login",
|
|
return_value={"logged_in": True, "cookie_names": ["SPC_ST"]},
|
|
) as detect_login, \
|
|
mock.patch("app.image_studio.editor.open_product", return_value=cdp) as open_product, \
|
|
mock.patch(
|
|
"app.image_studio.editor.read_product_image_urls",
|
|
return_value=images,
|
|
) as read_urls, \
|
|
mock.patch("app.image_studio.editor.close_readonly_product") as close_readonly, \
|
|
mock.patch("app.image_studio.editor.change_title") as change_title, \
|
|
mock.patch("app.image_studio.editor.replace_cover") as replace_cover, \
|
|
mock.patch("app.image_studio.editor.click_update") as click_update:
|
|
result = image_studio.pull_remote_main_image_urls(
|
|
"alias-a",
|
|
"51100639510",
|
|
path=cfg["db_path"],
|
|
config=cfg,
|
|
)
|
|
second = image_studio.pull_remote_main_image_urls(
|
|
"alias-a",
|
|
"51100639510",
|
|
path=cfg["db_path"],
|
|
config=cfg,
|
|
)
|
|
|
|
self.assertEqual("alias-a", result["project"].account_alias)
|
|
self.assertEqual("51100639510", result["project"].item_id)
|
|
self.assertTrue(result["readiness"]["reused"])
|
|
self.assertFalse(result["readiness"]["login_uncertain"])
|
|
self.assertEqual(images, result["images"])
|
|
self.assertEqual([asset.id for asset in result["assets"]], [asset.id for asset in second["assets"]])
|
|
self.assertEqual(9, len(result["assets"]))
|
|
self.assertEqual(
|
|
list(range(1, 10)),
|
|
[asset.source_order for asset in result["assets"]],
|
|
)
|
|
self.assertTrue(all(asset.kind == image_studio.ASSET_KIND_ORIGINAL for asset in result["assets"]))
|
|
self.assertTrue(all(asset.local_path is None for asset in result["assets"]))
|
|
is_running.assert_called_with(9222)
|
|
launch_for_login.assert_not_called()
|
|
detect_login.assert_called()
|
|
open_product.assert_called_with(account, "51100639510", on_step=None, bring_to_front=False)
|
|
self.assertEqual(2, read_urls.call_count)
|
|
self.assertEqual(2, close_readonly.call_count)
|
|
change_title.assert_not_called()
|
|
replace_cover.assert_not_called()
|
|
click_update.assert_not_called()
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_pull_remote_main_image_urls_launches_chrome_when_needed(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self._config(temp_dir)
|
|
accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
|
|
cdp = SimpleNamespace()
|
|
|
|
with mock.patch("app.image_studio.chrome.is_running", return_value=False) as is_running, \
|
|
mock.patch(
|
|
"app.image_studio.accounts.launch_for_login",
|
|
return_value={"launched": True, "reused": False, "debug_port": 9222},
|
|
) as launch_for_login, \
|
|
mock.patch(
|
|
"app.image_studio.accounts.detect_login",
|
|
return_value={"logged_in": True},
|
|
), \
|
|
mock.patch("app.image_studio.editor.open_product", return_value=cdp), \
|
|
mock.patch(
|
|
"app.image_studio.editor.read_product_image_urls",
|
|
return_value=[{"index": 1, "src": "https://susercontent.com/main.jpg"}],
|
|
), \
|
|
mock.patch("app.image_studio.editor.close_readonly_product"):
|
|
result = image_studio.pull_remote_main_image_urls(
|
|
"alias-a",
|
|
"51100639510",
|
|
path=cfg["db_path"],
|
|
config=cfg,
|
|
)
|
|
|
|
self.assertTrue(result["readiness"]["launched"])
|
|
is_running.assert_called_once_with(9222)
|
|
launch_for_login.assert_called_once()
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_pull_remote_main_image_urls_blocks_definitive_logged_out(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self._config(temp_dir)
|
|
accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
|
|
|
|
with mock.patch("app.image_studio.chrome.is_running", return_value=True), \
|
|
mock.patch(
|
|
"app.image_studio.accounts.detect_login",
|
|
return_value={
|
|
"logged_in": False,
|
|
"reason": "LOGIN_PAGE",
|
|
"url": "https://accounts.shopee.tw/seller/login",
|
|
"cookie_names": [],
|
|
},
|
|
), \
|
|
mock.patch("app.image_studio.editor.open_product") as open_product:
|
|
with self.assertRaisesRegex(image_studio.ImageStudioError, "未登录"):
|
|
image_studio.pull_remote_main_image_urls(
|
|
"alias-a",
|
|
"51100639510",
|
|
path=cfg["db_path"],
|
|
config=cfg,
|
|
)
|
|
|
|
open_product.assert_not_called()
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_pull_remote_main_image_urls_continues_when_login_status_uncertain(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self._config(temp_dir)
|
|
accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
|
|
cdp = SimpleNamespace()
|
|
|
|
with mock.patch("app.image_studio.chrome.is_running", return_value=True), \
|
|
mock.patch(
|
|
"app.image_studio.accounts.detect_login",
|
|
return_value={
|
|
"logged_in": False,
|
|
"reason": "NO_SESSION_COOKIE",
|
|
"url": "https://seller.shopee.tw/portal/",
|
|
"cookie_names": [],
|
|
},
|
|
), \
|
|
mock.patch("app.image_studio.editor.open_product", return_value=cdp) as open_product, \
|
|
mock.patch(
|
|
"app.image_studio.editor.read_product_image_urls",
|
|
return_value=[{"index": 1, "src": "https://susercontent.com/main.jpg"}],
|
|
), \
|
|
mock.patch("app.image_studio.editor.close_readonly_product"):
|
|
result = image_studio.pull_remote_main_image_urls(
|
|
"alias-a",
|
|
"51100639510",
|
|
path=cfg["db_path"],
|
|
config=cfg,
|
|
)
|
|
|
|
self.assertTrue(result["readiness"]["login_uncertain"])
|
|
open_product.assert_called_once()
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
def test_pull_remote_main_image_urls_preserves_product_unavailable_toast_error(self):
|
|
with self.make_temp_dir() as temp_dir:
|
|
cfg = self._config(temp_dir)
|
|
accounts.create_account("主店", "alias-a", debug_port=9222, config=cfg)
|
|
|
|
with mock.patch("app.image_studio.chrome.is_running", return_value=True), \
|
|
mock.patch(
|
|
"app.image_studio.accounts.detect_login",
|
|
return_value={"logged_in": True},
|
|
), \
|
|
mock.patch(
|
|
"app.image_studio.editor.open_product",
|
|
side_effect=editor.EditorError("商品失效：please input correct product id"),
|
|
), \
|
|
mock.patch("app.image_studio.editor.close_readonly_product") as close_readonly:
|
|
with self.assertRaisesRegex(image_studio.ImageStudioError, "please input correct product id"):
|
|
image_studio.pull_remote_main_image_urls(
|
|
"alias-a",
|
|
"bad-item",
|
|
path=cfg["db_path"],
|
|
config=cfg,
|
|
)
|
|
|
|
close_readonly.assert_not_called()
|
|
|
|
self.assert_removed(temp_dir)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|