test_server.py

#!/usr/bin/env python3
# Test script for HTTP server assignment
# Tests GET, POST, security, and error handling

import requests
import json
import time
import threading
import os
from concurrent.futures import ThreadPoolExecutor

def test_basic_functionality():
    # test basic server functionality
    print("🧪 Testing Basic Functionality...")
    
    base_url = "http://localhost:8080"
    
    # Test GET requests
    try:
        # Test home page
        response = requests.get(f"{base_url}/")
        assert response.status_code == 200
        assert "Multi-threaded HTTP Server" in response.text
        print("✅ GET / - Home page served correctly")
        
        # Test HTML files
        response = requests.get(f"{base_url}/about.html")
        assert response.status_code == 200
        print("✅ GET /about.html - HTML file served correctly")
        
        # Test text file download
        response = requests.get(f"{base_url}/sample.txt")
        assert response.status_code == 200
        assert response.headers['content-type'] == 'application/octet-stream'
        print("✅ GET /sample.txt - Text file download works")
        
        # Test image files
        response = requests.get(f"{base_url}/logo.png")
        assert response.status_code == 200
        assert response.headers['content-type'] == 'application/octet-stream'
        print("✅ GET /logo.png - PNG file download works")
        
        response = requests.get(f"{base_url}/photo.jpg")
        assert response.status_code == 200
        assert response.headers['content-type'] == 'application/octet-stream'
        print("✅ GET /photo.jpg - JPEG file download works")
        
    except Exception as e:
        print(f"❌ Basic functionality test failed: {e}")
        return False
    
    return True

def test_post_functionality():
    # test POST request functionality
    print("\n🧪 Testing POST Functionality...")
    
    base_url = "http://localhost:8080"
    
    try:
        # Test JSON upload
        test_data = {
            "name": "test_user",
            "email": "test@example.com",
            "message": "This is a test message",
            "timestamp": "2024-03-15T10:30:00Z"
        }
        
        response = requests.post(
            f"{base_url}/upload",
            headers={'Content-Type': 'application/json'},
            data=json.dumps(test_data)
        )
        
        assert response.status_code == 201
        result = response.json()
        assert result['status'] == 'success'
        assert 'filepath' in result
        print("✅ POST /upload - JSON upload works correctly")
        
        # Test invalid JSON
        response = requests.post(
            f"{base_url}/upload",
            headers={'Content-Type': 'application/json'},
            data="invalid json"
        )
        assert response.status_code == 400
        print("✅ POST /upload - Invalid JSON rejected correctly")
        
        # Test non-JSON content
        response = requests.post(
            f"{base_url}/upload",
            headers={'Content-Type': 'text/plain'},
            data="plain text"
        )
        assert response.status_code == 415
        print("✅ POST /upload - Non-JSON content rejected correctly")
        
    except Exception as e:
        print(f"❌ POST functionality test failed: {e}")
        return False
    
    return True

def test_security_features():
    # test security features
    print("\n🧪 Testing Security Features...")
    
    base_url = "http://localhost:8080"
    
    try:
        # Test path traversal protection
        response = requests.get(f"{base_url}/../etc/passwd")
        assert response.status_code == 403
        print("✅ Path traversal protection - ../etc/passwd blocked")
        
        response = requests.get(f"{base_url}/./././../config")
        assert response.status_code == 403
        print("✅ Path traversal protection - ./././../config blocked")
        
        # Test host validation
        response = requests.get(f"{base_url}/", headers={'Host': 'evil.com'})
        assert response.status_code == 403
        print("✅ Host validation - evil.com blocked")
        
        # Test missing host header
        response = requests.get(f"{base_url}/", headers={'Host': ''})
        assert response.status_code == 403
        print("✅ Host validation - empty host blocked")
        
    except Exception as e:
        print(f"❌ Security features test failed: {e}")
        return False
    
    return True

def test_error_handling():
    # test error handling
    print("\n🧪 Testing Error Handling...")
    
    base_url = "http://localhost:8080"
    
    try:
        # Test 404 Not Found
        response = requests.get(f"{base_url}/nonexistent.html")
        assert response.status_code == 404
        print("✅ 404 Not Found - Non-existent file handled correctly")
        
        # Test 405 Method Not Allowed
        response = requests.put(f"{base_url}/index.html")
        assert response.status_code == 405
        print("✅ 405 Method Not Allowed - PUT method rejected")
        
        response = requests.delete(f"{base_url}/index.html")
        assert response.status_code == 405
        print("✅ 405 Method Not Allowed - DELETE method rejected")
        
    except Exception as e:
        print(f"❌ Error handling test failed: {e}")
        return False
    
    return True

def test_concurrent_requests():
    # test concurrent request handling
    print("\n🧪 Testing Concurrent Requests...")
    
    base_url = "http://localhost:8080"
    
    def make_request():
        try:
            response = requests.get(f"{base_url}/sample.txt", timeout=10)
            return response.status_code == 200
        except:
            return False
    
    try:
        # Test 10 concurrent requests
        with ThreadPoolExecutor(max_workers=10) as executor:
            futures = [executor.submit(make_request) for _ in range(10)]
            results = [future.result() for future in futures]
        
        success_count = sum(results)
        print(f"✅ Concurrent requests - {success_count}/10 requests successful")
        
        if success_count >= 8:  # Allow for some failures
            return True
        else:
            print("❌ Too many concurrent request failures")
            return False
            
    except Exception as e:
        print(f"❌ Concurrent requests test failed: {e}")
        return False

def test_binary_file_integrity():
    # test binary file transfer integrity
    print("\n🧪 Testing Binary File Integrity...")
    
    base_url = "http://localhost:8080"
    
    try:
        # Test text file integrity
        response = requests.get(f"{base_url}/sample.txt")
        assert response.status_code == 200
        
        # Check if content contains expected text
        content = response.text
        assert "Multi-threaded HTTP Server Test File" in content
        assert "Binary file transfer" in content
        print("✅ Text file integrity - Content preserved correctly")
        
        # Test image file headers
        response = requests.get(f"{base_url}/logo.png")
        assert response.status_code == 200
        assert 'content-disposition' in response.headers
        assert 'attachment' in response.headers['content-disposition']
        print("✅ Image file headers - Download headers set correctly")
        
    except Exception as e:
        print(f"❌ Binary file integrity test failed: {e}")
        return False
    
    return True

def main():
    # run all tests
    print("🚀 Starting HTTP Server Tests")
    print("=" * 50)
    
    # Check if server is running
    try:
        response = requests.get("http://localhost:8080/", timeout=5)
        if response.status_code != 200:
            print("❌ Server is not responding correctly")
            return
    except:
        print("❌ Server is not running. Please start the server first:")
        print("   python3 server.py")
        return
    
    print("✅ Server is running and responding")
    
    # Run all tests
    tests = [
        test_basic_functionality,
        test_post_functionality,
        test_security_features,
        test_error_handling,
        test_concurrent_requests,
        test_binary_file_integrity
    ]
    
    passed = 0
    total = len(tests)
    
    for test in tests:
        try:
            if test():
                passed += 1
        except Exception as e:
            print(f"❌ Test {test.__name__} failed with exception: {e}")
    
    print("\n" + "=" * 50)
    print(f"🏁 Test Results: {passed}/{total} tests passed")
    
    if passed == total:
        print("🎉 All tests passed! Server is working correctly.")
    else:
        print("⚠️  Some tests failed. Check the server implementation.")

if __name__ == "__main__":
    main()