#!/usr/bin/env python3
"""
FAF File Tools - Comprehensive Test Suite
Tests all Python functionality with pytest
"""
import pytest
import asyncio
import json
import os
import tempfile
import shutil
from pathlib import Path
from datetime import datetime
import pandas as pd
import numpy as np
from unittest.mock import Mock, patch, AsyncMock
import sys
# Add parent directory to path for imports
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
# Import modules to test
from faf_file_tools import FAFPythonBridge, FileOperation
from faf_data_analyzer import FAFDataAnalyzer
class TestFAFPythonBridge:
"""Test suite for FAF Python Bridge"""
@pytest.fixture
def temp_dir(self):
"""Create temporary directory for testing"""
temp_path = tempfile.mkdtemp(prefix="faf_test_")
yield Path(temp_path)
shutil.rmtree(temp_path)
@pytest.fixture
def bridge(self, temp_dir):
"""Create FAF bridge instance"""
return FAFPythonBridge(base_path=str(temp_dir))
def test_initialization(self, bridge):
"""Test bridge initialization"""
assert bridge is not None
assert bridge.stats["files_read"] == 0
assert bridge.stats["files_written"] == 0
assert bridge.stats["bytes_processed"] == 0
assert bridge.stats["errors"] == 0
def test_read_json_config(self, bridge, temp_dir):
"""Test reading JSON configuration"""
# Create test JSON file
test_data = {"version": "2.0.0", "status": "active"}
test_file = temp_dir / "test_config.json"
with open(test_file, 'w') as f:
json.dump(test_data, f)
# Read using bridge
result = bridge.read_json_config("test_config.json")
assert result == test_data
assert bridge.stats["files_read"] == 1
assert bridge.stats["bytes_processed"] > 0
def test_write_python_report(self, bridge, temp_dir):
"""Test writing Python report"""
report_data = {
"title": "Test Report",
"version": "2.0.0",
"metrics": {"speed": 100, "accuracy": 99.9}
}
result = bridge.write_python_report(report_data, "test_report.py")
assert "Successfully wrote" in result
assert bridge.stats["files_written"] == 1
# Verify file exists and is valid Python
report_file = temp_dir / "test_report.py"
assert report_file.exists()
# Try to compile the Python file
with open(report_file, 'r') as f:
code = f.read()
compile(code, str(report_file), 'exec')
def test_generate_test_suite(self, bridge, temp_dir):
"""Test generating test suite"""
test_file = bridge.generate_test_suite()
assert os.path.exists(test_file)
assert bridge.stats["files_written"] == 1
# Verify it's valid Python
with open(test_file, 'r') as f:
code = f.read()
compile(code, test_file, 'exec')
def test_calculate_file_hash(self, bridge, temp_dir):
"""Test file hash calculation"""
test_file = temp_dir / "hash_test.txt"
test_content = "Hello FAF!"
test_file.write_text(test_content)
hash_result = bridge.calculate_file_hash("hash_test.txt")
assert len(hash_result) == 64 # SHA-256 hash length
assert all(c in '0123456789abcdef' for c in hash_result)
def test_get_statistics(self, bridge):
"""Test statistics retrieval"""
stats = bridge.get_statistics()
assert "stats" in stats
assert "operations_count" in stats
assert "total_bytes" in stats
assert "success_rate" in stats
assert stats["success_rate"] == 0 # No operations yet
def test_operation_logging(self, bridge):
"""Test operation logging"""
bridge._log_operation("read", "/test/file.txt", 1024, 35.5, True)
assert len(bridge.operations) == 1
op = bridge.operations[0]
assert op.operation == "read"
assert op.path == "/test/file.txt"
assert op.size_bytes == 1024
assert op.duration_ms == 35.5
assert op.success is True
def test_error_handling(self, bridge):
"""Test error handling for non-existent file"""
with pytest.raises(Exception) as exc_info:
bridge.read_json_config("non_existent.json")
assert "Failed to read" in str(exc_info.value)
assert bridge.stats["errors"] == 1
class TestFAFDataAnalyzer:
"""Test suite for FAF Data Analyzer"""
@pytest.fixture
def analyzer(self):
"""Create analyzer instance"""
return FAFDataAnalyzer()
def test_generate_sample_data(self, analyzer):
"""Test sample data generation"""
df = analyzer.generate_sample_data(50)
assert len(df) == 50
assert 'timestamp' in df.columns
assert 'operation' in df.columns
assert 'file_type' in df.columns
assert 'duration_ms' in df.columns
assert 'size_bytes' in df.columns
assert 'success' in df.columns
assert 'path' in df.columns
# Check data types
assert pd.api.types.is_datetime64_any_dtype(df['timestamp'])
assert pd.api.types.is_numeric_dtype(df['duration_ms'])
assert pd.api.types.is_numeric_dtype(df['size_bytes'])
assert pd.api.types.is_bool_dtype(df['success'])
def test_analyze_performance(self, analyzer):
"""Test performance analysis"""
df = analyzer.generate_sample_data(100)
analysis = analyzer.analyze_performance(df)
assert 'summary' in analysis
assert 'performance' in analysis
assert 'by_operation' in analysis
assert 'by_file_type' in analysis
# Check summary metrics
summary = analysis['summary']
assert summary['total_operations'] == 100
assert 0 <= summary['success_rate'] <= 100
# Check performance metrics
perf = analysis['performance']
assert 'avg_duration_ms' in perf
assert 'median_duration_ms' in perf
assert 'p95_duration_ms' in perf
assert 'p99_duration_ms' in perf
assert perf['avg_duration_ms'] > 0
def test_generate_performance_report(self, analyzer):
"""Test report generation"""
df = analyzer.generate_sample_data(50)
analysis = analyzer.analyze_performance(df)
report = analyzer.generate_performance_report(analysis)
assert isinstance(report, str)
assert "FAF FILE TOOLS - PERFORMANCE ANALYSIS REPORT" in report
assert "SUMMARY" in report
assert "PERFORMANCE METRICS" in report
assert "Total Operations:" in report
def test_export_metrics_csv(self, analyzer, tmp_path):
"""Test CSV export"""
df = analyzer.generate_sample_data(25)
csv_path = tmp_path / "test_metrics.csv"
result = analyzer.export_metrics_csv(df, str(csv_path))
assert f"Exported 25 records" in result
assert csv_path.exists()
# Read back and verify
df_read = pd.read_csv(csv_path)
assert len(df_read) == 25
assert list(df_read.columns) == list(df.columns)
def test_create_benchmark_suite(self, analyzer):
"""Test benchmark suite creation"""
benchmark_code = analyzer.create_benchmark_suite()
assert isinstance(benchmark_code, str)
assert "FAF File Tools - Benchmark Suite" in benchmark_code
assert "class FAFBenchmark:" in benchmark_code
assert "def benchmark_read_operations" in benchmark_code
# Verify it's valid Python
compile(benchmark_code, '<string>', 'exec')
class TestFileOperation:
"""Test FileOperation dataclass"""
def test_file_operation_creation(self):
"""Test creating FileOperation instance"""
op = FileOperation(
operation="read",
path="/test/file.txt",
size_bytes=1024,
timestamp="2025-09-15T08:00:00",
duration_ms=35.5,
success=True
)
assert op.operation == "read"
assert op.path == "/test/file.txt"
assert op.size_bytes == 1024
assert op.duration_ms == 35.5
assert op.success is True
def test_file_operation_to_dict(self):
"""Test converting FileOperation to dictionary"""
op = FileOperation(
operation="write",
path="/output/report.md",
size_bytes=2048,
timestamp="2025-09-15T08:00:00",
duration_ms=67.3,
success=True
)
result = op.to_dict()
assert isinstance(result, dict)
assert result["operation"] == "write"
assert result["path"] == "/output/report.md"
assert result["size_bytes"] == 2048
assert result["duration_ms"] == 67.3
assert result["success"] is True
class TestIntegration:
"""Integration tests for FAF File Tools"""
@pytest.fixture
def setup_env(self):
"""Setup test environment"""
temp_dir = tempfile.mkdtemp(prefix="faf_integration_")
bridge = FAFPythonBridge(base_path=temp_dir)
analyzer = FAFDataAnalyzer()
yield {
"temp_dir": Path(temp_dir),
"bridge": bridge,
"analyzer": analyzer
}
shutil.rmtree(temp_dir)
def test_full_workflow(self, setup_env):
"""Test complete workflow: generate, analyze, report"""
bridge = setup_env["bridge"]
analyzer = setup_env["analyzer"]
temp_dir = setup_env["temp_dir"]
# Step 1: Generate sample data
df = analyzer.generate_sample_data(100)
# Step 2: Export to CSV
csv_path = temp_dir / "metrics.csv"
analyzer.export_metrics_csv(df, str(csv_path))
# Step 3: Read CSV back
assert csv_path.exists()
df_read = pd.read_csv(csv_path)
assert len(df_read) == 100
# Step 4: Analyze performance
analysis = analyzer.analyze_performance(df_read)
# Step 5: Generate report
report = analyzer.generate_performance_report(analysis)
# Step 6: Write report using bridge
result = bridge.write_python_report(
{"analysis": "complete", "records": len(df)},
"final_report.py"
)
assert "Successfully wrote" in result
assert (temp_dir / "final_report.py").exists()
# Verify statistics
stats = bridge.get_statistics()
assert stats["stats"]["files_written"] == 1
assert stats["success_rate"] == 100
def test_error_recovery(self, setup_env):
"""Test error handling and recovery"""
bridge = setup_env["bridge"]
# Try to read non-existent file
with pytest.raises(Exception):
bridge.read_json_config("does_not_exist.json")
# Verify error was logged
assert bridge.stats["errors"] == 1
# Verify bridge still works after error
test_data = {"test": "data"}
result = bridge.write_python_report(test_data, "recovery_test.py")
assert "Successfully wrote" in result
class TestPerformance:
"""Performance tests for FAF File Tools"""
@pytest.fixture
def large_dataset(self):
"""Generate large dataset for performance testing"""
return pd.DataFrame({
'id': range(10000),
'value': np.random.randn(10000),
'category': np.random.choice(['A', 'B', 'C', 'D'], 10000),
'timestamp': pd.date_range('2025-01-01', periods=10000, freq='min')
})
def test_large_file_performance(self, tmp_path, large_dataset):
"""Test performance with large files"""
import time
# Write large CSV
csv_path = tmp_path / "large_file.csv"
start = time.time()
large_dataset.to_csv(csv_path, index=False)
write_duration = time.time() - start
# Should complete within 5 seconds
assert write_duration < 5.0
# Read large CSV
start = time.time()
df_read = pd.read_csv(csv_path)
read_duration = time.time() - start
# Should complete within 2 seconds
assert read_duration < 2.0
assert len(df_read) == 10000
def test_multiple_operations_performance(self, tmp_path):
"""Test performance of multiple operations"""
import time
bridge = FAFPythonBridge(base_path=str(tmp_path))
start = time.time()
# Perform 100 operations
for i in range(100):
data = {"iteration": i, "timestamp": datetime.now().isoformat()}
bridge.write_python_report(data, f"report_{i}.py")
duration = time.time() - start
# Should complete 100 operations within 10 seconds
assert duration < 10.0
assert bridge.stats["files_written"] == 100
# Calculate average operation time
avg_time = duration / 100 * 1000 # Convert to ms
assert avg_time < 100 # Should average under 100ms per operation
# Run tests with pytest
if __name__ == "__main__":
pytest.main([__file__, "-v", "--tb=short"])
