diff --git a/python/dao-voting-dashboard/.env.example b/python/dao-voting-dashboard/.env.example new file mode 100644 index 0000000..59ed563 --- /dev/null +++ b/python/dao-voting-dashboard/.env.example @@ -0,0 +1,22 @@ +# Environment Configuration + +# BNB Chain Testnet RPC +RPC_URL=https://data-seed-prebsc-1-s1.bnbchain.org:8545 + +# Governance Contract Address (mock) +GOVERNANCE_CONTRACT_ADDRESS=0x1234567890123456789012345678901234567890 + +# Database +DATABASE_PATH=./data/dao_votes.db + +# Server +HOST=0.0.0.0 +PORT=8000 + +# Governance Rules +QUORUM_PERCENTAGE=20 +APPROVAL_THRESHOLD_PERCENTAGE=50 + +# Demo Private Key (DO NOT USE IN PRODUCTION) +# Generate your own: from eth_account import Account; Account.create() +DEMO_PRIVATE_KEY=0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80 diff --git a/python/dao-voting-dashboard/.gitignore b/python/dao-voting-dashboard/.gitignore new file mode 100644 index 0000000..bf2fe07 --- /dev/null +++ b/python/dao-voting-dashboard/.gitignore @@ -0,0 +1,57 @@ +# Python +__pycache__/ +*.py[cod] +*$py.class +*.so +.Python +build/ +develop-eggs/ +dist/ +downloads/ +eggs/ +.eggs/ +lib/ +lib64/ +parts/ +sdist/ +var/ +wheels/ +*.egg-info/ +.installed.cfg +*.egg + +# Virtual environments +venv/ +ENV/ +env/ +.venv + +# IDEs +.vscode/ +.idea/ +*.swp +*.swo +*~ + +# Environment +.env +.env.local + +# Database +*.db +*.sqlite +*.sqlite3 +data/ + +# Testing +.pytest_cache/ +.coverage +htmlcov/ +.tox/ + +# Logs +*.log + +# OS +.DS_Store +Thumbs.db diff --git a/python/dao-voting-dashboard/README.md b/python/dao-voting-dashboard/README.md new file mode 100644 index 0000000..74b9bbb --- /dev/null +++ b/python/dao-voting-dashboard/README.md @@ -0,0 +1,344 @@ +# DAO Voting Dashboard + +Off-chain + on-chain DAO governance dashboard built with FastAPI and Web3.py for BNB Chain. + +## 🌟 Features + +- **Off-Chain Proposal Management**: Create and manage proposals stored in SQLite database +- **Cryptographic Vote Signing**: Sign votes with private keys using eth-account +- **Vote Aggregation**: Batch votes off-chain before syncing to reduce gas costs +- **On-Chain Integration**: Mock governance contract interaction (production-ready pattern) +- **Governance Rules**: Configurable quorum (20%) and approval threshold (50%) +- **Web Dashboard**: Clean Jinja2 templates for proposal creation, voting, and results +- **RESTful API**: FastAPI backend with comprehensive endpoints +- **Comprehensive Tests**: 100+ tests covering all components + +## πŸ—οΈ Architecture + +``` +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ Web UI β”‚ Jinja2 Templates +β”‚ (Frontend) β”‚ index.html, proposal.html, results.html +β””β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”˜ + β”‚ + β–Ό +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚ FastAPI App β”‚ REST API endpoints +β”‚ (Backend) β”‚ /api/proposals, /api/proposals/{id}/vote +β””β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”˜ + β”‚ + β”Œβ”€β”€β”€β”€β”΄β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”¬β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” + β–Ό β–Ό β–Ό β–Ό +β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β” β”Œβ”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β” +β”‚Databaseβ”‚ β”‚Governanceβ”‚ β”‚ Crypto β”‚ β”‚ Contract β”‚ +β”‚(SQLite)β”‚ β”‚ Rules β”‚ β”‚ Utils β”‚ β”‚(Web3.py) β”‚ +β””β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ β””β”€β”€β”€β”€β”€β”€β”€β”€β”€β”€β”˜ +``` + +## πŸ“¦ Installation + +### Using uv (Recommended) + +```bash +# Install uv +pip install uv + +# Install dependencies +uv pip install -e . + +# Or install in development mode +uv pip install -e ".[dev]" +``` + +### Using pip + +```bash +pip install -r requirements.txt +``` + +## βš™οΈ Configuration + +1. Copy `.env.example` to `.env`: +```bash +cp .env.example .env +``` + +2. Configure environment variables: +```env +RPC_URL=https://data-seed-prebsc-1-s1.bnbchain.org:8545 +GOVERNANCE_CONTRACT_ADDRESS=0x1234567890123456789012345678901234567890 +QUORUM_PERCENTAGE=20 +APPROVAL_THRESHOLD_PERCENTAGE=50 +DEMO_PRIVATE_KEY=0x0123456789abcdef... +``` + +## πŸš€ Quick Start + +### 1. Start the Server + +```bash +# Using uvicorn directly +uvicorn app:app --reload + +# Or using Python +python app.py +``` + +Server runs at `http://localhost:8000` + +### 2. Access the Dashboard + +Open your browser to: +- Dashboard: `http://localhost:8000` +- API Docs: `http://localhost:8000/docs` +- ReDoc: `http://localhost:8000/redoc` + +### 3. Create a Proposal + +```bash +curl -X POST "http://localhost:8000/api/proposals" \ + -H "Content-Type: application/json" \ + -d '{ + "title": "Increase Treasury Allocation", + "description": "Allocate 10,000 tokens to community development", + "creator": "0x1234567890123456789012345678901234567890", + "duration_hours": 168 + }' +``` + +### 4. Cast a Vote + +```bash +curl -X POST "http://localhost:8000/api/proposals/1/vote" \ + -H "Content-Type: application/json" \ + -d '{ + "voter_address": "0xabcdef0123456789012345678901234567890abc", + "vote_choice": "for", + "private_key": "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + }' +``` + +### 5. View Results + +```bash +curl "http://localhost:8000/api/proposals/1/results" +``` + +## πŸ“š API Documentation + +### Proposals + +#### Create Proposal +```http +POST /api/proposals +Content-Type: application/json + +{ + "title": "string", + "description": "string", + "creator": "0x...", + "duration_hours": 168 +} +``` + +#### List Proposals +```http +GET /api/proposals?status=active +``` + +#### Get Proposal Details +```http +GET /api/proposals/{proposal_id} +``` + +### Voting + +#### Cast Vote +```http +POST /api/proposals/{proposal_id}/vote +Content-Type: application/json + +{ + "voter_address": "0x...", + "vote_choice": "for|against", + "private_key": "0x..." +} +``` + +#### Get Results +```http +GET /api/proposals/{proposal_id}/results +``` + +#### Sync Votes to Chain +```http +POST /api/proposals/{proposal_id}/sync +``` + +#### Close Proposal +```http +POST /api/proposals/{proposal_id}/close +``` + +## πŸ§ͺ Testing + +### Run All Tests + +```bash +pytest +``` + +### Run with Coverage + +```bash +pytest --cov=. --cov-report=html --cov-report=term +``` + +### Run Specific Test Suite + +```bash +# Test database +pytest tests/test_database.py -v + +# Test API endpoints +pytest tests/test_api.py -v + +# Test governance logic +pytest tests/test_governance.py -v + +# Test cryptographic functions +pytest tests/test_crypto.py -v + +# Test contract interactions +pytest tests/test_contract.py -v +``` + +## 🎯 Governance Rules + +### Quorum Requirement +- **Default**: 20% of eligible voters must participate +- Formula: `required_votes = total_eligible_voters * 0.20` +- Configurable via `QUORUM_PERCENTAGE` in `.env` + +### Approval Threshold +- **Default**: 50% of votes must be "for" +- Formula: `approval_rate = votes_for / (votes_for + votes_against) * 100` +- Configurable via `APPROVAL_THRESHOLD_PERCENTAGE` in `.env` + +### Proposal Outcomes + +| Condition | Outcome | +|-----------|---------| +| Quorum met + Approval β‰₯ 50% | βœ… **Passed** | +| Quorum not met | ❌ **Rejected** (Quorum not met) | +| Quorum met + Approval < 50% | ❌ **Rejected** (Approval threshold not met) | + +## πŸ” Security Notes + +**⚠️ IMPORTANT**: This is a demo application. Do NOT use in production without: + +1. **Never expose private keys in requests**: Implement proper wallet connection (MetaMask, WalletConnect) +2. **Secure the API**: Add authentication, rate limiting, and input validation +3. **Use real on-chain contracts**: Replace mock contract with actual smart contract +4. **Implement proper Merkle proofs**: Current implementation is mock only +5. **Add voter eligibility checks**: Query token balances or NFT ownership +6. **Secure database**: Use proper database credentials and connection pooling +7. **HTTPS only**: Never run in production over HTTP + +## πŸ› οΈ Development + +### Project Structure + +``` +dao-voting-dashboard/ +β”œβ”€β”€ app.py # FastAPI application +β”œβ”€β”€ config.py # Configuration management +β”œβ”€β”€ database.py # SQLite database layer +β”œβ”€β”€ crypto_utils.py # Vote signing & Merkle proofs +β”œβ”€β”€ governance.py # Governance rules engine +β”œβ”€β”€ contract.py # Web3 contract interactions +β”œβ”€β”€ templates/ # Jinja2 HTML templates +β”‚ β”œβ”€β”€ index.html +β”‚ β”œβ”€β”€ proposal.html +β”‚ └── results.html +β”œβ”€β”€ tests/ # Test suite +β”‚ β”œβ”€β”€ test_database.py +β”‚ β”œβ”€β”€ test_crypto.py +β”‚ β”œβ”€β”€ test_governance.py +β”‚ β”œβ”€β”€ test_contract.py +β”‚ └── test_api.py +β”œβ”€β”€ pyproject.toml # Project metadata +β”œβ”€β”€ .env.example # Environment template +└── README.md +``` + +### Adding New Features + +1. **Database Changes**: Update schema in `database.py` +2. **API Endpoints**: Add routes in `app.py` +3. **Governance Logic**: Modify rules in `governance.py` +4. **Smart Contract**: Update ABI and methods in `contract.py` +5. **Tests**: Add test cases in `tests/` + +## 🌐 Deployment + +### Production Checklist + +- [ ] Set `DEBUG=False` in environment +- [ ] Use production database (PostgreSQL recommended) +- [ ] Configure reverse proxy (nginx) +- [ ] Enable HTTPS with SSL certificates +- [ ] Set up monitoring and logging +- [ ] Deploy smart contracts to mainnet +- [ ] Implement proper wallet connection +- [ ] Add rate limiting and API authentication +- [ ] Configure CORS properly +- [ ] Set up CI/CD pipeline + +### Deploy to Cloud + +```bash +# Example: Deploy with Gunicorn +gunicorn app:app -w 4 -k uvicorn.workers.UvicornWorker +``` + +## πŸ“Š Example Flow + +1. **Create Proposal**: DAO member creates proposal via UI or API +2. **Off-Chain Voting**: Users vote and signatures are stored in SQLite +3. **Vote Aggregation**: Votes accumulate off-chain to save gas +4. **Sync to Chain**: Batch sync aggregated votes to smart contract +5. **Finalize**: After deadline, proposal is closed and outcome determined +6. **Execution**: Passed proposals can be executed on-chain + +## 🀝 Contributing + +This is an example project for the BNB Chain Cookbook. Feel free to: +- Report issues +- Submit pull requests +- Suggest improvements +- Use as a template for your own DAO + +## πŸ“„ License + +MIT License - see LICENSE file for details + +## πŸ”— Resources + +- [BNB Chain Documentation](https://docs.bnbchain.org/) +- [FastAPI Documentation](https://fastapi.tiangolo.com/) +- [Web3.py Documentation](https://web3py.readthedocs.io/) +- [eth-account Documentation](https://eth-account.readthedocs.io/) + +## πŸ’‘ Tips + +- Use `uv` for faster dependency management +- Run tests before committing changes +- Check API docs at `/docs` for interactive testing +- Monitor database size in production +- Batch vote syncing to reduce gas costs +- Consider using Redis for caching in production + +--- + +Built with ❀️ for BNB Chain Cookbook diff --git a/python/dao-voting-dashboard/app.py b/python/dao-voting-dashboard/app.py new file mode 100644 index 0000000..9da79b6 --- /dev/null +++ b/python/dao-voting-dashboard/app.py @@ -0,0 +1,397 @@ +""" +FastAPI backend for DAO voting dashboard. +""" + +from fastapi import FastAPI, HTTPException, Request +from fastapi.responses import HTMLResponse, JSONResponse +from fastapi.staticfiles import StaticFiles +from fastapi.templating import Jinja2Templates +from pydantic import BaseModel +from typing import Optional +from datetime import datetime, timedelta +import uvicorn + +from database import Database +from crypto_utils import VoteSignature +from governance import GovernanceRules +from contract import governance_contract +from config import config + + +app = FastAPI( + title="DAO Voting Dashboard", + description="Off-chain + on-chain DAO governance dashboard", + version="1.0.0" +) + +# Templates +templates = Jinja2Templates(directory="templates") + +# Database instance +db = Database() + + +# Request/Response models +class CreateProposalRequest(BaseModel): + title: str + description: str + creator: str + duration_hours: int = 168 # Default 7 days + + +class VoteRequest(BaseModel): + voter_address: str + vote_choice: str # "for" or "against" + private_key: str # For signing (demo only) + + +class SyncRequest(BaseModel): + proposal_id: int + + +# Root endpoint +@app.get("/", response_class=HTMLResponse) +async def root(request: Request): + """Render main dashboard.""" + proposals = db.list_proposals() + return templates.TemplateResponse( + "index.html", + {"request": request, "proposals": proposals} + ) + + +# Health check +@app.get("/health") +async def health_check(): + """Health check endpoint.""" + return {"status": "healthy", "timestamp": datetime.now().isoformat()} + + +# Create proposal +@app.post("/api/proposals") +async def create_proposal(req: CreateProposalRequest): + """ + Create a new proposal (off-chain). + + In production, this would also create it on-chain. + """ + try: + end_time = datetime.now() + timedelta(hours=req.duration_hours) + + # Create off-chain + proposal_id = db.create_proposal( + title=req.title, + description=req.description, + creator=req.creator, + end_time=end_time + ) + + # Mock: Create on-chain (in production, use actual contract) + on_chain_id = governance_contract.create_proposal( + req.title, + req.description, + int(end_time.timestamp()) + ) + + # Update with on-chain ID + db.conn.execute( + "UPDATE proposals SET on_chain_id = ? WHERE id = ?", + (on_chain_id, proposal_id) + ) + db.conn.commit() + + return { + "success": True, + "proposal_id": proposal_id, + "on_chain_id": on_chain_id, + "message": "Proposal created successfully" + } + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +# List proposals +@app.get("/api/proposals") +async def list_proposals(status: Optional[str] = None): + """Get all proposals, optionally filtered by status.""" + try: + proposals = db.list_proposals(status) + return {"proposals": proposals} + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +# Get proposal details +@app.get("/api/proposals/{proposal_id}") +async def get_proposal(proposal_id: int): + """Get proposal details with vote counts.""" + try: + proposal = db.get_proposal(proposal_id) + if not proposal: + raise HTTPException(status_code=404, detail="Proposal not found") + + vote_counts = db.get_vote_counts(proposal_id) + votes = db.get_votes(proposal_id) + + return { + "proposal": proposal, + "vote_counts": vote_counts, + "votes": votes + } + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +# Cast vote +@app.post("/api/proposals/{proposal_id}/vote") +async def cast_vote(proposal_id: int, req: VoteRequest): + """ + Cast a vote on a proposal (off-chain). + + Vote is cryptographically signed and stored off-chain. + """ + try: + # Validate proposal exists and is active + proposal = db.get_proposal(proposal_id) + if not proposal: + raise HTTPException(status_code=404, detail="Proposal not found") + + if proposal["status"] != "active": + raise HTTPException( + status_code=400, + detail=f"Proposal is {proposal['status']}, voting closed" + ) + + # Check if already voted + existing_votes = db.get_votes(proposal_id) + if any(v["voter_address"] == req.voter_address for v in existing_votes): + raise HTTPException( + status_code=400, + detail="Address has already voted" + ) + + # Validate vote choice + if req.vote_choice not in ["for", "against"]: + raise HTTPException( + status_code=400, + detail="vote_choice must be 'for' or 'against'" + ) + + # Check voting eligibility + if not GovernanceRules.is_eligible_voter(req.voter_address): + raise HTTPException( + status_code=403, + detail="Address not eligible to vote" + ) + + # Sign vote + message = f"Vote {req.vote_choice} on proposal {proposal_id}" + signer_address, signature = VoteSignature.sign_vote( + req.private_key, + message + ) + + # Verify signature matches voter address + if signer_address.lower() != req.voter_address.lower(): + raise HTTPException( + status_code=400, + detail="Signature verification failed" + ) + + # Store vote + vote_id = db.create_vote( + proposal_id=proposal_id, + voter_address=req.voter_address, + vote_choice=req.vote_choice, + signature=signature + ) + + return { + "success": True, + "vote_id": vote_id, + "message": "Vote cast successfully" + } + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +# Get proposal results +@app.get("/api/proposals/{proposal_id}/results") +async def get_results(proposal_id: int): + """ + Get voting results and outcome determination. + """ + try: + proposal = db.get_proposal(proposal_id) + if not proposal: + raise HTTPException(status_code=404, detail="Proposal not found") + + vote_counts = db.get_vote_counts(proposal_id) + + # Mock: Total eligible voters (in production, query from contract) + total_eligible_voters = 100 + + # Determine outcome + outcome = GovernanceRules.determine_outcome( + votes_for=vote_counts["for"], + votes_against=vote_counts["against"], + total_eligible_voters=total_eligible_voters + ) + + return { + "proposal": proposal, + "outcome": outcome + } + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +# Sync votes to chain +@app.post("/api/proposals/{proposal_id}/sync") +async def sync_to_chain(proposal_id: int): + """ + Sync aggregated votes to on-chain contract. + + In production, this would submit votes in batches with Merkle proofs. + """ + try: + proposal = db.get_proposal(proposal_id) + if not proposal: + raise HTTPException(status_code=404, detail="Proposal not found") + + vote_counts = db.get_vote_counts(proposal_id) + + # Submit votes to mock contract + success = governance_contract.submit_votes( + proposal_id=proposal["on_chain_id"], + votes_for=vote_counts["for"], + votes_against=vote_counts["against"] + ) + + if success: + # Mark votes as synced + db.mark_votes_synced(proposal_id) + + return { + "success": True, + "message": "Votes synced to chain", + "votes_synced": vote_counts["total"] + } + else: + raise HTTPException( + status_code=500, + detail="Failed to sync votes to chain" + ) + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +# Close proposal +@app.post("/api/proposals/{proposal_id}/close") +async def close_proposal(proposal_id: int): + """ + Close voting and finalize proposal. + """ + try: + proposal = db.get_proposal(proposal_id) + if not proposal: + raise HTTPException(status_code=404, detail="Proposal not found") + + if proposal["status"] != "active": + raise HTTPException( + status_code=400, + detail=f"Proposal already {proposal['status']}" + ) + + # Get results + vote_counts = db.get_vote_counts(proposal_id) + total_eligible_voters = 100 # Mock + + outcome = GovernanceRules.determine_outcome( + votes_for=vote_counts["for"], + votes_against=vote_counts["against"], + total_eligible_voters=total_eligible_voters + ) + + # Close proposal + db.close_proposal(proposal_id, outcome["status"]) + + # Finalize on-chain (mock) + governance_contract.finalize_proposal(proposal["on_chain_id"]) + + return { + "success": True, + "message": "Proposal closed", + "outcome": outcome + } + except HTTPException: + raise + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +# Proposal detail page +@app.get("/proposals/{proposal_id}", response_class=HTMLResponse) +async def proposal_detail(request: Request, proposal_id: int): + """Render proposal detail page.""" + try: + proposal = db.get_proposal(proposal_id) + if not proposal: + raise HTTPException(status_code=404, detail="Proposal not found") + + vote_counts = db.get_vote_counts(proposal_id) + votes = db.get_votes(proposal_id) + + return templates.TemplateResponse( + "proposal.html", + { + "request": request, + "proposal": proposal, + "vote_counts": vote_counts, + "votes": votes + } + ) + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +# Results page +@app.get("/proposals/{proposal_id}/results", response_class=HTMLResponse) +async def results_page(request: Request, proposal_id: int): + """Render results page.""" + try: + proposal = db.get_proposal(proposal_id) + if not proposal: + raise HTTPException(status_code=404, detail="Proposal not found") + + vote_counts = db.get_vote_counts(proposal_id) + total_eligible_voters = 100 # Mock + + outcome = GovernanceRules.determine_outcome( + votes_for=vote_counts["for"], + votes_against=vote_counts["against"], + total_eligible_voters=total_eligible_voters + ) + + return templates.TemplateResponse( + "results.html", + { + "request": request, + "proposal": proposal, + "outcome": outcome + } + ) + except Exception as e: + raise HTTPException(status_code=500, detail=str(e)) + + +if __name__ == "__main__": + uvicorn.run(app, host="0.0.0.0", port=8000) diff --git a/python/dao-voting-dashboard/app/__init__.py b/python/dao-voting-dashboard/app/__init__.py new file mode 100644 index 0000000..6b57580 --- /dev/null +++ b/python/dao-voting-dashboard/app/__init__.py @@ -0,0 +1 @@ +"""DAO Voting Dashboard Application.""" diff --git a/python/dao-voting-dashboard/app/blockchain.py b/python/dao-voting-dashboard/app/blockchain.py new file mode 100644 index 0000000..57c73b8 --- /dev/null +++ b/python/dao-voting-dashboard/app/blockchain.py @@ -0,0 +1,163 @@ +"""Blockchain interaction for DAO voting dashboard.""" + +from typing import Dict, Any, Optional +from web3 import Web3 +from eth_account import Account +from eth_account.messages import encode_defunct +from app.config import config + +# Mock Governance Contract ABI +GOVERNANCE_ABI = [ + { + "inputs": [{"name": "proposalId", "type": "uint256"}], + "name": "getProposal", + "outputs": [ + {"name": "title", "type": "string"}, + {"name": "votesFor", "type": "uint256"}, + {"name": "votesAgainst", "type": "uint256"}, + {"name": "status", "type": "uint8"} + ], + "stateMutability": "view", + "type": "function" + }, + { + "inputs": [ + {"name": "title", "type": "string"}, + {"name": "description", "type": "string"}, + {"name": "endTime", "type": "uint256"} + ], + "name": "createProposal", + "outputs": [{"name": "proposalId", "type": "uint256"}], + "stateMutability": "nonpayable", + "type": "function" + }, + { + "inputs": [ + {"name": "proposalId", "type": "uint256"}, + {"name": "support", "type": "bool"} + ], + "name": "vote", + "outputs": [], + "stateMutability": "nonpayable", + "type": "function" + } +] + + +class BlockchainClient: + """Web3 client for interacting with BNB Chain.""" + + def __init__(self): + """Initialize Web3 connection.""" + self.w3 = Web3(Web3.HTTPProvider(config.RPC_URL)) + self.contract_address = config.GOVERNANCE_CONTRACT_ADDRESS + + # Mock contract instance + if self.contract_address != "0x0000000000000000000000000000000000000000": + self.contract = self.w3.eth.contract( + address=self.contract_address, + abi=GOVERNANCE_ABI + ) + else: + self.contract = None + + def is_connected(self) -> bool: + """Check if connected to BNB Chain.""" + try: + return self.w3.is_connected() + except Exception: + return False + + def get_chain_id(self) -> int: + """Get chain ID.""" + return self.w3.eth.chain_id if self.is_connected() else config.CHAIN_ID + + def sign_message(self, private_key: str, message: str) -> str: + """Sign a message with private key.""" + account = Account.from_key(private_key) + message_hash = encode_defunct(text=message) + signed = account.sign_message(message_hash) + return signed.signature.hex() + + def verify_signature( + self, + message: str, + signature: str, + expected_address: str + ) -> bool: + """Verify a signed message.""" + try: + message_hash = encode_defunct(text=message) + recovered_address = Account.recover_message( + message_hash, + signature=bytes.fromhex(signature.replace('0x', '')) + ) + return recovered_address.lower() == expected_address.lower() + except Exception: + return False + + # Mock on-chain operations + + def get_proposal_from_chain(self, proposal_id: int) -> Optional[Dict[str, Any]]: + """Get proposal from smart contract (mocked).""" + if not self.contract: + # Return mock data + return { + "proposal_id": proposal_id, + "title": f"Proposal #{proposal_id}", + "votes_for": 100, + "votes_against": 50, + "status": "active" + } + + try: + result = self.contract.functions.getProposal(proposal_id).call() + return { + "proposal_id": proposal_id, + "title": result[0], + "votes_for": result[1], + "votes_against": result[2], + "status": ["active", "passed", "rejected"][result[3]] + } + except Exception as e: + print(f"Error fetching proposal from chain: {e}") + return None + + def submit_proposal_to_chain( + self, + title: str, + description: str, + end_time: int, + private_key: Optional[str] = None + ) -> Optional[int]: + """Submit proposal to chain (mocked).""" + # In a real implementation, this would create a transaction + # For now, return a mock proposal ID + import random + return random.randint(1000, 9999) + + def submit_vote_to_chain( + self, + proposal_id: int, + support: bool, + private_key: Optional[str] = None + ) -> Optional[str]: + """Submit vote to chain (mocked).""" + # In a real implementation, this would create a transaction + # For now, return a mock transaction hash + return f"0x{'0' * 64}" + + def batch_submit_votes( + self, + proposal_id: int, + votes: list, + private_key: Optional[str] = None + ) -> Optional[str]: + """Batch submit multiple votes to chain (gasless simulation).""" + # Simulate Merkle proof generation and batch submission + # In production, this would use a relayer or meta-transaction + return f"0x{'1' * 64}" + + +# Global blockchain client +blockchain = BlockchainClient() diff --git a/python/dao-voting-dashboard/app/config.py b/python/dao-voting-dashboard/app/config.py new file mode 100644 index 0000000..794e18c --- /dev/null +++ b/python/dao-voting-dashboard/app/config.py @@ -0,0 +1,42 @@ +"""Configuration management for DAO voting dashboard.""" + +import os +from pathlib import Path +from dotenv import load_dotenv + +# Load environment variables +load_dotenv() + + +class Config: + """Application configuration.""" + + # BNB Chain + RPC_URL: str = os.getenv("RPC_URL", "https://data-seed-prebsc-1-s1.bnbchain.org:8545") + CHAIN_ID: int = int(os.getenv("CHAIN_ID", "97")) + GOVERNANCE_CONTRACT_ADDRESS: str = os.getenv( + "GOVERNANCE_CONTRACT_ADDRESS", + "0x0000000000000000000000000000000000000000" + ) + + # Database + DATABASE_PATH: str = os.getenv("DATABASE_PATH", "./data/dao.db") + + # Server + HOST: str = os.getenv("HOST", "0.0.0.0") + PORT: int = int(os.getenv("PORT", "8000")) + + # Voting Rules + QUORUM_PERCENTAGE: int = int(os.getenv("QUORUM_PERCENTAGE", "20")) + PASS_THRESHOLD_PERCENTAGE: int = int(os.getenv("PASS_THRESHOLD_PERCENTAGE", "50")) + VOTING_PERIOD_SECONDS: int = int(os.getenv("VOTING_PERIOD_SECONDS", "604800")) + + @classmethod + def get_database_path(cls) -> Path: + """Get database path and ensure directory exists.""" + db_path = Path(cls.DATABASE_PATH) + db_path.parent.mkdir(parents=True, exist_ok=True) + return db_path + + +config = Config() diff --git a/python/dao-voting-dashboard/app/database.py b/python/dao-voting-dashboard/app/database.py new file mode 100644 index 0000000..4b02abc --- /dev/null +++ b/python/dao-voting-dashboard/app/database.py @@ -0,0 +1,238 @@ +"""Database operations for DAO voting dashboard.""" + +import sqlite3 +import json +from typing import List, Optional, Dict, Any +from datetime import datetime +from pathlib import Path +from app.config import config + + +class Database: + """SQLite database manager for proposals and votes.""" + + def __init__(self, db_path: Optional[Path] = None): + """Initialize database connection.""" + self.db_path = db_path or config.get_database_path() + self._init_db() + + def _get_connection(self) -> sqlite3.Connection: + """Get database connection.""" + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row + return conn + + def _init_db(self): + """Initialize database schema.""" + conn = self._get_connection() + cursor = conn.cursor() + + # Proposals table + cursor.execute(""" + CREATE TABLE IF NOT EXISTS proposals ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + title TEXT NOT NULL, + description TEXT NOT NULL, + creator TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + end_time TIMESTAMP NOT NULL, + status TEXT DEFAULT 'active', + on_chain_id INTEGER DEFAULT NULL, + on_chain_synced BOOLEAN DEFAULT FALSE + ) + """) + + # Votes table + cursor.execute(""" + CREATE TABLE IF NOT EXISTS votes ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + proposal_id INTEGER NOT NULL, + voter_address TEXT NOT NULL, + vote_choice TEXT NOT NULL, + signature TEXT NOT NULL, + message TEXT NOT NULL, + voted_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + on_chain_synced BOOLEAN DEFAULT FALSE, + FOREIGN KEY (proposal_id) REFERENCES proposals (id), + UNIQUE(proposal_id, voter_address) + ) + """) + + conn.commit() + conn.close() + + # Proposal operations + + def create_proposal( + self, + title: str, + description: str, + creator: str, + end_time: datetime + ) -> int: + """Create a new proposal.""" + conn = self._get_connection() + cursor = conn.cursor() + + cursor.execute(""" + INSERT INTO proposals (title, description, creator, end_time) + VALUES (?, ?, ?, ?) + """, (title, description, creator, end_time)) + + proposal_id = cursor.lastrowid + conn.commit() + conn.close() + + return proposal_id + + def get_proposal(self, proposal_id: int) -> Optional[Dict[str, Any]]: + """Get proposal by ID.""" + conn = self._get_connection() + cursor = conn.cursor() + + cursor.execute("SELECT * FROM proposals WHERE id = ?", (proposal_id,)) + row = cursor.fetchone() + conn.close() + + if row: + return dict(row) + return None + + def list_proposals( + self, + status: Optional[str] = None, + limit: int = 100 + ) -> List[Dict[str, Any]]: + """List proposals with optional status filter.""" + conn = self._get_connection() + cursor = conn.cursor() + + if status: + cursor.execute( + "SELECT * FROM proposals WHERE status = ? ORDER BY created_at DESC LIMIT ?", + (status, limit) + ) + else: + cursor.execute( + "SELECT * FROM proposals ORDER BY created_at DESC LIMIT ?", + (limit,) + ) + + rows = cursor.fetchall() + conn.close() + + return [dict(row) for row in rows] + + def update_proposal_status(self, proposal_id: int, status: str) -> bool: + """Update proposal status.""" + conn = self._get_connection() + cursor = conn.cursor() + + cursor.execute( + "UPDATE proposals SET status = ? WHERE id = ?", + (status, proposal_id) + ) + + updated = cursor.rowcount > 0 + conn.commit() + conn.close() + + return updated + + def sync_proposal_to_chain(self, proposal_id: int, on_chain_id: int) -> bool: + """Mark proposal as synced to chain.""" + conn = self._get_connection() + cursor = conn.cursor() + + cursor.execute(""" + UPDATE proposals + SET on_chain_id = ?, on_chain_synced = TRUE + WHERE id = ? + """, (on_chain_id, proposal_id)) + + updated = cursor.rowcount > 0 + conn.commit() + conn.close() + + return updated + + # Vote operations + + def cast_vote( + self, + proposal_id: int, + voter_address: str, + vote_choice: str, + signature: str, + message: str + ) -> int: + """Cast a vote on a proposal.""" + conn = self._get_connection() + cursor = conn.cursor() + + cursor.execute(""" + INSERT INTO votes (proposal_id, voter_address, vote_choice, signature, message) + VALUES (?, ?, ?, ?, ?) + """, (proposal_id, voter_address, vote_choice, signature, message)) + + vote_id = cursor.lastrowid + conn.commit() + conn.close() + + return vote_id + + def get_votes(self, proposal_id: int) -> List[Dict[str, Any]]: + """Get all votes for a proposal.""" + conn = self._get_connection() + cursor = conn.cursor() + + cursor.execute( + "SELECT * FROM votes WHERE proposal_id = ? ORDER BY voted_at DESC", + (proposal_id,) + ) + + rows = cursor.fetchall() + conn.close() + + return [dict(row) for row in rows] + + def get_vote_counts(self, proposal_id: int) -> Dict[str, int]: + """Get vote counts for a proposal.""" + conn = self._get_connection() + cursor = conn.cursor() + + cursor.execute(""" + SELECT vote_choice, COUNT(*) as count + FROM votes + WHERE proposal_id = ? + GROUP BY vote_choice + """, (proposal_id,)) + + rows = cursor.fetchall() + conn.close() + + counts = {"for": 0, "against": 0, "abstain": 0} + for row in rows: + counts[row["vote_choice"]] = row["count"] + + return counts + + def has_voted(self, proposal_id: int, voter_address: str) -> bool: + """Check if address has already voted.""" + conn = self._get_connection() + cursor = conn.cursor() + + cursor.execute(""" + SELECT COUNT(*) as count + FROM votes + WHERE proposal_id = ? AND voter_address = ? + """, (proposal_id, voter_address.lower())) + + result = cursor.fetchone() + conn.close() + + return result["count"] > 0 + + +# Global database instance +db = Database() diff --git a/python/dao-voting-dashboard/app/governance.py b/python/dao-voting-dashboard/app/governance.py new file mode 100644 index 0000000..60309c0 --- /dev/null +++ b/python/dao-voting-dashboard/app/governance.py @@ -0,0 +1,109 @@ +"""Governance logic for DAO voting.""" + +from typing import Dict, Any +from datetime import datetime +from app.config import config +from app.database import db + + +class GovernanceEngine: + """Governance rules and decision logic.""" + + @staticmethod + def calculate_results(proposal_id: int, total_eligible_voters: int = 1000) -> Dict[str, Any]: + """Calculate voting results for a proposal.""" + counts = db.get_vote_counts(proposal_id) + + total_votes = counts["for"] + counts["against"] + counts["abstain"] + + # Calculate percentages + votes_for_pct = (counts["for"] / total_votes * 100) if total_votes > 0 else 0 + votes_against_pct = (counts["against"] / total_votes * 100) if total_votes > 0 else 0 + abstain_pct = (counts["abstain"] / total_votes * 100) if total_votes > 0 else 0 + + # Check quorum + quorum_needed = int(total_eligible_voters * config.QUORUM_PERCENTAGE / 100) + quorum_reached = total_votes >= quorum_needed + + # Check if passed + threshold_needed = config.PASS_THRESHOLD_PERCENTAGE + passed = quorum_reached and votes_for_pct > threshold_needed + + return { + "votes_for": counts["for"], + "votes_against": counts["against"], + "votes_abstain": counts["abstain"], + "total_votes": total_votes, + "votes_for_percentage": round(votes_for_pct, 2), + "votes_against_percentage": round(votes_against_pct, 2), + "abstain_percentage": round(abstain_pct, 2), + "quorum_needed": quorum_needed, + "quorum_reached": quorum_reached, + "threshold_needed": threshold_needed, + "passed": passed, + "status": "passed" if passed else "rejected" if quorum_reached else "pending" + } + + @staticmethod + def can_vote(proposal_id: int, voter_address: str) -> tuple[bool, str]: + """Check if a user can vote on a proposal.""" + # Check if proposal exists + proposal = db.get_proposal(proposal_id) + if not proposal: + return False, "Proposal not found" + + # Check if proposal is active + if proposal["status"] != "active": + return False, "Proposal is not active" + + # Check if voting period has ended + end_time = datetime.fromisoformat(proposal["end_time"]) + if datetime.now() > end_time: + return False, "Voting period has ended" + + # Check if already voted + if db.has_voted(proposal_id, voter_address): + return False, "Already voted on this proposal" + + return True, "Can vote" + + @staticmethod + def close_proposal(proposal_id: int) -> Dict[str, Any]: + """Close a proposal and finalize results.""" + proposal = db.get_proposal(proposal_id) + if not proposal: + return {"success": False, "error": "Proposal not found"} + + if proposal["status"] != "active": + return {"success": False, "error": "Proposal already closed"} + + # Calculate final results + results = GovernanceEngine.calculate_results(proposal_id) + + # Update proposal status + new_status = results["status"] + db.update_proposal_status(proposal_id, new_status) + + return { + "success": True, + "proposal_id": proposal_id, + "final_status": new_status, + "results": results + } + + @staticmethod + def auto_close_expired_proposals(): + """Automatically close proposals past their end time.""" + proposals = db.list_proposals(status="active") + closed_count = 0 + + for proposal in proposals: + end_time = datetime.fromisoformat(proposal["end_time"]) + if datetime.now() > end_time: + GovernanceEngine.close_proposal(proposal["id"]) + closed_count += 1 + + return closed_count + + +governance = GovernanceEngine() diff --git a/python/dao-voting-dashboard/app/main.py b/python/dao-voting-dashboard/app/main.py new file mode 100644 index 0000000..fb94d6c --- /dev/null +++ b/python/dao-voting-dashboard/app/main.py @@ -0,0 +1,253 @@ +"""FastAPI application for DAO voting dashboard.""" + +from fastapi import FastAPI, HTTPException, Request +from fastapi.responses import HTMLResponse +from fastapi.staticfiles import StaticFiles +from fastapi.templating import Jinja2Templates +from pydantic import BaseModel +from typing import Optional, List +from datetime import datetime, timedelta +import uvicorn + +from app.config import config +from app.database import db +from app.blockchain import blockchain +from app.governance import governance + +# Initialize FastAPI app +app = FastAPI( + title="DAO Voting Dashboard", + description="Off-chain + on-chain DAO voting dashboard for BNB Chain", + version="1.0.0" +) + +# Setup templates +templates = Jinja2Templates(directory="app/templates") + + +# Request/Response Models + +class CreateProposalRequest(BaseModel): + title: str + description: str + creator: str + duration_days: int = 7 + + +class VoteRequest(BaseModel): + proposal_id: int + voter_address: str + vote_choice: str # "for", "against", "abstain" + private_key: str # For signing (in production, use wallet integration) + + +class SyncToChainRequest(BaseModel): + proposal_id: int + private_key: Optional[str] = None + + +# API Routes + +@app.get("/", response_class=HTMLResponse) +async def home(request: Request): + """Render home page.""" + proposals = db.list_proposals(limit=10) + + # Auto-close expired proposals + governance.auto_close_expired_proposals() + + return templates.TemplateResponse( + "index.html", + {"request": request, "proposals": proposals} + ) + + +@app.get("/api/proposals") +async def list_proposals(status: Optional[str] = None, limit: int = 100): + """List all proposals.""" + proposals = db.list_proposals(status=status, limit=limit) + return {"proposals": proposals} + + +@app.post("/api/proposals") +async def create_proposal(request: CreateProposalRequest): + """Create a new proposal.""" + end_time = datetime.now() + timedelta(days=request.duration_days) + + proposal_id = db.create_proposal( + title=request.title, + description=request.description, + creator=request.creator, + end_time=end_time + ) + + return { + "success": True, + "proposal_id": proposal_id, + "message": "Proposal created successfully" + } + + +@app.get("/api/proposals/{proposal_id}") +async def get_proposal(proposal_id: int): + """Get proposal details.""" + proposal = db.get_proposal(proposal_id) + + if not proposal: + raise HTTPException(status_code=404, detail="Proposal not found") + + # Get voting results + results = governance.calculate_results(proposal_id) + + # Get votes + votes = db.get_votes(proposal_id) + + return { + "proposal": proposal, + "results": results, + "votes": votes + } + + +@app.post("/api/votes") +async def cast_vote(request: VoteRequest): + """Cast a vote on a proposal.""" + # Validate vote choice + if request.vote_choice not in ["for", "against", "abstain"]: + raise HTTPException(status_code=400, detail="Invalid vote choice") + + # Check if can vote + can_vote, message = governance.can_vote(request.proposal_id, request.voter_address) + if not can_vote: + raise HTTPException(status_code=400, detail=message) + + # Create vote message + proposal = db.get_proposal(request.proposal_id) + vote_message = f"Vote {request.vote_choice} on proposal #{request.proposal_id}: {proposal['title']}" + + # Sign the vote + try: + signature = blockchain.sign_message(request.private_key, vote_message) + except Exception as e: + raise HTTPException(status_code=400, detail=f"Failed to sign vote: {str(e)}") + + # Verify signature + if not blockchain.verify_signature(vote_message, signature, request.voter_address): + raise HTTPException(status_code=400, detail="Signature verification failed") + + # Store vote + vote_id = db.cast_vote( + proposal_id=request.proposal_id, + voter_address=request.voter_address, + vote_choice=request.vote_choice, + signature=signature, + message=vote_message + ) + + return { + "success": True, + "vote_id": vote_id, + "message": "Vote cast successfully", + "signature": signature + } + + +@app.post("/api/proposals/{proposal_id}/close") +async def close_proposal(proposal_id: int): + """Close a proposal and finalize results.""" + result = governance.close_proposal(proposal_id) + + if not result["success"]: + raise HTTPException(status_code=400, detail=result["error"]) + + return result + + +@app.post("/api/proposals/{proposal_id}/sync-to-chain") +async def sync_to_chain(proposal_id: int, request: SyncToChainRequest): + """Sync proposal to blockchain.""" + proposal = db.get_proposal(proposal_id) + + if not proposal: + raise HTTPException(status_code=404, detail="Proposal not found") + + if proposal["on_chain_synced"]: + return { + "success": True, + "message": "Proposal already synced", + "on_chain_id": proposal["on_chain_id"] + } + + # Submit to chain (mocked) + try: + end_time_timestamp = int(datetime.fromisoformat(proposal["end_time"]).timestamp()) + on_chain_id = blockchain.submit_proposal_to_chain( + title=proposal["title"], + description=proposal["description"], + end_time=end_time_timestamp, + private_key=request.private_key + ) + + # Update database + db.sync_proposal_to_chain(proposal_id, on_chain_id) + + return { + "success": True, + "on_chain_id": on_chain_id, + "message": "Proposal synced to chain" + } + except Exception as e: + raise HTTPException(status_code=500, detail=f"Failed to sync: {str(e)}") + + +@app.get("/api/chain/proposal/{on_chain_id}") +async def get_chain_proposal(on_chain_id: int): + """Get proposal from blockchain.""" + proposal = blockchain.get_proposal_from_chain(on_chain_id) + + if not proposal: + raise HTTPException(status_code=404, detail="Proposal not found on chain") + + return proposal + + +@app.get("/api/health") +async def health_check(): + """Health check endpoint.""" + return { + "status": "healthy", + "chain_connected": blockchain.is_connected(), + "chain_id": blockchain.get_chain_id(), + "database": "connected" + } + + +@app.get("/proposal/{proposal_id}", response_class=HTMLResponse) +async def proposal_detail(request: Request, proposal_id: int): + """Render proposal detail page.""" + proposal = db.get_proposal(proposal_id) + + if not proposal: + raise HTTPException(status_code=404, detail="Proposal not found") + + results = governance.calculate_results(proposal_id) + votes = db.get_votes(proposal_id) + + return templates.TemplateResponse( + "proposal.html", + { + "request": request, + "proposal": proposal, + "results": results, + "votes": votes + } + ) + + +if __name__ == "__main__": + uvicorn.run( + "app.main:app", + host=config.HOST, + port=config.PORT, + reload=True + ) diff --git a/python/dao-voting-dashboard/app/templates/index.html b/python/dao-voting-dashboard/app/templates/index.html new file mode 100644 index 0000000..8f688c9 --- /dev/null +++ b/python/dao-voting-dashboard/app/templates/index.html @@ -0,0 +1,247 @@ + + + + + + DAO Voting Dashboard + + + +
+
+

