remove part 1
This commit is contained in:
@@ -322,74 +322,3 @@ class TestAPIParameterValidation:
|
||||
# Should not grant admin from parameter
|
||||
data = response.json()
|
||||
assert not data.get("data", {}).get("is_admin", False)
|
||||
|
||||
|
||||
@pytest.mark.security
|
||||
class TestFileUploadSecurity:
|
||||
"""Security tests for file upload handling."""
|
||||
|
||||
@pytest.fixture
|
||||
async def client(self):
|
||||
"""Create async HTTP client for testing."""
|
||||
from httpx import ASGITransport
|
||||
|
||||
async with AsyncClient(
|
||||
transport=ASGITransport(app=app), base_url="http://test"
|
||||
) as ac:
|
||||
yield ac
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_malicious_file_extension(self, client):
|
||||
"""Test handling of dangerous file extensions."""
|
||||
dangerous_extensions = [
|
||||
".exe",
|
||||
".sh",
|
||||
".bat",
|
||||
".cmd",
|
||||
".php",
|
||||
".jsp",
|
||||
]
|
||||
|
||||
for ext in dangerous_extensions:
|
||||
files = {"file": (f"test{ext}", b"malicious content")}
|
||||
response = await client.post("/api/upload", files=files)
|
||||
|
||||
# Should reject dangerous files
|
||||
assert response.status_code in [400, 403, 415]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_file_size_limit(self, client):
|
||||
"""Test enforcement of file size limits."""
|
||||
# Try to upload very large file
|
||||
large_content = b"A" * (100 * 1024 * 1024) # 100MB
|
||||
|
||||
files = {"file": ("large.txt", large_content)}
|
||||
response = await client.post("/api/upload", files=files)
|
||||
|
||||
# Should reject oversized files
|
||||
assert response.status_code in [413, 422]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_double_extension_bypass(self, client):
|
||||
"""Test protection against double extension bypass."""
|
||||
files = {"file": ("image.jpg.php", b"<?php phpinfo(); ?>")}
|
||||
response = await client.post("/api/upload", files=files)
|
||||
|
||||
# Should detect and reject
|
||||
assert response.status_code in [400, 403, 415]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_mime_type_validation(self, client):
|
||||
"""Test MIME type validation."""
|
||||
# PHP file with image MIME type
|
||||
files = {
|
||||
"file": (
|
||||
"image.jpg",
|
||||
b"<?php phpinfo(); ?>",
|
||||
"image/jpeg",
|
||||
)
|
||||
}
|
||||
response = await client.post("/api/upload", files=files)
|
||||
|
||||
# Should validate actual content, not just MIME type
|
||||
assert response.status_code in [400, 403, 415]
|
||||
|
||||
@@ -1,227 +0,0 @@
|
||||
"""Unit tests for diagnostics endpoints."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
import pytest
|
||||
|
||||
from src.server.api.diagnostics import (
|
||||
NetworkTestResult,
|
||||
check_dns,
|
||||
check_host_connectivity,
|
||||
network_diagnostics,
|
||||
)
|
||||
|
||||
|
||||
class TestDiagnosticsEndpoint:
|
||||
"""Test diagnostics API endpoints."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_network_diagnostics_returns_standard_format(self):
|
||||
"""Test that network diagnostics returns the expected format."""
|
||||
# Mock authentication
|
||||
mock_auth = {"user_id": "test_user"}
|
||||
|
||||
# Mock the helper functions
|
||||
with patch(
|
||||
"src.server.api.diagnostics.check_dns",
|
||||
return_value=True
|
||||
), patch(
|
||||
"src.server.api.diagnostics.check_host_connectivity",
|
||||
side_effect=[
|
||||
NetworkTestResult(
|
||||
host="google.com",
|
||||
reachable=True,
|
||||
response_time_ms=50.5
|
||||
),
|
||||
NetworkTestResult(
|
||||
host="cloudflare.com",
|
||||
reachable=True,
|
||||
response_time_ms=30.2
|
||||
),
|
||||
NetworkTestResult(
|
||||
host="github.com",
|
||||
reachable=True,
|
||||
response_time_ms=100.0
|
||||
),
|
||||
NetworkTestResult(
|
||||
host="aniworld.to",
|
||||
reachable=True,
|
||||
response_time_ms=75.3
|
||||
),
|
||||
]
|
||||
):
|
||||
# Call the endpoint
|
||||
result = await network_diagnostics(auth=mock_auth)
|
||||
|
||||
# Verify response structure
|
||||
assert isinstance(result, dict)
|
||||
assert "status" in result
|
||||
assert "data" in result
|
||||
assert result["status"] == "success"
|
||||
|
||||
# Verify data structure
|
||||
data = result["data"]
|
||||
assert "internet_connected" in data
|
||||
assert "dns_working" in data
|
||||
assert "aniworld_reachable" in data
|
||||
assert "tests" in data
|
||||
|
||||
# Verify values
|
||||
assert data["internet_connected"] is True
|
||||
assert data["dns_working"] is True
|
||||
assert data["aniworld_reachable"] is True
|
||||
assert len(data["tests"]) == 4
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_network_diagnostics_aniworld_unreachable(self):
|
||||
"""Test diagnostics when aniworld.to is unreachable."""
|
||||
mock_auth = {"user_id": "test_user"}
|
||||
|
||||
with patch(
|
||||
"src.server.api.diagnostics.check_dns",
|
||||
return_value=True
|
||||
), patch(
|
||||
"src.server.api.diagnostics.check_host_connectivity",
|
||||
side_effect=[
|
||||
NetworkTestResult(
|
||||
host="google.com",
|
||||
reachable=True,
|
||||
response_time_ms=50.5
|
||||
),
|
||||
NetworkTestResult(
|
||||
host="cloudflare.com",
|
||||
reachable=True,
|
||||
response_time_ms=30.2
|
||||
),
|
||||
NetworkTestResult(
|
||||
host="github.com",
|
||||
reachable=True,
|
||||
response_time_ms=100.0
|
||||
),
|
||||
NetworkTestResult(
|
||||
host="aniworld.to",
|
||||
reachable=False,
|
||||
error="Connection timeout"
|
||||
),
|
||||
]
|
||||
):
|
||||
result = await network_diagnostics(auth=mock_auth)
|
||||
|
||||
# Verify aniworld is marked as unreachable
|
||||
assert result["status"] == "success"
|
||||
assert result["data"]["aniworld_reachable"] is False
|
||||
assert result["data"]["internet_connected"] is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_network_diagnostics_all_unreachable(self):
|
||||
"""Test diagnostics when all hosts are unreachable."""
|
||||
mock_auth = {"user_id": "test_user"}
|
||||
|
||||
with patch(
|
||||
"src.server.api.diagnostics.check_dns",
|
||||
return_value=False
|
||||
), patch(
|
||||
"src.server.api.diagnostics.check_host_connectivity",
|
||||
side_effect=[
|
||||
NetworkTestResult(
|
||||
host="google.com",
|
||||
reachable=False,
|
||||
error="Connection timeout"
|
||||
),
|
||||
NetworkTestResult(
|
||||
host="cloudflare.com",
|
||||
reachable=False,
|
||||
error="Connection timeout"
|
||||
),
|
||||
NetworkTestResult(
|
||||
host="github.com",
|
||||
reachable=False,
|
||||
error="Connection timeout"
|
||||
),
|
||||
NetworkTestResult(
|
||||
host="aniworld.to",
|
||||
reachable=False,
|
||||
error="Connection timeout"
|
||||
),
|
||||
]
|
||||
):
|
||||
result = await network_diagnostics(auth=mock_auth)
|
||||
|
||||
# Verify all are unreachable
|
||||
assert result["status"] == "success"
|
||||
assert result["data"]["internet_connected"] is False
|
||||
assert result["data"]["dns_working"] is False
|
||||
assert result["data"]["aniworld_reachable"] is False
|
||||
|
||||
|
||||
class TestNetworkHelpers:
|
||||
"""Test network helper functions."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_dns_success(self):
|
||||
"""Test DNS check when DNS is working."""
|
||||
with patch("socket.gethostbyname", return_value="142.250.185.78"):
|
||||
result = await check_dns()
|
||||
assert result is True
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_check_dns_failure(self):
|
||||
"""Test DNS check when DNS fails."""
|
||||
import socket
|
||||
with patch(
|
||||
"socket.gethostbyname",
|
||||
side_effect=socket.gaierror("DNS lookup failed")
|
||||
):
|
||||
result = await check_dns()
|
||||
assert result is False
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_host_connectivity_success(self):
|
||||
"""Test host connectivity check when host is reachable."""
|
||||
with patch(
|
||||
"socket.create_connection",
|
||||
return_value=MagicMock()
|
||||
):
|
||||
result = await check_host_connectivity("google.com", 80)
|
||||
assert result.host == "google.com"
|
||||
assert result.reachable is True
|
||||
assert result.response_time_ms is not None
|
||||
assert result.response_time_ms >= 0
|
||||
assert result.error is None
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_host_connectivity_timeout(self):
|
||||
"""Test host connectivity when connection times out."""
|
||||
import asyncio
|
||||
with patch(
|
||||
"socket.create_connection",
|
||||
side_effect=asyncio.TimeoutError()
|
||||
):
|
||||
result = await check_host_connectivity("example.com", 80, 1.0)
|
||||
assert result.host == "example.com"
|
||||
assert result.reachable is False
|
||||
assert result.error == "Connection timeout"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_host_connectivity_dns_failure(self):
|
||||
"""Test host connectivity when DNS resolution fails."""
|
||||
import socket
|
||||
with patch(
|
||||
"socket.create_connection",
|
||||
side_effect=socket.gaierror("Name resolution failed")
|
||||
):
|
||||
result = await check_host_connectivity("invalid.host", 80)
|
||||
assert result.host == "invalid.host"
|
||||
assert result.reachable is False
|
||||
assert "DNS resolution failed" in result.error
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_host_connectivity_connection_refused(self):
|
||||
"""Test host connectivity when connection is refused."""
|
||||
with patch(
|
||||
"socket.create_connection",
|
||||
side_effect=ConnectionRefusedError()
|
||||
):
|
||||
result = await check_host_connectivity("localhost", 12345)
|
||||
assert result.host == "localhost"
|
||||
assert result.reachable is False
|
||||
assert result.error == "Connection refused"
|
||||
Reference in New Issue
Block a user