import os import shutil import tempfile import time from datetime import datetime, timedelta import pytest # from unittest.mock import patch, MagicMock from functools import wraps # Import the code to test from transaction_tracker import TransactionProcessor, TransactionAlreadyProcessedError class TestTransactionProcessor: """Test suite for TransactionProcessor class.""" @pytest.fixture def temp_dir(self): """Create a temporary directory for testing.""" temp_dir = tempfile.mkdtemp() yield temp_dir # Clean up after the test shutil.rmtree(temp_dir) @pytest.fixture def processor(self, temp_dir): """Create a TransactionProcessor instance for testing.""" return TransactionProcessor(storage_dir=temp_dir, ttl_days=3) def test_unique_transaction_decorator(self, processor): """Test the unique transaction decorator works.""" # Define a function with the decorator @processor.unique_transaction() def sample_transaction(transaction_id, amount): return {"status": "success", "amount": amount} # First call should succeed result = sample_transaction("test_tx_1", 100.00) assert result == {"status": "success", "amount": 100.00} # Second call with same ID should raise exception with pytest.raises(TransactionAlreadyProcessedError): sample_transaction("test_tx_1", 200.00) # Different ID should work result = sample_transaction("test_tx_2", 300.00) assert result == {"status": "success", "amount": 300.00} def test_custom_id_arg(self, processor): """Test decorator with custom transaction ID argument name.""" @processor.unique_transaction(id_arg='custom_id') def custom_transaction(custom_id, amount): return {"status": "success", "amount": amount, "id": custom_id} # First call should succeed result = custom_transaction("custom_1", 100.00) assert result == {"status": "success", "amount": 100.00, "id": "custom_1"} # Second call with same ID should raise exception with pytest.raises(TransactionAlreadyProcessedError): custom_transaction("custom_1", 200.00) def test_keyword_arguments(self, processor): """Test decorator with keyword arguments.""" @processor.unique_transaction() def kw_transaction(transaction_id, amount): return {"status": "success", "amount": amount} # Using keyword arguments result = kw_transaction(transaction_id="kw_1", amount=100.00) assert result == {"status": "success", "amount": 100.00} # Second call should fail with pytest.raises(TransactionAlreadyProcessedError): kw_transaction(transaction_id="kw_1", amount=200.00) def test_cleanup_expired(self, temp_dir): """Test cleanup of expired transactions.""" # Create processor with short TTL for testing processor = TransactionProcessor(storage_dir=temp_dir, ttl_days=0.01) # ~15 minutes # Define a test function @processor.unique_transaction() def expired_transaction(transaction_id, amount): return {"status": "success", "amount": amount} # Process a transaction expired_transaction("expire_1", 100.00) # Verify record exists record_path = os.path.join(temp_dir, "expire_1.json") assert os.path.exists(record_path) # Manually modify the expiration time to make it expire immediately import json with open(record_path, 'r') as f: record = json.load(f) # Set expires_at to 1 second ago record['expires_at'] = (datetime.now() - timedelta(seconds=1)).isoformat() with open(record_path, 'w') as f: json.dump(record, f) # Run cleanup processor.cleanup() # Verify record was removed assert not os.path.exists(record_path) def test_reset(self, processor): """Test resetting all transaction records.""" @processor.unique_transaction() def reset_transaction(transaction_id, amount): return {"status": "success", "amount": amount} # Process multiple transactions reset_transaction("reset_1", 100.00) reset_transaction("reset_2", 200.00) # Reset all records processor.reset() # Should be able to process the same transactions again result1 = reset_transaction("reset_1", 100.00) result2 = reset_transaction("reset_2", 200.00) assert result1 == {"status": "success", "amount": 100.00} assert result2 == {"status": "success", "amount": 200.00} def test_missing_transaction_id(self, processor): """Test error handling when transaction ID argument is missing.""" @processor.unique_transaction(id_arg='missing_id') def missing_transaction(transaction_id, amount): return {"status": "success", "amount": amount} # Should raise ValueError because 'missing_id' is not in the function signature with pytest.raises(ValueError): missing_transaction("tx_1", 100.00) def test_corrupt_file_handling(self, temp_dir, processor): """Test handling of corrupt transaction record files.""" # Manually create a corrupt record file corrupt_path = os.path.join(temp_dir, "corrupt_tx.json") with open(corrupt_path, 'w') as f: f.write("This is not valid JSON") # Cleanup should remove the corrupt file without errors processor.cleanup() # File should be gone assert not os.path.exists(corrupt_path) @pytest.mark.slow def test_real_ttl_expiration(self, temp_dir): """Test actual TTL expiration (takes longer to run).""" # Create processor with very short TTL processor = TransactionProcessor(storage_dir=temp_dir, ttl_days=0.0007) # ~1 minute @processor.unique_transaction() def quick_expire_transaction(transaction_id, amount): return {"status": "success", "amount": amount} # Process a transaction quick_expire_transaction("quick_expire", 100.00) # Verify we can't process it again immediately with pytest.raises(TransactionAlreadyProcessedError): quick_expire_transaction("quick_expire", 100.00) # Wait for TTL to expire (slightly over 1 minute) time.sleep(65) # Run cleanup processor.cleanup() # Should be able to process it again result = quick_expire_transaction("quick_expire", 100.00) assert result == {"status": "success", "amount": 100.00} def test_multiple_decorators(self, processor): """Test stacking multiple decorators.""" calls = [] def another_decorator(func): @wraps(func) def wrapper(*args, **kwargs): calls.append("another_decorator") return func(*args, **kwargs) return wrapper @another_decorator @processor.unique_transaction() def multi_decorated(transaction_id, amount): calls.append("original_function") return {"status": "success", "amount": amount} # Call the function result = multi_decorated("multi_1", 100.00) # Check result assert result == {"status": "success", "amount": 100.00} # Check call order assert calls == ["another_decorator", "original_function"] # Second call should still be caught by our transaction decorator with pytest.raises(TransactionAlreadyProcessedError): multi_decorated("multi_1", 100.00) def test_non_string_transaction_id(self, processor): """Test handling of non-string transaction IDs.""" @processor.unique_transaction() def int_id_transaction(transaction_id, amount): return {"status": "success", "amount": amount} # Use an integer as transaction_id result = int_id_transaction(12345, 100.00) assert result == {"status": "success", "amount": 100.00} # Try again with same ID with pytest.raises(TransactionAlreadyProcessedError): int_id_transaction(12345, 200.00) # The string representation should also be caught with pytest.raises(TransactionAlreadyProcessedError): int_id_transaction("12345", 300.00)