"""
Test write operations for the Wealthfolio MCP server.
WARNING: These tests will modify the database. Use a test database copy!
"""
import os
import sqlite3
import uuid
from datetime import datetime
# Set the database path
DB_PATH = r'C:\Users\miche\AppData\Roaming\com.teymz.wealthfolio\app.db'
def test_create_asset():
"""Test creating a new asset."""
print("\n=== Test: Create Asset ===")
conn = sqlite3.connect(DB_PATH)
c = conn.cursor()
# Generate unique symbol to avoid conflicts
test_symbol = f"TEST{uuid.uuid4().hex[:6].upper()}"
try:
# Insert test asset
c.execute("""
INSERT INTO assets (
id, symbol, name, asset_type, currency, data_source, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?)
""", (
test_symbol,
test_symbol,
"Test Asset Inc.",
"Stock",
"USD",
"MANUAL",
datetime.now().isoformat(),
datetime.now().isoformat()
))
conn.commit()
# Verify it was created
c.execute("SELECT * FROM assets WHERE id = ?", (test_symbol,))
result = c.fetchone()
if result:
print(f"[PASS] Asset created: {test_symbol}")
# Cleanup: delete the test asset
c.execute("DELETE FROM assets WHERE id = ?", (test_symbol,))
conn.commit()
print(f"[CLEANUP] Test asset deleted")
return True
else:
print(f"[FAIL] Asset not found after creation")
return False
except Exception as e:
print(f"[FAIL] Error: {e}")
conn.rollback()
return False
finally:
conn.close()
def test_create_activity():
"""Test creating a new activity."""
print("\n=== Test: Create Activity ===")
conn = sqlite3.connect(DB_PATH)
conn.row_factory = sqlite3.Row
c = conn.cursor()
try:
# Get first account
c.execute("SELECT id FROM accounts LIMIT 1")
account = c.fetchone()
if not account:
print("[SKIP] No accounts found in database")
return None
account_id = account['id']
# Get first asset
c.execute("SELECT id FROM assets LIMIT 1")
asset = c.fetchone()
if not asset:
print("[SKIP] No assets found in database")
return None
asset_id = asset['id']
# Create test activity
activity_id = str(uuid.uuid4())
activity_date = "2025-01-01"
quantity = "1"
unit_price = "100.00"
currency = "EUR"
c.execute("""
INSERT INTO activities (
id, account_id, asset_id, activity_type, activity_date,
quantity, unit_price, currency, fee, amount, is_draft, comment, created_at, updated_at
) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", (
activity_id, account_id, asset_id, "BUY", activity_date,
quantity, unit_price, currency, "0", "100.00", 0, "Test activity",
datetime.now().isoformat(), datetime.now().isoformat()
))
conn.commit()
# Verify it was created
c.execute("SELECT * FROM activities WHERE id = ?", (activity_id,))
result = c.fetchone()
if result:
print(f"[PASS] Activity created: {activity_id}")
print(f" Account: {account_id}")
print(f" Asset: {asset_id}")
print(f" Type: BUY, Quantity: {quantity}, Price: {unit_price}")
# Cleanup: delete the test activity
c.execute("DELETE FROM activities WHERE id = ?", (activity_id,))
conn.commit()
print(f"[CLEANUP] Test activity deleted")
return True
else:
print(f"[FAIL] Activity not found after creation")
return False
except Exception as e:
print(f"[FAIL] Error: {e}")
conn.rollback()
return False
finally:
conn.close()
def test_validation():
"""Test validation rules."""
print("\n=== Test: Validation ===")
# Test invalid date format
try:
datetime.strptime("2025-13-01", "%Y-%m-%d")
print("[FAIL] Invalid date should have been rejected")
except ValueError:
print("[PASS] Invalid date format rejected")
# Test valid date format
try:
datetime.strptime("2025-01-27", "%Y-%m-%d")
print("[PASS] Valid date format accepted")
except ValueError:
print("[FAIL] Valid date should have been accepted")
def show_write_warning():
"""Display warning about write operations."""
print("=" * 60)
print("WARNING: WRITE OPERATIONS TEST")
print("=" * 60)
print("These tests will temporarily modify your Wealthfolio database.")
print("All changes will be cleaned up after testing.")
print("It's recommended to backup your database before running these tests.")
print("=" * 60)
print()
response = input("Do you want to continue? (yes/no): ")
return response.lower() in ['yes', 'y']
if __name__ == "__main__":
print("Wealthfolio MCP Server - Write Operations Tests\n")
if not show_write_warning():
print("\nTests cancelled by user.")
exit(0)
try:
# Run tests
test_validation()
asset_result = test_create_asset()
activity_result = test_create_activity()
print("\n" + "=" * 60)
print("SUMMARY")
print("=" * 60)
if asset_result and activity_result:
print("[SUCCESS] All write operation tests passed!")
elif asset_result is None or activity_result is None:
print("[PARTIAL] Some tests were skipped due to missing data")
else:
print("[FAILURE] Some tests failed")
except Exception as e:
print(f"\n[ERROR] Test suite failed: {e}")
import traceback
traceback.print_exc()
