390 lines
15 KiB
Python
390 lines
15 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
PacaBotManager TEST Client - Safe Testing Version
|
||
=================================================
|
||
|
||
This is a SAFE TEST VERSION of the PacaBotManager client that:
|
||
1. Uses BSC testnet or fork for testing
|
||
2. Shows what operations would do without executing them
|
||
3. Provides dry-run functionality
|
||
4. Only executes writes when explicitly confirmed
|
||
|
||
Usage: python python_scripts/test_client.py
|
||
"""
|
||
|
||
import os
|
||
import sys
|
||
from web3 import Web3
|
||
from eth_account import Account
|
||
from dotenv import load_dotenv
|
||
import json
|
||
|
||
# Load environment variables
|
||
load_dotenv()
|
||
|
||
class PacaBotManagerTestClient:
|
||
def __init__(self, use_testnet=True):
|
||
if use_testnet:
|
||
# BSC Testnet for safe testing
|
||
self.rpc_url = "https://data-seed-prebsc-1-s1.binance.org:8545"
|
||
self.chain_name = "BSC Testnet"
|
||
self.chain_id = 97
|
||
|
||
# You'll need to deploy contracts on testnet for full testing
|
||
self.bot_manager_address = None # Deploy on testnet
|
||
self.paca_bsc_address = None # Deploy on testnet
|
||
else:
|
||
# Local fork for testing (safer than mainnet)
|
||
self.rpc_url = "http://127.0.0.1:8545" # Local hardhat fork
|
||
self.chain_name = "Local Fork"
|
||
self.chain_id = 31337
|
||
|
||
# Mainnet addresses (only for fork testing)
|
||
self.bot_manager_address = "0x4E5d3cD7743934b61041ba2ac3E9df39a0A26dcC"
|
||
self.paca_bsc_address = "0x3fF44D639a4982A4436f6d737430141aBE68b4E1"
|
||
|
||
self.w3 = Web3(Web3.HTTPProvider(self.rpc_url))
|
||
|
||
# Test mode settings
|
||
self.dry_run_mode = True # Default to dry run
|
||
self.require_confirmation = True
|
||
|
||
# Load private key
|
||
self.private_key = os.getenv('PRIVATE_KEY')
|
||
if not self.private_key:
|
||
print("❌ Error: PRIVATE_KEY not found in environment variables")
|
||
print("Please add PRIVATE_KEY=your_private_key_here to your .env file")
|
||
sys.exit(1)
|
||
|
||
# Set up account
|
||
self.account = Account.from_key(self.private_key)
|
||
|
||
try:
|
||
balance = self.w3.eth.get_balance(self.account.address)
|
||
print(f"🔑 Using account: {self.account.address}")
|
||
print(f"💰 Balance: {self.w3.from_wei(balance, 'ether')} ETH")
|
||
print(f"🌐 Network: {self.chain_name}")
|
||
print(f"🔒 Dry Run Mode: {self.dry_run_mode}")
|
||
except Exception as e:
|
||
print(f"❌ Connection failed: {e}")
|
||
print(f" Make sure the RPC endpoint is accessible: {self.rpc_url}")
|
||
sys.exit(1)
|
||
|
||
# Contract ABIs (same as main client)
|
||
self.bot_manager_abi = [
|
||
{
|
||
"inputs": [],
|
||
"stateMutability": "nonpayable",
|
||
"type": "constructor"
|
||
},
|
||
{
|
||
"inputs": [
|
||
{"internalType": "address", "name": "pacaContract", "type": "address"},
|
||
{"internalType": "address", "name": "user", "type": "address"}
|
||
],
|
||
"name": "clearStakes",
|
||
"outputs": [],
|
||
"stateMutability": "nonpayable",
|
||
"type": "function"
|
||
},
|
||
{
|
||
"inputs": [
|
||
{"internalType": "address", "name": "pacaContract", "type": "address"},
|
||
{"internalType": "address", "name": "user", "type": "address"}
|
||
],
|
||
"name": "clearVesting",
|
||
"outputs": [],
|
||
"stateMutability": "nonpayable",
|
||
"type": "function"
|
||
},
|
||
{
|
||
"inputs": [],
|
||
"name": "owner",
|
||
"outputs": [
|
||
{"internalType": "address", "name": "", "type": "address"}
|
||
],
|
||
"stateMutability": "view",
|
||
"type": "function"
|
||
}
|
||
]
|
||
|
||
self.paca_abi = [
|
||
{
|
||
"inputs": [
|
||
{"internalType": "address", "name": "user", "type": "address"}
|
||
],
|
||
"name": "getStakes",
|
||
"outputs": [
|
||
{
|
||
"components": [
|
||
{"internalType": "uint256", "name": "amount", "type": "uint256"},
|
||
{"internalType": "uint256", "name": "lastClaimed", "type": "uint256"},
|
||
{"internalType": "uint256", "name": "dailyRewardRate", "type": "uint256"},
|
||
{"internalType": "uint256", "name": "unlockTime", "type": "uint256"},
|
||
{"internalType": "bool", "name": "complete", "type": "bool"}
|
||
],
|
||
"internalType": "struct PacaFinanceWithBoostAndScheduleBsc.Stake[]",
|
||
"name": "",
|
||
"type": "tuple[]"
|
||
}
|
||
],
|
||
"stateMutability": "view",
|
||
"type": "function"
|
||
},
|
||
{
|
||
"inputs": [],
|
||
"name": "owner",
|
||
"outputs": [
|
||
{"internalType": "address", "name": "", "type": "address"}
|
||
],
|
||
"stateMutability": "view",
|
||
"type": "function"
|
||
}
|
||
]
|
||
|
||
# Create contract instances if addresses are available
|
||
if self.bot_manager_address and self.paca_bsc_address:
|
||
try:
|
||
self.bot_manager = self.w3.eth.contract(
|
||
address=self.bot_manager_address,
|
||
abi=self.bot_manager_abi
|
||
)
|
||
|
||
self.paca_contract = self.w3.eth.contract(
|
||
address=self.paca_bsc_address,
|
||
abi=self.paca_abi
|
||
)
|
||
|
||
print(f"🤖 BotManager: {self.bot_manager_address}")
|
||
print(f"🔗 PACA Contract: {self.paca_bsc_address}")
|
||
except Exception as e:
|
||
print(f"⚠️ Contract connection failed: {e}")
|
||
self.bot_manager = None
|
||
self.paca_contract = None
|
||
else:
|
||
print("⚠️ Contract addresses not set - deploy contracts first for full testing")
|
||
self.bot_manager = None
|
||
self.paca_contract = None
|
||
|
||
print()
|
||
|
||
def confirm_action(self, action_description):
|
||
"""Ask for user confirmation before executing"""
|
||
if not self.require_confirmation:
|
||
return True
|
||
|
||
print(f"⚠️ CONFIRMATION REQUIRED:")
|
||
print(f" Action: {action_description}")
|
||
print(f" Network: {self.chain_name}")
|
||
print(f" Dry Run: {self.dry_run_mode}")
|
||
|
||
response = input(" Continue? (yes/no): ").lower().strip()
|
||
return response in ['yes', 'y']
|
||
|
||
def simulate_clear_stakes(self, user_address):
|
||
"""Simulate what clearStakes would do without executing"""
|
||
print(f"🧪 SIMULATING clearStakes for user: {user_address}")
|
||
|
||
if not self.paca_contract:
|
||
print("❌ PACA contract not available")
|
||
return
|
||
|
||
try:
|
||
stakes = self.paca_contract.functions.getStakes(user_address).call()
|
||
|
||
print(f"📊 Current stakes analysis:")
|
||
print(f" Total stakes: {len(stakes)}")
|
||
|
||
if len(stakes) == 0:
|
||
print(" 📭 No stakes found - nothing to clear")
|
||
return
|
||
|
||
stakes_to_clear = 0
|
||
amount_to_zero = 0
|
||
|
||
for i, stake in enumerate(stakes):
|
||
amount = stake[0]
|
||
complete = stake[4]
|
||
|
||
if not complete and amount > 0:
|
||
stakes_to_clear += 1
|
||
amount_to_zero += amount
|
||
print(f" 🎯 Would clear Stake {i + 1}: {self.w3.from_wei(amount, 'ether')} ETH")
|
||
else:
|
||
print(f" ⏭️ Stake {i + 1} already complete: {self.w3.from_wei(amount, 'ether')} ETH")
|
||
|
||
print(f"\n📋 SIMULATION RESULTS:")
|
||
print(f" Stakes to clear: {stakes_to_clear}")
|
||
print(f" Total amount to zero: {self.w3.from_wei(amount_to_zero, 'ether')} ETH")
|
||
print(f" All stakes would be marked complete: ✅")
|
||
|
||
return {
|
||
'stakes_to_clear': stakes_to_clear,
|
||
'amount_to_zero': amount_to_zero,
|
||
'total_stakes': len(stakes)
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"❌ Simulation failed: {e}")
|
||
return None
|
||
|
||
def get_stakes_safe(self, user_address):
|
||
"""Safely get stakes with error handling"""
|
||
print(f"📊 Getting stakes for user: {user_address}")
|
||
|
||
if not self.paca_contract:
|
||
print("❌ PACA contract not available")
|
||
return []
|
||
|
||
try:
|
||
stakes = self.paca_contract.functions.getStakes(user_address).call()
|
||
|
||
print(f"📈 User has {len(stakes)} stakes:")
|
||
|
||
if len(stakes) == 0:
|
||
print(" 📭 No stakes found")
|
||
return stakes
|
||
|
||
total_amount = 0
|
||
active_stakes = 0
|
||
|
||
for i, stake in enumerate(stakes):
|
||
amount = stake[0]
|
||
last_claimed = stake[1]
|
||
daily_reward_rate = stake[2]
|
||
unlock_time = stake[3]
|
||
complete = stake[4]
|
||
|
||
is_active = not complete and amount > 0
|
||
if is_active:
|
||
active_stakes += 1
|
||
total_amount += amount
|
||
|
||
print(f"\n 📌 Stake {i + 1}:")
|
||
print(f" Amount: {self.w3.from_wei(amount, 'ether')} ETH")
|
||
print(f" Daily Reward Rate: {self.w3.from_wei(daily_reward_rate, 'ether')} ETH")
|
||
print(f" Complete: {complete}")
|
||
print(f" Status: {'🟢 ACTIVE' if is_active else '🔴 COMPLETED'}")
|
||
|
||
if last_claimed > 0:
|
||
import datetime
|
||
last_claimed_date = datetime.datetime.fromtimestamp(last_claimed)
|
||
print(f" Last Claimed: {last_claimed_date}")
|
||
if unlock_time > 0:
|
||
import datetime
|
||
unlock_date = datetime.datetime.fromtimestamp(unlock_time)
|
||
print(f" Unlock Time: {unlock_date}")
|
||
|
||
print(f"\n💎 Summary:")
|
||
print(f" Total Stakes: {len(stakes)}")
|
||
print(f" Active Stakes: {active_stakes}")
|
||
print(f" Total Active Amount: {self.w3.from_wei(total_amount, 'ether')} ETH")
|
||
|
||
return stakes
|
||
|
||
except Exception as e:
|
||
print(f"❌ Error getting stakes: {e}")
|
||
return []
|
||
|
||
def test_clear_stakes(self, user_address, execute=False):
|
||
"""Test clearStakes with option to execute"""
|
||
print(f"\n{'='*60}")
|
||
print(f"🧪 TESTING clearStakes for {user_address}")
|
||
print(f" Execute: {execute}")
|
||
print(f" Dry Run Mode: {self.dry_run_mode}")
|
||
print(f"{'='*60}")
|
||
|
||
# Step 1: Get current stakes
|
||
print("\n1️⃣ Getting current stakes...")
|
||
stakes_before = self.get_stakes_safe(user_address)
|
||
|
||
if not stakes_before:
|
||
print("⚠️ No stakes to test with")
|
||
return
|
||
|
||
# Step 2: Simulate what would happen
|
||
print("\n2️⃣ Simulating clearStakes...")
|
||
simulation = self.simulate_clear_stakes(user_address)
|
||
|
||
if not simulation or simulation['stakes_to_clear'] == 0:
|
||
print("⚠️ No stakes to clear")
|
||
return
|
||
|
||
# Step 3: Execute if requested and confirmed
|
||
if execute and not self.dry_run_mode:
|
||
print("\n3️⃣ Preparing to execute clearStakes...")
|
||
|
||
action = f"Clear {simulation['stakes_to_clear']} stakes ({self.w3.from_wei(simulation['amount_to_zero'], 'ether')} ETH)"
|
||
|
||
if not self.confirm_action(action):
|
||
print("❌ Action cancelled by user")
|
||
return
|
||
|
||
try:
|
||
if not self.bot_manager:
|
||
print("❌ Bot manager contract not available")
|
||
return
|
||
|
||
print("⚡ Executing clearStakes...")
|
||
# This would execute the actual transaction
|
||
# Implementation depends on whether you're on testnet or fork
|
||
print(" (Actual execution code would go here)")
|
||
print("✅ Transaction would be executed")
|
||
|
||
except Exception as e:
|
||
print(f"❌ Execution failed: {e}")
|
||
|
||
else:
|
||
print("\n3️⃣ Execution skipped (dry run mode or not requested)")
|
||
print(" To execute: set execute=True and dry_run_mode=False")
|
||
|
||
def main():
|
||
print("🧪 PacaBotManager TEST Client")
|
||
print("=" * 40)
|
||
print("This is a SAFE testing version that simulates operations")
|
||
print()
|
||
|
||
# Ask user what type of testing they want
|
||
print("Testing options:")
|
||
print("1. Local fork testing (uses mainnet contracts on fork)")
|
||
print("2. Testnet testing (requires testnet contract deployment)")
|
||
|
||
choice = input("Choose testing mode (1 or 2): ").strip()
|
||
|
||
try:
|
||
if choice == "1":
|
||
print("\n🔄 Starting local fork testing...")
|
||
client = PacaBotManagerTestClient(use_testnet=False)
|
||
else:
|
||
print("\n🌐 Starting testnet testing...")
|
||
client = PacaBotManagerTestClient(use_testnet=True)
|
||
|
||
# Test user address
|
||
test_user = "0x41970Ce76b656030A79E7C1FA76FC4EB93980255"
|
||
|
||
print("\n📋 Available Test Operations:")
|
||
print("1. Get stakes (safe)")
|
||
print("2. Simulate clearStakes (safe)")
|
||
print("3. Test clearStakes with execution (requires confirmation)")
|
||
print()
|
||
|
||
# Run safe operations
|
||
print("🔍 Running safe operations...")
|
||
stakes = client.get_stakes_safe(test_user)
|
||
|
||
if stakes:
|
||
simulation = client.simulate_clear_stakes(test_user)
|
||
|
||
# Optionally test execution (will ask for confirmation)
|
||
execute_test = input("\nRun execution test? (yes/no): ").lower().strip()
|
||
if execute_test in ['yes', 'y']:
|
||
client.test_clear_stakes(test_user, execute=True)
|
||
|
||
print("\n✅ Test completed safely!")
|
||
|
||
except Exception as e:
|
||
print(f"💥 Test failed: {e}")
|
||
sys.exit(1)
|
||
|
||
if __name__ == "__main__":
|
||
main() |