mirror of
https://github.com/FuzzingLabs/fuzzforge_ai.git
synced 2026-07-18 10:27:25 +02:00
chore: improve code quality (backend package).
- add configuration file for 'ruff'.
- fix most of 'ruff' lints.
- format 'backend' package using 'ruff'.
This commit is contained in:
fztee
@@ -0,0 +1 @@
|
||||
"""Unit tests for modules."""
|
||||
|
||||
@@ -1,17 +1,26 @@
|
||||
"""
|
||||
Unit tests for AtherisFuzzer module
|
||||
"""
|
||||
"""Unit tests for AtherisFuzzer module."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from modules.fuzzer.atheris_fuzzer import AtherisFuzzer
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestAtherisFuzzerMetadata:
|
||||
"""Test AtherisFuzzer metadata"""
|
||||
"""Test AtherisFuzzer metadata."""
|
||||
|
||||
async def test_metadata_structure(self, atheris_fuzzer):
|
||||
"""Test that module metadata is properly defined"""
|
||||
async def test_metadata_structure(self, atheris_fuzzer: AtherisFuzzer) -> None:
|
||||
"""Test that module metadata is properly defined."""
|
||||
metadata = atheris_fuzzer.get_metadata()
|
||||
|
||||
assert metadata.name == "atheris_fuzzer"
|
||||
@@ -22,28 +31,28 @@ class TestAtherisFuzzerMetadata:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestAtherisFuzzerConfigValidation:
|
||||
"""Test configuration validation"""
|
||||
"""Test configuration validation."""
|
||||
|
||||
async def test_valid_config(self, atheris_fuzzer, atheris_config):
|
||||
"""Test validation of valid configuration"""
|
||||
async def test_valid_config(self, atheris_fuzzer: AtherisFuzzer, atheris_config: dict[str, Any]) -> None:
|
||||
"""Test validation of valid configuration."""
|
||||
assert atheris_fuzzer.validate_config(atheris_config) is True
|
||||
|
||||
async def test_invalid_max_iterations(self, atheris_fuzzer):
|
||||
"""Test validation fails with invalid max_iterations"""
|
||||
async def test_invalid_max_iterations(self, atheris_fuzzer: AtherisFuzzer) -> None:
|
||||
"""Test validation fails with invalid max_iterations."""
|
||||
config = {
|
||||
"target_file": "fuzz_target.py",
|
||||
"max_iterations": -1,
|
||||
"timeout_seconds": 10
|
||||
"timeout_seconds": 10,
|
||||
}
|
||||
with pytest.raises(ValueError, match="max_iterations"):
|
||||
atheris_fuzzer.validate_config(config)
|
||||
|
||||
async def test_invalid_timeout(self, atheris_fuzzer):
|
||||
"""Test validation fails with invalid timeout"""
|
||||
async def test_invalid_timeout(self, atheris_fuzzer: AtherisFuzzer) -> None:
|
||||
"""Test validation fails with invalid timeout."""
|
||||
config = {
|
||||
"target_file": "fuzz_target.py",
|
||||
"max_iterations": 1000,
|
||||
"timeout_seconds": 0
|
||||
"timeout_seconds": 0,
|
||||
}
|
||||
with pytest.raises(ValueError, match="timeout_seconds"):
|
||||
atheris_fuzzer.validate_config(config)
|
||||
@@ -51,10 +60,10 @@ class TestAtherisFuzzerConfigValidation:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestAtherisFuzzerDiscovery:
|
||||
"""Test fuzz target discovery"""
|
||||
"""Test fuzz target discovery."""
|
||||
|
||||
async def test_auto_discover(self, atheris_fuzzer, python_test_workspace):
|
||||
"""Test auto-discovery of Python fuzz targets"""
|
||||
async def test_auto_discover(self, atheris_fuzzer: AtherisFuzzer, python_test_workspace: Path) -> None:
|
||||
"""Test auto-discovery of Python fuzz targets."""
|
||||
# Create a fuzz target file
|
||||
(python_test_workspace / "fuzz_target.py").write_text("""
|
||||
import atheris
|
||||
@@ -69,7 +78,7 @@ if __name__ == "__main__":
|
||||
""")
|
||||
|
||||
# Pass None for auto-discovery
|
||||
target = atheris_fuzzer._discover_target(python_test_workspace, None)
|
||||
target = atheris_fuzzer._discover_target(python_test_workspace, None) # noqa: SLF001
|
||||
|
||||
assert target is not None
|
||||
assert "fuzz_target.py" in str(target)
|
||||
@@ -77,10 +86,14 @@ if __name__ == "__main__":
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestAtherisFuzzerExecution:
|
||||
"""Test fuzzer execution logic"""
|
||||
"""Test fuzzer execution logic."""
|
||||
|
||||
async def test_execution_creates_result(self, atheris_fuzzer, python_test_workspace, atheris_config):
|
||||
"""Test that execution returns a ModuleResult"""
|
||||
async def test_execution_creates_result(
|
||||
self,
|
||||
atheris_fuzzer: AtherisFuzzer,
|
||||
python_test_workspace: Path,
|
||||
) -> None:
|
||||
"""Test that execution returns a ModuleResult."""
|
||||
# Create a simple fuzz target
|
||||
(python_test_workspace / "fuzz_target.py").write_text("""
|
||||
import atheris
|
||||
@@ -99,11 +112,16 @@ if __name__ == "__main__":
|
||||
test_config = {
|
||||
"target_file": "fuzz_target.py",
|
||||
"max_iterations": 10,
|
||||
"timeout_seconds": 1
|
||||
"timeout_seconds": 1,
|
||||
}
|
||||
|
||||
# Mock the fuzzing subprocess to avoid actual execution
|
||||
with patch.object(atheris_fuzzer, '_run_fuzzing', new_callable=AsyncMock, return_value=([], {"total_executions": 10})):
|
||||
with patch.object(
|
||||
atheris_fuzzer,
|
||||
"_run_fuzzing",
|
||||
new_callable=AsyncMock,
|
||||
return_value=([], {"total_executions": 10}),
|
||||
):
|
||||
result = await atheris_fuzzer.execute(test_config, python_test_workspace)
|
||||
|
||||
assert result.module == "atheris_fuzzer"
|
||||
@@ -113,10 +131,16 @@ if __name__ == "__main__":
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestAtherisFuzzerStatsCallback:
|
||||
"""Test stats callback functionality"""
|
||||
"""Test stats callback functionality."""
|
||||
|
||||
async def test_stats_callback_invoked(self, atheris_fuzzer, python_test_workspace, atheris_config, mock_stats_callback):
|
||||
"""Test that stats callback is invoked during fuzzing"""
|
||||
async def test_stats_callback_invoked(
|
||||
self,
|
||||
atheris_fuzzer: AtherisFuzzer,
|
||||
python_test_workspace: Path,
|
||||
atheris_config: dict[str, Any],
|
||||
mock_stats_callback: Callable | None,
|
||||
) -> None:
|
||||
"""Test that stats callback is invoked during fuzzing."""
|
||||
(python_test_workspace / "fuzz_target.py").write_text("""
|
||||
import atheris
|
||||
import sys
|
||||
@@ -130,35 +154,45 @@ if __name__ == "__main__":
|
||||
""")
|
||||
|
||||
# Mock fuzzing to simulate stats
|
||||
async def mock_run_fuzzing(test_one_input, target_path, workspace, max_iterations, timeout_seconds, stats_callback):
|
||||
async def mock_run_fuzzing(
|
||||
test_one_input: Callable, # noqa: ARG001
|
||||
target_path: Path, # noqa: ARG001
|
||||
workspace: Path, # noqa: ARG001
|
||||
max_iterations: int, # noqa: ARG001
|
||||
timeout_seconds: int, # noqa: ARG001
|
||||
stats_callback: Callable | None,
|
||||
) -> None:
|
||||
if stats_callback:
|
||||
await stats_callback({
|
||||
"total_execs": 100,
|
||||
"execs_per_sec": 10.0,
|
||||
"crashes": 0,
|
||||
"coverage": 5,
|
||||
"corpus_size": 2,
|
||||
"elapsed_time": 10
|
||||
})
|
||||
return
|
||||
await stats_callback(
|
||||
{
|
||||
"total_execs": 100,
|
||||
"execs_per_sec": 10.0,
|
||||
"crashes": 0,
|
||||
"coverage": 5,
|
||||
"corpus_size": 2,
|
||||
"elapsed_time": 10,
|
||||
},
|
||||
)
|
||||
|
||||
with patch.object(atheris_fuzzer, '_run_fuzzing', side_effect=mock_run_fuzzing):
|
||||
with patch.object(atheris_fuzzer, '_load_target_module', return_value=lambda x: None):
|
||||
# Put stats_callback in config dict, not as kwarg
|
||||
atheris_config["target_file"] = "fuzz_target.py"
|
||||
atheris_config["stats_callback"] = mock_stats_callback
|
||||
await atheris_fuzzer.execute(atheris_config, python_test_workspace)
|
||||
with (
|
||||
patch.object(atheris_fuzzer, "_run_fuzzing", side_effect=mock_run_fuzzing),
|
||||
patch.object(atheris_fuzzer, "_load_target_module", return_value=lambda _x: None),
|
||||
):
|
||||
# Put stats_callback in config dict, not as kwarg
|
||||
atheris_config["target_file"] = "fuzz_target.py"
|
||||
atheris_config["stats_callback"] = mock_stats_callback
|
||||
await atheris_fuzzer.execute(atheris_config, python_test_workspace)
|
||||
|
||||
# Verify callback was invoked
|
||||
assert len(mock_stats_callback.stats_received) > 0
|
||||
# Verify callback was invoked
|
||||
assert len(mock_stats_callback.stats_received) > 0
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestAtherisFuzzerFindingGeneration:
|
||||
"""Test finding generation from crashes"""
|
||||
"""Test finding generation from crashes."""
|
||||
|
||||
async def test_create_crash_finding(self, atheris_fuzzer):
|
||||
"""Test crash finding creation"""
|
||||
async def test_create_crash_finding(self, atheris_fuzzer: AtherisFuzzer) -> None:
|
||||
"""Test crash finding creation."""
|
||||
finding = atheris_fuzzer.create_finding(
|
||||
title="Crash: Exception in TestOneInput",
|
||||
description="IndexError: list index out of range",
|
||||
@@ -167,8 +201,8 @@ class TestAtherisFuzzerFindingGeneration:
|
||||
file_path="fuzz_target.py",
|
||||
metadata={
|
||||
"crash_type": "IndexError",
|
||||
"stack_trace": "Traceback..."
|
||||
}
|
||||
"stack_trace": "Traceback...",
|
||||
},
|
||||
)
|
||||
|
||||
assert finding.title == "Crash: Exception in TestOneInput"
|
||||
|
||||
@@ -1,17 +1,26 @@
|
||||
"""
|
||||
Unit tests for CargoFuzzer module
|
||||
"""
|
||||
"""Unit tests for CargoFuzzer module."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import TYPE_CHECKING
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
import pytest
|
||||
from unittest.mock import AsyncMock, patch
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from collections.abc import Callable
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from modules.fuzzer.cargo_fuzzer import CargoFuzzer
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestCargoFuzzerMetadata:
|
||||
"""Test CargoFuzzer metadata"""
|
||||
"""Test CargoFuzzer metadata."""
|
||||
|
||||
async def test_metadata_structure(self, cargo_fuzzer):
|
||||
"""Test that module metadata is properly defined"""
|
||||
async def test_metadata_structure(self, cargo_fuzzer: CargoFuzzer) -> None:
|
||||
"""Test that module metadata is properly defined."""
|
||||
metadata = cargo_fuzzer.get_metadata()
|
||||
|
||||
assert metadata.name == "cargo_fuzz"
|
||||
@@ -23,38 +32,38 @@ class TestCargoFuzzerMetadata:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestCargoFuzzerConfigValidation:
|
||||
"""Test configuration validation"""
|
||||
"""Test configuration validation."""
|
||||
|
||||
async def test_valid_config(self, cargo_fuzzer, cargo_fuzz_config):
|
||||
"""Test validation of valid configuration"""
|
||||
async def test_valid_config(self, cargo_fuzzer: CargoFuzzer, cargo_fuzz_config: dict[str, Any]) -> None:
|
||||
"""Test validation of valid configuration."""
|
||||
assert cargo_fuzzer.validate_config(cargo_fuzz_config) is True
|
||||
|
||||
async def test_invalid_max_iterations(self, cargo_fuzzer):
|
||||
"""Test validation fails with invalid max_iterations"""
|
||||
async def test_invalid_max_iterations(self, cargo_fuzzer: CargoFuzzer) -> None:
|
||||
"""Test validation fails with invalid max_iterations."""
|
||||
config = {
|
||||
"max_iterations": -1,
|
||||
"timeout_seconds": 10,
|
||||
"sanitizer": "address"
|
||||
"sanitizer": "address",
|
||||
}
|
||||
with pytest.raises(ValueError, match="max_iterations"):
|
||||
cargo_fuzzer.validate_config(config)
|
||||
|
||||
async def test_invalid_timeout(self, cargo_fuzzer):
|
||||
"""Test validation fails with invalid timeout"""
|
||||
async def test_invalid_timeout(self, cargo_fuzzer: CargoFuzzer) -> None:
|
||||
"""Test validation fails with invalid timeout."""
|
||||
config = {
|
||||
"max_iterations": 1000,
|
||||
"timeout_seconds": 0,
|
||||
"sanitizer": "address"
|
||||
"sanitizer": "address",
|
||||
}
|
||||
with pytest.raises(ValueError, match="timeout_seconds"):
|
||||
cargo_fuzzer.validate_config(config)
|
||||
|
||||
async def test_invalid_sanitizer(self, cargo_fuzzer):
|
||||
"""Test validation fails with invalid sanitizer"""
|
||||
async def test_invalid_sanitizer(self, cargo_fuzzer: CargoFuzzer) -> None:
|
||||
"""Test validation fails with invalid sanitizer."""
|
||||
config = {
|
||||
"max_iterations": 1000,
|
||||
"timeout_seconds": 10,
|
||||
"sanitizer": "invalid_sanitizer"
|
||||
"sanitizer": "invalid_sanitizer",
|
||||
}
|
||||
with pytest.raises(ValueError, match="sanitizer"):
|
||||
cargo_fuzzer.validate_config(config)
|
||||
@@ -62,20 +71,20 @@ class TestCargoFuzzerConfigValidation:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestCargoFuzzerWorkspaceValidation:
|
||||
"""Test workspace validation"""
|
||||
"""Test workspace validation."""
|
||||
|
||||
async def test_valid_workspace(self, cargo_fuzzer, rust_test_workspace):
|
||||
"""Test validation of valid workspace"""
|
||||
async def test_valid_workspace(self, cargo_fuzzer: CargoFuzzer, rust_test_workspace: Path) -> None:
|
||||
"""Test validation of valid workspace."""
|
||||
assert cargo_fuzzer.validate_workspace(rust_test_workspace) is True
|
||||
|
||||
async def test_nonexistent_workspace(self, cargo_fuzzer, tmp_path):
|
||||
"""Test validation fails with nonexistent workspace"""
|
||||
async def test_nonexistent_workspace(self, cargo_fuzzer: CargoFuzzer, tmp_path: Path) -> None:
|
||||
"""Test validation fails with nonexistent workspace."""
|
||||
nonexistent = tmp_path / "does_not_exist"
|
||||
with pytest.raises(ValueError, match="does not exist"):
|
||||
cargo_fuzzer.validate_workspace(nonexistent)
|
||||
|
||||
async def test_workspace_is_file(self, cargo_fuzzer, tmp_path):
|
||||
"""Test validation fails when workspace is a file"""
|
||||
async def test_workspace_is_file(self, cargo_fuzzer: CargoFuzzer, tmp_path: Path) -> None:
|
||||
"""Test validation fails when workspace is a file."""
|
||||
file_path = tmp_path / "file.txt"
|
||||
file_path.write_text("test")
|
||||
with pytest.raises(ValueError, match="not a directory"):
|
||||
@@ -84,41 +93,58 @@ class TestCargoFuzzerWorkspaceValidation:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestCargoFuzzerDiscovery:
|
||||
"""Test fuzz target discovery"""
|
||||
"""Test fuzz target discovery."""
|
||||
|
||||
async def test_discover_targets(self, cargo_fuzzer, rust_test_workspace):
|
||||
"""Test discovery of fuzz targets"""
|
||||
targets = await cargo_fuzzer._discover_fuzz_targets(rust_test_workspace)
|
||||
async def test_discover_targets(self, cargo_fuzzer: CargoFuzzer, rust_test_workspace: Path) -> None:
|
||||
"""Test discovery of fuzz targets."""
|
||||
targets = await cargo_fuzzer._discover_fuzz_targets(rust_test_workspace) # noqa: SLF001
|
||||
|
||||
assert len(targets) == 1
|
||||
assert "fuzz_target_1" in targets
|
||||
|
||||
async def test_no_fuzz_directory(self, cargo_fuzzer, temp_workspace):
|
||||
"""Test discovery with no fuzz directory"""
|
||||
targets = await cargo_fuzzer._discover_fuzz_targets(temp_workspace)
|
||||
async def test_no_fuzz_directory(self, cargo_fuzzer: CargoFuzzer, temp_workspace: Path) -> None:
|
||||
"""Test discovery with no fuzz directory."""
|
||||
targets = await cargo_fuzzer._discover_fuzz_targets(temp_workspace) # noqa: SLF001
|
||||
|
||||
assert targets == []
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestCargoFuzzerExecution:
|
||||
"""Test fuzzer execution logic"""
|
||||
"""Test fuzzer execution logic."""
|
||||
|
||||
async def test_execution_creates_result(self, cargo_fuzzer, rust_test_workspace, cargo_fuzz_config):
|
||||
"""Test that execution returns a ModuleResult"""
|
||||
async def test_execution_creates_result(
|
||||
self,
|
||||
cargo_fuzzer: CargoFuzzer,
|
||||
rust_test_workspace: Path,
|
||||
cargo_fuzz_config: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test that execution returns a ModuleResult."""
|
||||
# Mock the build and run methods to avoid actual fuzzing
|
||||
with patch.object(cargo_fuzzer, '_build_fuzz_target', new_callable=AsyncMock, return_value=True):
|
||||
with patch.object(cargo_fuzzer, '_run_fuzzing', new_callable=AsyncMock, return_value=([], {"total_executions": 0, "crashes_found": 0})):
|
||||
with patch.object(cargo_fuzzer, '_parse_crash_artifacts', new_callable=AsyncMock, return_value=[]):
|
||||
result = await cargo_fuzzer.execute(cargo_fuzz_config, rust_test_workspace)
|
||||
with (
|
||||
patch.object(cargo_fuzzer, "_build_fuzz_target", new_callable=AsyncMock, return_value=True),
|
||||
patch.object(
|
||||
cargo_fuzzer,
|
||||
"_run_fuzzing",
|
||||
new_callable=AsyncMock,
|
||||
return_value=([], {"total_executions": 0, "crashes_found": 0}),
|
||||
),
|
||||
patch.object(cargo_fuzzer, "_parse_crash_artifacts", new_callable=AsyncMock, return_value=[]),
|
||||
):
|
||||
result = await cargo_fuzzer.execute(cargo_fuzz_config, rust_test_workspace)
|
||||
|
||||
assert result.module == "cargo_fuzz"
|
||||
assert result.status == "success"
|
||||
assert isinstance(result.execution_time, float)
|
||||
assert result.execution_time >= 0
|
||||
assert result.module == "cargo_fuzz"
|
||||
assert result.status == "success"
|
||||
assert isinstance(result.execution_time, float)
|
||||
assert result.execution_time >= 0
|
||||
|
||||
async def test_execution_with_no_targets(self, cargo_fuzzer, temp_workspace, cargo_fuzz_config):
|
||||
"""Test execution fails gracefully with no fuzz targets"""
|
||||
async def test_execution_with_no_targets(
|
||||
self,
|
||||
cargo_fuzzer: CargoFuzzer,
|
||||
temp_workspace: Path,
|
||||
cargo_fuzz_config: dict[str, Any],
|
||||
) -> None:
|
||||
"""Test execution fails gracefully with no fuzz targets."""
|
||||
result = await cargo_fuzzer.execute(cargo_fuzz_config, temp_workspace)
|
||||
|
||||
assert result.status == "failed"
|
||||
@@ -127,47 +153,67 @@ class TestCargoFuzzerExecution:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestCargoFuzzerStatsCallback:
|
||||
"""Test stats callback functionality"""
|
||||
"""Test stats callback functionality."""
|
||||
|
||||
async def test_stats_callback_invoked(
|
||||
self,
|
||||
cargo_fuzzer: CargoFuzzer,
|
||||
rust_test_workspace: Path,
|
||||
cargo_fuzz_config: dict[str, Any],
|
||||
mock_stats_callback: Callable | None,
|
||||
) -> None:
|
||||
"""Test that stats callback is invoked during fuzzing."""
|
||||
|
||||
async def test_stats_callback_invoked(self, cargo_fuzzer, rust_test_workspace, cargo_fuzz_config, mock_stats_callback):
|
||||
"""Test that stats callback is invoked during fuzzing"""
|
||||
# Mock build/run to simulate stats generation
|
||||
async def mock_run_fuzzing(workspace, target, config, callback):
|
||||
async def mock_run_fuzzing(
|
||||
_workspace: Path,
|
||||
_target: str,
|
||||
_config: dict[str, Any],
|
||||
callback: Callable | None,
|
||||
) -> tuple[list, dict[str, int]]:
|
||||
# Simulate stats callback
|
||||
if callback:
|
||||
await callback({
|
||||
"total_execs": 1000,
|
||||
"execs_per_sec": 100.0,
|
||||
"crashes": 0,
|
||||
"coverage": 10,
|
||||
"corpus_size": 5,
|
||||
"elapsed_time": 10
|
||||
})
|
||||
await callback(
|
||||
{
|
||||
"total_execs": 1000,
|
||||
"execs_per_sec": 100.0,
|
||||
"crashes": 0,
|
||||
"coverage": 10,
|
||||
"corpus_size": 5,
|
||||
"elapsed_time": 10,
|
||||
},
|
||||
)
|
||||
return [], {"total_executions": 1000}
|
||||
|
||||
with patch.object(cargo_fuzzer, '_build_fuzz_target', new_callable=AsyncMock, return_value=True):
|
||||
with patch.object(cargo_fuzzer, '_run_fuzzing', side_effect=mock_run_fuzzing):
|
||||
with patch.object(cargo_fuzzer, '_parse_crash_artifacts', new_callable=AsyncMock, return_value=[]):
|
||||
await cargo_fuzzer.execute(cargo_fuzz_config, rust_test_workspace, stats_callback=mock_stats_callback)
|
||||
with (
|
||||
patch.object(cargo_fuzzer, "_build_fuzz_target", new_callable=AsyncMock, return_value=True),
|
||||
patch.object(cargo_fuzzer, "_run_fuzzing", side_effect=mock_run_fuzzing),
|
||||
patch.object(cargo_fuzzer, "_parse_crash_artifacts", new_callable=AsyncMock, return_value=[]),
|
||||
):
|
||||
await cargo_fuzzer.execute(
|
||||
cargo_fuzz_config,
|
||||
rust_test_workspace,
|
||||
stats_callback=mock_stats_callback,
|
||||
)
|
||||
|
||||
# Verify callback was invoked
|
||||
assert len(mock_stats_callback.stats_received) > 0
|
||||
assert mock_stats_callback.stats_received[0]["total_execs"] == 1000
|
||||
# Verify callback was invoked
|
||||
assert len(mock_stats_callback.stats_received) > 0
|
||||
assert mock_stats_callback.stats_received[0]["total_execs"] == 1000
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestCargoFuzzerFindingGeneration:
|
||||
"""Test finding generation from crashes"""
|
||||
"""Test finding generation from crashes."""
|
||||
|
||||
async def test_create_finding_from_crash(self, cargo_fuzzer):
|
||||
"""Test finding creation"""
|
||||
async def test_create_finding_from_crash(self, cargo_fuzzer: CargoFuzzer) -> None:
|
||||
"""Test finding creation."""
|
||||
finding = cargo_fuzzer.create_finding(
|
||||
title="Crash: Segmentation Fault",
|
||||
description="Test crash",
|
||||
severity="critical",
|
||||
category="crash",
|
||||
file_path="fuzz/fuzz_targets/fuzz_target_1.rs",
|
||||
metadata={"crash_type": "SIGSEGV"}
|
||||
metadata={"crash_type": "SIGSEGV"},
|
||||
)
|
||||
|
||||
assert finding.title == "Crash: Segmentation Fault"
|
||||
|
||||
@@ -1,22 +1,25 @@
|
||||
"""
|
||||
Unit tests for FileScanner module
|
||||
"""
|
||||
"""Unit tests for FileScanner module."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
|
||||
if TYPE_CHECKING:
|
||||
from modules.scanner.file_scanner import FileScanner
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "toolbox"))
|
||||
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestFileScannerMetadata:
|
||||
"""Test FileScanner metadata"""
|
||||
"""Test FileScanner metadata."""
|
||||
|
||||
async def test_metadata_structure(self, file_scanner):
|
||||
"""Test that metadata has correct structure"""
|
||||
async def test_metadata_structure(self, file_scanner: FileScanner) -> None:
|
||||
"""Test that metadata has correct structure."""
|
||||
metadata = file_scanner.get_metadata()
|
||||
|
||||
assert metadata.name == "file_scanner"
|
||||
@@ -29,37 +32,37 @@ class TestFileScannerMetadata:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestFileScannerConfigValidation:
|
||||
"""Test configuration validation"""
|
||||
"""Test configuration validation."""
|
||||
|
||||
async def test_valid_config(self, file_scanner):
|
||||
"""Test that valid config passes validation"""
|
||||
async def test_valid_config(self, file_scanner: FileScanner) -> None:
|
||||
"""Test that valid config passes validation."""
|
||||
config = {
|
||||
"patterns": ["*.py", "*.js"],
|
||||
"max_file_size": 1048576,
|
||||
"check_sensitive": True,
|
||||
"calculate_hashes": False
|
||||
"calculate_hashes": False,
|
||||
}
|
||||
assert file_scanner.validate_config(config) is True
|
||||
|
||||
async def test_default_config(self, file_scanner):
|
||||
"""Test that empty config uses defaults"""
|
||||
async def test_default_config(self, file_scanner: FileScanner) -> None:
|
||||
"""Test that empty config uses defaults."""
|
||||
config = {}
|
||||
assert file_scanner.validate_config(config) is True
|
||||
|
||||
async def test_invalid_patterns_type(self, file_scanner):
|
||||
"""Test that non-list patterns raises error"""
|
||||
async def test_invalid_patterns_type(self, file_scanner: FileScanner) -> None:
|
||||
"""Test that non-list patterns raises error."""
|
||||
config = {"patterns": "*.py"}
|
||||
with pytest.raises(ValueError, match="patterns must be a list"):
|
||||
file_scanner.validate_config(config)
|
||||
|
||||
async def test_invalid_max_file_size(self, file_scanner):
|
||||
"""Test that invalid max_file_size raises error"""
|
||||
async def test_invalid_max_file_size(self, file_scanner: FileScanner) -> None:
|
||||
"""Test that invalid max_file_size raises error."""
|
||||
config = {"max_file_size": -1}
|
||||
with pytest.raises(ValueError, match="max_file_size must be a positive integer"):
|
||||
file_scanner.validate_config(config)
|
||||
|
||||
async def test_invalid_max_file_size_type(self, file_scanner):
|
||||
"""Test that non-integer max_file_size raises error"""
|
||||
async def test_invalid_max_file_size_type(self, file_scanner: FileScanner) -> None:
|
||||
"""Test that non-integer max_file_size raises error."""
|
||||
config = {"max_file_size": "large"}
|
||||
with pytest.raises(ValueError, match="max_file_size must be a positive integer"):
|
||||
file_scanner.validate_config(config)
|
||||
@@ -67,14 +70,14 @@ class TestFileScannerConfigValidation:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestFileScannerExecution:
|
||||
"""Test scanner execution"""
|
||||
"""Test scanner execution."""
|
||||
|
||||
async def test_scan_python_files(self, file_scanner, python_test_workspace):
|
||||
"""Test scanning Python files"""
|
||||
async def test_scan_python_files(self, file_scanner: FileScanner, python_test_workspace: Path) -> None:
|
||||
"""Test scanning Python files."""
|
||||
config = {
|
||||
"patterns": ["*.py"],
|
||||
"check_sensitive": False,
|
||||
"calculate_hashes": False
|
||||
"calculate_hashes": False,
|
||||
}
|
||||
|
||||
result = await file_scanner.execute(config, python_test_workspace)
|
||||
@@ -84,15 +87,15 @@ class TestFileScannerExecution:
|
||||
assert len(result.findings) > 0
|
||||
|
||||
# Check that Python files were found
|
||||
python_files = [f for f in result.findings if f.file_path.endswith('.py')]
|
||||
python_files = [f for f in result.findings if f.file_path.endswith(".py")]
|
||||
assert len(python_files) > 0
|
||||
|
||||
async def test_scan_all_files(self, file_scanner, python_test_workspace):
|
||||
"""Test scanning all files with wildcard"""
|
||||
async def test_scan_all_files(self, file_scanner: FileScanner, python_test_workspace: Path) -> None:
|
||||
"""Test scanning all files with wildcard."""
|
||||
config = {
|
||||
"patterns": ["*"],
|
||||
"check_sensitive": False,
|
||||
"calculate_hashes": False
|
||||
"calculate_hashes": False,
|
||||
}
|
||||
|
||||
result = await file_scanner.execute(config, python_test_workspace)
|
||||
@@ -101,12 +104,12 @@ class TestFileScannerExecution:
|
||||
assert len(result.findings) > 0
|
||||
assert result.summary["total_files"] > 0
|
||||
|
||||
async def test_scan_with_multiple_patterns(self, file_scanner, python_test_workspace):
|
||||
"""Test scanning with multiple patterns"""
|
||||
async def test_scan_with_multiple_patterns(self, file_scanner: FileScanner, python_test_workspace: Path) -> None:
|
||||
"""Test scanning with multiple patterns."""
|
||||
config = {
|
||||
"patterns": ["*.py", "*.txt"],
|
||||
"check_sensitive": False,
|
||||
"calculate_hashes": False
|
||||
"calculate_hashes": False,
|
||||
}
|
||||
|
||||
result = await file_scanner.execute(config, python_test_workspace)
|
||||
@@ -114,11 +117,11 @@ class TestFileScannerExecution:
|
||||
assert result.status == "success"
|
||||
assert len(result.findings) > 0
|
||||
|
||||
async def test_empty_workspace(self, file_scanner, temp_workspace):
|
||||
"""Test scanning empty workspace"""
|
||||
async def test_empty_workspace(self, file_scanner: FileScanner, temp_workspace: Path) -> None:
|
||||
"""Test scanning empty workspace."""
|
||||
config = {
|
||||
"patterns": ["*.py"],
|
||||
"check_sensitive": False
|
||||
"check_sensitive": False,
|
||||
}
|
||||
|
||||
result = await file_scanner.execute(config, temp_workspace)
|
||||
@@ -130,17 +133,17 @@ class TestFileScannerExecution:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestFileScannerSensitiveDetection:
|
||||
"""Test sensitive file detection"""
|
||||
"""Test sensitive file detection."""
|
||||
|
||||
async def test_detect_env_file(self, file_scanner, temp_workspace):
|
||||
"""Test detection of .env file"""
|
||||
async def test_detect_env_file(self, file_scanner: FileScanner, temp_workspace: Path) -> None:
|
||||
"""Test detection of .env file."""
|
||||
# Create .env file
|
||||
(temp_workspace / ".env").write_text("API_KEY=secret123")
|
||||
|
||||
config = {
|
||||
"patterns": ["*"],
|
||||
"check_sensitive": True,
|
||||
"calculate_hashes": False
|
||||
"calculate_hashes": False,
|
||||
}
|
||||
|
||||
result = await file_scanner.execute(config, temp_workspace)
|
||||
@@ -152,14 +155,14 @@ class TestFileScannerSensitiveDetection:
|
||||
assert len(sensitive_findings) > 0
|
||||
assert any(".env" in f.title for f in sensitive_findings)
|
||||
|
||||
async def test_detect_private_key(self, file_scanner, temp_workspace):
|
||||
"""Test detection of private key file"""
|
||||
async def test_detect_private_key(self, file_scanner: FileScanner, temp_workspace: Path) -> None:
|
||||
"""Test detection of private key file."""
|
||||
# Create private key file
|
||||
(temp_workspace / "id_rsa").write_text("-----BEGIN RSA PRIVATE KEY-----")
|
||||
|
||||
config = {
|
||||
"patterns": ["*"],
|
||||
"check_sensitive": True
|
||||
"check_sensitive": True,
|
||||
}
|
||||
|
||||
result = await file_scanner.execute(config, temp_workspace)
|
||||
@@ -168,13 +171,13 @@ class TestFileScannerSensitiveDetection:
|
||||
sensitive_findings = [f for f in result.findings if f.category == "sensitive_file"]
|
||||
assert len(sensitive_findings) > 0
|
||||
|
||||
async def test_no_sensitive_detection_when_disabled(self, file_scanner, temp_workspace):
|
||||
"""Test that sensitive detection can be disabled"""
|
||||
async def test_no_sensitive_detection_when_disabled(self, file_scanner: FileScanner, temp_workspace: Path) -> None:
|
||||
"""Test that sensitive detection can be disabled."""
|
||||
(temp_workspace / ".env").write_text("API_KEY=secret123")
|
||||
|
||||
config = {
|
||||
"patterns": ["*"],
|
||||
"check_sensitive": False
|
||||
"check_sensitive": False,
|
||||
}
|
||||
|
||||
result = await file_scanner.execute(config, temp_workspace)
|
||||
@@ -186,17 +189,17 @@ class TestFileScannerSensitiveDetection:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestFileScannerHashing:
|
||||
"""Test file hashing functionality"""
|
||||
"""Test file hashing functionality."""
|
||||
|
||||
async def test_hash_calculation(self, file_scanner, temp_workspace):
|
||||
"""Test SHA256 hash calculation"""
|
||||
async def test_hash_calculation(self, file_scanner: FileScanner, temp_workspace: Path) -> None:
|
||||
"""Test SHA256 hash calculation."""
|
||||
# Create test file
|
||||
test_file = temp_workspace / "test.txt"
|
||||
test_file.write_text("Hello World")
|
||||
|
||||
config = {
|
||||
"patterns": ["*.txt"],
|
||||
"calculate_hashes": True
|
||||
"calculate_hashes": True,
|
||||
}
|
||||
|
||||
result = await file_scanner.execute(config, temp_workspace)
|
||||
@@ -212,14 +215,14 @@ class TestFileScannerHashing:
|
||||
assert finding.metadata.get("file_hash") is not None
|
||||
assert len(finding.metadata["file_hash"]) == 64 # SHA256 hex length
|
||||
|
||||
async def test_no_hash_when_disabled(self, file_scanner, temp_workspace):
|
||||
"""Test that hashing can be disabled"""
|
||||
async def test_no_hash_when_disabled(self, file_scanner: FileScanner, temp_workspace: Path) -> None:
|
||||
"""Test that hashing can be disabled."""
|
||||
test_file = temp_workspace / "test.txt"
|
||||
test_file.write_text("Hello World")
|
||||
|
||||
config = {
|
||||
"patterns": ["*.txt"],
|
||||
"calculate_hashes": False
|
||||
"calculate_hashes": False,
|
||||
}
|
||||
|
||||
result = await file_scanner.execute(config, temp_workspace)
|
||||
@@ -234,10 +237,10 @@ class TestFileScannerHashing:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestFileScannerFileTypes:
|
||||
"""Test file type detection"""
|
||||
"""Test file type detection."""
|
||||
|
||||
async def test_detect_python_type(self, file_scanner, temp_workspace):
|
||||
"""Test detection of Python file type"""
|
||||
async def test_detect_python_type(self, file_scanner: FileScanner, temp_workspace: Path) -> None:
|
||||
"""Test detection of Python file type."""
|
||||
(temp_workspace / "script.py").write_text("print('hello')")
|
||||
|
||||
config = {"patterns": ["*.py"]}
|
||||
@@ -248,8 +251,8 @@ class TestFileScannerFileTypes:
|
||||
assert len(py_findings) > 0
|
||||
assert "python" in py_findings[0].metadata["file_type"]
|
||||
|
||||
async def test_detect_javascript_type(self, file_scanner, temp_workspace):
|
||||
"""Test detection of JavaScript file type"""
|
||||
async def test_detect_javascript_type(self, file_scanner: FileScanner, temp_workspace: Path) -> None:
|
||||
"""Test detection of JavaScript file type."""
|
||||
(temp_workspace / "app.js").write_text("console.log('hello')")
|
||||
|
||||
config = {"patterns": ["*.js"]}
|
||||
@@ -260,8 +263,8 @@ class TestFileScannerFileTypes:
|
||||
assert len(js_findings) > 0
|
||||
assert "javascript" in js_findings[0].metadata["file_type"]
|
||||
|
||||
async def test_file_type_summary(self, file_scanner, temp_workspace):
|
||||
"""Test that file type summary is generated"""
|
||||
async def test_file_type_summary(self, file_scanner: FileScanner, temp_workspace: Path) -> None:
|
||||
"""Test that file type summary is generated."""
|
||||
(temp_workspace / "script.py").write_text("print('hello')")
|
||||
(temp_workspace / "app.js").write_text("console.log('hello')")
|
||||
(temp_workspace / "readme.txt").write_text("Documentation")
|
||||
@@ -276,17 +279,17 @@ class TestFileScannerFileTypes:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestFileScannerSizeLimits:
|
||||
"""Test file size handling"""
|
||||
"""Test file size handling."""
|
||||
|
||||
async def test_skip_large_files(self, file_scanner, temp_workspace):
|
||||
"""Test that large files are skipped"""
|
||||
async def test_skip_large_files(self, file_scanner: FileScanner, temp_workspace: Path) -> None:
|
||||
"""Test that large files are skipped."""
|
||||
# Create a "large" file
|
||||
large_file = temp_workspace / "large.txt"
|
||||
large_file.write_text("x" * 1000)
|
||||
|
||||
config = {
|
||||
"patterns": ["*.txt"],
|
||||
"max_file_size": 500 # Set limit smaller than file
|
||||
"max_file_size": 500, # Set limit smaller than file
|
||||
}
|
||||
|
||||
result = await file_scanner.execute(config, temp_workspace)
|
||||
@@ -297,14 +300,14 @@ class TestFileScannerSizeLimits:
|
||||
# The file should still be counted but not have a detailed finding
|
||||
assert result.summary["total_files"] > 0
|
||||
|
||||
async def test_process_small_files(self, file_scanner, temp_workspace):
|
||||
"""Test that small files are processed"""
|
||||
async def test_process_small_files(self, file_scanner: FileScanner, temp_workspace: Path) -> None:
|
||||
"""Test that small files are processed."""
|
||||
small_file = temp_workspace / "small.txt"
|
||||
small_file.write_text("small content")
|
||||
|
||||
config = {
|
||||
"patterns": ["*.txt"],
|
||||
"max_file_size": 1048576 # 1MB
|
||||
"max_file_size": 1048576, # 1MB
|
||||
}
|
||||
|
||||
result = await file_scanner.execute(config, temp_workspace)
|
||||
@@ -316,10 +319,10 @@ class TestFileScannerSizeLimits:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestFileScannerSummary:
|
||||
"""Test result summary generation"""
|
||||
"""Test result summary generation."""
|
||||
|
||||
async def test_summary_structure(self, file_scanner, python_test_workspace):
|
||||
"""Test that summary has correct structure"""
|
||||
async def test_summary_structure(self, file_scanner: FileScanner, python_test_workspace: Path) -> None:
|
||||
"""Test that summary has correct structure."""
|
||||
config = {"patterns": ["*"]}
|
||||
result = await file_scanner.execute(config, python_test_workspace)
|
||||
|
||||
@@ -334,8 +337,8 @@ class TestFileScannerSummary:
|
||||
assert isinstance(result.summary["file_types"], dict)
|
||||
assert isinstance(result.summary["patterns_scanned"], list)
|
||||
|
||||
async def test_summary_counts(self, file_scanner, temp_workspace):
|
||||
"""Test that summary counts are accurate"""
|
||||
async def test_summary_counts(self, file_scanner: FileScanner, temp_workspace: Path) -> None:
|
||||
"""Test that summary counts are accurate."""
|
||||
# Create known files
|
||||
(temp_workspace / "file1.py").write_text("content1")
|
||||
(temp_workspace / "file2.py").write_text("content2")
|
||||
|
||||
@@ -1,28 +1,25 @@
|
||||
"""
|
||||
Unit tests for SecurityAnalyzer module
|
||||
"""
|
||||
"""Unit tests for SecurityAnalyzer module."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import pytest
|
||||
import sys
|
||||
from pathlib import Path
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
import pytest
|
||||
|
||||
sys.path.insert(0, str(Path(__file__).resolve().parents[3] / "toolbox"))
|
||||
|
||||
from modules.analyzer.security_analyzer import SecurityAnalyzer
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def security_analyzer():
|
||||
"""Create SecurityAnalyzer instance"""
|
||||
return SecurityAnalyzer()
|
||||
if TYPE_CHECKING:
|
||||
from modules.analyzer.security_analyzer import SecurityAnalyzer
|
||||
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestSecurityAnalyzerMetadata:
|
||||
"""Test SecurityAnalyzer metadata"""
|
||||
"""Test SecurityAnalyzer metadata."""
|
||||
|
||||
async def test_metadata_structure(self, security_analyzer):
|
||||
"""Test that metadata has correct structure"""
|
||||
async def test_metadata_structure(self, security_analyzer: SecurityAnalyzer) -> None:
|
||||
"""Test that metadata has correct structure."""
|
||||
metadata = security_analyzer.get_metadata()
|
||||
|
||||
assert metadata.name == "security_analyzer"
|
||||
@@ -35,25 +32,25 @@ class TestSecurityAnalyzerMetadata:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestSecurityAnalyzerConfigValidation:
|
||||
"""Test configuration validation"""
|
||||
"""Test configuration validation."""
|
||||
|
||||
async def test_valid_config(self, security_analyzer):
|
||||
"""Test that valid config passes validation"""
|
||||
async def test_valid_config(self, security_analyzer: SecurityAnalyzer) -> None:
|
||||
"""Test that valid config passes validation."""
|
||||
config = {
|
||||
"file_extensions": [".py", ".js"],
|
||||
"check_secrets": True,
|
||||
"check_sql": True,
|
||||
"check_dangerous_functions": True
|
||||
"check_dangerous_functions": True,
|
||||
}
|
||||
assert security_analyzer.validate_config(config) is True
|
||||
|
||||
async def test_default_config(self, security_analyzer):
|
||||
"""Test that empty config uses defaults"""
|
||||
async def test_default_config(self, security_analyzer: SecurityAnalyzer) -> None:
|
||||
"""Test that empty config uses defaults."""
|
||||
config = {}
|
||||
assert security_analyzer.validate_config(config) is True
|
||||
|
||||
async def test_invalid_extensions_type(self, security_analyzer):
|
||||
"""Test that non-list extensions raises error"""
|
||||
async def test_invalid_extensions_type(self, security_analyzer: SecurityAnalyzer) -> None:
|
||||
"""Test that non-list extensions raises error."""
|
||||
config = {"file_extensions": ".py"}
|
||||
with pytest.raises(ValueError, match="file_extensions must be a list"):
|
||||
security_analyzer.validate_config(config)
|
||||
@@ -61,10 +58,10 @@ class TestSecurityAnalyzerConfigValidation:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestSecurityAnalyzerSecretDetection:
|
||||
"""Test hardcoded secret detection"""
|
||||
"""Test hardcoded secret detection."""
|
||||
|
||||
async def test_detect_api_key(self, security_analyzer, temp_workspace):
|
||||
"""Test detection of hardcoded API key"""
|
||||
async def test_detect_api_key(self, security_analyzer: SecurityAnalyzer, temp_workspace: Path) -> None:
|
||||
"""Test detection of hardcoded API key."""
|
||||
code_file = temp_workspace / "config.py"
|
||||
code_file.write_text("""
|
||||
# Configuration file
|
||||
@@ -76,7 +73,7 @@ database_url = "postgresql://localhost/db"
|
||||
"file_extensions": [".py"],
|
||||
"check_secrets": True,
|
||||
"check_sql": False,
|
||||
"check_dangerous_functions": False
|
||||
"check_dangerous_functions": False,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -86,8 +83,8 @@ database_url = "postgresql://localhost/db"
|
||||
assert len(secret_findings) > 0
|
||||
assert any("API Key" in f.title for f in secret_findings)
|
||||
|
||||
async def test_detect_password(self, security_analyzer, temp_workspace):
|
||||
"""Test detection of hardcoded password"""
|
||||
async def test_detect_password(self, security_analyzer: SecurityAnalyzer, temp_workspace: Path) -> None:
|
||||
"""Test detection of hardcoded password."""
|
||||
code_file = temp_workspace / "auth.py"
|
||||
code_file.write_text("""
|
||||
def connect():
|
||||
@@ -99,7 +96,7 @@ def connect():
|
||||
"file_extensions": [".py"],
|
||||
"check_secrets": True,
|
||||
"check_sql": False,
|
||||
"check_dangerous_functions": False
|
||||
"check_dangerous_functions": False,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -108,8 +105,8 @@ def connect():
|
||||
secret_findings = [f for f in result.findings if f.category == "hardcoded_secret"]
|
||||
assert len(secret_findings) > 0
|
||||
|
||||
async def test_detect_aws_credentials(self, security_analyzer, temp_workspace):
|
||||
"""Test detection of AWS credentials"""
|
||||
async def test_detect_aws_credentials(self, security_analyzer: SecurityAnalyzer, temp_workspace: Path) -> None:
|
||||
"""Test detection of AWS credentials."""
|
||||
code_file = temp_workspace / "aws_config.py"
|
||||
code_file.write_text("""
|
||||
aws_access_key = "AKIAIOSFODNN7REALKEY"
|
||||
@@ -118,7 +115,7 @@ aws_secret_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYREALKEY"
|
||||
|
||||
config = {
|
||||
"file_extensions": [".py"],
|
||||
"check_secrets": True
|
||||
"check_secrets": True,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -127,14 +124,18 @@ aws_secret_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYREALKEY"
|
||||
aws_findings = [f for f in result.findings if "AWS" in f.title]
|
||||
assert len(aws_findings) >= 2 # Both access key and secret key
|
||||
|
||||
async def test_no_secret_detection_when_disabled(self, security_analyzer, temp_workspace):
|
||||
"""Test that secret detection can be disabled"""
|
||||
async def test_no_secret_detection_when_disabled(
|
||||
self,
|
||||
security_analyzer: SecurityAnalyzer,
|
||||
temp_workspace: Path,
|
||||
) -> None:
|
||||
"""Test that secret detection can be disabled."""
|
||||
code_file = temp_workspace / "config.py"
|
||||
code_file.write_text('api_key = "sk_live_1234567890abcdef"')
|
||||
|
||||
config = {
|
||||
"file_extensions": [".py"],
|
||||
"check_secrets": False
|
||||
"check_secrets": False,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -146,10 +147,10 @@ aws_secret_key = "wJalrXUtnFEMI/K7MDENG/bPxRfiCYREALKEY"
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestSecurityAnalyzerSQLInjection:
|
||||
"""Test SQL injection detection"""
|
||||
"""Test SQL injection detection."""
|
||||
|
||||
async def test_detect_string_concatenation(self, security_analyzer, temp_workspace):
|
||||
"""Test detection of SQL string concatenation"""
|
||||
async def test_detect_string_concatenation(self, security_analyzer: SecurityAnalyzer, temp_workspace: Path) -> None:
|
||||
"""Test detection of SQL string concatenation."""
|
||||
code_file = temp_workspace / "db.py"
|
||||
code_file.write_text("""
|
||||
def get_user(user_id):
|
||||
@@ -161,7 +162,7 @@ def get_user(user_id):
|
||||
"file_extensions": [".py"],
|
||||
"check_secrets": False,
|
||||
"check_sql": True,
|
||||
"check_dangerous_functions": False
|
||||
"check_dangerous_functions": False,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -170,8 +171,8 @@ def get_user(user_id):
|
||||
sql_findings = [f for f in result.findings if f.category == "sql_injection"]
|
||||
assert len(sql_findings) > 0
|
||||
|
||||
async def test_detect_f_string_sql(self, security_analyzer, temp_workspace):
|
||||
"""Test detection of f-string in SQL"""
|
||||
async def test_detect_f_string_sql(self, security_analyzer: SecurityAnalyzer, temp_workspace: Path) -> None:
|
||||
"""Test detection of f-string in SQL."""
|
||||
code_file = temp_workspace / "db.py"
|
||||
code_file.write_text("""
|
||||
def get_user(name):
|
||||
@@ -181,7 +182,7 @@ def get_user(name):
|
||||
|
||||
config = {
|
||||
"file_extensions": [".py"],
|
||||
"check_sql": True
|
||||
"check_sql": True,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -190,8 +191,12 @@ def get_user(name):
|
||||
sql_findings = [f for f in result.findings if f.category == "sql_injection"]
|
||||
assert len(sql_findings) > 0
|
||||
|
||||
async def test_detect_dynamic_query_building(self, security_analyzer, temp_workspace):
|
||||
"""Test detection of dynamic query building"""
|
||||
async def test_detect_dynamic_query_building(
|
||||
self,
|
||||
security_analyzer: SecurityAnalyzer,
|
||||
temp_workspace: Path,
|
||||
) -> None:
|
||||
"""Test detection of dynamic query building."""
|
||||
code_file = temp_workspace / "queries.py"
|
||||
code_file.write_text("""
|
||||
def search(keyword):
|
||||
@@ -201,7 +206,7 @@ def search(keyword):
|
||||
|
||||
config = {
|
||||
"file_extensions": [".py"],
|
||||
"check_sql": True
|
||||
"check_sql": True,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -210,14 +215,18 @@ def search(keyword):
|
||||
sql_findings = [f for f in result.findings if f.category == "sql_injection"]
|
||||
assert len(sql_findings) > 0
|
||||
|
||||
async def test_no_sql_detection_when_disabled(self, security_analyzer, temp_workspace):
|
||||
"""Test that SQL detection can be disabled"""
|
||||
async def test_no_sql_detection_when_disabled(
|
||||
self,
|
||||
security_analyzer: SecurityAnalyzer,
|
||||
temp_workspace: Path,
|
||||
) -> None:
|
||||
"""Test that SQL detection can be disabled."""
|
||||
code_file = temp_workspace / "db.py"
|
||||
code_file.write_text('query = "SELECT * FROM users WHERE id = " + user_id')
|
||||
|
||||
config = {
|
||||
"file_extensions": [".py"],
|
||||
"check_sql": False
|
||||
"check_sql": False,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -229,10 +238,10 @@ def search(keyword):
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestSecurityAnalyzerDangerousFunctions:
|
||||
"""Test dangerous function detection"""
|
||||
"""Test dangerous function detection."""
|
||||
|
||||
async def test_detect_eval(self, security_analyzer, temp_workspace):
|
||||
"""Test detection of eval() usage"""
|
||||
async def test_detect_eval(self, security_analyzer: SecurityAnalyzer, temp_workspace: Path) -> None:
|
||||
"""Test detection of eval() usage."""
|
||||
code_file = temp_workspace / "dangerous.py"
|
||||
code_file.write_text("""
|
||||
def process_input(user_input):
|
||||
@@ -244,7 +253,7 @@ def process_input(user_input):
|
||||
"file_extensions": [".py"],
|
||||
"check_secrets": False,
|
||||
"check_sql": False,
|
||||
"check_dangerous_functions": True
|
||||
"check_dangerous_functions": True,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -254,8 +263,8 @@ def process_input(user_input):
|
||||
assert len(dangerous_findings) > 0
|
||||
assert any("eval" in f.title.lower() for f in dangerous_findings)
|
||||
|
||||
async def test_detect_exec(self, security_analyzer, temp_workspace):
|
||||
"""Test detection of exec() usage"""
|
||||
async def test_detect_exec(self, security_analyzer: SecurityAnalyzer, temp_workspace: Path) -> None:
|
||||
"""Test detection of exec() usage."""
|
||||
code_file = temp_workspace / "runner.py"
|
||||
code_file.write_text("""
|
||||
def run_code(code):
|
||||
@@ -264,7 +273,7 @@ def run_code(code):
|
||||
|
||||
config = {
|
||||
"file_extensions": [".py"],
|
||||
"check_dangerous_functions": True
|
||||
"check_dangerous_functions": True,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -273,8 +282,8 @@ def run_code(code):
|
||||
dangerous_findings = [f for f in result.findings if f.category == "dangerous_function"]
|
||||
assert len(dangerous_findings) > 0
|
||||
|
||||
async def test_detect_os_system(self, security_analyzer, temp_workspace):
|
||||
"""Test detection of os.system() usage"""
|
||||
async def test_detect_os_system(self, security_analyzer: SecurityAnalyzer, temp_workspace: Path) -> None:
|
||||
"""Test detection of os.system() usage."""
|
||||
code_file = temp_workspace / "commands.py"
|
||||
code_file.write_text("""
|
||||
import os
|
||||
@@ -285,7 +294,7 @@ def run_command(cmd):
|
||||
|
||||
config = {
|
||||
"file_extensions": [".py"],
|
||||
"check_dangerous_functions": True
|
||||
"check_dangerous_functions": True,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -295,8 +304,8 @@ def run_command(cmd):
|
||||
assert len(dangerous_findings) > 0
|
||||
assert any("os.system" in f.title for f in dangerous_findings)
|
||||
|
||||
async def test_detect_pickle_loads(self, security_analyzer, temp_workspace):
|
||||
"""Test detection of pickle.loads() usage"""
|
||||
async def test_detect_pickle_loads(self, security_analyzer: SecurityAnalyzer, temp_workspace: Path) -> None:
|
||||
"""Test detection of pickle.loads() usage."""
|
||||
code_file = temp_workspace / "serializer.py"
|
||||
code_file.write_text("""
|
||||
import pickle
|
||||
@@ -307,7 +316,7 @@ def deserialize(data):
|
||||
|
||||
config = {
|
||||
"file_extensions": [".py"],
|
||||
"check_dangerous_functions": True
|
||||
"check_dangerous_functions": True,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -316,8 +325,8 @@ def deserialize(data):
|
||||
dangerous_findings = [f for f in result.findings if f.category == "dangerous_function"]
|
||||
assert len(dangerous_findings) > 0
|
||||
|
||||
async def test_detect_javascript_eval(self, security_analyzer, temp_workspace):
|
||||
"""Test detection of eval() in JavaScript"""
|
||||
async def test_detect_javascript_eval(self, security_analyzer: SecurityAnalyzer, temp_workspace: Path) -> None:
|
||||
"""Test detection of eval() in JavaScript."""
|
||||
code_file = temp_workspace / "app.js"
|
||||
code_file.write_text("""
|
||||
function processInput(userInput) {
|
||||
@@ -327,7 +336,7 @@ function processInput(userInput) {
|
||||
|
||||
config = {
|
||||
"file_extensions": [".js"],
|
||||
"check_dangerous_functions": True
|
||||
"check_dangerous_functions": True,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -336,8 +345,8 @@ function processInput(userInput) {
|
||||
dangerous_findings = [f for f in result.findings if f.category == "dangerous_function"]
|
||||
assert len(dangerous_findings) > 0
|
||||
|
||||
async def test_detect_innerHTML(self, security_analyzer, temp_workspace):
|
||||
"""Test detection of innerHTML (XSS risk)"""
|
||||
async def test_detect_inner_html(self, security_analyzer: SecurityAnalyzer, temp_workspace: Path) -> None:
|
||||
"""Test detection of innerHTML (XSS risk)."""
|
||||
code_file = temp_workspace / "dom.js"
|
||||
code_file.write_text("""
|
||||
function updateContent(html) {
|
||||
@@ -347,7 +356,7 @@ function updateContent(html) {
|
||||
|
||||
config = {
|
||||
"file_extensions": [".js"],
|
||||
"check_dangerous_functions": True
|
||||
"check_dangerous_functions": True,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -356,14 +365,18 @@ function updateContent(html) {
|
||||
dangerous_findings = [f for f in result.findings if f.category == "dangerous_function"]
|
||||
assert len(dangerous_findings) > 0
|
||||
|
||||
async def test_no_dangerous_detection_when_disabled(self, security_analyzer, temp_workspace):
|
||||
"""Test that dangerous function detection can be disabled"""
|
||||
async def test_no_dangerous_detection_when_disabled(
|
||||
self,
|
||||
security_analyzer: SecurityAnalyzer,
|
||||
temp_workspace: Path,
|
||||
) -> None:
|
||||
"""Test that dangerous function detection can be disabled."""
|
||||
code_file = temp_workspace / "code.py"
|
||||
code_file.write_text('result = eval(user_input)')
|
||||
code_file.write_text("result = eval(user_input)")
|
||||
|
||||
config = {
|
||||
"file_extensions": [".py"],
|
||||
"check_dangerous_functions": False
|
||||
"check_dangerous_functions": False,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -375,10 +388,14 @@ function updateContent(html) {
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestSecurityAnalyzerMultipleIssues:
|
||||
"""Test detection of multiple issues in same file"""
|
||||
"""Test detection of multiple issues in same file."""
|
||||
|
||||
async def test_detect_multiple_vulnerabilities(self, security_analyzer, temp_workspace):
|
||||
"""Test detection of multiple vulnerability types"""
|
||||
async def test_detect_multiple_vulnerabilities(
|
||||
self,
|
||||
security_analyzer: SecurityAnalyzer,
|
||||
temp_workspace: Path,
|
||||
) -> None:
|
||||
"""Test detection of multiple vulnerability types."""
|
||||
code_file = temp_workspace / "vulnerable.py"
|
||||
code_file.write_text("""
|
||||
import os
|
||||
@@ -404,7 +421,7 @@ def process_query(user_input):
|
||||
"file_extensions": [".py"],
|
||||
"check_secrets": True,
|
||||
"check_sql": True,
|
||||
"check_dangerous_functions": True
|
||||
"check_dangerous_functions": True,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -423,10 +440,10 @@ def process_query(user_input):
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestSecurityAnalyzerSummary:
|
||||
"""Test result summary generation"""
|
||||
"""Test result summary generation."""
|
||||
|
||||
async def test_summary_structure(self, security_analyzer, temp_workspace):
|
||||
"""Test that summary has correct structure"""
|
||||
async def test_summary_structure(self, security_analyzer: SecurityAnalyzer, temp_workspace: Path) -> None:
|
||||
"""Test that summary has correct structure."""
|
||||
(temp_workspace / "test.py").write_text("print('hello')")
|
||||
|
||||
config = {"file_extensions": [".py"]}
|
||||
@@ -441,16 +458,16 @@ class TestSecurityAnalyzerSummary:
|
||||
assert isinstance(result.summary["total_findings"], int)
|
||||
assert isinstance(result.summary["extensions_scanned"], list)
|
||||
|
||||
async def test_empty_workspace(self, security_analyzer, temp_workspace):
|
||||
"""Test analyzing empty workspace"""
|
||||
async def test_empty_workspace(self, security_analyzer: SecurityAnalyzer, temp_workspace: Path) -> None:
|
||||
"""Test analyzing empty workspace."""
|
||||
config = {"file_extensions": [".py"]}
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
|
||||
assert result.status == "partial" # No files found
|
||||
assert result.summary["files_analyzed"] == 0
|
||||
|
||||
async def test_analyze_multiple_file_types(self, security_analyzer, temp_workspace):
|
||||
"""Test analyzing multiple file types"""
|
||||
async def test_analyze_multiple_file_types(self, security_analyzer: SecurityAnalyzer, temp_workspace: Path) -> None:
|
||||
"""Test analyzing multiple file types."""
|
||||
(temp_workspace / "app.py").write_text("print('hello')")
|
||||
(temp_workspace / "script.js").write_text("console.log('hello')")
|
||||
(temp_workspace / "index.php").write_text("<?php echo 'hello'; ?>")
|
||||
@@ -464,10 +481,10 @@ class TestSecurityAnalyzerSummary:
|
||||
|
||||
@pytest.mark.asyncio
|
||||
class TestSecurityAnalyzerFalsePositives:
|
||||
"""Test false positive filtering"""
|
||||
"""Test false positive filtering."""
|
||||
|
||||
async def test_skip_test_secrets(self, security_analyzer, temp_workspace):
|
||||
"""Test that test/example secrets are filtered"""
|
||||
async def test_skip_test_secrets(self, security_analyzer: SecurityAnalyzer, temp_workspace: Path) -> None:
|
||||
"""Test that test/example secrets are filtered."""
|
||||
code_file = temp_workspace / "test_config.py"
|
||||
code_file.write_text("""
|
||||
# Test configuration - should be filtered
|
||||
@@ -478,7 +495,7 @@ token = "sample_token_placeholder"
|
||||
|
||||
config = {
|
||||
"file_extensions": [".py"],
|
||||
"check_secrets": True
|
||||
"check_secrets": True,
|
||||
}
|
||||
|
||||
result = await security_analyzer.execute(config, temp_workspace)
|
||||
@@ -488,6 +505,6 @@ token = "sample_token_placeholder"
|
||||
secret_findings = [f for f in result.findings if f.category == "hardcoded_secret"]
|
||||
# Should have fewer or no findings due to false positive filtering
|
||||
assert len(secret_findings) == 0 or all(
|
||||
not any(fp in f.description.lower() for fp in ['test', 'example', 'dummy', 'sample'])
|
||||
not any(fp in f.description.lower() for fp in ["test", "example", "dummy", "sample"])
|
||||
for f in secret_findings
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user