test_server.py

#!/usr/bin/env python3
"""
Comprehensive test suite for the Multi-threaded HTTP Server
Tests all requirements including GET/POST, binary transfer, security, and concurrency
"""

import unittest
import requests
import json
import os
import threading
import time
import hashlib
import tempfile
from concurrent.futures import ThreadPoolExecutor
from pathlib import Path


class HTTPServerTest(unittest.TestCase):
    """Test suite for HTTP Server requirements compliance"""
    
    BASE_URL = "http://127.0.0.1:8080"
    RESOURCES_DIR = "resources"
    
    @classmethod
    def setUpClass(cls):
        """Set up test environment"""
        print("=" * 60)
        print("HTTP Server Test Suite")
        print("=" * 60)
        print("Make sure the server is running on 127.0.0.1:8080")
        print("Run: python server.py")
        print("=" * 60)
    
    def test_01_server_connectivity(self):
        """Test 1: Basic server connectivity"""
        print("\n1. Testing server connectivity...")
        try:
            response = requests.get(f"{self.BASE_URL}/", timeout=5)
            self.assertTrue(response.status_code in [200, 404], "Server should be reachable")
            print("   ✅ Server is reachable")
        except requests.ConnectionError:
            self.fail("❌ Server is not running or not reachable")
    
    def test_02_get_root_path(self):
        """Test 2: GET / → Serves resources/index.html"""
        print("\n2. Testing GET / (root path)...")
        response = requests.get(f"{self.BASE_URL}/")
        
        self.assertEqual(response.status_code, 200, "Root path should return 200 OK")
        self.assertIn("text/html", response.headers.get("content-type", ""))
        self.assertIn("index", response.text.lower())
        print("   ✅ GET / serves index.html with correct content-type")
    
    def test_03_get_html_files(self):
        """Test 3: GET HTML files → Serves with text/html content-type"""
        print("\n3. Testing HTML file serving...")
        html_files = ["about.html", "contact.html"]
        
        for filename in html_files:
            with self.subTest(filename=filename):
                response = requests.get(f"{self.BASE_URL}/{filename}")
                self.assertEqual(response.status_code, 200)
                self.assertIn("text/html", response.headers.get("content-type", ""))
                print(f"   ✅ {filename} served correctly")
    
    def test_04_get_png_binary_download(self):
        """Test 4: GET PNG files → Downloads as binary with correct headers"""
        print("\n4. Testing PNG file binary download...")
        
        # Test existing PNG files
        png_files = ["image.png", "http.png"]
        
        for filename in png_files:
            with self.subTest(filename=filename):
                response = requests.get(f"{self.BASE_URL}/{filename}")
                
                if response.status_code == 200:
                    # Check content type
                    content_type = response.headers.get("content-type", "")
                    self.assertIn("application/octet-stream", content_type)
                    
                    # Check content disposition for download
                    content_disposition = response.headers.get("content-disposition", "")
                    self.assertIn("attachment", content_disposition)
                    self.assertIn(filename, content_disposition)
                    
                    # Verify binary content
                    self.assertTrue(len(response.content) > 0)
                    print(f"   ✅ {filename} downloaded correctly ({len(response.content)} bytes)")
                else:
                    print(f"   ⚠️  {filename} not found (status: {response.status_code})")
    
    def test_05_get_jpeg_binary_download(self):
        """Test 5: GET JPEG files → Downloads as binary"""
        print("\n5. Testing JPEG file binary download...")
        
        # We'll create test JPEG files if they don't exist
        jpeg_files = ["photo.jpg", "test.jpeg"]
        
        for filename in jpeg_files:
            with self.subTest(filename=filename):
                response = requests.get(f"{self.BASE_URL}/{filename}")
                
                if response.status_code == 200:
                    content_type = response.headers.get("content-type", "")
                    self.assertIn("application/octet-stream", content_type)
                    print(f"   ✅ {filename} downloaded correctly")
                elif response.status_code == 404:
                    print(f"   ⚠️  {filename} not found - this is expected if file doesn't exist")
                else:
                    print(f"   ❌ Unexpected status for {filename}: {response.status_code}")
    
    def test_06_get_text_binary_download(self):
        """Test 6: GET TXT files → Downloads as binary (not displayed)"""
        print("\n6. Testing TXT file binary download...")
        
        response = requests.get(f"{self.BASE_URL}/sample.txt")
        
        if response.status_code == 200:
            content_type = response.headers.get("content-type", "")
            self.assertIn("application/octet-stream", content_type)
            
            content_disposition = response.headers.get("content-disposition", "")
            self.assertIn("attachment", content_disposition)
            print("   ✅ TXT file downloaded as binary")
        else:
            self.fail("sample.txt should be available for download")
    
    def test_07_get_nonexistent_file(self):
        """Test 7: GET nonexistent file → Returns 404"""
        print("\n7. Testing 404 for nonexistent files...")
        
        response = requests.get(f"{self.BASE_URL}/nonexistent.png")
        self.assertEqual(response.status_code, 404)
        print("   ✅ Nonexistent file returns 404")
    
    def test_08_unsupported_methods(self):
        """Test 8: PUT/DELETE methods → Returns 405 Method Not Allowed"""
        print("\n8. Testing unsupported HTTP methods...")
        
        methods_to_test = ["PUT", "DELETE", "PATCH"]
        
        for method in methods_to_test:
            with self.subTest(method=method):
                response = requests.request(method, f"{self.BASE_URL}/index.html")
                self.assertEqual(response.status_code, 405)
                print(f"   ✅ {method} method returns 405")
    
    def test_09_post_json_upload(self):
        """Test 9: POST with JSON → Creates file in uploads directory"""
        print("\n9. Testing POST JSON upload...")
        
        test_data = {
            "name": "Test User",
            "email": "test@example.com",
            "message": "This is a test message",
            "timestamp": time.time()
        }
        
        headers = {"Content-Type": "application/json"}
        response = requests.post(f"{self.BASE_URL}/upload", 
                               json=test_data, 
                               headers=headers)
        
        if response.status_code == 201:
            response_data = response.json()
            self.assertIn("status", response_data)
            self.assertEqual(response_data["status"], "success")
            self.assertIn("filepath", response_data)
            print(f"   ✅ JSON uploaded successfully: {response_data['filepath']}")
        else:
            print(f"   ❌ POST upload failed with status: {response.status_code}")
            if response.text:
                print(f"   Response: {response.text}")
    
    def test_10_post_non_json_content(self):
        """Test 10: POST with non-JSON → Returns 415 Unsupported Media Type"""
        print("\n10. Testing POST with non-JSON content...")
        
        headers = {"Content-Type": "text/plain"}
        response = requests.post(f"{self.BASE_URL}/upload", 
                               data="This is not JSON",
                               headers=headers)
        
        self.assertEqual(response.status_code, 415)
        print("   ✅ Non-JSON content returns 415")
    
    def test_11_post_invalid_json(self):
        """Test 11: POST with invalid JSON → Returns 400 Bad Request"""
        print("\n11. Testing POST with invalid JSON...")
        
        headers = {"Content-Type": "application/json"}
        invalid_json = '{"name": "test", "incomplete": '
        
        response = requests.post(f"{self.BASE_URL}/upload",
                               data=invalid_json,
                               headers=headers)
        
        self.assertEqual(response.status_code, 400)
        print("   ✅ Invalid JSON returns 400")
    
    def test_12_security_path_traversal(self):
        """Test 12: Path traversal attacks → Returns 403 Forbidden"""
        print("\n12. Testing path traversal protection...")
        
        malicious_paths = [
            "/../etc/passwd",
            "/../../etc/hosts", 
            "/../../../sensitive.txt",
            "/./././../config",
            "//etc/passwd",
            "/resources/../server.py"
        ]
        
        for path in malicious_paths:
            with self.subTest(path=path):
                response = requests.get(f"{self.BASE_URL}{path}")
                self.assertEqual(response.status_code, 403, 
                               f"Path traversal should be blocked: {path}")
        
        print("   ✅ Path traversal attacks properly blocked")
    
    def test_13_security_host_validation(self):
        """Test 13: Host header validation → Returns 403 for wrong hosts"""
        print("\n13. Testing Host header validation...")
        
        # Test with wrong host header
        headers = {"Host": "evil.com:8080"}
        response = requests.get(f"{self.BASE_URL}/", headers=headers)
        
        # Should return 403 for wrong host
        if response.status_code == 403:
            print("   ✅ Wrong Host header properly rejected")
        else:
            print(f"   ⚠️  Host validation may need improvement (got {response.status_code})")
    
    def test_14_security_missing_host_header(self):
        """Test 14: Missing Host header → Returns 400 Bad Request"""
        print("\n14. Testing missing Host header...")
        
        # This is harder to test with requests library as it automatically adds Host header
        # We'll just note this requirement
        print("   ℹ️  Note: Missing Host header should return 400 (hard to test with requests)")
    
    def test_15_unsupported_file_types(self):
        """Test 15: Unsupported file types → Returns 415"""
        print("\n15. Testing unsupported file types...")
        
        # We'll test with a hypothetical .exe file
        response = requests.get(f"{self.BASE_URL}/test.exe")
        
        if response.status_code == 415:
            print("   ✅ Unsupported file types return 415")
        elif response.status_code == 404:
            print("   ℹ️  File not found (would return 415 if file existed)")
        else:
            print(f"   ⚠️  Unexpected status for unsupported file: {response.status_code}")
    
    def test_16_binary_data_integrity(self):
        """Test 16: Binary data integrity → Files match exactly"""
        print("\n16. Testing binary data integrity...")
        
        # Test with existing PNG file
        local_file = Path(f"{self.RESOURCES_DIR}/image.png")
        if local_file.exists():
            # Calculate local file hash
            with open(local_file, 'rb') as f:
                local_hash = hashlib.md5(f.read()).hexdigest()
            
            # Download file and calculate hash
            response = requests.get(f"{self.BASE_URL}/image.png")
            if response.status_code == 200:
                downloaded_hash = hashlib.md5(response.content).hexdigest()
                self.assertEqual(local_hash, downloaded_hash)
                print("   ✅ Binary data integrity verified (checksums match)")
            else:
                print("   ⚠️  Could not download file for integrity check")
        else:
            print("   ⚠️  image.png not found for integrity test")
    
    def test_17_concurrent_requests(self):
        """Test 17: Concurrent requests handling"""
        print("\n17. Testing concurrent request handling...")
        
        def make_request(url):
            try:
                response = requests.get(url, timeout=10)
                return response.status_code
            except:
                return None
        
        # Make 5 concurrent requests
        with ThreadPoolExecutor(max_workers=5) as executor:
            futures = []
            for i in range(5):
                future = executor.submit(make_request, f"{self.BASE_URL}/")
                futures.append(future)
            
            results = [f.result() for f in futures]
            successful_requests = sum(1 for r in results if r == 200)
            
            self.assertGreaterEqual(successful_requests, 3, 
                                  "At least 3 out of 5 concurrent requests should succeed")
            print(f"   ✅ Concurrent requests handled: {successful_requests}/5 successful")
    
    def test_18_persistent_connections(self):
        """Test 18: Connection keep-alive support"""
        print("\n18. Testing persistent connections...")
        
        # Use session to maintain connection
        session = requests.Session()
        
        # Make multiple requests with same session
        responses = []
        for i in range(3):
            response = session.get(f"{self.BASE_URL}/")
            responses.append(response.status_code)
        
        successful_requests = sum(1 for r in responses if r == 200)
        self.assertEqual(successful_requests, 3)
        print("   ✅ Persistent connections working")
    
    def test_19_response_headers(self):
        """Test 19: Required response headers present"""
        print("\n19. Testing required response headers...")
        
        response = requests.get(f"{self.BASE_URL}/")
        headers = response.headers
        
        required_headers = ["date", "server", "connection"]
        missing_headers = []
        
        for header in required_headers:
            if header not in headers:
                missing_headers.append(header)
        
        if not missing_headers:
            print("   ✅ All required headers present")
            print(f"      Server: {headers.get('server', 'N/A')}")
            print(f"      Connection: {headers.get('connection', 'N/A')}")
        else:
            print(f"   ⚠️  Missing headers: {missing_headers}")
    
    def test_20_large_file_transfer(self):
        """Test 20: Large file transfer capability"""
        print("\n20. Testing large file transfer...")
        
        # Test with the largest available file
        response = requests.get(f"{self.BASE_URL}/image.png")
        
        if response.status_code == 200:
            file_size = len(response.content)
            print(f"   ✅ Large file transfer successful ({file_size} bytes)")
            
            if file_size > 1024 * 1024:  # > 1MB
                print("   ✅ File is larger than 1MB as required")
            else:
                print("   ⚠️  Consider adding a file > 1MB for better testing")
        else:
            print("   ⚠️  Could not test large file transfer")