πŸ—³οΈ DAO Voting Dashboard

+

Off-chain + On-chain Governance for BNB Chain

+
+ +
+

Create New Proposal

+
+
+ + +
+ +
+ + +
+ +
+ + +
+ +
+ + +
+ + +
+
+ +
+

Active Proposals

+
+ {% if proposals %} + {% for proposal in proposals %} +
+

{{ proposal.title }}

+

{{ proposal.description[:150] }}{% if proposal.description|length > 150 %}...{% endif %}

+
+ {{ proposal.status.upper() }} + Created: {{ proposal.created_at }} + Ends: {{ proposal.end_time }} +
+
+ View Details +
+
+ {% endfor %} + {% else %} +
+

No proposals yet. Create one to get started!

+
+ {% endif %} +
+
+
+ + + + diff --git a/python/dao-voting-dashboard/app/templates/proposal.html b/python/dao-voting-dashboard/app/templates/proposal.html new file mode 100644 index 0000000..fdab6f8 --- /dev/null +++ b/python/dao-voting-dashboard/app/templates/proposal.html @@ -0,0 +1,377 @@ + + + + + + {{ proposal.title }} - DAO Voting + + + +
+
+ ← Back to Dashboard +

{{ proposal.title }}

+
+ +
+ {{ proposal.status.upper() }} +
+ Creator: {{ proposal.creator }}
+ Created: {{ proposal.created_at }}
+ Voting Ends: {{ proposal.end_time }}
+ {% if proposal.on_chain_synced %} + On-Chain ID: #{{ proposal.on_chain_id }} + {% endif %} +
+ +

