backend_test.py

#!/usr/bin/env python3

import requests
import sys
import json
from datetime import datetime, timezone

class SeftaliAPITester:
    def __init__(self, base_url="https://b2b-delivery-2.preview.emergentagent.com/api"):
        self.base_url = base_url
        self.tokens = {}  # Store tokens for different users
        self.tests_run = 0
        self.tests_passed = 0
        self.customer_id = None
        self.product_ids = []
        self.order_id = None
        self.delivery_id = None

    def run_test(self, name, method, endpoint, expected_status, data=None, token_type=None):
        """Run a single API test"""
        url = f"{self.base_url}/{endpoint}"
        headers = {'Content-Type': 'application/json'}
        
        if token_type and token_type in self.tokens:
            headers['Authorization'] = f'Bearer {self.tokens[token_type]}'

        self.tests_run += 1
        print(f"\n🔍 Testing {name}...")
        
        try:
            if method == 'GET':
                response = requests.get(url, headers=headers)
            elif method == 'POST':
                response = requests.post(url, json=data, headers=headers)
            elif method == 'PATCH':
                response = requests.patch(url, json=data, headers=headers)

            success = response.status_code == expected_status
            if success:
                self.tests_passed += 1
                print(f"✅ Passed - Status: {response.status_code}")
                try:
                    return True, response.json()
                except:
                    return True, {}
            else:
                print(f"❌ Failed - Expected {expected_status}, got {response.status_code}")
                try:
                    print(f"   Response: {response.json()}")
                except:
                    print(f"   Response: {response.text}")
                return False, {}

        except Exception as e:
            print(f"❌ Failed - Error: {str(e)}")
            return False, {}

    def test_seed_database(self):
        """Seed the database with initial data"""
        print("\n🌱 Seeding database...")
        success, response = self.run_test(
            "Seed Database",
            "POST",
            "seed",
            200
        )
        if success:
            print("   Database seeded successfully")
            # Extract product IDs for later use
            if 'products' in response:
                self.product_ids = [p['id'] for p in response['products']]
        return success

    def test_login(self, username, password, user_type):
        """Test login and store token"""
        success, response = self.run_test(
            f"Login ({user_type})",
            "POST",
            "auth/login",
            200,
            data={"username": username, "password": password}
        )
        if success and 'access_token' in response:
            self.tokens[user_type] = response['access_token']
            if user_type == 'customer' and 'user' in response:
                self.customer_id = response['user'].get('customer_id')
            return True
        return False

    def test_auth_me(self, user_type):
        """Test /auth/me endpoint"""
        success, response = self.run_test(
            f"Get Current User ({user_type})",
            "GET",
            "auth/me",
            200,
            token_type=user_type
        )
        return success

    def test_customer_draft(self):
        """Test customer draft endpoint"""
        success, response = self.run_test(
            "Get Customer Draft",
            "GET",
            "customer/draft",
            200,
            token_type='customer'
        )
        return success

    def test_start_working_copy(self):
        """Test starting a working copy"""
        success, response = self.run_test(
            "Start Working Copy",
            "POST",
            "customer/working-copy/start",
            200,
            token_type='customer'
        )
        return success

    def test_get_working_copy(self):
        """Test getting working copy"""
        success, response = self.run_test(
            "Get Working Copy",
            "GET",
            "customer/working-copy",
            200,
            token_type='customer'
        )
        return success

    def test_update_working_copy(self):
        """Test updating working copy with quantities"""
        if not self.product_ids:
            print("❌ No product IDs available for working copy update")
            return False
            
        # Create items with quantities
        items = []
        for i, product_id in enumerate(self.product_ids[:2]):  # Use first 2 products
            items.append({
                "product_id": product_id,
                "suggested_qty": 0,
                "user_qty": (i + 1) * 5,  # 5, 10
                "removed": False
            })
        
        success, response = self.run_test(
            "Update Working Copy",
            "PATCH",
            "customer/working-copy",
            200,
            data={"items": items},
            token_type='customer'
        )
        return success

    def test_submit_working_copy(self):
        """Test submitting working copy as order"""
        success, response = self.run_test(
            "Submit Working Copy",
            "POST",
            "customer/working-copy/submit",
            200,
            token_type='customer'
        )
        if success and 'order_id' in response:
            self.order_id = response['order_id']
        return success

    def test_sales_orders(self):
        """Test salesperson orders list"""
        success, response = self.run_test(
            "Get Sales Orders",
            "GET",
            "sales/orders",
            200,
            token_type='salesperson'
        )
        return success

    def test_get_order_detail(self):
        """Test getting order detail"""
        if not self.order_id:
            print("❌ No order ID available for detail test")
            return False
            
        success, response = self.run_test(
            "Get Order Detail",
            "GET",
            f"sales/orders/{self.order_id}",
            200,
            token_type='salesperson'
        )
        return success

    def test_approve_order(self):
        """Test approving an order"""
        if not self.order_id:
            print("❌ No order ID available for approval test")
            return False
            
        success, response = self.run_test(
            "Approve Order",
            "POST",
            f"sales/orders/{self.order_id}/approve",
            200,
            token_type='salesperson'
        )
        return success

    def test_create_delivery(self):
        """Test creating a delivery"""
        if not self.customer_id or not self.product_ids:
            print("❌ Missing customer ID or product IDs for delivery test")
            return False
            
        delivery_data = {
            "customer_id": self.customer_id,
            "delivery_type": "route",
            "delivered_at": datetime.now(timezone.utc).isoformat(),
            "invoice_no": "INV-001",
            "items": [
                {"product_id": self.product_ids[0], "quantity": 10},
                {"product_id": self.product_ids[1], "quantity": 5}
            ]
        }
        
        success, response = self.run_test(
            "Create Delivery",
            "POST",
            "sales/deliveries",
            200,
            data=delivery_data,
            token_type='salesperson'
        )
        if success and 'delivery_id' in response:
            self.delivery_id = response['delivery_id']
        return success

    def test_get_deliveries(self):
        """Test getting deliveries list"""
        success, response = self.run_test(
            "Get Deliveries",
            "GET",
            "sales/deliveries",
            200,
            token_type='salesperson'
        )
        return success

    def test_admin_health_summary(self):
        """Test admin health summary"""
        success, response = self.run_test(
            "Admin Health Summary",
            "GET",
            "admin/health/summary",
            200,
            token_type='admin'
        )
        return success

    def test_admin_audit_events(self):
        """Test admin audit events"""
        success, response = self.run_test(
            "Admin Audit Events",
            "GET",
            "admin/audit-events",
            200,
            token_type='admin'
        )
        return success

    def test_consumption_calculation(self):
        """Test that consumption calculation happened after delivery"""
        if not self.customer_id:
            print("❌ No customer ID available for consumption test")
            return False
            
        success, response = self.run_test(
            "Get Customer Consumption Stats",
            "GET",
            f"admin/customers/{self.customer_id}/consumption",
            200,
            token_type='admin'
        )
        
        if success and response:
            print(f"   Found {len(response)} consumption records")
            for record in response:
                if record.get('daily_avg', 0) >= 0:
                    print(f"   ✅ Product {record.get('product', {}).get('code', 'Unknown')}: daily_avg = {record.get('daily_avg', 0)}")
        
        return success

