cswine
/
point-of-sales
1
1
Fork 0
Code Issues 2 Pull Requests Packages Projects 1 Releases Wiki Activity

autocommit 17-07-2024-07-54

This commit is contained in:
Jasen Qin 2024-07-17 07:54:02 +10:00
parent 2c201e47cd
commit 242c5c08f0
10 changed files with 645 additions and 36 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

534
py-kivy/API.md Normal file
View File

@ -0,0 +1,534 @@
# POS System API Documentation
## Base URL
All URLs referenced in the documentation have the following base:
```
http://localhost:8000/api/v1
```
## Authentication
Most endpoints require authentication. Use the following endpoint to obtain a JWT token:
### Login for Access Token
```
POST /token
```
**Request Body:**
```json
{
"username": "string",
"password": "string"
}
```
**Response:**
```json
{
"access_token": "string",
"token_type": "bearer"
}
```
Use the received token in the Authorization header for subsequent requests:
```
Authorization: Bearer <access_token>
```
## Items
### Create a new item
```
POST /items
```
**Request Body:**
```json
{
"name": "string",
"price": 0,
"quantity": 0,
"unit": "string",
"related_items": ["string"]
}
```
**Response:**
```json
{
"id": "string",
"name": "string",
"price": 0,
"quantity": 0,
"unit": "string",
"related_items": ["string"]
}
```
### Get all items
```
GET /items
```
**Query Parameters:**
- `skip` (optional): number of items to skip
- `limit` (optional): maximum number of items to return
**Response:**
```json
[
{
"id": "string",
"name": "string",
"price": 0,
"quantity": 0,
"unit": "string",
"related_items": ["string"]
}
]
```
### Get a specific item
```
GET /items/{item_id}
```
**Response:**
```json
{
"id": "string",
"name": "string",
"price": 0,
"quantity": 0,
"unit": "string",
"related_items": ["string"]
}
```
### Update an item
```
PUT /items/{item_id}
```
**Request Body:**
```json
{
"name": "string",
"price": 0,
"quantity": 0,
"unit": "string",
"related_items": ["string"]
}
```
**Response:**
```json
{
"id": "string",
"name": "string",
"price": 0,
"quantity": 0,
"unit": "string",
"related_items": ["string"]
}
```
### Delete an item
```
DELETE /items/{item_id}
```
**Response:**
```json
{
"message": "Item successfully deleted"
}
```
## Orders
### Create a new order
```
POST /orders
```
**Request Body:**
```json
{
"user_id": "string",
"items": [
{
"item_id": "string",
"quantity": 0,
"price_at_order": 0
}
],
"total_amount": 0,
"payment_method": "string",
"notes": "string"
}
```
**Response:**
```json
{
"id": "string",
"user_id": "string",
"items": [
{
"item_id": "string",
"quantity": 0,
"price_at_order": 0
}
],
"total_amount": 0,
"payment_method": "string",
"payment_status": "string",
"order_status": "string",
"created_at": "string",
"updated_at": "string",
"discount_applied": 0,
"notes": "string"
}
```
### Get all orders
```
GET /orders
```
**Query Parameters:**
- `skip` (optional): number of orders to skip
- `limit` (optional): maximum number of orders to return
**Response:**
```json
[
{
"id": "string",
"user_id": "string",
"items": [
{
"item_id": "string",
"quantity": 0,
"price_at_order": 0
}
],
"total_amount": 0,
"payment_method": "string",
"payment_status": "string",
"order_status": "string",
"created_at": "string",
"updated_at": "string",
"discount_applied": 0,
"notes": "string"
}
]
```
### Get a specific order
```
GET /orders/{order_id}
```
**Response:**
```json
{
"id": "string",
"user_id": "string",
"items": [
{
"item_id": "string",
"quantity": 0,
"price_at_order": 0
}
],
"total_amount": 0,
"payment_method": "string",
"payment_status": "string",
"order_status": "string",
"created_at": "string",
"updated_at": "string",
"discount_applied": 0,
"notes": "string"
}
```
### Update an order
```
PUT /orders/{order_id}
```
**Request Body:**
```json
{
"items": [
{
"item_id": "string",
"quantity": 0,
"price_at_order": 0
}
],
"total_amount": 0,
"payment_method": "string",
"payment_status": "string",
"order_status": "string",
"discount_applied": 0,
"notes": "string"
}
```
**Response:**
```json
{
"id": "string",
"user_id": "string",
"items": [
{
"item_id": "string",
"quantity": 0,
"price_at_order": 0
}
],
"total_amount": 0,
"payment_method": "string",
"payment_status": "string",
"order_status": "string",
"created_at": "string",
"updated_at": "string",
"discount_applied": 0,
"notes": "string"
}
```
### Delete an order
```
DELETE /orders/{order_id}
```
**Response:**
```json
{
"message": "Order successfully deleted"
}
```
### Process payment for an order
```
POST /orders/{order_id}/process_payment
```
**Request Body:**
```json
{
"payment_method": "string"
}
```
**Response:**
```json
{
"message": "Payment processed successfully"
}
```
### Apply discount to an order
```
POST /orders/{order_id}/apply_discount
```
**Request Body:**
```json
{
"discount_percentage": 0
}
```
**Response:**
```json
{
"id": "string",
"total_amount": 0,
"discount_applied": 0
}
```
## Users
### Register a new user
```
POST /users
```
**Request Body:**
```json
{
"username": "string",
"email": "string",
"full_name": "string",
"password": "string"
}
```
**Response:**
```json
{
"id": "string",
"username": "string",
"email": "string",
"full_name": "string",
"is_active": true,
"is_superuser": false
}
```
### Get current user
```
GET /users/me
```
**Response:**
```json
{
"id": "string",
"username": "string",
"email": "string",
"full_name": "string",
"is_active": true,
"is_superuser": false
}
```
### Get all users
```
GET /users
```
**Query Parameters:**
- `skip` (optional): number of users to skip
- `limit` (optional): maximum number of users to return
**Response:**
```json
[
{
"id": "string",
"username": "string",
"email": "string",
"full_name": "string",
"is_active": true,
"is_superuser": false
}
]
```
### Update a user
```
PUT /users/{user_id}
```
**Request Body:**
```json
{
"email": "string",
"full_name": "string",
"password": "string",
"is_active": true,
"is_superuser": false
}
```
**Response:**
```json
{
"id": "string",
"username": "string",
"email": "string",
"full_name": "string",
"is_active": true,
"is_superuser": false
}
```
### Delete a user
```
DELETE /users/{user_id}
```
**Response:**
```json
{
"message": "User successfully deleted"
}
```
## Error Responses
All endpoints can return the following error responses:
- 400 Bad Request
- 401 Unauthorized
- 403 Forbidden
- 404 Not Found
- 422 Unprocessable Entity
- 500 Internal Server Error
Error response body:
```json
{
"detail": "Error message"
}
```