Description

+

{{ proposal.description }}

+
+ +
+

Voting Results

+
+
+

{{ results.for_votes }}

+

FOR

+
+
+

{{ results.against_votes }}

+

AGAINST

+
+
+

{{ results.abstain_votes }}

+

ABSTAIN

+
+
+ +
+ Total Votes: {{ results.total_votes }} +
+ + {% if results.quorum_reached %} +
+ βœ… Quorum Reached ({{ results.participation_rate }}%) +
+ {% else %} +
+ ⚠️ Quorum Not Reached ({{ results.participation_rate }}% / {{ results.quorum_required }}% required) +
+ {% endif %} + + {% if results.vote_passed %} +
+ βœ… Proposal Passed ({{ results.approval_rate }}% approval) +
+ {% else %} +
+ ❌ Proposal Rejected ({{ results.approval_rate }}% approval / {{ results.pass_threshold }}% required) +
+ {% endif %} +
+ + {% if proposal.status == 'active' %} +
+

Cast Your Vote

+
+
+ + +
+ +
+ + + ⚠️ For demo only. In production, use wallet integration. +
+ +
+ + +
+ + +
+
+ {% endif %} + +
+

Vote History ({{ votes|length }} votes)

+
+ {% if votes %} + {% for vote in votes %} +
+ {{ vote.voter_address[:10] }}...{{ vote.voter_address[-8:] }} + {{ vote.vote_choice.upper() }} +
+ {{ vote.created_at }} +
+
+ {% endfor %} + {% else %} +

