ach_ui_dbtl_file_based/LOCAL_TESTING.md

Local Testing Without Docker

This guide shows how to test the ACH processing system locally without Docker or SFTP server.

Option 1: Direct File Testing (Simplest)

This approach tests the core processing logic by using local files directly.

1. Setup Test Files

# Create local test directories
mkdir -p test_files/HDFC/NACH
mkdir -p test_files/ICICI/NACH

# Copy sample ACH file
cp ACH_99944_19012026103217_001.txt test_files/HDFC/NACH/
cp ACH_99944_19012026103217_001.txt test_files/ICICI/NACH/ACH_12345_05122025102947_001.txt

2. Create Local Testing Script

Create test_local.py:

cat > test_local.py << 'EOF'
#!/usr/bin/env python3
"""
Local testing script - test core processing without SFTP/Database.
"""

import sys
import os
from pathlib import Path

# Test data mapper
print("\n" + "="*80)
print("TEST 1: Data Mapper")
print("="*80)

from processors.data_mapper import DataMapper
from datetime import date
from decimal import Decimal

# Test date conversion
d = DataMapper.convert_date('19/01/26')
assert d == date(2026, 1, 19), f"Expected 2026-01-19, got {d}"
print("✓ Date conversion: '19/01/26' → 2026-01-19")

# Test TXNIND
assert DataMapper.calculate_txnind('100.50') == 'CR'
assert DataMapper.calculate_txnind('-50.00') == 'DR'
print("✓ TXNIND calculation: 100.50 → CR, -50.00 → DR")

# Test amount
amt = DataMapper.convert_amount('-100.50')
assert amt == Decimal('100.50')
print("✓ Amount conversion: -100.50 → 100.50 (absolute)")

# Test transaction mapping
from ach_parser import ACHParser
parser = ACHParser('ACH_99944_19012026103217_001.txt')
transactions, metadata, summary = parser.parse()
print(f"✓ ACH Parser: Extracted {len(transactions)} transactions")

mapped = DataMapper.map_transaction(transactions[0], 'HDFC')
print(f"✓ Transaction mapping: Single transaction mapped to DB format")

all_mapped = DataMapper.map_transactions(transactions, 'HDFC')
print(f"✓ Batch mapping: {len(all_mapped)} transactions mapped")

# Test file monitor
print("\n" + "="*80)
print("TEST 2: File Monitor")
print("="*80)

from sftp.file_monitor import FileMonitor

# Test filename parsing
filename = 'ACH_99944_05122025102947_001.txt'
parsed = FileMonitor.parse_filename(filename)
assert parsed['branch'] == '99944'
assert parsed['day'] == '05'
assert parsed['month'] == '12'
assert parsed['year'] == '2025'
print(f"✓ Filename parsing: {filename}")
print(f"  Branch: {parsed['branch']}")
print(f"  Timestamp: {parsed['timestamp']}")

# Test filename validation
invalid = 'invalid_file.txt'
parsed = FileMonitor.parse_filename(invalid)
assert parsed == {}
print(f"✓ Invalid filename rejected: {invalid}")

# Test local file discovery
print("\n" + "="*80)
print("TEST 3: Local File Discovery")
print("="*80)

# Find ACH files locally
test_dir = Path('test_files')
if test_dir.exists():
    ach_files = list(test_dir.glob('**/ACH_*.txt'))
    print(f"✓ Found {len(ach_files)} test ACH files locally:")
    for f in ach_files:
        print(f"  - {f.relative_to(test_dir)}")

# Test configuration
print("\n" + "="*80)
print("TEST 4: Configuration")
print("="*80)

from config import get_config
cfg = get_config()
print(f"✓ Bank codes: {cfg.bank_codes}")
print(f"✓ Poll interval: {cfg.poll_interval_minutes} minutes")
print(f"✓ Batch size: {cfg.batch_size}")

# Summary
print("\n" + "="*80)
print("ALL TESTS PASSED ✓")
print("="*80)
print("\nCore processing logic is working correctly.")
print("Ready for database and SFTP integration testing.")
print("\nNext steps:")
print("1. Install Oracle Instant Client (for DB testing)")
print("2. Create database tables")
print("3. Configure .env with actual credentials")
print("4. Test with actual SFTP server")
print("5. Deploy to production")
EOF

python test_local.py

3. Run the Test

python test_local.py

Expected output:

================================================================================
TEST 1: Data Mapper
================================================================================
✓ Date conversion: '19/01/26' → 2026-01-19
✓ TXNIND calculation: 100.50 → CR, -50.00 → DR
✓ Amount conversion: -100.50 → 100.50 (absolute)
✓ ACH Parser: Extracted 178 transactions
✓ Transaction mapping: Single transaction mapped to DB format
✓ Batch mapping: 178 transactions mapped

================================================================================
TEST 2: File Monitor
================================================================================
✓ Filename parsing: ACH_99944_05122025102947_001.txt
  Branch: 99944
  Timestamp: 05/12/2025 10:29:47
✓ Invalid filename rejected: invalid_file.txt

