fred-dev/comfy_ui_ali

1import pytest2import os3from aiohttp import web4from app.user_manager import UserManager5from unittest.mock import patch6 7pytestmark = (8    pytest.mark.asyncio9)  # This applies the asyncio mark to all test functions in the module10 11 12@pytest.fixture13def user_manager(tmp_path):14    um = UserManager()15    um.get_request_user_filepath = lambda req, file, **kwargs: os.path.join(16        tmp_path, file17    ) if file else tmp_path18    return um19 20 21@pytest.fixture22def app(user_manager):23    app = web.Application()24    routes = web.RouteTableDef()25    user_manager.add_routes(routes)26    app.add_routes(routes)27    return app28 29 30async def test_listuserdata_empty_directory(aiohttp_client, app, tmp_path):31    client = await aiohttp_client(app)32    resp = await client.get("/userdata?dir=test_dir")33    assert resp.status == 40434 35 36async def test_listuserdata_with_files(aiohttp_client, app, tmp_path):37    os.makedirs(tmp_path / "test_dir")38    with open(tmp_path / "test_dir" / "file1.txt", "w") as f:39        f.write("test content")40 41    client = await aiohttp_client(app)42    resp = await client.get("/userdata?dir=test_dir")43    assert resp.status == 20044    assert await resp.json() == ["file1.txt"]45 46 47async def test_listuserdata_recursive(aiohttp_client, app, tmp_path):48    os.makedirs(tmp_path / "test_dir" / "subdir")49    with open(tmp_path / "test_dir" / "file1.txt", "w") as f:50        f.write("test content")51    with open(tmp_path / "test_dir" / "subdir" / "file2.txt", "w") as f:52        f.write("test content")53 54    client = await aiohttp_client(app)55    resp = await client.get("/userdata?dir=test_dir&recurse=true")56    assert resp.status == 20057    assert set(await resp.json()) == {"file1.txt", "subdir/file2.txt"}58 59 60async def test_listuserdata_full_info(aiohttp_client, app, tmp_path):61    os.makedirs(tmp_path / "test_dir")62    with open(tmp_path / "test_dir" / "file1.txt", "w") as f:63        f.write("test content")64 65    client = await aiohttp_client(app)66    resp = await client.get("/userdata?dir=test_dir&full_info=true")67    assert resp.status == 20068    result = await resp.json()69    assert len(result) == 170    assert result[0]["path"] == "file1.txt"71    assert "size" in result[0]72    assert "modified" in result[0]73 74 75async def test_listuserdata_split_path(aiohttp_client, app, tmp_path):76    os.makedirs(tmp_path / "test_dir" / "subdir")77    with open(tmp_path / "test_dir" / "subdir" / "file1.txt", "w") as f:78        f.write("test content")79 80    client = await aiohttp_client(app)81    resp = await client.get("/userdata?dir=test_dir&recurse=true&split=true")82    assert resp.status == 20083    assert await resp.json() == [["subdir/file1.txt", "subdir", "file1.txt"]]84 85 86async def test_listuserdata_invalid_directory(aiohttp_client, app):87    client = await aiohttp_client(app)88    resp = await client.get("/userdata?dir=")89    assert resp.status == 40090 91 92async def test_listuserdata_normalized_separator(aiohttp_client, app, tmp_path):93    os_sep = "\\"94    with patch("os.sep", os_sep):95        with patch("os.path.sep", os_sep):96            os.makedirs(tmp_path / "test_dir" / "subdir")97            with open(tmp_path / "test_dir" / "subdir" / "file1.txt", "w") as f:98                f.write("test content")99 100            client = await aiohttp_client(app)101            resp = await client.get("/userdata?dir=test_dir&recurse=true")102            assert resp.status == 200103            result = await resp.json()104            assert len(result) == 1105            assert "/" in result[0]  # Ensure forward slash is used106            assert "\\" not in result[0]  # Ensure backslash is not present107            assert result[0] == "subdir/file1.txt"108 109            # Test with full_info110            resp = await client.get(111                "/userdata?dir=test_dir&recurse=true&full_info=true"112            )113            assert resp.status == 200114            result = await resp.json()115            assert len(result) == 1116            assert "/" in result[0]["path"]  # Ensure forward slash is used117            assert "\\" not in result[0]["path"]  # Ensure backslash is not present118            assert result[0]["path"] == "subdir/file1.txt"119 120 121async def test_post_userdata_new_file(aiohttp_client, app, tmp_path):122    client = await aiohttp_client(app)123    content = b"test content"124    resp = await client.post("/userdata/test.txt", data=content)125 126    assert resp.status == 200127    assert await resp.text() == '"test.txt"'128 129    # Verify file was created with correct content130    with open(tmp_path / "test.txt", "rb") as f:131        assert f.read() == content132 133 134async def test_post_userdata_overwrite_existing(aiohttp_client, app, tmp_path):135    # Create initial file136    with open(tmp_path / "test.txt", "w") as f:137        f.write("initial content")138 139    client = await aiohttp_client(app)140    new_content = b"updated content"141    resp = await client.post("/userdata/test.txt", data=new_content)142 143    assert resp.status == 200144    assert await resp.text() == '"test.txt"'145 146    # Verify file was overwritten147    with open(tmp_path / "test.txt", "rb") as f:148        assert f.read() == new_content149 150 151async def test_post_userdata_no_overwrite(aiohttp_client, app, tmp_path):152    # Create initial file153    with open(tmp_path / "test.txt", "w") as f:154        f.write("initial content")155 156    client = await aiohttp_client(app)157    resp = await client.post("/userdata/test.txt?overwrite=false", data=b"new content")158 159    assert resp.status == 409160 161    # Verify original content unchanged162    with open(tmp_path / "test.txt", "r") as f:163        assert f.read() == "initial content"164 165 166async def test_post_userdata_full_info(aiohttp_client, app, tmp_path):167    client = await aiohttp_client(app)168    content = b"test content"169    resp = await client.post("/userdata/test.txt?full_info=true", data=content)170 171    assert resp.status == 200172    result = await resp.json()173    assert result["path"] == "test.txt"174    assert result["size"] == len(content)175    assert "modified" in result176 177 178async def test_move_userdata(aiohttp_client, app, tmp_path):179    # Create initial file180    with open(tmp_path / "source.txt", "w") as f:181        f.write("test content")182 183    client = await aiohttp_client(app)184    resp = await client.post("/userdata/source.txt/move/dest.txt")185 186    assert resp.status == 200187    assert await resp.text() == '"dest.txt"'188 189    # Verify file was moved190    assert not os.path.exists(tmp_path / "source.txt")191    with open(tmp_path / "dest.txt", "r") as f:192        assert f.read() == "test content"193 194 195async def test_move_userdata_no_overwrite(aiohttp_client, app, tmp_path):196    # Create source and destination files197    with open(tmp_path / "source.txt", "w") as f:198        f.write("source content")199    with open(tmp_path / "dest.txt", "w") as f:200        f.write("destination content")201 202    client = await aiohttp_client(app)203    resp = await client.post("/userdata/source.txt/move/dest.txt?overwrite=false")204 205    assert resp.status == 409206 207    # Verify files remain unchanged208    with open(tmp_path / "source.txt", "r") as f:209        assert f.read() == "source content"210    with open(tmp_path / "dest.txt", "r") as f:211        assert f.read() == "destination content"212 213 214async def test_move_userdata_full_info(aiohttp_client, app, tmp_path):215    # Create initial file216    with open(tmp_path / "source.txt", "w") as f:217        f.write("test content")218 219    client = await aiohttp_client(app)220    resp = await client.post("/userdata/source.txt/move/dest.txt?full_info=true")221 222    assert resp.status == 200223    result = await resp.json()224    assert result["path"] == "dest.txt"225    assert result["size"] == len("test content")226    assert "modified" in result227 228    # Verify file was moved229    assert not os.path.exists(tmp_path / "source.txt")230    with open(tmp_path / "dest.txt", "r") as f:231        assert f.read() == "test content"232