No votes yet.

+ {% endif %} +
+
+
+ + + + diff --git a/python/dao-voting-dashboard/config.py b/python/dao-voting-dashboard/config.py new file mode 100644 index 0000000..4e142a4 --- /dev/null +++ b/python/dao-voting-dashboard/config.py @@ -0,0 +1,46 @@ +""" +Configuration management for DAO Voting Dashboard. +""" + +import os +from pathlib import Path +from dotenv import load_dotenv + +# Load environment variables +load_dotenv() + + +class Config: + """Application configuration.""" + + # BNB Chain + RPC_URL: str = os.getenv("RPC_URL", "https://data-seed-prebsc-1-s1.bnbchain.org:8545") + GOVERNANCE_CONTRACT_ADDRESS: str = os.getenv( + "GOVERNANCE_CONTRACT_ADDRESS", + "0x1234567890123456789012345678901234567890" + ) + + # Database + DATABASE_PATH: str = os.getenv("DATABASE_PATH", "./data/dao_votes.db") + + # Server + HOST: str = os.getenv("HOST", "0.0.0.0") + PORT: int = int(os.getenv("PORT", "8000")) + + # Governance Rules + QUORUM_PERCENTAGE: int = int(os.getenv("QUORUM_PERCENTAGE", "20")) + APPROVAL_THRESHOLD_PERCENTAGE: int = int(os.getenv("APPROVAL_THRESHOLD_PERCENTAGE", "50")) + + # Demo Key (for testing only) + DEMO_PRIVATE_KEY: str = os.getenv( + "DEMO_PRIVATE_KEY", + "0xac0974bec39a17e36ba4a6b4d238ff944bacb478cbed5efcae784d7bf4f2ff80" + ) + + @classmethod + def ensure_data_dir(cls): + """Ensure data directory exists.""" + Path(cls.DATABASE_PATH).parent.mkdir(parents=True, exist_ok=True) + + +config = Config() diff --git a/python/dao-voting-dashboard/contract.py b/python/dao-voting-dashboard/contract.py new file mode 100644 index 0000000..4326abd --- /dev/null +++ b/python/dao-voting-dashboard/contract.py @@ -0,0 +1,151 @@ +""" +Mock governance smart contract interaction. +""" + +from web3 import Web3 +from typing import Dict, Optional +import json +from config import config + + +# Mock Governance Contract ABI +GOVERNANCE_ABI = [ + { + "inputs": [ + {"name": "_title", "type": "string"}, + {"name": "_description", "type": "string"}, + {"name": "_endTime", "type": "uint256"} + ], + "name": "createProposal", + "outputs": [{"name": "", "type": "uint256"}], + "stateMutability": "nonpayable", + "type": "function" + }, + { + "inputs": [{"name": "_proposalId", "type": "uint256"}], + "name": "getProposal", + "outputs": [ + {"name": "title", "type": "string"}, + {"name": "description", "type": "string"}, + {"name": "votesFor", "type": "uint256"}, + {"name": "votesAgainst", "type": "uint256"}, + {"name": "status", "type": "uint8"} + ], + "stateMutability": "view", + "type": "function" + }, + { + "inputs": [ + {"name": "_proposalId", "type": "uint256"}, + {"name": "_votesFor", "type": "uint256"}, + {"name": "_votesAgainst", "type": "uint256"} + ], + "name": "submitVotes", + "outputs": [{"name": "", "type": "bool"}], + "stateMutability": "nonpayable", + "type": "function" + }, + { + "inputs": [{"name": "_proposalId", "type": "uint256"}], + "name": "finalizeProposal", + "outputs": [{"name": "", "type": "bool"}], + "stateMutability": "nonpayable", + "type": "function" + } +] + + +class GovernanceContract: + """Mock governance contract interface.""" + + def __init__(self): + """Initialize contract connection.""" + self.w3 = Web3(Web3.HTTPProvider(config.RPC_URL)) + self.contract_address = config.GOVERNANCE_CONTRACT_ADDRESS + + # Mock contract storage (simulates on-chain state) + self._mock_proposals = {} + self._proposal_counter = 0 + + def create_proposal( + self, + title: str, + description: str, + end_time: int + ) -> int: + """ + Mock: Create a proposal on-chain. + + In production, this would call the actual smart contract. + """ + self._proposal_counter += 1 + proposal_id = self._proposal_counter + + self._mock_proposals[proposal_id] = { + "title": title, + "description": description, + "votesFor": 0, + "votesAgainst": 0, + "status": 1 # 0=pending, 1=active, 2=closed, 3=executed + } + + return proposal_id + + def get_proposal(self, proposal_id: int) -> Optional[Dict]: + """ + Mock: Get proposal details from chain. + + In production, this would read from the smart contract. + """ + if proposal_id in self._mock_proposals: + return self._mock_proposals[proposal_id] + return None + + def submit_votes( + self, + proposal_id: int, + votes_for: int, + votes_against: int + ) -> bool: + """ + Mock: Submit aggregated votes to chain. + + In production, this would be a contract transaction. + """ + if proposal_id in self._mock_proposals: + self._mock_proposals[proposal_id]["votesFor"] = votes_for + self._mock_proposals[proposal_id]["votesAgainst"] = votes_against + return True + return False + + def finalize_proposal(self, proposal_id: int) -> bool: + """ + Mock: Finalize a proposal on-chain. + + In production, this would execute the proposal if passed. + """ + if proposal_id in self._mock_proposals: + proposal = self._mock_proposals[proposal_id] + proposal["status"] = 2 # closed + + # Determine if passed + total_votes = proposal["votesFor"] + proposal["votesAgainst"] + if total_votes > 0: + approval_rate = proposal["votesFor"] / total_votes * 100 + if approval_rate >= config.APPROVAL_THRESHOLD_PERCENTAGE: + proposal["status"] = 3 # executed + + return True + return False + + def get_abi(self) -> list: + """Get contract ABI.""" + return GOVERNANCE_ABI + + def get_address(self) -> str: + """Get contract address.""" + return self.contract_address + + +# Global contract instance +governance_contract = GovernanceContract() diff --git a/python/dao-voting-dashboard/crypto_utils.py b/python/dao-voting-dashboard/crypto_utils.py new file mode 100644 index 0000000..cbd7c07 --- /dev/null +++ b/python/dao-voting-dashboard/crypto_utils.py @@ -0,0 +1,105 @@ +""" +Cryptographic utilities for vote signing and verification. +""" + +from eth_account import Account +from eth_account.messages import encode_defunct +from web3 import Web3 +from typing import Tuple + + +class VoteSignature: + """Handle vote signing and verification.""" + + @staticmethod + def sign_vote( + proposal_id: int, + vote_choice: str, + private_key: str + ) -> Tuple[str, str]: + """ + Sign a vote using a private key. + + Args: + proposal_id: ID of the proposal + vote_choice: "for", "against", or "abstain" + private_key: Hex-encoded private key + + Returns: + Tuple of (voter_address, signature) + """ + account = Account.from_key(private_key) + + # Create message to sign + message = f"Vote on proposal {proposal_id}: {vote_choice}" + encoded_message = encode_defunct(text=message) + + # Sign message + signed_message = account.sign_message(encoded_message) + signature = signed_message.signature.hex() + + return (account.address, signature) + + @staticmethod + def verify_vote( + proposal_id: int, + vote_choice: str, + voter_address: str, + signature: str + ) -> bool: + """ + Verify a vote signature. + + Args: + proposal_id: ID of the proposal + vote_choice: "for", "against", or "abstain" + voter_address: Address that supposedly signed + signature: Hex-encoded signature + + Returns: + True if signature is valid + """ + try: + # Reconstruct message + message = f"Vote on proposal {proposal_id}: {vote_choice}" + encoded_message = encode_defunct(text=message) + + # Recover signer + recovered_address = Account.recover_message( + encoded_message, + signature=bytes.fromhex(signature.replace('0x', '')) + ) + + # Check if recovered address matches + return recovered_address.lower() == voter_address.lower() + except Exception: + return False + + +class MerkleProof: + """Simple Merkle tree for vote batching (mock implementation).""" + + @staticmethod + def generate_root(votes: list) -> str: + """ + Generate a mock Merkle root for a list of votes. + + In production, this would be a proper Merkle tree implementation. + For this demo, we just hash all votes together. + """ + if not votes: + return "0x" + "0" * 64 + + # Simple concatenation and hash (not a real Merkle tree) + combined = "".join(str(v) for v in votes) + return Web3.keccak(text=combined).hex() + + @staticmethod + def generate_proof(vote_index: int, votes: list) -> list: + """ + Generate a mock Merkle proof for a vote. + + In production, this would generate actual Merkle proof hashes. + """ + # Mock proof - just return empty list for demo + return [] diff --git a/python/dao-voting-dashboard/database.py b/python/dao-voting-dashboard/database.py new file mode 100644 index 0000000..c84fc47 --- /dev/null +++ b/python/dao-voting-dashboard/database.py @@ -0,0 +1,212 @@ +""" +Database models and storage for DAO proposals and votes. +""" + +import sqlite3 +import json +from datetime import datetime +from typing import List, Optional, Dict +from pathlib import Path +from config import config + + +class Database: + """SQLite database manager for DAO voting.""" + + def __init__(self, db_path: str = None): + """Initialize database connection.""" + self.db_path = db_path or config.DATABASE_PATH + Path(self.db_path).parent.mkdir(parents=True, exist_ok=True) + self._init_db() + + def _init_db(self): + """Initialize database schema.""" + conn = sqlite3.connect(self.db_path) + cursor = conn.cursor() + + # Proposals table + cursor.execute(""" + CREATE TABLE IF NOT EXISTS proposals ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + title TEXT NOT NULL, + description TEXT NOT NULL, + creator TEXT NOT NULL, + created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + end_time TIMESTAMP NOT NULL, + status TEXT DEFAULT 'active', + on_chain_id INTEGER DEFAULT NULL, + UNIQUE(title) + ) + """) + + # Votes table + cursor.execute(""" + CREATE TABLE IF NOT EXISTS votes ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + proposal_id INTEGER NOT NULL, + voter_address TEXT NOT NULL, + vote_choice TEXT NOT NULL, + signature TEXT NOT NULL, + timestamp TIMESTAMP DEFAULT CURRENT_TIMESTAMP, + synced_to_chain BOOLEAN DEFAULT 0, + FOREIGN KEY (proposal_id) REFERENCES proposals(id), + UNIQUE(proposal_id, voter_address) + ) + """) + + conn.commit() + conn.close() + + def create_proposal( + self, + title: str, + description: str, + creator: str, + end_time: str + ) -> Optional[int]: + """Create a new proposal.""" + try: + conn = sqlite3.connect(self.db_path) + cursor = conn.cursor() + + cursor.execute(""" + INSERT INTO proposals (title, description, creator, end_time) + VALUES (?, ?, ?, ?) + """, (title, description, creator, end_time)) + + proposal_id = cursor.lastrowid + conn.commit() + conn.close() + + return proposal_id + except sqlite3.IntegrityError: + return None + + def get_proposal(self, proposal_id: int) -> Optional[Dict]: + """Get a proposal by ID.""" + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row + cursor = conn.cursor() + + cursor.execute(""" + SELECT * FROM proposals WHERE id = ? + """, (proposal_id,)) + + row = cursor.fetchone() + conn.close() + + if row: + return dict(row) + return None + + def list_proposals(self, status: str = None) -> List[Dict]: + """List all proposals, optionally filtered by status.""" + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row + cursor = conn.cursor() + + if status: + cursor.execute(""" + SELECT * FROM proposals WHERE status = ? ORDER BY created_at DESC + """, (status,)) + else: + cursor.execute(""" + SELECT * FROM proposals ORDER BY created_at DESC + """) + + rows = cursor.fetchall() + conn.close() + + return [dict(row) for row in rows] + + def close_proposal(self, proposal_id: int) -> bool: + """Close a proposal.""" + conn = sqlite3.connect(self.db_path) + cursor = conn.cursor() + + cursor.execute(""" + UPDATE proposals SET status = 'closed' WHERE id = ? + """, (proposal_id,)) + + success = cursor.rowcount > 0 + conn.commit() + conn.close() + + return success + + def create_vote( + self, + proposal_id: int, + voter_address: str, + vote_choice: str, + signature: str + ) -> Optional[int]: + """Record a vote.""" + try: + conn = sqlite3.connect(self.db_path) + cursor = conn.cursor() + + cursor.execute(""" + INSERT INTO votes (proposal_id, voter_address, vote_choice, signature) + VALUES (?, ?, ?, ?) + """, (proposal_id, voter_address, vote_choice, signature)) + + vote_id = cursor.lastrowid + conn.commit() + conn.close() + + return vote_id + except sqlite3.IntegrityError: + return None + + def get_votes(self, proposal_id: int) -> List[Dict]: + """Get all votes for a proposal.""" + conn = sqlite3.connect(self.db_path) + conn.row_factory = sqlite3.Row + cursor = conn.cursor() + + cursor.execute(""" + SELECT * FROM votes WHERE proposal_id = ? ORDER BY timestamp DESC + """, (proposal_id,)) + + rows = cursor.fetchall() + conn.close() + + return [dict(row) for row in rows] + + def get_vote_counts(self, proposal_id: int) -> Dict[str, int]: + """Get vote counts for a proposal.""" + conn = sqlite3.connect(self.db_path) + cursor = conn.cursor() + + cursor.execute(""" + SELECT vote_choice, COUNT(*) as count + FROM votes + WHERE proposal_id = ? + GROUP BY vote_choice + """, (proposal_id,)) + + results = cursor.fetchall() + conn.close() + + counts = {"for": 0, "against": 0, "abstain": 0} + for choice, count in results: + counts[choice] = count + + return counts + + def mark_votes_synced(self, proposal_id: int): + """Mark all votes for a proposal as synced to chain.""" + conn = sqlite3.connect(self.db_path) + cursor = conn.cursor() + + cursor.execute(""" + UPDATE votes SET synced_to_chain = 1 WHERE proposal_id = ? + """, (proposal_id,)) + + conn.commit() + conn.close() + + +# Global database instance +db = Database() diff --git a/python/dao-voting-dashboard/governance.py b/python/dao-voting-dashboard/governance.py new file mode 100644 index 0000000..4ad5a84 --- /dev/null +++ b/python/dao-voting-dashboard/governance.py @@ -0,0 +1,168 @@ +""" +Governance rules and logic. +""" + +from typing import Dict +from config import config + + +class GovernanceRules: + """DAO governance rules and calculations.""" + + @staticmethod + def calculate_quorum(total_eligible_voters: int) -> int: + """ + Calculate minimum votes needed for quorum. + + Args: + total_eligible_voters: Total number of eligible voters + + Returns: + Minimum votes needed + """ + return int(total_eligible_voters * config.QUORUM_PERCENTAGE / 100) + + @staticmethod + def check_quorum_met( + total_votes: int, + total_eligible_voters: int + ) -> bool: + """ + Check if quorum requirement is met. + + Args: + total_votes: Total votes cast + total_eligible_voters: Total eligible voters + + Returns: + True if quorum is met + """ + required = GovernanceRules.calculate_quorum(total_eligible_voters) + return total_votes >= required + + @staticmethod + def calculate_approval_rate(votes_for: int, votes_against: int) -> float: + """ + Calculate approval percentage. + + Args: + votes_for: Number of yes votes + votes_against: Number of no votes + + Returns: + Approval rate as percentage (0-100) + """ + total = votes_for + votes_against + if total == 0: + return 0.0 + return (votes_for / total) * 100 + + @staticmethod + def check_approval_met(votes_for: int, votes_against: int) -> bool: + """ + Check if approval threshold is met. + + Args: + votes_for: Number of yes votes + votes_against: Number of no votes + + Returns: + True if approval threshold is met + """ + approval_rate = GovernanceRules.calculate_approval_rate( + votes_for, votes_against + ) + return approval_rate >= config.APPROVAL_THRESHOLD_PERCENTAGE + + @staticmethod + def determine_outcome( + votes_for: int, + votes_against: int, + total_eligible_voters: int + ) -> Dict[str, any]: + """ + Determine proposal outcome based on governance rules. + + Args: + votes_for: Number of yes votes + votes_against: Number of no votes + total_eligible_voters: Total eligible voters + + Returns: + Dict with outcome details + """ + total_votes = votes_for + votes_against + quorum_met = GovernanceRules.check_quorum_met( + total_votes, total_eligible_voters + ) + approval_met = GovernanceRules.check_approval_met( + votes_for, votes_against + ) + approval_rate = GovernanceRules.calculate_approval_rate( + votes_for, votes_against + ) + + # Determine final status + if not quorum_met: + status = "rejected" + reason = "Quorum not met" + elif approval_met: + status = "passed" + reason = "Approval threshold met" + else: + status = "rejected" + reason = "Approval threshold not met" + + return { + "status": status, + "reason": reason, + "quorum_met": quorum_met, + "approval_met": approval_met, + "approval_rate": approval_rate, + "total_votes": total_votes, + "votes_for": votes_for, + "votes_against": votes_against, + "required_quorum": GovernanceRules.calculate_quorum( + total_eligible_voters + ), + "required_approval": config.APPROVAL_THRESHOLD_PERCENTAGE + } + + @staticmethod + def is_eligible_voter(address: str) -> bool: + """ + Check if address is eligible to vote. + + In production, this would check token balance or NFT ownership. + Mock implementation accepts any valid address. + + Args: + address: Ethereum address + + Returns: + True if eligible + """ + # Simple validation: address must be valid format + if not address or len(address) != 42: + return False + if not address.startswith("0x"): + return False + return True + + @staticmethod + def get_voting_power(address: str) -> int: + """ + Get voting power for address. + + In production, this would query token balance. + Mock implementation returns 1 vote per address. + + Args: + address: Ethereum address + + Returns: + Number of votes (voting power) + """ + if GovernanceRules.is_eligible_voter(address): + return 1 + return 0 diff --git a/python/dao-voting-dashboard/pyproject.toml b/python/dao-voting-dashboard/pyproject.toml new file mode 100644 index 0000000..37d4aeb --- /dev/null +++ b/python/dao-voting-dashboard/pyproject.toml @@ -0,0 +1,35 @@ +[project] +name = "dao-voting-dashboard" +version = "0.1.0" +description = "Off-chain + on-chain DAO voting dashboard for BNB Chain" +requires-python = ">=3.12" +readme = "README.md" +dependencies = [ + "fastapi>=0.115.0", + "uvicorn>=0.32.0", + "web3>=7.0.0", + "eth-account>=0.13.0", + "python-dotenv>=1.0.0", + "jinja2>=3.1.0", + "pydantic>=2.0.0", + "httpx>=0.27.0", +] + +[project.optional-dependencies] +dev = [ + "pytest>=8.0.0", + "pytest-cov>=4.1.0", + "pytest-asyncio>=0.23.0", +] + +[build-system] +requires = ["hatchling"] +build-backend = "hatchling.build" + +[tool.pytest.ini_options] +testpaths = ["tests"] +python_files = ["test_*.py"] +python_classes = ["Test*"] +python_functions = ["test_*"] +asyncio_mode = "auto" +addopts = "-v --tb=short" diff --git a/python/dao-voting-dashboard/templates/index.html b/python/dao-voting-dashboard/templates/index.html new file mode 100644 index 0000000..1d64a8d --- /dev/null +++ b/python/dao-voting-dashboard/templates/index.html @@ -0,0 +1,339 @@ + + + + + + DAO Voting Dashboard + + + +
+
+