36
py-kivy/poetry.lock generated
View File

@ -401,6 +401,20 @@ files = [
dnspython = ">=2.0.0"
idna = ">=2.0.0"
[[package]]
name = "execnet"
version = "2.1.1"
description = "execnet: rapid multi-Python deployment"
optional = false
python-versions = ">=3.8"
files = [
{file = "execnet-2.1.1-py3-none-any.whl", hash = "sha256:26dee51f1b80cebd6d0ca8e74dd8745419761d3bef34163928cbebbdc4749fdc"},
{file = "execnet-2.1.1.tar.gz", hash = "sha256:5189b52c6121c24feae288166ab41b32549c7e2348652736540b9e6e7d4e72e3"},
]
[package.extras]
testing = ["hatch", "pre-commit", "pytest", "tox"]
[[package]]
name = "executing"
version = "2.0.1"
@ -1511,6 +1525,26 @@ pluggy = ">=1.5,<2.0"
[package.extras]
dev = ["argcomplete", "attrs (>=19.2)", "hypothesis (>=3.56)", "mock", "pygments (>=2.7.2)", "requests", "setuptools", "xmlschema"]
[[package]]
name = "pytest-xdist"
version = "3.6.1"
description = "pytest xdist plugin for distributed testing, most importantly across multiple CPUs"
optional = false
python-versions = ">=3.8"
files = [
{file = "pytest_xdist-3.6.1-py3-none-any.whl", hash = "sha256:9ed4adfb68a016610848639bb7e02c9352d5d9f03d04809919e2dafc3be4cca7"},
{file = "pytest_xdist-3.6.1.tar.gz", hash = "sha256:ead156a4db231eec769737f57668ef58a2084a34b2e55c4a8fa20d861107300d"},
]
[package.dependencies]
execnet = ">=2.1"
pytest = ">=7.0.0"
[package.extras]
psutil = ["psutil (>=3.0)"]
setproctitle = ["setproctitle"]
testing = ["filelock"]
[[package]]
name = "python-dateutil"
version = "2.9.0.post0"
@ -2308,4 +2342,4 @@ files = [
[metadata]
lock-version = "2.0"
python-versions = "^3.12"
content-hash = "e206ba057f9668e090364cc695646a317f0896de3d81b60e05450b7498324d84"
content-hash = "0d6489d261e19e3bffaafb61a05fdc436dae4801beefcaf640e9abc4b7bad8fd"

31
py-kivy/pos_system/models.py
View File

@ -1,13 +1,7 @@
from pydantic import BaseModel, Field, EmailStr, field_validator, validator, ConfigDict
from pydantic import BaseModel, Field, EmailStr, field_validator, ConfigDict
from typing import List, Optional
from datetime import datetime
from bson import ObjectId
def validate_object_id(value: str) -> ObjectId:
if not ObjectId.is_valid(value):
raise ValueError("Invalid ObjectId")
return ObjectId(value)
from datetime import datetime
class MongoBaseModel(BaseModel):
@ -20,11 +14,12 @@ class MongoBaseModel(BaseModel):
json_encoders={ObjectId: str}
)
# @field_validator("id", pre=True)
# def validate_id(cls, v):
# if isinstance(v, ObjectId):
# return str(v)
# return v
@field_validator('id', mode='before')
@classmethod
def convert_object_id_to_string(cls, v):
if isinstance(v, ObjectId):
return str(v)
return v
class Item(MongoBaseModel):
@ -40,9 +35,11 @@ class OrderItem(BaseModel):
quantity: int
price_at_order: float
# @field_validator("item_id")
# def validate_item_id(cls, v):
# return validate_object_id(v)
@field_validator("item_id")
def validate_id(cls, v):
if isinstance(v, ObjectId):
return str(v)
return v
class Order(MongoBaseModel):
@ -57,7 +54,7 @@ class Order(MongoBaseModel):
discount_applied: Optional[float] = None
notes: Optional[str] = None
# @validator("user_id")
# @field_validator("user_id")
# def validate_user_id(cls, v):
# return validate_object_id(v)

8
py-kivy/pos_system/routers/items.py
View File

@ -16,14 +16,14 @@ async def create_item(item: Item):
item_dict = item.model_dump(exclude={"id"})
result = db.items.insert_one(item_dict)
created_item = db.items.find_one({"_id": result.inserted_id})
return Item(**created_item)
return Item.model_validate(created_item)
@router.get("/", response_model=List[Item])
async def read_items():
db = get_db()
items = list(db.items.find())
return [Item(**item) for item in items]
return [Item.model_validate(item) for item in items]
@router.get("/{item_id}", response_model=Item)
@ -32,7 +32,7 @@ async def read_item(item_id: str):
item = db.items.find_one({"_id": ObjectId(item_id)})
if item is None:
raise HTTPException(status_code=404, detail="Item not found")
return Item(**item)
return Item.model_validate(item)
@router.put("/{item_id}", response_model=Item)
@ -44,7 +44,7 @@ async def update_item(item_id: str, item: Item):
if result.modified_count == 0:
raise HTTPException(status_code=404, detail="Item not found")
updated_item = db.items.find_one({"_id": ObjectId(item_id)})
return Item(**updated_item)
return Item.model_validate(updated_item)
@router.delete("/{item_id}", response_model=dict)

2
py-kivy/pyproject.toml
View File

@ -17,6 +17,7 @@ python-jose = "^3.3.0"
passlib = "^1.7.4"
pydantic = "^2.8.2"
pytest-xdist = "^3.6.1"
[tool.poetry.group.dev.dependencies]
ipykernel = "^6.29.5"
@ -30,3 +31,4 @@ indent-size = 2
[tool.poetry.scripts]
server = "pos_system.server:main"
ui = "pos_system.ui:main"
test-items = "tests.test_items_fastapi:test"

11
py-kivy/pytest.ini Normal file
View File

@ -0,0 +1,11 @@
# File: pytest.ini (place this in your project root directory)
[pytest]
log_cli = true
log_cli_level = INFO
log_cli_format = %(asctime)s [%(levelname)8s] %(message)s (%(filename)s:%(lineno)s)
log_cli_date_format=%Y-%m-%d %H:%M:%S
log_file = tests/output/pytest.log
log_file_level = DEBUG
log_file_format = %(asctime)s [%(levelname)8s] %(message)s (%(filename)s:%(lineno)s)
log_file_date_format=%Y-%m-%d %H:%M:%S

BIN
py-kivy/tests/.DS_Store vendored Normal file
View File

Binary file not shown.

1
py-kivy/tests/.gitignore vendored Normal file
View File

@ -0,0 +1 @@
output/

0
py-kivy/tests/__init__.py Normal file
View File

56
py-kivy/tests/test_fastapi.py → py-kivy/tests/test_items_fastapi.py
View File

@ -1,11 +1,13 @@
import pytest
import logging
from fastapi.testclient import TestClient
from hypothesis import given, strategies as st
from hypothesis import given, strategies as st, settings
from bson import ObjectId
from pos_system.server import app
from pos_system.database import get_db
client = TestClient(app)
logger = logging.getLogger(__name__)
@pytest.fixture(autouse=True)
@ -17,15 +19,18 @@ def clear_db():
def test_create_item():
logger.info("Testing create item")
response = client.post(
"/items/",
json={"name": "Test Item", "price": 10.99, "quantity": 5, "unit": "piece"}
)
assert response.status_code == 200
assert "id" in response.json()
assert "_id" in response.json()
logger.info(f"Created item with ID: {response.json()['_id']}")
def test_read_items():
logger.info("Testing read items")
client.post("/items/", json={"name": "Test Item",
"price": 10.99, "quantity": 5, "unit": "piece"})
response = client.get("/items/")
@ -33,54 +38,62 @@ def test_read_items():
items = response.json()
assert len(items) == 1
assert items[0]["name"] == "Test Item"
logger.info(f"Retrieved {len(items)} items")
def test_read_item():
logger.info("Testing read single item")
create_response = client.post(
"/items/", json={"name": "Test Item", "price": 10.99, "quantity": 5, "unit": "piece"})
item_id = create_response.json()["id"]
item_id = create_response.json()["_id"]
response = client.get(f"/items/{item_id}")
assert response.status_code == 200
assert response.json()["name"] == "Test Item"
logger.info(f"Retrieved item with ID: {item_id}")
def test_update_item():
logger.info("Testing update item")
create_response = client.post(
"/items/", json={"name": "Test Item", "price": 10.99, "quantity": 5, "unit": "piece"})
item_id = create_response.json()["id"]
item_id = create_response.json()["_id"]
update_data = {"name": "Updated Item",
"price": 15.99, "quantity": 10, "unit": "piece"}
response = client.put(f"/items/{item_id}", json=update_data)
assert response.status_code == 200
assert response.json()["name"] == "Updated Item"
logger.info(f"Updated item with ID: {item_id}")
def test_delete_item():
logger.info("Testing delete item")
create_response = client.post(
"/items/", json={"name": "Test Item", "price": 10.99, "quantity": 5, "unit": "piece"})
item_id = create_response.json()["id"]
item_id = create_response.json()["_id"]
response = client.delete(f"/items/{item_id}")
assert response.status_code == 200
assert response.json()["message"] == "Item deleted successfully"
get_response = client.get(f"/items/{item_id}")
assert get_response.status_code == 404
logger.info(f"Deleted item with ID: {item_id}")
@given(
name=st.text(min_size=1, max_size=50),
price=st.floats(min_value=0.01, max_value=1000000,
name=st.text(min_size=1, max_size=20),
price=st.floats(min_value=0.01, max_value=100,
allow_nan=False, allow_infinity=False),
quantity=st.integers(min_value=0, max_value=1000000),
quantity=st.integers(min_value=0, max_value=100),
unit=st.text(min_size=1, max_size=10)
)
@settings(max_examples=50) # Limit the number of examples to 50
def test_create_item_property(name, price, quantity, unit):
response = client.post(
"/items/",
json={"name": name, "price": price, "quantity": quantity, "unit": unit}
)
assert response.status_code == 200
assert "id" in response.json()
item_id = response.json()["id"]
assert "_id" in response.json()
item_id = response.json()["_id"]
get_response = client.get(f"/items/{item_id}")
assert get_response.status_code == 200
item = get_response.json()
@ -93,15 +106,16 @@ def test_create_item_property(name, price, quantity, unit):
@given(
st.lists(
st.fixed_dictionaries({
"name": st.text(min_size=1, max_size=50),
"price": st.floats(min_value=0.01, max_value=1000000, allow_nan=False, allow_infinity=False),
"quantity": st.integers(min_value=0, max_value=1000000),
"name": st.text(min_size=1, max_size=20),
"price": st.floats(min_value=0.01, max_value=100, allow_nan=False, allow_infinity=False),
"quantity": st.integers(min_value=0, max_value=100),
"unit": st.text(min_size=1, max_size=10)
}),
min_size=1,
max_size=10
)
)
@settings(max_examples=20) # Limit the number of examples to 20
def test_read_items_property(items):
for item in items:
client.post("/items/", json=item)
@ -115,6 +129,22 @@ def test_read_items_property(items):
assert "quantity" in retrieved_item
assert "unit" in retrieved_item
# Add this function to log only failed property tests
@pytest.hookimpl(hookwrapper=True)
def pytest_runtest_makereport(item, call):
outcome = yield
report = outcome.get_result()
if report.when == "call" and report.failed:
if "hypothesis" in item.keywords:
logger.error(f"Property test failed: {item.name}")
logger.error(f"Falsifying example: {call.excinfo.value}")
if __name__ == "__main__":
pytest.main([__file__])
def test():
pytest.main([__file__])