backend_test.py

import requests
import sys
import json
from datetime import datetime
import time

class VendorLinkAPITester:
    def __init__(self, base_url="https://vendor-link-10.preview.emergentagent.com"):
        self.base_url = base_url
        self.session_token = None
        self.user_id = None
        self.tests_run = 0
        self.tests_passed = 0
        self.test_results = []

    def log_test(self, name, success, details=""):
        """Log test result"""
        self.tests_run += 1
        if success:
            self.tests_passed += 1
            print(f"✅ {name}")
        else:
            print(f"❌ {name} - {details}")
        
        self.test_results.append({
            "test": name,
            "success": success,
            "details": details
        })

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

        try:
            if method == 'GET':
                response = requests.get(url, headers=test_headers)
            elif method == 'POST':
                response = requests.post(url, json=data, headers=test_headers)
            elif method == 'PUT':
                response = requests.put(url, json=data, headers=test_headers)
            elif method == 'DELETE':
                response = requests.delete(url, headers=test_headers)

            success = response.status_code == expected_status
            details = f"Status: {response.status_code}"
            
            if not success:
                details += f", Expected: {expected_status}"
                try:
                    error_data = response.json()
                    details += f", Response: {error_data}"
                except:
                    details += f", Response: {response.text[:200]}"

            self.log_test(name, success, details)
            
            if success:
                try:
                    return response.json()
                except:
                    return {}
            return None

        except Exception as e:
            self.log_test(name, False, f"Exception: {str(e)}")
            return None

    def create_test_session(self):
        """Create test session using mongosh approach"""
        print("\n🔧 Creating test session with mongosh...")
        
        # Generate test user data
        timestamp = int(time.time())
        user_id = f"test-user-{timestamp}"
        session_token = f"test_session_{timestamp}"
        email = f"test.user.{timestamp}@example.com"
        
        # Create mongosh command
        mongosh_cmd = f"""
mongosh --eval "
use('test_database');
var userId = '{user_id}';
var sessionToken = '{session_token}';
db.users.insertOne({{
  user_id: userId,
  email: '{email}',
  name: 'Test User',
  picture: 'https://via.placeholder.com/150',
  role: 'buyer',
  created_at: new Date()
}});
db.user_sessions.insertOne({{
  user_id: userId,
  session_token: sessionToken,
  expires_at: new Date(Date.now() + 7*24*60*60*1000),
  created_at: new Date()
}});
print('Session created successfully');
"
        """
        
        import subprocess
        try:
            result = subprocess.run(mongosh_cmd, shell=True, capture_output=True, text=True)
            if "Session created successfully" in result.stdout:
                self.session_token = session_token
                self.user_id = user_id
                print(f"✅ Test session created: {session_token}")
                return True
            else:
                print(f"❌ Failed to create session: {result.stderr}")
                return False
        except Exception as e:
            print(f"❌ Mongosh error: {str(e)}")
            return False

    def test_auth_endpoints(self):
        """Test authentication endpoints"""
        print("\n🔐 Testing Authentication Endpoints...")
        
        # Test /auth/me with valid session
        user_data = self.run_test(
            "Get current user (/auth/me)",
            "GET",
            "auth/me",
            200
        )
        
        if user_data and 'user_id' in user_data:
            print(f"   User ID: {user_data['user_id']}")
            print(f"   Email: {user_data['email']}")
            print(f"   Role: {user_data['role']}")

        # Test role update
        self.run_test(
            "Update user role to supplier",
            "PUT",
            "auth/role?role=supplier",
            200
        )

        # Test role update back to buyer
        self.run_test(
            "Update user role to buyer",
            "PUT",
            "auth/role?role=buyer",
            200
        )

    def test_contract_endpoints(self):
        """Test contract management endpoints"""
        print("\n📋 Testing Contract Endpoints...")
        
        # Create a test contract
        contract_data = {
            "product": "Steel Beams",
            "quantity": 1000,
            "location": "New York",
            "budget": 50000.0,
            "timeline": "30 days"
        }
        
        created_contract = self.run_test(
            "Create contract",
            "POST",
            "contracts",
            200,
            contract_data
        )
        
        if created_contract and 'contract_id' in created_contract:
            contract_id = created_contract['contract_id']
            print(f"   Created contract: {contract_id}")
            
            # Get all contracts
            self.run_test(
                "Get all contracts",
                "GET",
                "contracts",
                200
            )
            
            # Get specific contract
            self.run_test(
                f"Get contract {contract_id}",
                "GET",
                f"contracts/{contract_id}",
                200
            )
            
            # Update contract status
            self.run_test(
                "Update contract status",
                "PUT",
                f"contracts/{contract_id}/status?status=matched",
                200
            )
            
            return contract_id
        
        return None

    def test_supplier_profile_endpoints(self):
        """Test supplier profile endpoints"""
        print("\n👤 Testing Supplier Profile Endpoints...")
        
        # First update role to supplier
        self.run_test(
            "Update role to supplier for profile test",
            "PUT",
            "auth/role?role=supplier",
            200
        )
        
        # Create supplier profile
        profile_data = {
            "categories": ["Steel", "Construction Materials"],
            "location": "New York",
            "capacity": 5000,
            "company_name": "Test Steel Co"
        }
        
        created_profile = self.run_test(
            "Create supplier profile",
            "POST",
            "suppliers/profile",
            200,
            profile_data
        )
        
        if created_profile:
            # Get supplier profile
            self.run_test(
                "Get supplier profile",
                "GET",
                "suppliers/profile",
                200
            )
            
            # Update supplier profile
            updated_profile_data = {
                "categories": ["Steel", "Construction Materials", "Raw Materials"],
                "location": "New York",
                "capacity": 7500,
                "company_name": "Test Steel Co Updated"
            }
            
            self.run_test(
                "Update supplier profile",
                "PUT",
                "suppliers/profile",
                200,
                updated_profile_data
            )

    def test_bid_endpoints(self, contract_id):
        """Test bidding endpoints"""
        print("\n💰 Testing Bid Endpoints...")
        
        if not contract_id:
            print("   Skipping bid tests - no contract available")
            return
        
        # Ensure we're a supplier
        self.run_test(
            "Update role to supplier for bidding",
            "PUT",
            "auth/role?role=supplier",
            200
        )
        
        # Create a bid
        bid_data = {
            "contract_id": contract_id,
            "amount": 45000.0,
            "timeline": "25 days",
            "notes": "High quality steel beams with fast delivery"
        }
        
        created_bid = self.run_test(
            "Create bid",
            "POST",
            "bids",
            200,
            bid_data
        )
        
        if created_bid and 'bid_id' in created_bid:
            bid_id = created_bid['bid_id']
            print(f"   Created bid: {bid_id}")
            
            # Get all bids
            self.run_test(
                "Get all bids",
                "GET",
                "bids",
                200
            )
            
            # Get bids for specific contract
            self.run_test(
                f"Get bids for contract {contract_id}",
                "GET",
                f"bids?contract_id={contract_id}",
                200
            )
            
            return bid_id
        
        return None

    def test_ai_recommendations(self, contract_id):
        """Test AI bid recommendations"""
        print("\n🤖 Testing AI Recommendations...")
        
        if not contract_id:
            print("   Skipping AI tests - no contract available")
            return
        
        # Switch back to buyer role
        self.run_test(
            "Update role to buyer for AI recommendations",
            "PUT",
            "auth/role?role=buyer",
            200
        )
        
        # Get AI recommendations (might take a few seconds)
        print("   Waiting for AI analysis...")
        time.sleep(3)  # Give AI time to process
        
        self.run_test(
            f"Get AI bid recommendations for contract {contract_id}",
            "GET",
            f"bids/ai-recommendations/{contract_id}",
            200
        )

    def test_matching_endpoints(self):
        """Test AI matching endpoints"""
        print("\n🎯 Testing Matching Endpoints...")
        
        # Get matches as supplier
        self.run_test(
            "Update role to supplier for matches",
            "PUT",
            "auth/role?role=supplier",
            200
        )
        
        self.run_test(
            "Get matches as supplier",
            "GET",
            "matches",
            200
        )
        
        # Get matches as admin
        self.run_test(
            "Update role to admin for matches",
            "PUT",
            "auth/role?role=admin",
            200
        )
        
        matches_data = self.run_test(
            "Get matches as admin",
            "GET",
            "matches",
            200
        )
        
        # Test match validation if matches exist
        if matches_data and len(matches_data) > 0:
            match_id = matches_data[0]['match_id']
            
            # Validate a match
            validation_data = {
                "status": "validated",
                "admin_notes": "Good match approved by admin"
            }
            
            self.run_test(
                f"Validate match {match_id}",
                "PUT",
                f"matches/{match_id}",
                200,
                validation_data
            )

    def test_notification_endpoints(self):
        """Test notification preference endpoints"""
        print("\n🔔 Testing Notification Endpoints...")
        
        # Get notification preferences
        self.run_test(
            "Get notification preferences",
            "GET",
            "notifications/preferences",
            200
        )
        
        # Update notification preferences
        pref_data = {"mode": "batch"}
        
        self.run_test(
            "Update notification preferences to batch",
            "PUT",
            "notifications/preferences",
            200,
            pref_data
        )
        
        # Update back to real-time
        pref_data = {"mode": "real-time"}
        
        self.run_test(
            "Update notification preferences to real-time",
            "PUT",
            "notifications/preferences",
            200,
            pref_data
        )

    def test_admin_endpoints(self, contract_id):
        """Test admin-specific endpoints"""
        print("\n👑 Testing Admin Endpoints...")
        
        if not contract_id:
            print("   Skipping admin tests - no contract available")
            return
        
        # Ensure admin role
        self.run_test(
            "Update role to admin",
            "PUT",
            "auth/role?role=admin",
            200
        )
        
        # Test contract rerouting
        reroute_data = {
            "supplier_ids": [self.user_id],  # Reroute to ourselves for testing
            "reason": "Manual rerouting for testing purposes"
        }
        
        self.run_test(
            f"Reroute contract {contract_id}",
            "POST",
            f"contracts/{contract_id}/reroute",
            200,
            reroute_data
        )

    def cleanup_test_data(self):
        """Clean up test data"""
        print("\n🧹 Cleaning up test data...")
        
        mongosh_cleanup = f"""
mongosh --eval "
use('test_database');
db.users.deleteMany({{email: /test\\.user\\./}});
db.user_sessions.deleteMany({{session_token: /test_session/}});
db.contracts.deleteMany({{buyer_name: 'Test User'}});
db.bids.deleteMany({{supplier_name: 'Test User'}});
db.matches.deleteMany({{supplier_name: 'Test User'}});
db.supplier_profiles.deleteMany({{company_name: /Test Steel Co/}});
print('Cleanup completed');
"
        """
        
        import subprocess
        try:
            result = subprocess.run(mongosh_cleanup, shell=True, capture_output=True, text=True)
            if "Cleanup completed" in result.stdout:
                print("✅ Test data cleaned up")
            else:
                print(f"⚠️  Cleanup warning: {result.stderr}")
        except Exception as e:
            print(f"⚠️  Cleanup error: {str(e)}")

    def run_all_tests(self):
        """Run complete test suite"""
        print("🚀 Starting VendorLink API Test Suite")
        print(f"Backend URL: {self.base_url}")
        
        # Create test session
        if not self.create_test_session():
            print("❌ Failed to create test session. Exiting.")
            return 1
        
        try:
            # Run all test categories
            self.test_auth_endpoints()
            contract_id = self.test_contract_endpoints()
            self.test_supplier_profile_endpoints()
            bid_id = self.test_bid_endpoints(contract_id)
            self.test_ai_recommendations(contract_id)
            self.test_matching_endpoints()
            self.test_notification_endpoints()
            self.test_admin_endpoints(contract_id)
            
            # Print summary
            print(f"\n📊 Test Summary:")
            print(f"Tests run: {self.tests_run}")
            print(f"Tests passed: {self.tests_passed}")
            print(f"Success rate: {(self.tests_passed/self.tests_run)*100:.1f}%")
            
            # Print failed tests
            failed_tests = [t for t in self.test_results if not t['success']]
            if failed_tests:
                print(f"\n❌ Failed Tests ({len(failed_tests)}):")
                for test in failed_tests:
                    print(f"   • {test['test']}: {test['details']}")
            
            return 0 if self.tests_passed == self.tests_run else 1
            
        finally:
            # Always cleanup
            self.cleanup_test_data()

def main():
    tester = VendorLinkAPITester()
    return tester.run_all_tests()

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