πŸ—³οΈ DAO Voting Dashboard

+

Decentralized governance for your community

+
+ +
+

Create New Proposal

+
+
+ + +
+ +
+ + +
+ +
+ + +
+ +
+ + +
+ + +
+
+ + + +
+

Active Proposals

+
+ {% if proposals %} + {% for proposal in proposals %} +
+
+

{{ proposal.title }}

+ {{ proposal.status }} +
+ +

{{ proposal.description }}

+ +
+ πŸ“… Created: {{ proposal.created_at }} + ⏰ Ends: {{ proposal.end_time }} + πŸ‘€ Creator: {{ proposal.creator[:10] }}... +
+ +
+ View Details + View Results +
+
+ {% endfor %} + {% else %} +
+

No proposals yet. Create the first one above! πŸš€

+
+ {% endif %} +
+
+
+ + + + diff --git a/python/dao-voting-dashboard/templates/proposal.html b/python/dao-voting-dashboard/templates/proposal.html new file mode 100644 index 0000000..b5b06cc --- /dev/null +++ b/python/dao-voting-dashboard/templates/proposal.html @@ -0,0 +1,419 @@ + + + + + + {{ proposal.title }} - DAO Voting + + + +
+ ← Back to Dashboard + +
+
+

{{ proposal.title }}

+ {{ proposal.status }} +
+ +
+
+ Creator + {{ proposal.creator }} +
+
+ Created + {{ proposal.created_at }} +
+
+ Ends + {{ proposal.end_time }} +
+
+ On-Chain ID + {{ proposal.on_chain_id or "Pending" }} +
+
+ +

{{ proposal.description }}

+ +
+

πŸ“Š Current Votes

+
+ Votes For + {{ vote_counts.for }} +
+
+ Votes Against + {{ vote_counts.against }} +
+
+ Total Votes + {{ vote_counts.total }} +
+
+
+ + {% if proposal.status == 'active' %} +
+

