linmihigo
c086f64363
- Add 7 core API endpoints: users, transactions, partners, products, inventory, payments, credit - Implement role-based authentication (admin/write/read-only access) - Add comprehensive database models with proper relationships - Include full test coverage for all endpoints and business logic - Set up Alembic migrations and Docker configuration - Configure FastAPI app with CORS and database integration
465 lines
20 KiB
Python
465 lines
20 KiB
Python
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from app.schemas.base import TransactionType, TransactionStatus
|
|
from app.schemas.models import Transaction, Credit
|
|
from sqlmodel import Session
|
|
|
|
|
|
@pytest.fixture(name="sample_credit")
|
|
def sample_credit_fixture(session: Session, sample_partner, sample_transaction, admin_user):
|
|
"""Create a sample credit for testing."""
|
|
credit = Credit(
|
|
partner_id=sample_partner.id,
|
|
transaction_id=sample_transaction.id,
|
|
credit_amount=5000,
|
|
credit_limit=10000,
|
|
balance=5000,
|
|
created_by=admin_user.id,
|
|
updated_by=admin_user.id
|
|
)
|
|
session.add(credit)
|
|
session.commit()
|
|
session.refresh(credit)
|
|
return credit
|
|
|
|
|
|
@pytest.fixture(name="multiple_credits")
|
|
def multiple_credits_fixture(session: Session, multiple_partners, admin_user):
|
|
"""Create multiple credits for testing."""
|
|
# Create transactions for each partner first
|
|
transactions = []
|
|
for partner in multiple_partners:
|
|
transaction = Transaction(
|
|
partner_id=partner.id,
|
|
transcation_type=TransactionType.SALE,
|
|
transaction_status=TransactionStatus.UNPAID,
|
|
total_amount=1000,
|
|
created_by=admin_user.id,
|
|
updated_by=admin_user.id
|
|
)
|
|
session.add(transaction)
|
|
transactions.append(transaction)
|
|
session.commit()
|
|
for transaction in transactions:
|
|
session.refresh(transaction)
|
|
|
|
# Create credits
|
|
credits = [
|
|
Credit(
|
|
partner_id=multiple_partners[0].id,
|
|
transaction_id=transactions[0].id,
|
|
credit_amount=3000,
|
|
credit_limit=5000,
|
|
balance=3000,
|
|
created_by=admin_user.id,
|
|
updated_by=admin_user.id
|
|
),
|
|
Credit(
|
|
partner_id=multiple_partners[1].id,
|
|
transaction_id=transactions[1].id,
|
|
credit_amount=7000,
|
|
credit_limit=10000,
|
|
balance=7000,
|
|
created_by=admin_user.id,
|
|
updated_by=admin_user.id
|
|
),
|
|
Credit(
|
|
partner_id=multiple_partners[2].id,
|
|
transaction_id=transactions[2].id,
|
|
credit_amount=2000,
|
|
credit_limit=8000,
|
|
balance=2000,
|
|
created_by=admin_user.id,
|
|
updated_by=admin_user.id
|
|
)
|
|
]
|
|
for credit in credits:
|
|
session.add(credit)
|
|
session.commit()
|
|
for credit in credits:
|
|
session.refresh(credit)
|
|
return credits
|
|
|
|
|
|
class TestCreditCreation:
|
|
"""Test credit creation endpoints."""
|
|
|
|
def test_create_credit_with_admin_access(self, client: TestClient, admin_token: str, sample_partner, sample_transaction):
|
|
"""Test credit creation with admin token."""
|
|
credit_data = {
|
|
"partner_id": sample_partner.id,
|
|
"transaction_id": sample_transaction.id,
|
|
"credit_amount": 5000,
|
|
"credit_limit": 10000,
|
|
"balance": 5000
|
|
}
|
|
response = client.post("/api/v1/credit/",
|
|
json=credit_data,
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["partner_id"] == sample_partner.id
|
|
assert data["transaction_id"] == sample_transaction.id
|
|
assert data["credit_amount"] == 5000
|
|
assert data["credit_limit"] == 10000
|
|
assert data["balance"] == 5000
|
|
assert "id" in data
|
|
assert "created_by" in data
|
|
assert "updated_by" in data
|
|
|
|
def test_create_credit_with_write_access(self, client: TestClient, write_token: str, multiple_partners, admin_user, session):
|
|
"""Test credit creation with write token."""
|
|
# Create a transaction for this test
|
|
transaction = Transaction(
|
|
partner_id=multiple_partners[0].id,
|
|
transcation_type=TransactionType.PURCHASE,
|
|
transaction_status=TransactionStatus.UNPAID,
|
|
total_amount=2000,
|
|
created_by=admin_user.id,
|
|
updated_by=admin_user.id
|
|
)
|
|
session.add(transaction)
|
|
session.commit()
|
|
session.refresh(transaction)
|
|
|
|
credit_data = {
|
|
"partner_id": multiple_partners[0].id,
|
|
"transaction_id": transaction.id,
|
|
"credit_amount": 3000,
|
|
"credit_limit": 7500,
|
|
"balance": 3000
|
|
}
|
|
response = client.post("/api/v1/credit/",
|
|
json=credit_data,
|
|
headers={"Authorization": f"Bearer {write_token}"})
|
|
assert response.status_code == 201
|
|
data = response.json()
|
|
assert data["credit_amount"] == 3000
|
|
assert data["credit_limit"] == 7500
|
|
|
|
def test_create_credit_unauthorized(self, client: TestClient, sample_partner, sample_transaction):
|
|
"""Test credit creation without authentication."""
|
|
credit_data = {
|
|
"partner_id": sample_partner.id,
|
|
"transaction_id": sample_transaction.id,
|
|
"credit_amount": 5000,
|
|
"credit_limit": 10000,
|
|
"balance": 5000
|
|
}
|
|
response = client.post("/api/v1/credit/", json=credit_data)
|
|
assert response.status_code == 403
|
|
|
|
def test_create_credit_read_only_forbidden(self, client: TestClient, read_only_token: str, sample_partner, sample_transaction):
|
|
"""Test credit creation with read-only access should fail."""
|
|
credit_data = {
|
|
"partner_id": sample_partner.id,
|
|
"transaction_id": sample_transaction.id,
|
|
"credit_amount": 5000,
|
|
"credit_limit": 10000,
|
|
"balance": 5000
|
|
}
|
|
response = client.post("/api/v1/credit/",
|
|
json=credit_data,
|
|
headers={"Authorization": f"Bearer {read_only_token}"})
|
|
assert response.status_code == 403
|
|
|
|
def test_create_credit_invalid_partner(self, client: TestClient, admin_token: str, sample_transaction):
|
|
"""Test creation with non-existent partner should fail."""
|
|
credit_data = {
|
|
"partner_id": 99999, # Non-existent partner
|
|
"transaction_id": sample_transaction.id,
|
|
"credit_amount": 5000,
|
|
"credit_limit": 10000,
|
|
"balance": 5000
|
|
}
|
|
response = client.post("/api/v1/credit/",
|
|
json=credit_data,
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 400
|
|
assert "Partner not found" in response.json()["detail"]
|
|
|
|
def test_create_credit_invalid_transaction(self, client: TestClient, admin_token: str, sample_partner):
|
|
"""Test creation with non-existent transaction should fail."""
|
|
credit_data = {
|
|
"partner_id": sample_partner.id,
|
|
"transaction_id": 99999, # Non-existent transaction
|
|
"credit_amount": 5000,
|
|
"credit_limit": 10000,
|
|
"balance": 5000
|
|
}
|
|
response = client.post("/api/v1/credit/",
|
|
json=credit_data,
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 400
|
|
assert "Transaction not found" in response.json()["detail"]
|
|
|
|
def test_create_credit_duplicate_partner(self, client: TestClient, admin_token: str, sample_credit):
|
|
"""Test creation with duplicate partner should fail."""
|
|
credit_data = {
|
|
"partner_id": sample_credit.partner_id, # Duplicate partner
|
|
"transaction_id": sample_credit.transaction_id,
|
|
"credit_amount": 3000,
|
|
"credit_limit": 8000,
|
|
"balance": 3000
|
|
}
|
|
response = client.post("/api/v1/credit/",
|
|
json=credit_data,
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 409
|
|
assert "Credit account already exists for this partner" in response.json()["detail"]
|
|
|
|
|
|
class TestCreditRetrieval:
|
|
"""Test credit retrieval endpoints."""
|
|
|
|
def test_get_all_credits_with_auth(self, client: TestClient, admin_token: str, multiple_credits):
|
|
"""Test retrieving all credits with authentication."""
|
|
response = client.get("/api/v1/credit/",
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
assert len(data) >= 3 # At least the fixture credits
|
|
|
|
def test_get_all_credits_read_only_access(self, client: TestClient, read_only_token: str, multiple_credits):
|
|
"""Test read-only user can retrieve credits."""
|
|
response = client.get("/api/v1/credit/",
|
|
headers={"Authorization": f"Bearer {read_only_token}"})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert isinstance(data, list)
|
|
|
|
def test_get_credits_unauthorized(self, client: TestClient):
|
|
"""Test retrieving credits without authentication."""
|
|
response = client.get("/api/v1/credit/")
|
|
assert response.status_code == 403
|
|
|
|
def test_get_credits_with_pagination(self, client: TestClient, admin_token: str, multiple_credits):
|
|
"""Test credit retrieval with pagination."""
|
|
response = client.get("/api/v1/credit/?skip=0&limit=2",
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data) <= 2
|
|
|
|
def test_get_single_credit_by_id(self, client: TestClient, admin_token: str, sample_credit):
|
|
"""Test retrieving a single credit by ID."""
|
|
response = client.get(f"/api/v1/credit/{sample_credit.id}",
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["id"] == sample_credit.id
|
|
assert data["partner_id"] == sample_credit.partner_id
|
|
assert data["transaction_id"] == sample_credit.transaction_id
|
|
assert data["credit_amount"] == sample_credit.credit_amount
|
|
assert data["credit_limit"] == sample_credit.credit_limit
|
|
assert data["balance"] == sample_credit.balance
|
|
|
|
def test_get_nonexistent_credit(self, client: TestClient, admin_token: str):
|
|
"""Test retrieving a non-existent credit."""
|
|
response = client.get("/api/v1/credit/99999",
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 404
|
|
assert "Credit account not found" in response.json()["detail"]
|
|
|
|
def test_get_credit_by_partner(self, client: TestClient, admin_token: str, sample_credit):
|
|
"""Test retrieving credit for specific partner."""
|
|
response = client.get(f"/api/v1/credit/partner/{sample_credit.partner_id}",
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["partner_id"] == sample_credit.partner_id
|
|
assert data["id"] == sample_credit.id
|
|
|
|
def test_get_credit_by_nonexistent_partner(self, client: TestClient, admin_token: str):
|
|
"""Test retrieving credit for non-existent partner."""
|
|
response = client.get("/api/v1/credit/partner/99999",
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 404
|
|
assert "Partner not found" in response.json()["detail"]
|
|
|
|
def test_get_credit_by_partner_no_credit(self, client: TestClient, admin_token: str):
|
|
"""Test retrieving credit for partner with no credit account."""
|
|
# Just test with a high partner ID that likely doesn't exist
|
|
response = client.get("/api/v1/credit/partner/99998",
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 404
|
|
|
|
|
|
class TestCreditUpdate:
|
|
"""Test credit update endpoints."""
|
|
|
|
def test_update_credit_with_write_access(self, client: TestClient, write_token: str, sample_credit):
|
|
"""Test updating credit with write access."""
|
|
update_data = {
|
|
"credit_amount": 6000,
|
|
"credit_limit": 12000,
|
|
"balance": 6000
|
|
}
|
|
response = client.put(f"/api/v1/credit/{sample_credit.id}",
|
|
json=update_data,
|
|
headers={"Authorization": f"Bearer {write_token}"})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["credit_amount"] == 6000
|
|
assert data["credit_limit"] == 12000
|
|
assert data["balance"] == 6000
|
|
assert data["partner_id"] == sample_credit.partner_id # Unchanged
|
|
|
|
def test_update_credit_balance_only(self, client: TestClient, admin_token: str, sample_credit):
|
|
"""Test updating only credit balance."""
|
|
update_data = {
|
|
"balance": 3500
|
|
}
|
|
response = client.put(f"/api/v1/credit/{sample_credit.id}",
|
|
json=update_data,
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["balance"] == 3500
|
|
assert data["credit_amount"] == sample_credit.credit_amount # Unchanged
|
|
|
|
def test_update_credit_read_only_forbidden(self, client: TestClient, read_only_token: str, sample_credit):
|
|
"""Test updating credit with read-only access should fail."""
|
|
update_data = {
|
|
"balance": 4000
|
|
}
|
|
response = client.put(f"/api/v1/credit/{sample_credit.id}",
|
|
json=update_data,
|
|
headers={"Authorization": f"Bearer {read_only_token}"})
|
|
assert response.status_code == 403
|
|
|
|
def test_update_credit_invalid_partner(self, client: TestClient, admin_token: str, sample_credit):
|
|
"""Test updating credit with invalid partner should fail."""
|
|
update_data = {
|
|
"partner_id": 99999 # Non-existent partner
|
|
}
|
|
response = client.put(f"/api/v1/credit/{sample_credit.id}",
|
|
json=update_data,
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 400
|
|
assert "Partner not found" in response.json()["detail"]
|
|
|
|
def test_update_credit_invalid_transaction(self, client: TestClient, admin_token: str, sample_credit):
|
|
"""Test updating credit with invalid transaction should fail."""
|
|
update_data = {
|
|
"transaction_id": 99999 # Non-existent transaction
|
|
}
|
|
response = client.put(f"/api/v1/credit/{sample_credit.id}",
|
|
json=update_data,
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 400
|
|
assert "Transaction not found" in response.json()["detail"]
|
|
|
|
def test_update_credit_duplicate_partner(self, client: TestClient, admin_token: str, multiple_credits):
|
|
"""Test updating credit with duplicate partner should fail."""
|
|
credit_to_update = multiple_credits[0]
|
|
existing_partner_id = multiple_credits[1].partner_id
|
|
|
|
update_data = {
|
|
"partner_id": existing_partner_id
|
|
}
|
|
response = client.put(f"/api/v1/credit/{credit_to_update.id}",
|
|
json=update_data,
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 409
|
|
assert "Credit account already exists for this partner" in response.json()["detail"]
|
|
|
|
def test_update_nonexistent_credit(self, client: TestClient, admin_token: str):
|
|
"""Test updating a non-existent credit."""
|
|
update_data = {
|
|
"balance": 5000
|
|
}
|
|
response = client.put("/api/v1/credit/99999",
|
|
json=update_data,
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 404
|
|
assert "Credit account not found" in response.json()["detail"]
|
|
|
|
|
|
class TestCreditDeletion:
|
|
"""Test credit deletion endpoints."""
|
|
|
|
def test_delete_credit_with_admin_access(self, client: TestClient, admin_token: str, sample_credit):
|
|
"""Test deleting credit with admin access."""
|
|
response = client.delete(f"/api/v1/credit/{sample_credit.id}",
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 204
|
|
|
|
# Verify credit is deleted
|
|
get_response = client.get(f"/api/v1/credit/{sample_credit.id}",
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert get_response.status_code == 404
|
|
|
|
def test_delete_credit_write_access_forbidden(self, client: TestClient, write_token: str, sample_credit):
|
|
"""Test deleting credit with write access should fail."""
|
|
response = client.delete(f"/api/v1/credit/{sample_credit.id}",
|
|
headers={"Authorization": f"Bearer {write_token}"})
|
|
assert response.status_code == 403
|
|
|
|
def test_delete_credit_read_only_forbidden(self, client: TestClient, read_only_token: str, sample_credit):
|
|
"""Test deleting credit with read-only access should fail."""
|
|
response = client.delete(f"/api/v1/credit/{sample_credit.id}",
|
|
headers={"Authorization": f"Bearer {read_only_token}"})
|
|
assert response.status_code == 403
|
|
|
|
def test_delete_nonexistent_credit(self, client: TestClient, admin_token: str):
|
|
"""Test deleting a non-existent credit."""
|
|
response = client.delete("/api/v1/credit/99999",
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 404
|
|
assert "Credit account not found" in response.json()["detail"]
|
|
|
|
def test_delete_credit_unauthorized(self, client: TestClient, sample_credit):
|
|
"""Test deleting credit without authentication."""
|
|
response = client.delete(f"/api/v1/credit/{sample_credit.id}")
|
|
assert response.status_code == 403
|
|
|
|
|
|
class TestCreditValidation:
|
|
"""Test credit data validation."""
|
|
|
|
def test_create_credit_missing_required_fields(self, client: TestClient, admin_token: str):
|
|
"""Test creating credit with missing required fields."""
|
|
# Missing partner_id
|
|
credit_data = {
|
|
"transaction_id": 1,
|
|
"credit_amount": 5000,
|
|
"credit_limit": 10000,
|
|
"balance": 5000
|
|
}
|
|
response = client.post("/api/v1/credit/",
|
|
json=credit_data,
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
assert response.status_code == 422 # Validation error
|
|
|
|
def test_create_credit_negative_amounts(self, client: TestClient, admin_token: str, sample_partner, sample_transaction):
|
|
"""Test creating credit with negative amounts."""
|
|
credit_data = {
|
|
"partner_id": sample_partner.id,
|
|
"transaction_id": sample_transaction.id,
|
|
"credit_amount": -1000, # Negative amount
|
|
"credit_limit": 10000,
|
|
"balance": 5000
|
|
}
|
|
response = client.post("/api/v1/credit/",
|
|
json=credit_data,
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
# This might pass depending on validation rules, but business logic should prevent it
|
|
# You might want to add validation in the endpoint for this
|
|
|
|
def test_create_credit_balance_exceeds_limit(self, client: TestClient, admin_token: str, sample_partner, sample_transaction):
|
|
"""Test creating credit where balance exceeds limit."""
|
|
credit_data = {
|
|
"partner_id": sample_partner.id,
|
|
"transaction_id": sample_transaction.id,
|
|
"credit_amount": 5000,
|
|
"credit_limit": 3000, # Limit less than amount
|
|
"balance": 5000 # Balance exceeds limit
|
|
}
|
|
response = client.post("/api/v1/credit/",
|
|
json=credit_data,
|
|
headers={"Authorization": f"Bearer {admin_token}"})
|
|
# This might pass depending on validation rules, but business logic should prevent it
|
|
# You might want to add validation in the endpoint for this
|