mirror of
https://git.datalinker.icu/comfyanonymous/ComfyUI
synced 2026-10-29 10:04:25 +08:00
Add tests
This commit is contained in:
parent
38dd4b419d
commit
b4c650b56e
@ -22,7 +22,7 @@ class FileInfo(TypedDict):
|
|||||||
|
|
||||||
def get_file_info(path: str, relative_to: str) -> FileInfo:
|
def get_file_info(path: str, relative_to: str) -> FileInfo:
|
||||||
return {
|
return {
|
||||||
"path": os.path.relpath(path, relative_to),
|
"path": os.path.relpath(path, relative_to).replace(os.sep, '/'),
|
||||||
"size": os.path.getsize(path),
|
"size": os.path.getsize(path),
|
||||||
"modified": os.path.getmtime(path)
|
"modified": os.path.getmtime(path)
|
||||||
}
|
}
|
||||||
@ -178,11 +178,10 @@ class UserManager():
|
|||||||
pattern = os.path.join(glob.escape(path), '*')
|
pattern = os.path.join(glob.escape(path), '*')
|
||||||
|
|
||||||
def process_full_path(full_path: str) -> FileInfo | str | list[str]:
|
def process_full_path(full_path: str) -> FileInfo | str | list[str]:
|
||||||
full_path = full_path.replace(os.sep, '/')
|
|
||||||
if full_info:
|
if full_info:
|
||||||
return get_file_info(full_path, path)
|
return get_file_info(full_path, path)
|
||||||
|
|
||||||
rel_path = os.path.relpath(full_path, path)
|
rel_path = os.path.relpath(full_path, path).replace(os.sep, '/')
|
||||||
if split_path:
|
if split_path:
|
||||||
return [rel_path] + rel_path.split('/')
|
return [rel_path] + rel_path.split('/')
|
||||||
|
|
||||||
|
|||||||
@ -80,9 +80,7 @@ async def test_listuserdata_split_path(aiohttp_client, app, tmp_path):
|
|||||||
client = await aiohttp_client(app)
|
client = await aiohttp_client(app)
|
||||||
resp = await client.get("/userdata?dir=test_dir&recurse=true&split=true")
|
resp = await client.get("/userdata?dir=test_dir&recurse=true&split=true")
|
||||||
assert resp.status == 200
|
assert resp.status == 200
|
||||||
assert await resp.json() == [
|
assert await resp.json() == [["subdir/file1.txt", "subdir", "file1.txt"]]
|
||||||
["subdir/file1.txt", "subdir", "file1.txt"]
|
|
||||||
]
|
|
||||||
|
|
||||||
|
|
||||||
async def test_listuserdata_invalid_directory(aiohttp_client, app):
|
async def test_listuserdata_invalid_directory(aiohttp_client, app):
|
||||||
@ -118,3 +116,60 @@ async def test_listuserdata_normalized_separator(aiohttp_client, app, tmp_path):
|
|||||||
assert "/" in result[0]["path"] # Ensure forward slash is used
|
assert "/" in result[0]["path"] # Ensure forward slash is used
|
||||||
assert "\\" not in result[0]["path"] # Ensure backslash is not present
|
assert "\\" not in result[0]["path"] # Ensure backslash is not present
|
||||||
assert result[0]["path"] == "subdir/file1.txt"
|
assert result[0]["path"] == "subdir/file1.txt"
|
||||||
|
|
||||||
|
|
||||||
|
async def test_post_userdata_new_file(aiohttp_client, app, tmp_path):
|
||||||
|
client = await aiohttp_client(app)
|
||||||
|
content = b"test content"
|
||||||
|
resp = await client.post("/userdata/test.txt", data=content)
|
||||||
|
|
||||||
|
assert resp.status == 200
|
||||||
|
assert await resp.text() == '"test.txt"'
|
||||||
|
|
||||||
|
# Verify file was created with correct content
|
||||||
|
with open(tmp_path / "test.txt", "rb") as f:
|
||||||
|
assert f.read() == content
|
||||||
|
|
||||||
|
|
||||||
|
async def test_post_userdata_overwrite_existing(aiohttp_client, app, tmp_path):
|
||||||
|
# Create initial file
|
||||||
|
with open(tmp_path / "test.txt", "w") as f:
|
||||||
|
f.write("initial content")
|
||||||
|
|
||||||
|
client = await aiohttp_client(app)
|
||||||
|
new_content = b"updated content"
|
||||||
|
resp = await client.post("/userdata/test.txt", data=new_content)
|
||||||
|
|
||||||
|
assert resp.status == 200
|
||||||
|
assert await resp.text() == '"test.txt"'
|
||||||
|
|
||||||
|
# Verify file was overwritten
|
||||||
|
with open(tmp_path / "test.txt", "rb") as f:
|
||||||
|
assert f.read() == new_content
|
||||||
|
|
||||||
|
|
||||||
|
async def test_post_userdata_no_overwrite(aiohttp_client, app, tmp_path):
|
||||||
|
# Create initial file
|
||||||
|
with open(tmp_path / "test.txt", "w") as f:
|
||||||
|
f.write("initial content")
|
||||||
|
|
||||||
|
client = await aiohttp_client(app)
|
||||||
|
resp = await client.post("/userdata/test.txt?overwrite=false", data=b"new content")
|
||||||
|
|
||||||
|
assert resp.status == 409
|
||||||
|
|
||||||
|
# Verify original content unchanged
|
||||||
|
with open(tmp_path / "test.txt", "r") as f:
|
||||||
|
assert f.read() == "initial content"
|
||||||
|
|
||||||
|
|
||||||
|
async def test_post_userdata_full_info(aiohttp_client, app, tmp_path):
|
||||||
|
client = await aiohttp_client(app)
|
||||||
|
content = b"test content"
|
||||||
|
resp = await client.post("/userdata/test.txt?full_info=true", data=content)
|
||||||
|
|
||||||
|
assert resp.status == 200
|
||||||
|
result = await resp.json()
|
||||||
|
assert result["path"] == "test.txt"
|
||||||
|
assert result["size"] == len(content)
|
||||||
|
assert "modified" in result
|
||||||
|
|||||||
Loading…
x
Reference in New Issue
Block a user