def main():
    print("🍑 ŞEFTALİ B2B Dairy Distribution System - API Testing")
    print("=" * 60)
    
    tester = SeftaliAPITester()
    
    # Test credentials
    credentials = [
        ("market", "market123", "customer"),
        ("plasiyer", "plasiyer123", "salesperson"), 
        ("admin", "admin123", "admin")
    ]
    
    # 1. Seed database
    if not tester.test_seed_database():
        print("❌ Database seeding failed, stopping tests")
        return 1
    
    # 2. Test login for all user types
    print("\n" + "="*40)
    print("AUTHENTICATION TESTS")
    print("="*40)
    
    for username, password, user_type in credentials:
        if not tester.test_login(username, password, user_type):
            print(f"❌ Login failed for {user_type}, stopping tests")
            return 1
    
    # Test /auth/me for all users
    for _, _, user_type in credentials:
        tester.test_auth_me(user_type)
    
    # 3. Customer workflow tests
    print("\n" + "="*40)
    print("CUSTOMER WORKFLOW TESTS")
    print("="*40)
    
    tester.test_customer_draft()
    tester.test_start_working_copy()
    tester.test_get_working_copy()
    tester.test_update_working_copy()
    tester.test_submit_working_copy()
    
    # 4. Salesperson workflow tests
    print("\n" + "="*40)
    print("SALESPERSON WORKFLOW TESTS")
    print("="*40)
    
    tester.test_sales_orders()
    tester.test_get_order_detail()
    tester.test_approve_order()
    tester.test_create_delivery()
    tester.test_get_deliveries()
    
    # 5. Admin tests
    print("\n" + "="*40)
    print("ADMIN TESTS")
    print("="*40)
    
    tester.test_admin_health_summary()
    tester.test_admin_audit_events()
    tester.test_consumption_calculation()
    
    # Print final results
    print("\n" + "="*60)
    print(f"📊 FINAL RESULTS: {tester.tests_passed}/{tester.tests_run} tests passed")
    
    if tester.tests_passed == tester.tests_run:
        print("🎉 All tests passed!")
        return 0
    else:
        print(f"⚠️  {tester.tests_run - tester.tests_passed} tests failed")
        return 1

if __name__ == "__main__":
    sys.exit(main())