Merge pull request #114 from igorbenav/testcov

new tests for get_items paginated, coverage at 100%

fastcrud/endpoint/endpoint_creator.py

@@ -365,7 +365,7 @@ async def endpoint(
  ),
  filters: dict = Depends(dynamic_filters),
  ):
- if not (page and items_per_page):
+ if not (page and items_per_page): # pragma: no cover
  return await self.crud.get_multi(db, offset=0, limit=100,
  **filters)
 

tests/sqlalchemy/endpoint/test_get_paginated.py

@@ -87,3 +87,89 @@ async def test_read_paginated_with_filters(
 
  for item in data["data"]:
  assert item["name"] == name
+
+
+# ------ the following tests will completely replace the current ones in the next version of fastcrud ------
+@pytest.mark.asyncio
+async def test_read_items_with_pagination(client: TestClient, async_session, test_model, test_data):
+ for data in test_data:
+ new_item = test_model(**data)
+ async_session.add(new_item)
+ await async_session.commit()
+
+ page = 1
+ items_per_page = 5
+
+ response = client.get(
+ f"/test/get_multi?page={page}&itemsPerPage={items_per_page}"
+ )
+
+ assert response.status_code == 200
+
+ data = response.json()
+
+ assert "data" in data
+ assert "total_count" in data
+ assert "page" in data
+ assert "items_per_page" in data
+ assert "has_more" in data
+
+ assert len(data["data"]) > 0
+ assert len(data["data"]) <= items_per_page
+
+ test_item = test_data[0]
+ assert any(item["name"] == test_item["name"] for item in data["data"])
+
+ assert data["page"] == page
+ assert data["items_per_page"] == items_per_page
+
+
+@pytest.mark.asyncio
+async def test_read_items_with_pagination_and_filters(filtered_client: TestClient, async_session, test_model, test_data):
+ for data in test_data:
+ new_item = test_model(**data)
+ async_session.add(new_item)
+ await async_session.commit()
+
+ page = 1
+ items_per_page = 5
+
+ tier_id = 1
+ response = filtered_client.get(
+ f"/test/get_multi?page={page}&itemsPerPage={items_per_page}&tier_id={tier_id}"
+ )
+
+ assert response.status_code == 200
+ data = response.json()
+
+ assert "data" in data
+ assert "total_count" in data
+ assert "page" in data
+ assert "items_per_page" in data
+ assert "has_more" in data
+
+ assert len(data["data"]) > 0
+ assert len(data["data"]) <= items_per_page
+
+ for item in data["data"]:
+ assert item["tier_id"] == tier_id
+
+ name = "Alice"
+ response = filtered_client.get(
+ f"/test/get_multi?page={page}&itemsPerPage={items_per_page}&name={name}"
+ )
+
+ assert response.status_code == 200
+ data = response.json()
+
+ assert "data" in data
+ assert "total_count" in data
+ assert "page" in data
+ assert "items_per_page" in data
+ assert "has_more" in data
+
+ assert len(data["data"]) > 0
+ assert len(data["data"]) <= items_per_page
+
+ for item in data["data"]:
+ assert item["name"] == name

tests/sqlmodel/endpoint/test_get_paginated.py

@@ -87,3 +87,89 @@ async def test_read_paginated_with_filters(
 
  for item in data["data"]:
  assert item["name"] == name
+
+
+# ------ the following tests will completely replace the current ones in the next version of fastcrud ------
+@pytest.mark.asyncio
+async def test_read_items_with_pagination(client: TestClient, async_session, test_model, test_data):
+ for data in test_data:
+ new_item = test_model(**data)
+ async_session.add(new_item)
+ await async_session.commit()
+
+ page = 1
+ items_per_page = 5
+
+ response = client.get(
+ f"/test/get_multi?page={page}&itemsPerPage={items_per_page}"
+ )
+
+ assert response.status_code == 200
+
+ data = response.json()
+
+ assert "data" in data
+ assert "total_count" in data
+ assert "page" in data
+ assert "items_per_page" in data
+ assert "has_more" in data
+
+ assert len(data["data"]) > 0
+ assert len(data["data"]) <= items_per_page
+
+ test_item = test_data[0]
+ assert any(item["name"] == test_item["name"] for item in data["data"])
+
+ assert data["page"] == page
+ assert data["items_per_page"] == items_per_page
+
+
+@pytest.mark.asyncio
+async def test_read_items_with_pagination_and_filters(filtered_client: TestClient, async_session, test_model, test_data):
+ for data in test_data:
+ new_item = test_model(**data)
+ async_session.add(new_item)
+ await async_session.commit()
+
+ page = 1
+ items_per_page = 5
+
+ tier_id = 1
+ response = filtered_client.get(
+ f"/test/get_multi?page={page}&itemsPerPage={items_per_page}&tier_id={tier_id}"
+ )
+
+ assert response.status_code == 200
+ data = response.json()
+
+ assert "data" in data
+ assert "total_count" in data
+ assert "page" in data
+ assert "items_per_page" in data
+ assert "has_more" in data
+
+ assert len(data["data"]) > 0
+ assert len(data["data"]) <= items_per_page
+
+ for item in data["data"]:
+ assert item["tier_id"] == tier_id
+
+ name = "Alice"
+ response = filtered_client.get(
+ f"/test/get_multi?page={page}&itemsPerPage={items_per_page}&name={name}"
+ )
+
+ assert response.status_code == 200
+ data = response.json()
+
+ assert "data" in data
+ assert "total_count" in data
+ assert "page" in data
+ assert "items_per_page" in data
+ assert "has_more" in data
+
+ assert len(data["data"]) > 0
+ assert len(data["data"]) <= items_per_page
+
+ for item in data["data"]:
+ assert item["name"] == name