def run_individual_test():
    """Run individual tests interactively"""
    print("\nIndividual Test Options:")
    print("1. Basic connectivity")
    print("2. HTML serving") 
    print("3. Binary downloads")
    print("4. Security tests")
    print("5. POST requests")
    print("6. Concurrency")
    print("7. Run all tests")
    
    choice = input("\nEnter test number (1-7): ").strip()
    
    suite = unittest.TestSuite()
    
    if choice == "1":
        suite.addTest(HTTPServerTest('test_01_server_connectivity'))
    elif choice == "2":
        suite.addTest(HTTPServerTest('test_02_get_root_path'))
        suite.addTest(HTTPServerTest('test_03_get_html_files'))
    elif choice == "3":
        suite.addTest(HTTPServerTest('test_04_get_png_binary_download'))
        suite.addTest(HTTPServerTest('test_05_get_jpeg_binary_download'))
        suite.addTest(HTTPServerTest('test_06_get_text_binary_download'))
        suite.addTest(HTTPServerTest('test_16_binary_data_integrity'))
    elif choice == "4":
        suite.addTest(HTTPServerTest('test_12_security_path_traversal'))
        suite.addTest(HTTPServerTest('test_13_security_host_validation'))
        suite.addTest(HTTPServerTest('test_14_security_missing_host_header'))
    elif choice == "5":
        suite.addTest(HTTPServerTest('test_09_post_json_upload'))
        suite.addTest(HTTPServerTest('test_10_post_non_json_content'))
        suite.addTest(HTTPServerTest('test_11_post_invalid_json'))
    elif choice == "6":
        suite.addTest(HTTPServerTest('test_17_concurrent_requests'))
        suite.addTest(HTTPServerTest('test_18_persistent_connections'))
    elif choice == "7":
        return unittest.main(argv=[''], exit=False)
    else:
        print("Invalid choice")
        return
    
    runner = unittest.TextTestRunner(verbosity=2)
    runner.run(suite)


if __name__ == "__main__":
    print("HTTP Server Test Suite")
    print("=" * 50)
    print("Make sure your server is running on 127.0.0.1:8080")
    print("Run: python server.py")
    print("=" * 50)
    
    mode = input("\n1. Run all tests\n2. Run individual tests\nChoose (1 or 2): ").strip()
    
    if mode == "2":
        run_individual_test()
    else:
        # Run all tests
        unittest.main(argv=[''], exit=False)