transaction-tracker/tests/test_transact_cache.py

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)