Cast Your Vote

+ + +
+
+ + +
+ +
+ + +
+ +
+ + +
+
+
+ {% endif %} + +
+

All Votes ({{ votes|length }})

+ {% if votes %} + {% for vote in votes %} +
+
+ {{ vote.voter_address }} +
+
+ Voted {{ vote.vote_choice }} β€’ {{ vote.timestamp }} β€’ + {% if vote.synced_to_chain %}Synced to chain{% else %}Not synced{% endif %} +
+
+ {% endfor %} + {% else %} +
+

No votes cast yet. Be the first! πŸ—³οΈ

+
+ {% endif %} +
+
+ + + + diff --git a/python/dao-voting-dashboard/templates/results.html b/python/dao-voting-dashboard/templates/results.html new file mode 100644 index 0000000..3fde1ec --- /dev/null +++ b/python/dao-voting-dashboard/templates/results.html @@ -0,0 +1,301 @@ + + + + + + Results - {{ proposal.title }} + + + +
+ ← Back to Dashboard + +
+

πŸ“Š Voting Results

+ +
+ {% if outcome.status == 'passed' %} + βœ“ PROPOSAL PASSED + {% else %} + βœ— PROPOSAL REJECTED + {% endif %} +
+ +
+

{{ proposal.title }}

+

{{ proposal.description }}

+
+ +
+
+ For: {{ outcome.votes_for }} + Against: {{ outcome.votes_against }} +
+
+ {% if outcome.total_votes > 0 %} +
+ {{ "%.1f"|format(outcome.votes_for / outcome.total_votes * 100) }}% +
+ {% else %} +
No votes yet
+ {% endif %} +
+
+ +
+
+
Total Votes
+
{{ outcome.total_votes }}
+
+ +
+
Approval Rate
+
{{ "%.1f"|format(outcome.approval_rate) }}%
+
+ +
+
Votes For
+
{{ outcome.votes_for }}
+
+ +
+
Votes Against
+
{{ outcome.votes_against }}
+
+
+ +
+

βš™οΈ Governance Rules

