add test

Author: Chise1

Makefile

@@ -34,8 +34,8 @@ test:
 	pytest tests/ --cov-report=
 
 cov: deps
-	coverage run -m pytest tests/
-	coverage report --show-missing -m
+	coverage run --branch -m pytest tests/
+	coverage report --show-missing -m --omit=tests/*,conftest.py
 	coverage xml
 	coverage html
 

fast_tmp/admin/depends.py

@@ -26,7 +26,6 @@ def decode_access_token_from_data(
 
 
 def get_user(username: str, session: Session) -> Optional[User]:
-
     res = session.execute(
         select(User)  # type: ignore
         .where(User.username == username)

fast_tmp/admin/endpoint.py

@@ -1,4 +1,4 @@
-from typing import Any, List, Optional
+from typing import List, Optional
 
 from fastapi import APIRouter, Depends
 from pydantic import BaseModel
@@ -13,6 +13,7 @@
 from ..models import User
 from ..site.utils import clean_data_to_model, get_pk
 from .depends import decode_access_token_from_data
+from .responses import BaseRes, key_error, not_found_instance
 
 router = APIRouter()
 
@@ -29,12 +30,6 @@ class ListDataWithPage(BaseModel):  # 带分页的数据
     total: int = 0
 
 
-class BaseRes(BaseModel):
-    status: int = 0
-    msg: str = ""
-    data: Any = {}
-
-
 @router.get("/{resource}/list")
 def list_view(
     request: Request,
@@ -76,13 +71,12 @@ def update_data(
     if not user:
         return RedirectResponse(request.url_for("admin:login"))
     data = clean_data_to_model(page_model.update_fields, data)
-    params = dict(request.query_params)
-    pks = get_pk(page_model.model)
-    w = []
-    for k, v in params.items():
-        if pks.get(k) is not None:
-            w.append(pks[k] == v)
+    w = get_pks(page_model, request)
+    if isinstance(w, BaseRes):
+        return w
     old_data = session.execute(select(page_model.model).where(*w)).scalar_one_or_none()
+    if not old_data:
+        return not_found_instance
     page_model.update_model(old_data, data)
     session.commit()
     return BaseRes()
@@ -98,15 +92,13 @@ def update_view(
     if not user:
         return RedirectResponse(request.url_for("admin:login"))
 
-    params = dict(request.query_params)
-    pks = get_pk(page_model.model)
-    w = []
-    for k, v in params.items():
-        if pks.get(k) is not None:
-            w.append(pks[k] == v)
-    data = session.execute(select(page_model.update_fields).where(*w))
-    for i in data:
-        return BaseRes(data=dict(i))
+    w = get_pks(page_model, request)
+    if isinstance(w, BaseRes):
+        return w
+    data = session.execute(select(page_model.update_fields).where(*w)).fetchone()
+    if not data:
+        return not_found_instance
+    return BaseRes(data=dict(data))
 
 
 @router.post("/{resource}/create")
@@ -137,17 +129,24 @@ def delete_one(
     if not user:
         return RedirectResponse(request.url_for("admin:login"))
 
+    w = get_pks(page_model, request)
+    if isinstance(w, BaseRes):
+        return w
+    session.execute(delete(page_model.model).where(*w))
+    session.commit()
+    return BaseRes()
+
+
+def get_pks(page_model: ModelAdmin, request: Request):
     params = dict(request.query_params)
     pks = get_pk(page_model.model)
     w = []
     for k, v in params.items():
         if pks.get(k) is not None:
             w.append(pks[k] == v)
         else:
-            return BaseRes(status=400, msg="主键错误")
-    session.execute(delete(page_model.model).where(*w))
-    session.commit()
-    return BaseRes()
+            return key_error
+    return w
 
 
 class DIDS(BaseModel):

fast_tmp/admin/responses.py

@@ -1,10 +1,13 @@
-from starlette.requests import Request
-from starlette.responses import RedirectResponse
-from starlette.status import HTTP_303_SEE_OTHER
+from typing import Any
 
+from pydantic import BaseModel
 
-def redirect(request: Request, view: str, **params):
-    return RedirectResponse(
-        url=request.app.admin_path + request.app.url_path_for(view, **params),
-        status_code=HTTP_303_SEE_OTHER,
-    )
+
+class BaseRes(BaseModel):
+    status: int = 0
+    msg: str = ""
+    data: Any = {}
+
+
+key_error = BaseRes(status=400, msg="主键错误")
+not_found_instance = BaseRes(status=404, msg="找不到对象")

fast_tmp/admin/schema/page.py

@@ -1,7 +1,5 @@
 from typing import List, Optional, Tuple, Union
 
-from pydantic import BaseModel, Field
-
 from fast_tmp.admin.schema.abstract_schema import BaseAmisModel
 
 from .enums import TypeEnum
@@ -31,31 +29,32 @@ class HBox(BaseAmisModel):
     columns: List[BaseAmisModel]
 
 
-class AppPage(BaseModel):
-    """
-    app的子页
-    """
-
-    label: str
-    icon: str
-    url: str
-    schema_: Page = Field(..., alias="schema")
-    # schemaApi: Optional[str] = None
-    link: Optional[str] = None
-    redirect: Optional[str] = None
-    rewrite: Optional[str] = None
-    isDefaultPage: Optional[str] = None
-
-    # visible:Optional[bool]=None
-    def dict(self, *args, **kwargs):
-        kwargs["exclude_none"] = True
-        res: dict = super().dict(*args, **kwargs)
-        return res
-
-
-class AppPageGroup(BaseModel):
-    label: str
-    children: List[AppPage]
+#
+# class AppPage(BaseModel):
+#     """
+#     app的子页
+#     """
+#
+#     label: str
+#     icon: str
+#     url: str
+#     schema_: Page = Field(..., alias="schema")
+#     # schemaApi: Optional[str] = None
+#     link: Optional[str] = None
+#     redirect: Optional[str] = None
+#     rewrite: Optional[str] = None
+#     isDefaultPage: Optional[str] = None
+#
+#     # visible:Optional[bool]=None
+#     def dict(self, *args, **kwargs):
+#         kwargs["exclude_none"] = True
+#         res: dict = super().dict(*args, **kwargs)
+#         return res
+
+
+# class AppPageGroup(BaseModel):
+#     label: str
+#     children: List[AppPage]
 
 
 # class App(BaseAmisModel):

fast_tmp/admin/server.py

@@ -18,7 +18,8 @@
 from fast_tmp.utils.token import create_access_token
 
 from ..jinja_extension.tags import register_tags
-from .endpoint import BaseRes, router
+from .endpoint import router
+from .responses import BaseRes
 
 oauth2_scheme = OAuth2PasswordBearer(tokenUrl=settings.LOGIN_URL)
 base_path = os.path.dirname(__file__)

setup.cfg

@@ -1,5 +1,5 @@
 [flake8]
-ignore = D203,E501,C901,W503
+ignore = D203,E501,C901,W503,F841
 exclude =
     .git,
     __pycache__,

tests/test_a.py

@@ -50,7 +50,44 @@ def test_crud():
     # login
     login_post_res = client.post("/admin/login", data={"username": "crud_user", "password": "crud_user"})
     assert login_post_res.status_code == 307
+    assert login_post_res.next.path_url == "/admin/"
     # list
     user_list = client.get("/admin/User/list")
     assert user_list.status_code == 200
-    assert user_list.text == '{"status":0,"msg":"","data":{"items":[{"id":1,"username":"root","is_active":true},{"id":2,"username":"crud_user","is_active":true}],"total":2}}'
+    assert user_list.text.count('"username":"crud_user"') == 1
+    user_data = user_list.json()["data"]
+    assert user_data["total"] == 2
+    for i in user_data["items"]:
+        if i.get("username") == "crud_user":
+            user_id = i.get("id")
+            break
+    else:
+        raise Exception("not found user id")
+    # update
+    user_update = client.get(f"/admin/User/update?id={user_id}")
+    assert user_update.status_code == 200
+    assert user_update.text.count('"password":') == 1
+    user_update_p = client.put(f"/admin/User/update?id={user_id}", json={"password": "root"})
+    assert user_update_p.status_code == 200
+    assert user_update_p.json()["status"] == 0
+    # test error pk
+    user_delete_err = client.delete(f"/admin/User/delete?ids={user_id}")
+    error_data = user_delete_err.json()
+    assert error_data["status"] == 400
+    assert error_data["msg"] == "主键错误"
+    error_data = client.put(f"/admin/User/update?ids={user_id}", json={"password": "root"}).json()
+    assert error_data["status"] == 400
+    assert error_data["msg"] == "主键错误"
+    # delete
+    user_delete = client.delete(f"/admin/User/delete?id={user_id}")
+    assert user_delete.status_code == 200
+
+
+def test_not_singin():
+    client = TestClient(app)
+    assert client.get("/admin/User/list").text.count("Login to your account") == 1
+    assert client.post("/admin/User/create", json={"username": "crud_user1", "password": "tt"}).status_code == 307
+    assert client.get("/admin/User/update?id=1").text.count("Login to your account") == 1
+    assert client.put("/admin/User/update?id=1", json={"password": "asdfadf"}).status_code == 307
+    assert client.delete("/admin/User/delete?id=1").status_code == 307
+    assert client.get("/admin/User/schema").text.count("Login to your account") == 1