================================================================================
TEST 3: Local File Discovery
================================================================================
✓ Found 2 test ACH files locally:
  - HDFC/NACH/ACH_99944_19012026103217_001.txt
  - ICICI/NACH/ACH_12345_05122025102947_001.txt

================================================================================
TEST 4: Configuration
================================================================================
✓ Bank codes: ['HDFC', 'ICICI', 'SBI', 'AXIS', 'PNB']
✓ Poll interval: 1 minutes
✓ Batch size: 100

================================================================================
ALL TESTS PASSED ✓
================================================================================

---

Option 2: Python Mock SFTP Server (Local)

If you want to test SFTP locally without Docker, use the included mock SFTP server.

1. Start Mock SFTP Server

# Start the server in one terminal
python tests/mock_sftp_server.py

Expected output:

================================================================================
Mock SFTP Server for Testing
================================================================================
✓ Created ./sftp_data/HDFC/NACH
✓ Created ./sftp_data/ICICI/NACH
✓ Created ./sftp_data/SBI/NACH

Starting mock SFTP server...
[INFO] Mock SFTP server listening on 127.0.0.1:2222
[INFO] SFTP root: /home/asif/projects/ach_ui_dbtl_file_based/sftp_data
[INFO] Username: ipks, Password: ipks_password

================================================================================
Server running. Press CTRL+C to stop.

To test connection:
  sftp -P 2222 ipks@127.0.0.1
  Password: ipks_password

To use with application:
  SFTP_HOST=127.0.0.1
  SFTP_PORT=2222
  SFTP_USERNAME=ipks
  SFTP_PASSWORD=ipks_password
================================================================================

2. Test SFTP Connection (in another terminal)

# Test connection
sftp -P 2222 ipks@127.0.0.1
# Password: ipks_password

# Commands to try:
sftp> ls
sftp> cd HDFC/NACH
sftp> ls
sftp> put ACH_99944_19012026103217_001.txt
sftp> quit

3. Configure for Testing

Edit .env:

SFTP_HOST=127.0.0.1
SFTP_PORT=2222
SFTP_USERNAME=ipks
SFTP_PASSWORD=ipks_password
SFTP_BASE_PATH=/home/ipks/IPKS_FILES/REPORTS
POLL_INTERVAL_MINUTES=1
BANK_CODES=HDFC,ICICI,SBI

4. Copy Test Files to Mock SFTP

mkdir -p sftp_data/HDFC/NACH
cp ACH_99944_19012026103217_001.txt sftp_data/HDFC/NACH/

5. Run Application

In another terminal:

source venv/bin/activate
python main.py

Note: This will try to connect to the database. Without a real database, it will fail, but you can see SFTP operations working.

---

Option 3: Unit Tests Only

Test without SFTP or Database - just the logic.

# Run unit tests
pytest tests/ -v

# Output:
# tests/test_data_mapper.py::TestDataMapper::test_convert_date_valid PASSED
# tests/test_data_mapper.py::TestDataMapper::test_calculate_txnind_credit PASSED
# tests/test_data_mapper.py::TestDataMapper::test_convert_amount PASSED
# tests/test_data_mapper.py::TestDataMapper::test_map_transaction PASSED
# tests/test_file_monitor.py::TestFileMonitor::test_parse_filename_valid PASSED
# ...

---

Option 4: Database-Only Testing (Local SQLite for testing)

Test database logic without Oracle. Use SQLite for testing first.

1. Create Test Database Module

Create tests/test_with_sqlite.py:

cat > tests/test_with_sqlite.py << 'EOF'
#!/usr/bin/env python3
"""
Test database operations with SQLite (no Oracle required).
"""

import sqlite3
import tempfile
from pathlib import Path
from datetime import datetime
from decimal import Decimal

print("\n" + "="*80)
print("SQLite Database Testing")
print("="*80)

# Create temporary database
temp_db = tempfile.mktemp(suffix='.db')
conn = sqlite3.connect(temp_db)
cursor = conn.cursor()

print(f"✓ Created test database: {temp_db}")

# Create test tables
cursor.execute("""
    CREATE TABLE ach_api_log (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        narration TEXT,
        status TEXT,
        bankcode TEXT,
        jrnl_id TEXT,
        tran_date DATE,
        cbs_acct TEXT,
        tran_amt DECIMAL(15, 2),
        TXNIND TEXT,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
""")

cursor.execute("""
    CREATE TABLE ach_processed_files (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        filename TEXT UNIQUE NOT NULL,
        bankcode TEXT,
        file_path TEXT,
        processed_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
        transaction_count INTEGER,
        status TEXT DEFAULT 'SUCCESS',
        error_message TEXT
    )
""")

conn.commit()
print("✓ Created tables: ach_api_log, ach_processed_files")

# Test data insertion
test_data = [
    ('Test Remark 1', '23-DEP-PROCESSED', 'HDFC', '001', '2026-01-19', '1001', 100.50, 'CR'),
    ('Test Remark 2', '23-DEP-PROCESSED', 'HDFC', '002', '2026-01-19', '1002', 50.00, 'CR'),
    ('Test Remark 3', '23-DEP-PROCESSED', 'ICICI', '003', '2026-01-20', '2001', 75.75, 'CR'),
]