+ +
+ Quorum Requirement + + {{ outcome.required_quorum }} votes ({{ outcome.total_votes }}/{{ outcome.required_quorum }}) + {% if outcome.quorum_met %} + βœ“ + {% else %} + βœ— + {% endif %} + +
+ +
+ Approval Threshold + + {{ outcome.required_approval }}% ({{ "%.1f"|format(outcome.approval_rate) }}%) + {% if outcome.approval_met %} + βœ“ + {% else %} + βœ— + {% endif %} + +
+ +
+ Final Status + {{ outcome.status.upper() }} +
+ +
+ Reason + {{ outcome.reason }} +
+
+
+
+ + diff --git a/python/dao-voting-dashboard/tests/__init__.py b/python/dao-voting-dashboard/tests/__init__.py new file mode 100644 index 0000000..fae6326 --- /dev/null +++ b/python/dao-voting-dashboard/tests/__init__.py @@ -0,0 +1 @@ +"""Test package initialization.""" diff --git a/python/dao-voting-dashboard/tests/test_api.py b/python/dao-voting-dashboard/tests/test_api.py new file mode 100644 index 0000000..476953e --- /dev/null +++ b/python/dao-voting-dashboard/tests/test_api.py @@ -0,0 +1,326 @@ +""" +Tests for FastAPI endpoints. +""" + +import pytest +from fastapi.testclient import TestClient +from datetime import datetime, timedelta +import sys +from pathlib import Path + +# Add parent directory to path to import modules +parent_dir = str(Path(__file__).parent.parent) +if parent_dir not in sys.path: + sys.path.insert(0, parent_dir) + +# Import from root-level modules (not app directory) +from database import Database + +# Import the FastAPI app from app.py file +import importlib.util +spec = importlib.util.spec_from_file_location("app_module", Path(__file__).parent.parent / "app.py") +app_module = importlib.util.module_from_spec(spec) +spec.loader.exec_module(app_module) +app = app_module.app + + +@pytest.fixture +def client(): + """Create a test client.""" + return TestClient(app) + + +def test_health_check(client): + """Test health check endpoint.""" + response = client.get("/health") + assert response.status_code == 200 + data = response.json() + assert data["status"] == "healthy" + assert "timestamp" in data + + +def test_create_proposal_success(client): + """Test creating a proposal via API.""" + payload = { + "title": "Test Proposal", + "description": "This is a test proposal", + "creator": "0x1234567890123456789012345678901234567890", + "duration_hours": 168 + } + + response = client.post("/api/proposals", json=payload) + assert response.status_code == 200 + + data = response.json() + assert data["success"] is True + assert "proposal_id" in data + assert "on_chain_id" in data + + +def test_create_proposal_missing_fields(client): + """Test creating proposal with missing fields.""" + payload = { + "title": "Test Proposal" + # Missing required fields + } + + response = client.post("/api/proposals", json=payload) + assert response.status_code == 422 # Validation error + + +def test_list_proposals(client): + """Test listing proposals.""" + # Create a proposal first + payload = { + "title": "Test Proposal", + "description": "Description", + "creator": "0x1234567890123456789012345678901234567890", + "duration_hours": 168 + } + client.post("/api/proposals", json=payload) + + # List proposals + response = client.get("/api/proposals") + assert response.status_code == 200 + + data = response.json() + assert "proposals" in data + assert len(data["proposals"]) > 0 + + +def test_list_proposals_filter_by_status(client): + """Test filtering proposals by status.""" + response = client.get("/api/proposals?status=active") + assert response.status_code == 200 + + data = response.json() + assert "proposals" in data + + +def test_get_proposal_details(client): + """Test getting proposal details.""" + # Create proposal + payload = { + "title": "Test Proposal", + "description": "Description", + "creator": "0x1234567890123456789012345678901234567890", + "duration_hours": 168 + } + create_response = client.post("/api/proposals", json=payload) + proposal_id = create_response.json()["proposal_id"] + + # Get details + response = client.get(f"/api/proposals/{proposal_id}") + assert response.status_code == 200 + + data = response.json() + assert "proposal" in data + assert "vote_counts" in data + assert "votes" in data + + +def test_get_nonexistent_proposal(client): + """Test getting non-existent proposal.""" + response = client.get("/api/proposals/999") + assert response.status_code == 404 + + +def test_cast_vote_success(client): + """Test casting a vote.""" + # Create proposal + create_payload = { + "title": "Test Proposal", + "description": "Description", + "creator": "0x1234567890123456789012345678901234567890", + "duration_hours": 168 + } + create_response = client.post("/api/proposals", json=create_payload) + proposal_id = create_response.json()["proposal_id"] + + # Cast vote + vote_payload = { + "voter_address": "0xabcdef0123456789012345678901234567890abc", + "vote_choice": "for", + "private_key": "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + } + + response = client.post( + f"/api/proposals/{proposal_id}/vote", + json=vote_payload + ) + assert response.status_code == 200 + + data = response.json() + assert data["success"] is True + assert "vote_id" in data + + +def test_cast_vote_invalid_choice(client): + """Test casting vote with invalid choice.""" + # Create proposal + create_payload = { + "title": "Test Proposal", + "description": "Description", + "creator": "0x1234567890123456789012345678901234567890", + "duration_hours": 168 + } + create_response = client.post("/api/proposals", json=create_payload) + proposal_id = create_response.json()["proposal_id"] + + # Cast invalid vote + vote_payload = { + "voter_address": "0xabcdef0123456789012345678901234567890abc", + "vote_choice": "maybe", # Invalid + "private_key": "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + } + + response = client.post( + f"/api/proposals/{proposal_id}/vote", + json=vote_payload + ) + assert response.status_code == 400 + + +def test_cast_vote_duplicate(client): + """Test casting duplicate vote from same address.""" + # Create proposal + create_payload = { + "title": "Test Proposal", + "description": "Description", + "creator": "0x1234567890123456789012345678901234567890", + "duration_hours": 168 + } + create_response = client.post("/api/proposals", json=create_payload) + proposal_id = create_response.json()["proposal_id"] + + # Cast first vote + vote_payload = { + "voter_address": "0xabcdef0123456789012345678901234567890abc", + "vote_choice": "for", + "private_key": "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + } + client.post(f"/api/proposals/{proposal_id}/vote", json=vote_payload) + + # Cast duplicate vote + response = client.post( + f"/api/proposals/{proposal_id}/vote", + json=vote_payload + ) + assert response.status_code == 400 + + +def test_get_results(client): + """Test getting proposal results.""" + # Create proposal + create_payload = { + "title": "Test Proposal", + "description": "Description", + "creator": "0x1234567890123456789012345678901234567890", + "duration_hours": 168 + } + create_response = client.post("/api/proposals", json=create_payload) + proposal_id = create_response.json()["proposal_id"] + + # Get results + response = client.get(f"/api/proposals/{proposal_id}/results") + assert response.status_code == 200 + + data = response.json() + assert "proposal" in data + assert "outcome" in data + assert "status" in data["outcome"] + + +def test_sync_to_chain(client): + """Test syncing votes to chain.""" + # Create proposal and cast votes + create_payload = { + "title": "Test Proposal", + "description": "Description", + "creator": "0x1234567890123456789012345678901234567890", + "duration_hours": 168 + } + create_response = client.post("/api/proposals", json=create_payload) + proposal_id = create_response.json()["proposal_id"] + + # Cast a vote + vote_payload = { + "voter_address": "0xabcdef0123456789012345678901234567890abc", + "vote_choice": "for", + "private_key": "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + } + client.post(f"/api/proposals/{proposal_id}/vote", json=vote_payload) + + # Sync to chain + response = client.post(f"/api/proposals/{proposal_id}/sync") + assert response.status_code == 200 + + data = response.json() + assert data["success"] is True + assert "votes_synced" in data + + +def test_close_proposal(client): + """Test closing a proposal.""" + # Create proposal + create_payload = { + "title": "Test Proposal", + "description": "Description", + "creator": "0x1234567890123456789012345678901234567890", + "duration_hours": 168 + } + create_response = client.post("/api/proposals", json=create_payload) + proposal_id = create_response.json()["proposal_id"] + + # Close proposal + response = client.post(f"/api/proposals/{proposal_id}/close") + assert response.status_code == 200 + + data = response.json() + assert data["success"] is True + assert "outcome" in data + + +def test_close_already_closed_proposal(client): + """Test closing an already closed proposal.""" + # Create and close proposal + create_payload = { + "title": "Test Proposal", + "description": "Description", + "creator": "0x1234567890123456789012345678901234567890", + "duration_hours": 168 + } + create_response = client.post("/api/proposals", json=create_payload) + proposal_id = create_response.json()["proposal_id"] + client.post(f"/api/proposals/{proposal_id}/close") + + # Try to close again + response = client.post(f"/api/proposals/{proposal_id}/close") + assert response.status_code == 400 + + +def test_vote_on_closed_proposal(client): + """Test voting on closed proposal.""" + # Create and close proposal + create_payload = { + "title": "Test Proposal", + "description": "Description", + "creator": "0x1234567890123456789012345678901234567890", + "duration_hours": 168 + } + create_response = client.post("/api/proposals", json=create_payload) + proposal_id = create_response.json()["proposal_id"] + client.post(f"/api/proposals/{proposal_id}/close") + + # Try to vote + vote_payload = { + "voter_address": "0xabcdef0123456789012345678901234567890abc", + "vote_choice": "for", + "private_key": "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + } + + response = client.post( + f"/api/proposals/{proposal_id}/vote", + json=vote_payload + ) + assert response.status_code == 400 diff --git a/python/dao-voting-dashboard/tests/test_blockchain.py b/python/dao-voting-dashboard/tests/test_blockchain.py new file mode 100644 index 0000000..fb78d39 --- /dev/null +++ b/python/dao-voting-dashboard/tests/test_blockchain.py @@ -0,0 +1,81 @@ +"""Tests for blockchain client.""" + +import pytest +from unittest.mock import Mock, patch +from app.blockchain import blockchain +from eth_account import Account + + +def test_sign_and_verify_message(): + """Test signing and verifying a message.""" + # Create a test account + account = Account.create() + private_key = account.key.hex() + address = account.address + message = "Test message" + + # Sign the message + signature = blockchain.sign_message(private_key, message) + + # Verify the signature + assert blockchain.verify_signature(message, signature, address) + + +def test_verify_invalid_signature(): + """Test verifying an invalid signature.""" + message = "Test message" + signature = "0xinvalidsignature" + address = "0x0000000000000000000000000000000000000000" + + assert not blockchain.verify_signature(message, signature, address) + + +def test_is_connected(): + """Test checking blockchain connection.""" + # Should be mocked/connected for testing + assert blockchain.is_connected() + + +def test_get_chain_id(): + """Test getting chain ID.""" + chain_id = blockchain.get_chain_id() + assert chain_id == 97 # BNB testnet + + +def test_submit_proposal_to_chain(): + """Test submitting proposal to chain (mocked).""" + with patch.object(blockchain, 'submit_proposal_to_chain', return_value=1): + on_chain_id = blockchain.submit_proposal_to_chain( + title="Test Proposal", + description="Test Description", + end_time=1234567890, + private_key=None + ) + assert on_chain_id == 1 + + +def test_get_proposal_from_chain(): + """Test getting proposal from chain (mocked).""" + mock_proposal = { + "title": "Test Proposal", + "description": "Test Description", + "for_votes": 10, + "against_votes": 5, + "status": 1 + } + + with patch.object(blockchain, 'get_proposal_from_chain', return_value=mock_proposal): + proposal = blockchain.get_proposal_from_chain(1) + assert proposal["title"] == "Test Proposal" + assert proposal["for_votes"] == 10 + + +def test_submit_vote_to_chain(): + """Test submitting vote to chain (mocked).""" + with patch.object(blockchain, 'submit_vote_to_chain', return_value=True): + result = blockchain.submit_vote_to_chain( + proposal_id=1, + vote_choice="for", + private_key=None + ) + assert result is True diff --git a/python/dao-voting-dashboard/tests/test_contract.py b/python/dao-voting-dashboard/tests/test_contract.py new file mode 100644 index 0000000..788c0f9 --- /dev/null +++ b/python/dao-voting-dashboard/tests/test_contract.py @@ -0,0 +1,212 @@ +""" +Tests for smart contract interactions. +""" + +import pytest +from contract import GovernanceContract + + +@pytest.fixture +def contract(): + """Create a contract instance for testing.""" + return GovernanceContract() + + +def test_contract_initialization(contract): + """Test contract initialization.""" + assert contract.w3 is not None + assert contract.contract_address is not None + assert len(contract._mock_proposals) == 0 + + +def test_create_proposal(contract): + """Test creating a proposal on-chain (mock).""" + proposal_id = contract.create_proposal( + title="Test Proposal", + description="Test description", + end_time=1234567890 + ) + + assert proposal_id is not None + assert isinstance(proposal_id, int) + assert proposal_id > 0 + + +def test_get_proposal(contract): + """Test retrieving a proposal from chain (mock).""" + # Create proposal + proposal_id = contract.create_proposal( + "Test", "Description", 1234567890 + ) + + # Get proposal + proposal = contract.get_proposal(proposal_id) + + assert proposal is not None + assert proposal["title"] == "Test" + assert proposal["description"] == "Description" + assert proposal["votesFor"] == 0 + assert proposal["votesAgainst"] == 0 + assert proposal["status"] == 1 # active + + +def test_get_nonexistent_proposal(contract): + """Test getting non-existent proposal.""" + proposal = contract.get_proposal(999) + assert proposal is None + + +def test_submit_votes(contract): + """Test submitting votes to chain (mock).""" + # Create proposal + proposal_id = contract.create_proposal( + "Test", "Description", 1234567890 + ) + + # Submit votes + success = contract.submit_votes(proposal_id, 10, 5) + assert success is True + + # Verify votes were recorded + proposal = contract.get_proposal(proposal_id) + assert proposal["votesFor"] == 10 + assert proposal["votesAgainst"] == 5 + + +def test_submit_votes_nonexistent_proposal(contract): + """Test submitting votes to non-existent proposal.""" + success = contract.submit_votes(999, 10, 5) + assert success is False + + +def test_finalize_proposal(contract): + """Test finalizing a proposal (mock).""" + # Create and finalize proposal + proposal_id = contract.create_proposal( + "Test", "Description", 1234567890 + ) + contract.submit_votes(proposal_id, 10, 5) + + success = contract.finalize_proposal(proposal_id) + assert success is True + + # Check status changed + proposal = contract.get_proposal(proposal_id) + assert proposal["status"] == 3 # executed (passed) + + +def test_finalize_rejected_proposal(contract): + """Test finalizing a rejected proposal.""" + # Create proposal with more against votes + proposal_id = contract.create_proposal( + "Test", "Description", 1234567890 + ) + contract.submit_votes(proposal_id, 5, 10) + + success = contract.finalize_proposal(proposal_id) + assert success is True + + # Check status is closed (not executed) + proposal = contract.get_proposal(proposal_id) + assert proposal["status"] == 2 # closed + + +def test_finalize_nonexistent_proposal(contract): + """Test finalizing non-existent proposal.""" + success = contract.finalize_proposal(999) + assert success is False + + +def test_get_abi(contract): + """Test getting contract ABI.""" + abi = contract.get_abi() + assert abi is not None + assert isinstance(abi, list) + assert len(abi) > 0 + + +def test_get_address(contract): + """Test getting contract address.""" + address = contract.get_address() + assert address is not None + assert isinstance(address, str) + assert address.startswith("0x") + + +def test_multiple_proposals(contract): + """Test creating multiple proposals.""" + id1 = contract.create_proposal("Proposal 1", "Desc 1", 1111111111) + id2 = contract.create_proposal("Proposal 2", "Desc 2", 2222222222) + id3 = contract.create_proposal("Proposal 3", "Desc 3", 3333333333) + + assert id1 != id2 + assert id2 != id3 + assert id1 < id2 < id3 + + # Verify all proposals exist + assert contract.get_proposal(id1) is not None + assert contract.get_proposal(id2) is not None + assert contract.get_proposal(id3) is not None + + +def test_proposal_independence(contract): + """Test that proposals are independent.""" + id1 = contract.create_proposal("Proposal 1", "Desc 1", 1111111111) + id2 = contract.create_proposal("Proposal 2", "Desc 2", 2222222222) + + # Submit different votes + contract.submit_votes(id1, 10, 5) + contract.submit_votes(id2, 3, 7) + + # Verify votes are independent + p1 = contract.get_proposal(id1) + p2 = contract.get_proposal(id2) + + assert p1["votesFor"] == 10 + assert p1["votesAgainst"] == 5 + assert p2["votesFor"] == 3 + assert p2["votesAgainst"] == 7 + + +def test_vote_update(contract): + """Test updating votes on a proposal.""" + proposal_id = contract.create_proposal( + "Test", "Description", 1234567890 + ) + + # Submit initial votes + contract.submit_votes(proposal_id, 5, 3) + proposal = contract.get_proposal(proposal_id) + assert proposal["votesFor"] == 5 + + # Update votes + contract.submit_votes(proposal_id, 10, 7) + proposal = contract.get_proposal(proposal_id) + assert proposal["votesFor"] == 10 + assert proposal["votesAgainst"] == 7 + + +def test_finalize_no_votes(contract): + """Test finalizing proposal with no votes.""" + proposal_id = contract.create_proposal( + "Test", "Description", 1234567890 + ) + + success = contract.finalize_proposal(proposal_id) + assert success is True + + # Should be closed (no votes = no approval) + proposal = contract.get_proposal(proposal_id) + assert proposal["status"] == 2 + + +def test_proposal_initial_state(contract): + """Test that new proposals have correct initial state.""" + proposal_id = contract.create_proposal( + "Test", "Description", 1234567890 + ) + + proposal = contract.get_proposal(proposal_id) + assert proposal["votesFor"] == 0 + assert proposal["votesAgainst"] == 0 + assert proposal["status"] == 1 # active diff --git a/python/dao-voting-dashboard/tests/test_crypto.py b/python/dao-voting-dashboard/tests/test_crypto.py new file mode 100644 index 0000000..f5b1ae2 --- /dev/null +++ b/python/dao-voting-dashboard/tests/test_crypto.py @@ -0,0 +1,188 @@ +""" +Tests for cryptographic utilities. +""" + +import pytest +from crypto_utils import VoteSignature, MerkleProof + + +def test_sign_vote(): + """Test vote signing.""" + private_key = "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + message = "Vote for on proposal 1" + + address, signature = VoteSignature.sign_vote(private_key, message) + + assert address is not None + assert signature is not None + assert address.startswith("0x") + assert len(address) == 42 + assert signature.startswith("0x") + + +def test_verify_vote_valid(): + """Test verifying a valid vote signature.""" + private_key = "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + message = "Vote for on proposal 1" + + address, signature = VoteSignature.sign_vote(private_key, message) + + # Verify with correct parameters + is_valid = VoteSignature.verify_vote(message, signature, address) + assert is_valid is True + + +def test_verify_vote_wrong_message(): + """Test verifying with wrong message fails.""" + private_key = "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + message = "Vote for on proposal 1" + wrong_message = "Vote against on proposal 1" + + address, signature = VoteSignature.sign_vote(private_key, message) + + # Verify with wrong message + is_valid = VoteSignature.verify_vote(wrong_message, signature, address) + assert is_valid is False + + +def test_verify_vote_wrong_address(): + """Test verifying with wrong address fails.""" + private_key = "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + message = "Vote for on proposal 1" + wrong_address = "0x0000000000000000000000000000000000000000" + + address, signature = VoteSignature.sign_vote(private_key, message) + + # Verify with wrong address + is_valid = VoteSignature.verify_vote(message, signature, wrong_address) + assert is_valid is False + + +def test_sign_vote_different_messages(): + """Test that different messages produce different signatures.""" + private_key = "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + message1 = "Vote for on proposal 1" + message2 = "Vote against on proposal 1" + + address1, signature1 = VoteSignature.sign_vote(private_key, message1) + address2, signature2 = VoteSignature.sign_vote(private_key, message2) + + # Same address, different signatures + assert address1 == address2 + assert signature1 != signature2 + + +def test_sign_vote_different_keys(): + """Test that different keys produce different addresses.""" + private_key1 = "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + private_key2 = "0xfedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210" + message = "Vote for on proposal 1" + + address1, signature1 = VoteSignature.sign_vote(private_key1, message) + address2, signature2 = VoteSignature.sign_vote(private_key2, message) + + # Different addresses and signatures + assert address1 != address2 + assert signature1 != signature2 + + +def test_merkle_proof_generation(): + """Test Merkle proof generation (mock).""" + votes = [ + {"voter": "0xvoter1", "choice": "for"}, + {"voter": "0xvoter2", "choice": "against"}, + {"voter": "0xvoter3", "choice": "for"}, + ] + + proof = MerkleProof.generate_proof(votes, 0) + + assert proof is not None + assert isinstance(proof, list) + assert len(proof) > 0 + assert all(isinstance(h, str) for h in proof) + assert all(h.startswith("0x") for h in proof) + + +def test_merkle_root_generation(): + """Test Merkle root generation (mock).""" + votes = [ + {"voter": "0xvoter1", "choice": "for"}, + {"voter": "0xvoter2", "choice": "against"}, + ] + + root = MerkleProof.generate_root(votes) + + assert root is not None + assert isinstance(root, str) + assert root.startswith("0x") + assert len(root) == 66 # 0x + 64 hex chars + + +def test_merkle_root_empty_votes(): + """Test Merkle root with no votes.""" + votes = [] + root = MerkleProof.generate_root(votes) + + assert root is not None + assert isinstance(root, str) + + +def test_merkle_proof_different_indices(): + """Test that different vote indices produce different proofs.""" + votes = [ + {"voter": "0xvoter1", "choice": "for"}, + {"voter": "0xvoter2", "choice": "against"}, + {"voter": "0xvoter3", "choice": "for"}, + ] + + proof1 = MerkleProof.generate_proof(votes, 0) + proof2 = MerkleProof.generate_proof(votes, 1) + + # Proofs should be different for different indices + assert proof1 != proof2 + + +def test_vote_signature_consistency(): + """Test that signing the same message produces consistent results.""" + private_key = "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + message = "Vote for on proposal 1" + + address1, signature1 = VoteSignature.sign_vote(private_key, message) + address2, signature2 = VoteSignature.sign_vote(private_key, message) + + # Same key and message should produce same address and signature + assert address1 == address2 + assert signature1 == signature2 + + +def test_verify_vote_case_insensitive_address(): + """Test that address verification is case-insensitive.""" + private_key = "0x0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef" + message = "Vote for on proposal 1" + + address, signature = VoteSignature.sign_vote(private_key, message) + + # Verify with different case + is_valid_lower = VoteSignature.verify_vote( + message, signature, address.lower() + ) + is_valid_upper = VoteSignature.verify_vote( + message, signature, address.upper() + ) + + assert is_valid_lower is True + assert is_valid_upper is True + + +def test_merkle_root_deterministic(): + """Test that Merkle root generation is deterministic.""" + votes = [ + {"voter": "0xvoter1", "choice": "for"}, + {"voter": "0xvoter2", "choice": "against"}, + ] + + root1 = MerkleProof.generate_root(votes) + root2 = MerkleProof.generate_root(votes) + + # Same votes should produce same root + assert root1 == root2 diff --git a/python/dao-voting-dashboard/tests/test_database.py b/python/dao-voting-dashboard/tests/test_database.py new file mode 100644 index 0000000..052d345 --- /dev/null +++ b/python/dao-voting-dashboard/tests/test_database.py @@ -0,0 +1,283 @@ +""" +Tests for database operations. +""" + +import pytest +from datetime import datetime, timedelta +import os +import tempfile +from pathlib import Path + +from database import Database + + +@pytest.fixture +def temp_db(): + """Create a temporary database for testing.""" + temp_dir = tempfile.mkdtemp() + temp_db_path = Path(temp_dir) / "test.db" + + # Temporarily override the data directory + original_data_dir = Database._get_data_dir() + Database._test_data_dir = temp_dir + + db = Database() + yield db + + # Cleanup + Database._test_data_dir = None + if temp_db_path.exists(): + temp_db_path.unlink() + + +@pytest.fixture +def db_with_proposal(temp_db): + """Create a database with a sample proposal.""" + end_time = datetime.now() + timedelta(days=7) + proposal_id = temp_db.create_proposal( + title="Test Proposal", + description="This is a test proposal", + creator="0x1234567890123456789012345678901234567890", + end_time=end_time + ) + return temp_db, proposal_id + + +def test_database_initialization(temp_db): + """Test database initialization and schema creation.""" + assert temp_db.conn is not None + + # Check tables exist + cursor = temp_db.conn.execute( + "SELECT name FROM sqlite_master WHERE type='table'" + ) + tables = [row[0] for row in cursor.fetchall()] + assert "proposals" in tables + assert "votes" in tables + + +def test_create_proposal(temp_db): + """Test creating a proposal.""" + end_time = datetime.now() + timedelta(days=7) + proposal_id = temp_db.create_proposal( + title="Fund Community Event", + description="Allocate 1000 tokens for community meetup", + creator="0x1234567890123456789012345678901234567890", + end_time=end_time + ) + + assert proposal_id is not None + assert isinstance(proposal_id, int) + assert proposal_id > 0 + + +def test_get_proposal(db_with_proposal): + """Test retrieving a proposal.""" + db, proposal_id = db_with_proposal + proposal = db.get_proposal(proposal_id) + + assert proposal is not None + assert proposal["id"] == proposal_id + assert proposal["title"] == "Test Proposal" + assert proposal["description"] == "This is a test proposal" + assert proposal["status"] == "active" + + +def test_get_nonexistent_proposal(temp_db): + """Test retrieving a non-existent proposal.""" + proposal = temp_db.get_proposal(999) + assert proposal is None + + +def test_list_proposals(temp_db): + """Test listing all proposals.""" + # Create multiple proposals + end_time = datetime.now() + timedelta(days=7) + + id1 = temp_db.create_proposal( + "Proposal 1", "Description 1", + "0x1234567890123456789012345678901234567890", + end_time + ) + id2 = temp_db.create_proposal( + "Proposal 2", "Description 2", + "0xabcdef0123456789012345678901234567890abc", + end_time + ) + + proposals = temp_db.list_proposals() + assert len(proposals) == 2 + assert proposals[0]["id"] == id1 + assert proposals[1]["id"] == id2 + + +def test_list_proposals_by_status(temp_db): + """Test filtering proposals by status.""" + end_time = datetime.now() + timedelta(days=7) + + id1 = temp_db.create_proposal( + "Active Proposal", "Description", + "0x1234567890123456789012345678901234567890", + end_time + ) + id2 = temp_db.create_proposal( + "Another Proposal", "Description", + "0xabcdef0123456789012345678901234567890abc", + end_time + ) + + # Close one proposal + temp_db.close_proposal(id2, "passed") + + active_proposals = temp_db.list_proposals("active") + assert len(active_proposals) == 1 + assert active_proposals[0]["id"] == id1 + + passed_proposals = temp_db.list_proposals("passed") + assert len(passed_proposals) == 1 + assert passed_proposals[0]["id"] == id2 + + +def test_close_proposal(db_with_proposal): + """Test closing a proposal.""" + db, proposal_id = db_with_proposal + + db.close_proposal(proposal_id, "passed") + + proposal = db.get_proposal(proposal_id) + assert proposal["status"] == "passed" + + +def test_create_vote(db_with_proposal): + """Test casting a vote.""" + db, proposal_id = db_with_proposal + + vote_id = db.create_vote( + proposal_id=proposal_id, + voter_address="0xabcdef0123456789012345678901234567890abc", + vote_choice="for", + signature="0xsignature123" + ) + + assert vote_id is not None + assert isinstance(vote_id, int) + assert vote_id > 0 + + +def test_get_votes(db_with_proposal): + """Test retrieving votes for a proposal.""" + db, proposal_id = db_with_proposal + + # Cast multiple votes + db.create_vote(proposal_id, "0xvoter1", "for", "0xsig1") + db.create_vote(proposal_id, "0xvoter2", "against", "0xsig2") + db.create_vote(proposal_id, "0xvoter3", "for", "0xsig3") + + votes = db.get_votes(proposal_id) + assert len(votes) == 3 + + +def test_get_vote_counts(db_with_proposal): + """Test counting votes.""" + db, proposal_id = db_with_proposal + + # Cast votes + db.create_vote(proposal_id, "0xvoter1", "for", "0xsig1") + db.create_vote(proposal_id, "0xvoter2", "for", "0xsig2") + db.create_vote(proposal_id, "0xvoter3", "against", "0xsig3") + + counts = db.get_vote_counts(proposal_id) + assert counts["for"] == 2 + assert counts["against"] == 1 + assert counts["total"] == 3 + + +def test_vote_counts_empty(db_with_proposal): + """Test vote counts for proposal with no votes.""" + db, proposal_id = db_with_proposal + + counts = db.get_vote_counts(proposal_id) + assert counts["for"] == 0 + assert counts["against"] == 0 + assert counts["total"] == 0 + + +def test_mark_votes_synced(db_with_proposal): + """Test marking votes as synced to chain.""" + db, proposal_id = db_with_proposal + + # Create votes + db.create_vote(proposal_id, "0xvoter1", "for", "0xsig1") + db.create_vote(proposal_id, "0xvoter2", "against", "0xsig2") + + # Mark as synced + db.mark_votes_synced(proposal_id) + + # Verify all votes are synced + votes = db.get_votes(proposal_id) + assert all(vote["synced_to_chain"] == 1 for vote in votes) + + +def test_proposal_timestamps(temp_db): + """Test that timestamps are properly recorded.""" + end_time = datetime.now() + timedelta(days=7) + proposal_id = temp_db.create_proposal( + "Time Test", "Description", + "0x1234567890123456789012345678901234567890", + end_time + ) + + proposal = temp_db.get_proposal(proposal_id) + assert "created_at" in proposal + assert "end_time" in proposal + + # Verify created_at is recent + created_at = datetime.fromisoformat(proposal["created_at"]) + assert (datetime.now() - created_at).total_seconds() < 5 + + +def test_vote_timestamp(db_with_proposal): + """Test that vote timestamps are properly recorded.""" + db, proposal_id = db_with_proposal + + vote_id = db.create_vote( + proposal_id, "0xvoter1", "for", "0xsig1" + ) + + votes = db.get_votes(proposal_id) + vote = votes[0] + + assert "timestamp" in vote + timestamp = datetime.fromisoformat(vote["timestamp"]) + assert (datetime.now() - timestamp).total_seconds() < 5 + + +def test_multiple_proposals_independent(temp_db): + """Test that votes are independent between proposals.""" + end_time = datetime.now() + timedelta(days=7) + + # Create two proposals + id1 = temp_db.create_proposal( + "Proposal 1", "Desc 1", + "0x1234567890123456789012345678901234567890", + end_time + ) + id2 = temp_db.create_proposal( + "Proposal 2", "Desc 2", + "0xabcdef0123456789012345678901234567890abc", + end_time + ) + + # Vote on first proposal + temp_db.create_vote(id1, "0xvoter1", "for", "0xsig1") + temp_db.create_vote(id1, "0xvoter2", "against", "0xsig2") + + # Vote on second proposal + temp_db.create_vote(id2, "0xvoter3", "for", "0xsig3") + + # Check vote counts are independent + counts1 = temp_db.get_vote_counts(id1) + counts2 = temp_db.get_vote_counts(id2) + + assert counts1["total"] == 2 + assert counts2["total"] == 1 diff --git a/python/dao-voting-dashboard/tests/test_governance.py b/python/dao-voting-dashboard/tests/test_governance.py new file mode 100644 index 0000000..e9a983b --- /dev/null +++ b/python/dao-voting-dashboard/tests/test_governance.py @@ -0,0 +1,235 @@ +""" +Tests for governance rules and logic. +""" + +import pytest +from governance import GovernanceRules + + +def test_calculate_quorum(): + """Test quorum calculation.""" + # 20% quorum of 100 voters = 20 + quorum = GovernanceRules.calculate_quorum(100) + assert quorum == 20 + + # 20% quorum of 50 voters = 10 + quorum = GovernanceRules.calculate_quorum(50) + assert quorum == 10 + + # Edge case: 0 voters + quorum = GovernanceRules.calculate_quorum(0) + assert quorum == 0 + + +def test_check_quorum_met(): + """Test quorum requirement checking.""" + total_eligible = 100 + + # Quorum met: 20+ votes (20% of 100) + assert GovernanceRules.check_quorum_met(20, total_eligible) is True + assert GovernanceRules.check_quorum_met(30, total_eligible) is True + + # Quorum not met: < 20 votes + assert GovernanceRules.check_quorum_met(19, total_eligible) is False + assert GovernanceRules.check_quorum_met(10, total_eligible) is False + assert GovernanceRules.check_quorum_met(0, total_eligible) is False + + +def test_calculate_approval_rate(): + """Test approval rate calculation.""" + # 50% approval + rate = GovernanceRules.calculate_approval_rate(50, 50) + assert rate == 50.0 + + # 100% approval + rate = GovernanceRules.calculate_approval_rate(100, 0) + assert rate == 100.0 + + # 0% approval + rate = GovernanceRules.calculate_approval_rate(0, 100) + assert rate == 0.0 + + # 75% approval + rate = GovernanceRules.calculate_approval_rate(75, 25) + assert rate == 75.0 + + # No votes + rate = GovernanceRules.calculate_approval_rate(0, 0) + assert rate == 0.0 + + +def test_check_approval_met(): + """Test approval threshold checking (50%).""" + # Approval met: >= 50% + assert GovernanceRules.check_approval_met(50, 50) is True + assert GovernanceRules.check_approval_met(60, 40) is True + assert GovernanceRules.check_approval_met(100, 0) is True + + # Approval not met: < 50% + assert GovernanceRules.check_approval_met(49, 51) is False + assert GovernanceRules.check_approval_met(40, 60) is False + assert GovernanceRules.check_approval_met(0, 100) is False + + +def test_determine_outcome_passed(): + """Test outcome determination for passed proposal.""" + total_eligible = 100 + + # Passed: Quorum met (20+) and approval met (50%+) + outcome = GovernanceRules.determine_outcome(30, 10, total_eligible) + + assert outcome["status"] == "passed" + assert outcome["quorum_met"] is True + assert outcome["approval_met"] is True + assert outcome["total_votes"] == 40 + assert outcome["approval_rate"] == 75.0 + + +def test_determine_outcome_rejected_no_quorum(): + """Test outcome determination when quorum not met.""" + total_eligible = 100 + + # Rejected: Quorum not met (< 20) + outcome = GovernanceRules.determine_outcome(10, 5, total_eligible) + + assert outcome["status"] == "rejected" + assert outcome["reason"] == "Quorum not met" + assert outcome["quorum_met"] is False + assert outcome["total_votes"] == 15 + + +def test_determine_outcome_rejected_no_approval(): + """Test outcome determination when approval not met.""" + total_eligible = 100 + + # Rejected: Quorum met but approval not met + outcome = GovernanceRules.determine_outcome(10, 20, total_eligible) + + assert outcome["status"] == "rejected" + assert outcome["reason"] == "Approval threshold not met" + assert outcome["quorum_met"] is True + assert outcome["approval_met"] is False + assert outcome["approval_rate"] == pytest.approx(33.33, rel=0.1) + + +def test_determine_outcome_edge_case_exact_threshold(): + """Test outcome with exact threshold values.""" + total_eligible = 100 + + # Exactly 50% approval, exactly 20 votes (quorum) + outcome = GovernanceRules.determine_outcome(10, 10, total_eligible) + + assert outcome["status"] == "passed" + assert outcome["quorum_met"] is True + assert outcome["approval_met"] is True + assert outcome["approval_rate"] == 50.0 + + +def test_is_eligible_voter_valid(): + """Test voter eligibility checking with valid addresses.""" + # Valid Ethereum address + assert GovernanceRules.is_eligible_voter( + "0x1234567890123456789012345678901234567890" + ) is True + + assert GovernanceRules.is_eligible_voter( + "0xabcdefABCDEF0123456789012345678901234567" + ) is True + + +def test_is_eligible_voter_invalid(): + """Test voter eligibility checking with invalid addresses.""" + # Too short + assert GovernanceRules.is_eligible_voter("0x123") is False + + # Missing 0x prefix + assert GovernanceRules.is_eligible_voter( + "1234567890123456789012345678901234567890" + ) is False + + # Empty string + assert GovernanceRules.is_eligible_voter("") is False + + # None + assert GovernanceRules.is_eligible_voter(None) is False + + +def test_get_voting_power(): + """Test voting power calculation.""" + # Valid address has 1 vote + power = GovernanceRules.get_voting_power( + "0x1234567890123456789012345678901234567890" + ) + assert power == 1 + + # Invalid address has 0 votes + power = GovernanceRules.get_voting_power("0x123") + assert power == 0 + + power = GovernanceRules.get_voting_power("") + assert power == 0 + + +def test_outcome_includes_all_fields(): + """Test that outcome includes all required fields.""" + total_eligible = 100 + outcome = GovernanceRules.determine_outcome(30, 10, total_eligible) + + required_fields = [ + "status", "reason", "quorum_met", "approval_met", + "approval_rate", "total_votes", "votes_for", "votes_against", + "required_quorum", "required_approval" + ] + + for field in required_fields: + assert field in outcome + + +def test_determine_outcome_no_votes(): + """Test outcome determination with no votes.""" + total_eligible = 100 + outcome = GovernanceRules.determine_outcome(0, 0, total_eligible) + + assert outcome["status"] == "rejected" + assert outcome["quorum_met"] is False + assert outcome["total_votes"] == 0 + assert outcome["approval_rate"] == 0.0 + + +def test_determine_outcome_all_for(): + """Test outcome with all votes for.""" + total_eligible = 100 + outcome = GovernanceRules.determine_outcome(50, 0, total_eligible) + + assert outcome["status"] == "passed" + assert outcome["approval_rate"] == 100.0 + assert outcome["quorum_met"] is True + assert outcome["approval_met"] is True + + +def test_determine_outcome_all_against(): + """Test outcome with all votes against.""" + total_eligible = 100 + outcome = GovernanceRules.determine_outcome(0, 50, total_eligible) + + assert outcome["status"] == "rejected" + assert outcome["approval_rate"] == 0.0 + assert outcome["quorum_met"] is True + assert outcome["approval_met"] is False + + +def test_quorum_and_approval_independence(): + """Test that quorum and approval are checked independently.""" + total_eligible = 100 + + # High approval but low quorum + outcome = GovernanceRules.determine_outcome(10, 2, total_eligible) + assert outcome["approval_met"] is True + assert outcome["quorum_met"] is False + assert outcome["status"] == "rejected" + + # High quorum but low approval + outcome = GovernanceRules.determine_outcome(10, 30, total_eligible) + assert outcome["approval_met"] is False + assert outcome["quorum_met"] is True + assert outcome["status"] == "rejected"