Fix File and Image fields checkbox and input (#761)

sqladmin/widgets.py

@@ -75,15 +75,23 @@ class FileInputWidget(widgets.FileInput):
     """
 
     def __call__(self, field: Field, **kwargs: Any) -> str:
-        file_input = super().__call__(field, **kwargs)
-        checkbox_id = f"{field.id}_checkbox"
-        checkbox_label = Markup(
-            f'<label class="form-check-label" for="{checkbox_id}">Clear</label>'
-        )
-        checkbox_input = Markup(
-            f'<input class="form-check-input" type="checkbox" id="{checkbox_id}" name="{checkbox_id}">'  # noqa: E501
-        )
-        checkbox = Markup(
-            f'<div class="form-check">{checkbox_input}{checkbox_label}</div>'
-        )
-        return file_input + checkbox
+        if not field.flags.required:
+            checkbox_id = f"{field.id}_checkbox"
+            checkbox_label = Markup(
+                f'<label class="form-check-label" for="{checkbox_id}">Clear</label>'
+            )
+            checkbox_input = Markup(
+                f'<input class="form-check-input" type="checkbox" id="{checkbox_id}" name="{checkbox_id}">'  # noqa: E501
+            )
+            checkbox = Markup(
+                f'<div class="form-check">{checkbox_input}{checkbox_label}</div>'
+            )
+        else:
+            checkbox = Markup()
+
+        if field.data:
+            current_value = Markup(f"<p>Currently: {field.data}</p>")
+            field.flags.required = False
+            return current_value + checkbox + super().__call__(field, **kwargs)
+        else:
+            return super().__call__(field, **kwargs)

tests/test_file_upload.py

@@ -1,21 +1,18 @@
-from typing import Any, AsyncGenerator
+from typing import Any, Generator
 
 import pytest
 from fastapi_storages import FileSystemStorage, StorageFile
-from fastapi_storages.integrations.sqlalchemy import FileType, ImageType
-from httpx import AsyncClient
+from fastapi_storages.integrations.sqlalchemy import FileType
 from sqlalchemy import Column, Integer, select
-from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import declarative_base, sessionmaker
 from starlette.applications import Starlette
+from starlette.testclient import TestClient
 
 from sqladmin import Admin, ModelView
-from tests.common import async_engine as engine
-
-pytestmark = pytest.mark.anyio
+from tests.common import sync_engine as engine
 
 Base = declarative_base()  # type: Any
-session_maker = sessionmaker(bind=engine, class_=AsyncSession, expire_on_commit=False)
+session_maker = sessionmaker(bind=engine)
 
 app = Starlette()
 admin = Admin(app=app, engine=engine)
@@ -25,24 +22,20 @@ class User(Base):
     __tablename__ = "users"
 
     id = Column(Integer, primary_key=True)
-    file = Column(FileType(FileSystemStorage(".uploads")))
-    image = Column(ImageType(FileSystemStorage(".uploads")))
+    file = Column(FileType(FileSystemStorage(".uploads")), nullable=False)
+    optional_file = Column(FileType(FileSystemStorage(".uploads")), nullable=True)
 
 
 @pytest.fixture
-async def prepare_database() -> AsyncGenerator[None, None]:
-    async with engine.begin() as conn:
-        await conn.run_sync(Base.metadata.create_all)
+def prepare_database() -> Generator[None, None, None]:
+    Base.metadata.create_all(engine)
     yield
-    async with engine.begin() as conn:
-        await conn.run_sync(Base.metadata.drop_all)
-
-    await engine.dispose()
+    Base.metadata.drop_all(engine)
 
 
 @pytest.fixture
-async def client(prepare_database: Any) -> AsyncGenerator[AsyncClient, None]:
-    async with AsyncClient(app=app, base_url="http://testserver") as c:
+def client(prepare_database: Any) -> Generator[TestClient, None, None]:
+    with TestClient(app=app, base_url="http://testserver") as c:
         yield c
 
 
@@ -53,66 +46,95 @@ class UserAdmin(ModelView, model=User):
 admin.add_view(UserAdmin)
 
 
-async def _query_user() -> Any:
+def _query_user() -> User:
     stmt = select(User).limit(1)
-    async with session_maker() as s:
-        result = await s.execute(stmt)
-    return result.scalar_one()
+    with session_maker() as s:
+        return s.scalar(stmt)
 
 
-async def test_create_form_fields(client: AsyncClient) -> None:
-    response = await client.get("/admin/user/create")
+def test_create_form_fields(client: TestClient) -> None:
+    response = client.get("/admin/user/create")
 
     assert response.status_code == 200
     assert (
-        '<input class="form-control" id="file" name="file" type="file">'
+        '<input class="form-control" id="file" name="file" required type="file">'
         in response.text
     )
-    assert '<input class="form-check-input" type="checkbox"' in response.text
     assert (
-        '<label class="form-check-label" for="file_checkbox">Clear</label>'
+        '<input class="form-control" id="optional_file" name="optional_file" type="file">'  # noqa: E501
         in response.text
     )
 
 
-async def test_create_form_post(client: AsyncClient) -> None:
-    files = {"file": ("upload.txt", b"abc")}
-    response = await client.post("/admin/user/create", files=files)
+def test_create_form_post(client: TestClient) -> None:
+    files = {
+        "file": ("file.txt", b"abc"),
+        "optional_file": ("optional_file.txt", b"cdb"),
+    }
+    client.post("/admin/user/create", files=files)
 
-    user = await _query_user()
+    user = _query_user()
 
-    assert response.status_code == 302
     assert isinstance(user.file, StorageFile) is True
-    assert user.file.name == "upload.txt"
-    assert user.file.path == ".uploads/upload.txt"
+    assert user.file.name == "file.txt"
+    assert user.file.path == ".uploads/file.txt"
     assert user.file.open().read() == b"abc"
+    assert user.optional_file.name == "optional_file.txt"
+    assert user.optional_file.path == ".uploads/optional_file.txt"
+    assert user.optional_file.open().read() == b"cdb"
+
+
+def test_create_form_update(client: TestClient) -> None:
+    files = {
+        "file": ("file.txt", b"abc"),
+        "optional_file": ("optional_file.txt", b"cdb"),
+    }
+    client.post("/admin/user/create", files=files)
+
+    files = {
+        "file": ("new_file.txt", b"xyz"),
+        "optional_file": ("new_optional_file.txt", b"zyx"),
+    }
+    client.post("/admin/user/edit/1", files=files)
+
+    user = _query_user()
+    assert user.file.name == "new_file.txt"
+    assert user.file.path == ".uploads/new_file.txt"
+    assert user.file.open().read() == b"xyz"
+    assert user.optional_file.name == "new_optional_file.txt"
+    assert user.optional_file.path == ".uploads/new_optional_file.txt"
+    assert user.optional_file.open().read() == b"zyx"
+
+    files = {"file": ("file.txt", b"abc")}
+    client.post(
+        "/admin/user/edit/1", files=files, data={"optional_file_checkbox": "true"}
+    )
 
+    user = _query_user()
+    assert user.file.name == "file.txt"
+    assert user.file.path == ".uploads/file.txt"
+    assert user.file.open().read() == b"abc"
+    assert user.optional_file is None
 
-async def test_create_form_update(client: AsyncClient) -> None:
-    files = {"file": ("upload.txt", b"abc")}
-    response = await client.post("/admin/user/create", files=files)
-
-    user = await _query_user()
-
-    files = {"file": ("new_upload.txt", b"abc")}
-    response = await client.post("/admin/user/edit/1", files=files)
-
-    user = await _query_user()
-    assert response.status_code == 302
-    assert user.file.name == "new_upload.txt"
-    assert user.file.path == ".uploads/new_upload.txt"
-
-    files = {"file": ("empty.txt", b"")}
-    response = await client.post("/admin/user/edit/1", files=files)
 
-    user = await _query_user()
-    assert user.file.name == "new_upload.txt"
-    assert user.file.path == ".uploads/new_upload.txt"
+def test_get_form_update(client: TestClient) -> None:
+    files = {
+        "file": ("file.txt", b"abc"),
+        "optional_file": ("optional_file.txt", b"cdb"),
+    }
+    client.post("/admin/user/create", files=files)
+    response = client.get("/admin/user/edit/1")
 
-    files = {"file": ("new_upload.txt", b"abc")}
-    response = await client.post(
-        "/admin/user/edit/1", files=files, data={"file_checkbox": True}
+    assert response.text.count("Currently:") == 2
+    assert '<input class="form-check-input" type="checkbox"' in response.text
+    assert (
+        '<label class="form-check-label" for="optional_file_checkbox">Clear</label>'
+        in response.text
     )
 
-    user = await _query_user()
-    assert user.file is None
+    files = {"file": ("file.txt", b"abc")}
+    client.post("/admin/user/edit/1", files=files)
+    response = client.get("/admin/user/edit/1")
+
+    assert response.text.count("Currently:") == 1
+    assert response.text.count("checkbox") == 0