insert_sql = """
    INSERT INTO ach_api_log (narration, status, bankcode, jrnl_id, tran_date, cbs_acct, tran_amt, TXNIND)
    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
"""

cursor.executemany(insert_sql, test_data)
conn.commit()
print(f"✓ Inserted {len(test_data)} test transactions")

# Query test
cursor.execute("SELECT COUNT(*) FROM ach_api_log")
count = cursor.fetchone()[0]
assert count == 3, f"Expected 3 records, got {count}"
print(f"✓ Query test: Found {count} transactions")

cursor.execute("SELECT * FROM ach_api_log WHERE bankcode = 'HDFC'")
hdfc_records = cursor.fetchall()
assert len(hdfc_records) == 2, f"Expected 2 HDFC records, got {len(hdfc_records)}"
print(f"✓ Bank filter: Found {len(hdfc_records)} HDFC transactions")

# Test processed files tracking
file_data = ('ACH_99944_19012026103217_001.txt', 'HDFC', '/path/to/file', 3, 'SUCCESS', None)
cursor.execute("""
    INSERT INTO ach_processed_files (filename, bankcode, file_path, transaction_count, status, error_message)
    VALUES (?, ?, ?, ?, ?, ?)
""", file_data)
conn.commit()
print("✓ File tracking: Marked file as processed")

# Test duplicate detection
cursor.execute("SELECT COUNT(*) FROM ach_processed_files WHERE filename = 'ACH_99944_19012026103217_001.txt'")
dup_count = cursor.fetchone()[0]
assert dup_count == 1, "Duplicate detection failed"
print("✓ Duplicate detection: Working correctly")

# Test transaction with rollback
cursor.execute("BEGIN TRANSACTION")
cursor.execute("""
    INSERT INTO ach_api_log (narration, status, bankcode, jrnl_id, tran_date, cbs_acct, tran_amt, TXNIND)
    VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", ('Rollback Test', '23-DEP-PROCESSED', 'SBI', '099', '2026-01-20', '9001', 999.99, 'CR'))
cursor.execute("ROLLBACK")
conn.commit()

cursor.execute("SELECT COUNT(*) FROM ach_api_log WHERE narration = 'Rollback Test'")
rb_count = cursor.fetchone()[0]
assert rb_count == 0, "Rollback did not work"
print("✓ Transaction safety: Rollback works correctly")

# Summary
print("\n" + "="*80)
print("DATABASE TESTS PASSED ✓")
print("="*80)
print("\nSQLite testing confirms:")
print("  ✓ Table structure works")
print("  ✓ Data insertion works")
print("  ✓ Queries work")
print("  ✓ Duplicate detection works")
print("  ✓ Transactions work")
print("\nReady for Oracle integration.")

# Cleanup
cursor.close()
conn.close()
Path(temp_db).unlink()
print(f"\n✓ Cleaned up test database")
EOF

python tests/test_with_sqlite.py

---

Testing Summary

Without Docker/SFTP/Database:

python test_local.py    # Tests data mapper, file monitor, config
pytest tests/ -v         # Unit tests

With Local Mock SFTP (Optional):

# Terminal 1
python tests/mock_sftp_server.py

# Terminal 2
python main.py  # Will test SFTP but fail on DB

With SQLite Database (Optional):

python tests/test_with_sqlite.py  # Tests database logic

---

What Gets Tested in Each Scenario

Scenario	Data Mapper	File Monitor	SFTP	Database	Full Pipeline
Option 1 (Local)	✓	✓	✗	✗	✗
Option 2 (SFTP)	✓	✓	✓	✗	✗
Option 3 (Unit)	✓	✓	✗	✗	✗
Option 4 (SQLite)	✓	✓	✗	✓	✗
Full (With Oracle)	✓	✓	✓	✓	✓

---

Recommended Testing Path

1. Start: python test_local.py (verify core logic)
2. Unit Tests: pytest tests/ -v (verify edge cases)
3. SFTP: python tests/mock_sftp_server.py (verify file operations)
4. Database: Setup Oracle & test with real database
5. Full Pipeline: Deploy and monitor in production

---

Troubleshooting

ImportError: No module named 'paramiko'

Mock SFTP server requires paramiko. Install it:

pip install paramiko cryptography

"Address already in use" on port 2222

Either:

• Change port in mock_sftp_server.py
• Kill previous server process
• Wait a minute for socket to reset

Test files not found

Make sure test_files directory exists:

mkdir -p test_files/HDFC/NACH test_files/ICICI/NACH
cp ACH_99944_19012026103217_001.txt test_files/HDFC/NACH/

Permission Denied errors

Ensure directory permissions are correct:

chmod -R 755 test_files/
chmod -R 755 sftp_data/

---

Next Steps After Testing

Once core logic is verified locally:

1. Install Oracle Instant Client
2. Create database tables
3. Update .env with real credentials
4. Test with actual SFTP server
5. Deploy to production

See SETUP.md for detailed Oracle setup instructions.