KillerKing93/Transformers-InferenceServer-OpenAPI
0
1#!/usr/bin/env python2# -*- coding: utf-8 -*-3"""4Comprehensive test suite for marketplace API endpoints5 6Tests cover:7- Supplier registration and management8- Product creation and search9- User registration and management10- Location-aware product search11- AI-powered natural language search12 13Run with: pytest tests/test_marketplace.py -v14"""15 16import pytest17import sys18import os19from datetime import datetime20 21# Add parent directory to path for imports22sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))23 24from fastapi.testclient import TestClient25from sqlalchemy import create_engine26from sqlalchemy.orm import sessionmaker27 28from main import app29from database import get_db30from models import Base, Supplier, Product, User31 32# Test database33SQLALCHEMY_DATABASE_URL = "sqlite:///./test_marketplace.db"34engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})35TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)36 37 38def override_get_db():39 """Override database dependency for testing."""40 try:41 db = TestingSessionLocal()42 yield db43 finally:44 db.close()45 46 47app.dependency_overrides[get_db] = override_get_db48 49client = TestClient(app)50 51 52@pytest.fixture(scope="module")53def setup_database():54 """Create test database tables before tests."""55 Base.metadata.create_all(bind=engine)56 yield57 Base.metadata.drop_all(bind=engine)58 59 60@pytest.fixture(scope="function")61def clean_database():62 """Clean database before each test."""63 db = TestingSessionLocal()64 try:65 db.query(Product).delete()66 db.query(Supplier).delete()67 db.query(User).delete()68 db.commit()69 finally:70 db.close()71 yield72 73 74class TestSupplierEndpoints:75 """Test supplier management endpoints."""76 77 def test_register_supplier_success(self, setup_database, clean_database):78 """Test successful supplier registration."""79 response = client.post(80 "/api/suppliers/register",81 json={82 "name": "Test Supplier",83 "business_name": "Test Electronics",84 "email": "test@supplier.com",85 "phone": "+62812345678",86 "address": "Jl. Test No. 1",87 "latitude": -6.2088,88 "longitude": 106.8456,89 "city": "Jakarta",90 "province": "DKI Jakarta"91 }92 )93 assert response.status_code == 20094 data = response.json()95 assert data["email"] == "test@supplier.com"96 assert data["business_name"] == "Test Electronics"97 assert data["city"] == "Jakarta"98 assert "id" in data99 assert data["is_active"] is True100 101 def test_register_supplier_duplicate_email(self, setup_database, clean_database):102 """Test supplier registration with duplicate email."""103 # Register first supplier104 client.post(105 "/api/suppliers/register",106 json={107 "name": "Supplier 1",108 "business_name": "Store 1",109 "email": "duplicate@test.com",110 "latitude": -6.2088,111 "longitude": 106.8456,112 "city": "Jakarta"113 }114 )115 116 # Try to register with same email117 response = client.post(118 "/api/suppliers/register",119 json={120 "name": "Supplier 2",121 "business_name": "Store 2",122 "email": "duplicate@test.com",123 "latitude": -6.2088,124 "longitude": 106.8456,125 "city": "Bandung"126 }127 )128 assert response.status_code == 400129 assert "already registered" in response.json()["detail"]130 131 def test_list_suppliers(self, setup_database, clean_database):132 """Test listing suppliers."""133 # Create test suppliers134 for i in range(3):135 client.post(136 "/api/suppliers/register",137 json={138 "name": f"Supplier {i}",139 "business_name": f"Store {i}",140 "email": f"supplier{i}@test.com",141 "latitude": -6.2088,142 "longitude": 106.8456,143 "city": "Jakarta" if i < 2 else "Bandung"144 }145 )146 147 # List all suppliers148 response = client.get("/api/suppliers")149 assert response.status_code == 200150 data = response.json()151 assert len(data) == 3152 153 # List suppliers by city154 response = client.get("/api/suppliers?city=Jakarta")155 assert response.status_code == 200156 data = response.json()157 assert len(data) == 2158 159 def test_get_supplier_by_id(self, setup_database, clean_database):160 """Test getting supplier by ID."""161 # Create supplier162 create_response = client.post(163 "/api/suppliers/register",164 json={165 "name": "Test Supplier",166 "business_name": "Test Store",167 "email": "test@supplier.com",168 "latitude": -6.2088,169 "longitude": 106.8456,170 "city": "Jakarta"171 }172 )173 supplier_id = create_response.json()["id"]174 175 # Get supplier176 response = client.get(f"/api/suppliers/{supplier_id}")177 assert response.status_code == 200178 data = response.json()179 assert data["id"] == supplier_id180 assert data["business_name"] == "Test Store"181 182 def test_get_nonexistent_supplier(self, setup_database, clean_database):183 """Test getting non-existent supplier."""184 response = client.get("/api/suppliers/99999")185 assert response.status_code == 404186 187 188class TestProductEndpoints:189 """Test product management endpoints."""190 191 def test_create_product_success(self, setup_database, clean_database):192 """Test successful product creation."""193 # Create supplier first194 supplier_response = client.post(195 "/api/suppliers/register",196 json={197 "name": "Supplier",198 "business_name": "Electronics Store",199 "email": "supplier@test.com",200 "latitude": -6.2088,201 "longitude": 106.8456,202 "city": "Jakarta"203 }204 )205 supplier_id = supplier_response.json()["id"]206 207 # Create product208 response = client.post(209 f"/api/suppliers/{supplier_id}/products",210 json={211 "name": "Test Laptop",212 "description": "A test laptop",213 "price": 10000000,214 "stock_quantity": 5,215 "category": "laptop",216 "tags": "test,laptop,gaming",217 "sku": "TEST-LAP-001"218 }219 )220 assert response.status_code == 200221 data = response.json()222 assert data["name"] == "Test Laptop"223 assert data["price"] == 10000000224 assert data["supplier_id"] == supplier_id225 assert "supplier_name" in data226 227 def test_create_product_invalid_supplier(self, setup_database, clean_database):228 """Test creating product with invalid supplier ID."""229 response = client.post(230 "/api/suppliers/99999/products",231 json={232 "name": "Test Product",233 "price": 1000000,234 "stock_quantity": 1,235 "category": "test"236 }237 )238 assert response.status_code == 404239 240 def test_update_product(self, setup_database, clean_database):241 """Test updating product."""242 # Create supplier and product243 supplier_response = client.post(244 "/api/suppliers/register",245 json={246 "name": "Supplier",247 "business_name": "Store",248 "email": "supplier@test.com",249 "latitude": -6.2088,250 "longitude": 106.8456,251 "city": "Jakarta"252 }253 )254 supplier_id = supplier_response.json()["id"]255 256 product_response = client.post(257 f"/api/suppliers/{supplier_id}/products",258 json={259 "name": "Product",260 "price": 1000000,261 "stock_quantity": 10,262 "category": "test"263 }264 )265 product_id = product_response.json()["id"]266 267 # Update product268 response = client.put(269 f"/api/products/{product_id}",270 json={271 "price": 1500000,272 "stock_quantity": 5273 }274 )275 assert response.status_code == 200276 data = response.json()277 assert data["price"] == 1500000278 assert data["stock_quantity"] == 5279 280 def test_list_products_with_filters(self, setup_database, clean_database):281 """Test listing products with filters."""282 # Create supplier283 supplier_response = client.post(284 "/api/suppliers/register",285 json={286 "name": "Supplier",287 "business_name": "Store",288 "email": "supplier@test.com",289 "latitude": -6.2088,290 "longitude": 106.8456,291 "city": "Jakarta"292 }293 )294 supplier_id = supplier_response.json()["id"]295 296 # Create products297 products = [298 {"name": "Laptop 1", "price": 5000000, "stock_quantity": 5, "category": "laptop"},299 {"name": "Laptop 2", "price": 15000000, "stock_quantity": 3, "category": "laptop"},300 {"name": "Phone 1", "price": 8000000, "stock_quantity": 10, "category": "smartphone"},301 ]302 for product in products:303 client.post(f"/api/suppliers/{supplier_id}/products", json=product)304 305 # List all products306 response = client.get("/api/products")307 assert response.status_code == 200308 assert len(response.json()) == 3309 310 # Filter by category311 response = client.get("/api/products?category=laptop")312 assert response.status_code == 200313 assert len(response.json()) == 2314 315 # Filter by price range316 response = client.get("/api/products?min_price=6000000&max_price=12000000")317 assert response.status_code == 200318 data = response.json()319 assert len(data) == 2320 for product in data:321 assert 6000000 <= product["price"] <= 12000000322 323 def test_search_products_by_keyword(self, setup_database, clean_database):324 """Test searching products by keyword."""325 # Create supplier and products326 supplier_response = client.post(327 "/api/suppliers/register",328 json={329 "name": "Supplier",330 "business_name": "Store",331 "email": "supplier@test.com",332 "latitude": -6.2088,333 "longitude": 106.8456,334 "city": "Jakarta"335 }336 )337 supplier_id = supplier_response.json()["id"]338 339 client.post(340 f"/api/suppliers/{supplier_id}/products",341 json={342 "name": "ASUS Gaming Laptop",343 "price": 10000000,344 "stock_quantity": 5,345 "category": "laptop",346 "tags": "gaming,asus,rtx"347 }348 )349 client.post(350 f"/api/suppliers/{supplier_id}/products",351 json={352 "name": "Samsung Smartphone",353 "price": 5000000,354 "stock_quantity": 10,355 "category": "smartphone",356 "tags": "samsung,android"357 }358 )359 360 # Search by name361 response = client.get("/api/products/search?q=gaming")362 assert response.status_code == 200363 data = response.json()364 assert len(data) == 1365 assert "Gaming" in data[0]["name"]366 367 # Search by tags368 response = client.get("/api/products/search?q=samsung")369 assert response.status_code == 200370 data = response.json()371 assert len(data) == 1372 373 def test_search_products_with_location_sorting(self, setup_database, clean_database):374 """Test location-aware product search."""375 # Create suppliers at different locations376 suppliers = [377 {"name": "Supplier 1", "business_name": "Store 1", "email": "s1@test.com",378 "latitude": -6.2088, "longitude": 106.8456, "city": "Jakarta"},379 {"name": "Supplier 2", "business_name": "Store 2", "email": "s2@test.com",380 "latitude": -6.9175, "longitude": 107.6191, "city": "Bandung"},381 ]382 383 for supplier_data in suppliers:384 supplier_response = client.post("/api/suppliers/register", json=supplier_data)385 supplier_id = supplier_response.json()["id"]386 387 # Create product for each supplier388 client.post(389 f"/api/suppliers/{supplier_id}/products",390 json={391 "name": "Test Laptop",392 "price": 10000000,393 "stock_quantity": 5,394 "category": "laptop"395 }396 )397 398 # Search from Jakarta location399 response = client.get(400 "/api/products/search?q=laptop&user_lat=-6.2088&user_lon=106.8456"401 )402 assert response.status_code == 200403 data = response.json()404 assert len(data) == 2405 # First result should be Jakarta (closer)406 assert data[0]["distance_km"] < data[1]["distance_km"]407 408 409class TestUserEndpoints:410 """Test user management endpoints."""411 412 def test_register_user_success(self, setup_database, clean_database):413 """Test successful user registration."""414 response = client.post(415 "/api/users/register",416 json={417 "name": "Test User",418 "email": "user@test.com",419 "phone": "+62812345678",420 "latitude": -6.2088,421 "longitude": 106.8456,422 "city": "Jakarta",423 "province": "DKI Jakarta",424 "ai_access_enabled": True425 }426 )427 assert response.status_code == 200428 data = response.json()429 assert data["email"] == "user@test.com"430 assert data["name"] == "Test User"431 assert data["ai_access_enabled"] is True432 433 def test_register_user_duplicate_email(self, setup_database, clean_database):434 """Test user registration with duplicate email."""435 client.post(436 "/api/users/register",437 json={438 "name": "User 1",439 "email": "duplicate@test.com",440 "city": "Jakarta"441 }442 )443 444 response = client.post(445 "/api/users/register",446 json={447 "name": "User 2",448 "email": "duplicate@test.com",449 "city": "Bandung"450 }451 )452 assert response.status_code == 400453 454 def test_get_user_by_id(self, setup_database, clean_database):455 """Test getting user by ID."""456 create_response = client.post(457 "/api/users/register",458 json={459 "name": "Test User",460 "email": "user@test.com",461 "city": "Jakarta"462 }463 )464 user_id = create_response.json()["id"]465 466 response = client.get(f"/api/users/{user_id}")467 assert response.status_code == 200468 data = response.json()469 assert data["id"] == user_id470 471 def test_get_nonexistent_user(self, setup_database, clean_database):472 """Test getting non-existent user."""473 response = client.get("/api/users/99999")474 assert response.status_code == 404475 476 477class TestAISearchEndpoint:478 """Test AI-powered search endpoint."""479 480 def test_ai_search_without_ai_access(self, setup_database, clean_database):481 """Test AI search for user without AI access."""482 # Create user without AI access483 user_response = client.post(484 "/api/users/register",485 json={486 "name": "Regular User",487 "email": "regular@test.com",488 "city": "Jakarta",489 "ai_access_enabled": False490 }491 )492 user_id = user_response.json()["id"]493 494 # Try AI search495 response = client.post(496 "/api/chat/search",497 json={498 "user_id": user_id,499 "query": "laptop gaming"500 }501 )502 assert response.status_code == 403503 assert "AI access not enabled" in response.json()["detail"]504 505 def test_ai_search_with_nonexistent_user(self, setup_database, clean_database):506 """Test AI search with non-existent user."""507 response = client.post(508 "/api/chat/search",509 json={510 "user_id": 99999,511 "query": "laptop gaming"512 }513 )514 assert response.status_code == 404515 516 def test_ai_search_no_products_found(self, setup_database, clean_database):517 """Test AI search when no products match."""518 # Create user with AI access519 user_response = client.post(520 "/api/users/register",521 json={522 "name": "Premium User",523 "email": "premium@test.com",524 "city": "Jakarta",525 "ai_access_enabled": True526 }527 )528 user_id = user_response.json()["id"]529 530 # Mock get_engine to avoid loading actual model in tests531 # In real scenario, this would return AI response532 # For now, we just test the endpoint structure533 response = client.post(534 "/api/chat/search",535 json={536 "user_id": user_id,537 "query": "nonexistent product xyz123"538 }539 )540 541 # This will fail on AI inference if model not loaded542 # In production tests, mock the engine543 # For now, we verify the request structure is valid544 assert response.status_code in [200, 500] # 500 if model not loaded545 546 547class TestUtilityFunctions:548 """Test utility functions."""549 550 def test_haversine_distance(self):551 """Test Haversine distance calculation."""552 from utils import haversine_distance553 554 # Jakarta to Bandung (approximately 126 km)555 distance = haversine_distance(-6.2088, 106.8456, -6.9175, 107.6191)556 assert 120 < distance < 135 # Allow some margin557 558 # Same location559 distance = haversine_distance(-6.2088, 106.8456, -6.2088, 106.8456)560 assert distance < 0.1 # Should be very close to 0561 562 def test_extract_location_query(self):563 """Test location extraction from query."""564 from utils import extract_location_query565 566 assert extract_location_query("laptop di Jakarta") == "Jakarta"567 assert extract_location_query("smartphone Bandung murah") == "Bandung"568 assert extract_location_query("cari laptop Jakarta Selatan") == "Jakarta Selatan"569 assert extract_location_query("laptop gaming") is None570 571 def test_format_price_idr(self):572 """Test IDR price formatting."""573 from utils import format_price_idr574 575 assert format_price_idr(10000000) == "Rp 10.000.000"576 assert format_price_idr(1500000) == "Rp 1.500.000"577 assert format_price_idr(999) == "Rp 999"578 579 580if __name__ == "__main__":581 pytest.main([__file__, "-v"])582