diff --git a/tests/conftest.py b/tests/conftest.py index 55ee13d..6056335 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -1,129 +1,56 @@ import pytest -import pytest_asyncio -from unittest.mock import AsyncMock, MagicMock -from telegram import Update, Chat, Message, User -from telegram.ext import CallbackContext -from dotenv import load_dotenv -import os import logging +import os +from pathlib import Path +from dotenv import load_dotenv -# Configure logging -logging.basicConfig( - format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', - level=logging.INFO -) -logger = logging.getLogger(__name__) - -def pytest_configure(config): - """Configure pytest markers""" - config.addinivalue_line("markers", "telegram: mark test as a telegram test") - config.addinivalue_line("markers", "claude: mark test as a claude test") - config.addinivalue_line("markers", "integration: mark as integration test") - -@pytest.fixture(scope="session", autouse=True) -def load_env(): - """Load environment variables""" - load_dotenv() - # Verify critical environment variables - required_vars = ['TELEGRAM_TOKEN', 'CLAUDE_KEY'] - for var in required_vars: - if not os.getenv(var): - logger.warning(f"{var} is not set in environment") - -@pytest.fixture -def telegram_token(): - """Get Telegram token""" - return os.getenv('TELEGRAM_TOKEN') - -@pytest.fixture -def claude_key(): - """Get Claude API key""" - return os.getenv('CLAUDE_KEY') - -@pytest.fixture -def test_chat_id(): - """Get test chat ID""" - return 1978731049 - -@pytest.fixture -async def mock_message(): - """Create mock telegram message""" - message = AsyncMock(spec=Message) - message.chat = AsyncMock(spec=Chat) - message.chat.id = 1978731049 - message.from_user = AsyncMock(spec=User) - message.from_user.id = 67950696 # Default admin ID - return message - -@pytest.fixture -async def mock_update(mock_message): - """Create mock telegram update""" - update = AsyncMock(spec=Update) - update.effective_chat = mock_message.chat - update.message = mock_message - return update - -@pytest.fixture -async def mock_context(): - """Create mock context with bot""" - context = AsyncMock(spec=CallbackContext) - context.bot = AsyncMock() - return context - -@pytest.fixture -def mock_claude_response(): - """Create mock Claude API response""" - response = MagicMock() - response.content = [MagicMock(text="Test response")] - return response +# Load environment variables from .env file +load_dotenv() @pytest.fixture -def sample_knowledge_base(): - """Provide sample knowledge base content""" - return """ - # Sample Ape Documentation - Ape is a tool for smart contract development... - """ - -@pytest.fixture(autouse=True) def setup_logging(): """Configure logging for tests""" logging.basicConfig(level=logging.INFO) return logging.getLogger(__name__) -# Helper fixtures for common test scenarios @pytest.fixture -async def send_command(mock_update, mock_context): - """Helper to simulate sending a command""" - async def _send_command(command: str, *args): - mock_update.message.text = f"/{command} {' '.join(args)}" - return mock_update, mock_context - return _send_command - -@pytest.fixture -async def admin_context(mock_context): - """Context with admin privileges""" - mock_context.user_data = {'is_admin': True} - return mock_context - -@pytest.fixture -async def group_context(mock_context): - """Context for group chat""" - mock_context.chat_data = {'is_group': True, 'messages_today': 0} - return mock_context - -@pytest_asyncio.fixture(scope='function') -async def event_loop(): - """Create event loop""" - import asyncio - loop = asyncio.get_event_loop_policy().new_event_loop() - yield loop - loop.close() - -def pytest_configure(config): - """Configure pytest""" - # Add asyncio marker - config.addinivalue_line( - "markers", - "asyncio: mark test as async" - ) \ No newline at end of file +def tokens(): + """Bot tokens and IDs from environment variables""" + return { + "telegram": os.getenv('TELEGRAM_TOKEN'), + "claude": os.getenv('CLAUDE_KEY'), + "admin_id": 1978731049, # Chris | ApeWorX + "group_id": -4718382612, # ApeClaudeCouncil + "bot_id": 7879249317 # ApeCluade bot + } + +@pytest.fixture(scope="session") +def cached_knowledge_base(request): + """Cache the knowledge base content using file hash as key""" + kb_path = Path("knowledge-base/all.txt") + + # Generate hash of file content for cache key + file_hash = hashlib.md5(kb_path.read_bytes()).hexdigest() + cache_key = f"knowledge_base_content_{file_hash}" + + # Try to get content from cache + cached_content = request.config.cache.get(cache_key, None) + + if cached_content is None: + # Cache miss - read file and store in cache + content = kb_path.read_text(encoding='utf-8') + request.config.cache.set(cache_key, content) + return content + + return cached_content + +@pytest.fixture(scope="session") +def knowledge_base_stats(cached_knowledge_base): + """Provide basic stats about the knowledge base""" + content = cached_knowledge_base + return { + "total_length": len(content), + "line_count": len(content.splitlines()), + "has_python_code": "```python" in content, + "has_vyper_code": "```vyper" in content + } \ No newline at end of file diff --git a/tests/knowledge-base/all.txt b/tests/knowledge-base/all.txt new file mode 100644 index 0000000..3edf689 --- /dev/null +++ b/tests/knowledge-base/all.txt @@ -0,0 +1,126 @@ +SAMPLE CODE FOR BOTS BUILT WITH SILVERBACK AND ApeWorX + +``` +import os + +from ape import Contract, chain +from silverback import SilverbackBot + + +bot = SilverbackBot() + +vault = Contract(os.environ["ERC4626_VAULT_ADDRESS"]) + +one_share = 10 ** vault.decimals() + + +@bot.on_(chain.blocks) +def update_shareprice(_): + """ + Conver share to set price unit. + """ + price = vault.convertToAssets(one_share) / one_share + # Total number of shares in the vault divide by + # the current unit price of a share = gwei + print(f"Price Event: {price}") +``` + +``` +from math import log10, floor + +from ape import project +from silverback import SilverbackApp +from ape_farcaster import Warpcast +import time + +import os +import tempfile +import requests + +from wonderwords import RandomWord + +STABILITY_KEY = os.environ.get("STABILITY_KEY") +PINATA_API_KEY = os.environ.get("PINATA_API_KEY") +PINATA_SECRET_API_KEY = os.environ.get("PINATA_SECRET_API_KEY") + +app = SilverbackApp() + +attempts = 0 +while attempts < 11: + try: + client = Warpcast(app.signer) + break + except Exception as e: + print(e) + attempts+=1 + time.sleep(1) + +my_contract = project.Echo.at(os.environ.get("ECHO_CONTRACT")) + +def create_prompt(number_adj: int) -> str: + w = RandomWord() + adjectives = w.random_words(number_adj, include_categories=["adjective"]) + adjectives_string = ", ".join(adjectives) + prompt = f"An {adjectives_string} ape that is screaming AHHHHH." + return prompt + + +@app.on_(my_contract.Received) +def payment_received(log): + print(log) + prompt = create_prompt(max(floor(log10(log.amount)) - 12, 1)) + for _ in range(10): + print(f"\n{prompt}\n") + # create image from AI using scream + try: + fileName = createImage(prompt) + break + except Exception as e: + print(e) + prompt = create_prompt(max(floor(log10(log.amount)) - 12, 1)) + + # open the created image + print(fileName + " has beeen created") + image = open(fileName, 'rb') + # upload image to Pinata + ipfsHash = uploadToPinata(image) + # cast the message with the scream and ipfs link + ipfsLink = "https://jade-slow-pigeon-687.mypinata.cloud/ipfs/" + ipfsHash + warpcast_text = f"A picture of {log.sender}:\n" + client.post_cast(text=warpcast_text, embeds=[ipfsLink]) + + +def createImage(prompt): + response = requests.post( + "https://api.stability.ai/v2beta/stable-image/generate/ultra", + headers={ + "authorization": STABILITY_KEY, + "accept": "image/*" + }, + files={"none": ''}, + data={ + "prompt": prompt, + "output_format": "jpeg", + }, + ) + if response.status_code == 200: + with tempfile.NamedTemporaryFile(delete=False, suffix=".jpeg") as fp: + fp.write(response.content) + else: + raise Exception(str(response.json())) + return fp.name + + +def uploadToPinata(image): + url = "https://api.pinata.cloud/pinning/pinFileToIPFS" + headers = { + 'pinata_api_key': PINATA_API_KEY, + 'pinata_secret_api_key': PINATA_SECRET_API_KEY, + } + + files = {'file': ("ApeScream", image)} + response = requests.request("POST", url, files=files, headers=headers) + response.raise_for_status() + return response.json()["IpfsHash"] + +``` \ No newline at end of file diff --git a/tests/knowledge-base/all.txt:Zone.Identifier b/tests/knowledge-base/all.txt:Zone.Identifier new file mode 100644 index 0000000..e69de29 diff --git a/tests/test_claude.py b/tests/test_claude.py index db13032..3797e4b 100644 --- a/tests/test_claude.py +++ b/tests/test_claude.py @@ -1,102 +1,93 @@ -import unittest import os -from unittest.mock import patch, MagicMock -from anthropic import Anthropic -import tempfile -import yaml -import base64 +import logging +from pathlib import Path +from dotenv import load_dotenv +import anthropic -class TestAnthropicPrompt(unittest.TestCase): - def setUp(self): - """Set up test environment""" - self.temp_dir = tempfile.mkdtemp() - self.config_file = os.path.join(self.temp_dir, 'anthropic_config.yml') - self.test_key = "test-api-key" - - # Create test config file - with open(self.config_file, 'w') as f: - yaml.dump({'api_key': base64.b64encode(self.test_key.encode('utf-8')).decode('utf-8')}, f) - - # Create test source files - self.source_dir = os.path.join(self.temp_dir, 'test_source') - os.makedirs(self.source_dir) - with open(os.path.join(self.source_dir, 'test.py'), 'w') as f: - f.write('def test_function():\n return "test"') - - # Set up Anthropic client - self.client = Anthropic(api_key=self.test_key) +# Load environment variables +load_dotenv() - def tearDown(self): - """Clean up test environment""" - import shutil - shutil.rmtree(self.temp_dir) +# Configure logging +logging.basicConfig( + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + level=logging.INFO +) +logger = logging.getLogger(__name__) - def test_anthropic_client(self): - """Test Anthropic client initialization""" - self.assertIsInstance(self.client, Anthropic) - self.assertEqual(self.client.api_key, self.test_key) +def load_knowledge_base(): + """Load the knowledge base content""" + try: + kb_path = Path("knowledge-base/all.txt") + return kb_path.read_text(encoding='utf-8') + except Exception as e: + logger.error(f"Error loading knowledge base: {e}") + raise - def test_concatenate_sources(self): - """Test source file concatenation""" - def concatenate_sources(source_dirs): - content = "" - for dir_path in source_dirs: - for root, _, files in os.walk(dir_path): - for file in files: - file_path = os.path.join(root, file) - with open(file_path, 'r') as f: - content += f"# {file_path}\n{f.read()}\n\n" - return content +def test_claude_queries(): + """Test Claude's responses to various queries""" + + # Initialize Claude + client = anthropic.Anthropic(api_key=os.getenv('CLAUDE_KEY')) + + # Load knowledge base + logger.info("Loading knowledge base...") + knowledge_base = load_knowledge_base() + logger.info("Knowledge base loaded successfully") - result = concatenate_sources([self.source_dir]) - self.assertIn('test_function', result) - self.assertIn('return "test"', result) + # Test queries + test_queries = [ + "What is ApeWorX?", + "Write a simple Silverback bot", + "How do I deploy a smart contract with Ape?", + ] - @patch('anthropic.resources.messages.Messages.create') - def test_send_prompt(self, mock_create): - """Test sending prompt to Anthropic API""" - # Mock API response - mock_response = MagicMock() - mock_response.content = [MagicMock(text="Test response")] - mock_create.return_value = mock_response + system_prompt = """You are a technical assistant for ApeWorX, specializing in smart contract development and blockchain tooling. +Use ONLY the provided documentation to answer questions. +If the answer cannot be found in the documentation, say so clearly.""" - # Test data - test_content = "Test content" - test_prompt = "Test prompt" - - # Send prompt - response = self.client.messages.create( - model="claude-3-opus-20240229", - max_tokens=1000, - messages=[{ - "role": "user", - "content": f"{test_prompt}\n\n{test_content}" - }] - ) - - # Verify response - self.assertEqual(response.content[0].text, "Test response") - mock_create.assert_called_once() + for query in test_queries: + print("\n" + "="*80) + print(f"\nTesting query: {query}") + print("="*80) - @patch('anthropic.resources.messages.Messages.create') - def test_api_error(self, mock_create): - """Test API error handling""" - mock_create.side_effect = Exception("API Error") - - with self.assertRaises(Exception) as context: - self.client.messages.create( + try: + # Construct full prompt + full_prompt = f"""Documentation: +{knowledge_base} + +Question: {query} + +Please provide a clear and specific answer based solely on the documentation provided.""" + + print("\nPrompt sent to Claude:") + print("-"*40) + print(full_prompt) + print("-"*40) + + # Get Claude's response using correct message structure + response = client.messages.create( model="claude-3-opus-20240229", - max_tokens=1000, - messages=[{"role": "user", "content": "test"}] + max_tokens=2000, + temperature=0, + system=system_prompt, # System prompt goes here + messages=[ + {"role": "user", "content": full_prompt} # Only user message in the messages array + ] ) - - self.assertEqual(str(context.exception), "API Error") - def test_environment_config(self): - """Test environment configuration""" - with patch.dict(os.environ, {'CLAUDE_KEY': self.test_key}): - client = Anthropic(api_key=os.getenv('CLAUDE_KEY')) - self.assertEqual(client.api_key, self.test_key) + print("\nClaude's response:") + print("-"*40) + print(response.content[0].text) + print("-"*40) + + except Exception as e: + logger.error(f"Error processing query '{query}': {e}") + print(f"Error: {e}") -if __name__ == '__main__': - unittest.main() \ No newline at end of file +if __name__ == "__main__": + print("\nStarting Claude test...") + try: + test_claude_queries() + print("\nTest completed") + except Exception as e: + print(f"\nTest failed: {e}") \ No newline at end of file diff --git a/tests/test_knowledge.py b/tests/test_knowledge.py new file mode 100644 index 0000000..458423f --- /dev/null +++ b/tests/test_knowledge.py @@ -0,0 +1,40 @@ +import pytest + +def test_knowledge_base_content(cached_knowledge_base, setup_logging): + """Test knowledge base content structure using cached content""" + content = cached_knowledge_base + + # Check for required sections + required_keywords = [ + "apeworx", + "vyper", + "silverback", + "contract", + "deploy", + "test" + ] + + for keyword in required_keywords: + assert keyword.lower() in content.lower(), f"Missing content about: {keyword}" + + setup_logging.info("Knowledge base content validation successful") + +def test_knowledge_base_stats(knowledge_base_stats, setup_logging): + """Test knowledge base statistics""" + assert knowledge_base_stats["total_length"] > 0, "Knowledge base is empty" + assert knowledge_base_stats["line_count"] > 0, "Knowledge base has no lines" + + setup_logging.info(f"""Knowledge base stats: + - Total length: {knowledge_base_stats['total_length']} characters + - Line count: {knowledge_base_stats['line_count']} lines + - Has Python code: {knowledge_base_stats['has_python_code']} + - Has Vyper code: {knowledge_base_stats['has_vyper_code']} + """) + +def test_code_examples(cached_knowledge_base): + """Test if knowledge base contains code examples""" + content = cached_knowledge_base + + code_indicators = ["```python", "```vyper", "@external", "def "] + found_code = any(indicator in content for indicator in code_indicators) + assert found_code, "No code examples found in knowledge base" \ No newline at end of file diff --git a/tests/test_manual.py b/tests/test_manual.py deleted file mode 100644 index 0dc5630..0000000 --- a/tests/test_manual.py +++ /dev/null @@ -1,105 +0,0 @@ -import os -from anthropic import Anthropic -import argparse -from dotenv import load_dotenv - -def test_claude_api(): - """Test Claude API connection and response""" - print("\nšŸ“” Testing Claude API connection...") - - client = Anthropic(api_key=os.getenv('CLAUDE_KEY')) - try: - response = client.messages.create( - model="claude-3-opus-20240229", - max_tokens=1000, - messages=[{ - "role": "user", - "content": "Please respond with 'Hello, test successful!'" - }] - ) - print("āœ… Claude API test successful!") - print(f"Response: {response.content[0].text}") - except Exception as e: - print(f"āŒ Claude API test failed: {str(e)}") - -def test_telegram_token(): - """Test Telegram token validity""" - print("\nšŸ¤– Testing Telegram token...") - - import telegram - try: - bot = telegram.Bot(token=os.getenv('TELEGRAM_TOKEN')) - bot_info = bot.get_me() - print("āœ… Telegram token valid!") - print(f"Bot username: @{bot_info.username}") - except Exception as e: - print(f"āŒ Telegram token test failed: {str(e)}") - -def test_file_system(): - """Test file system setup""" - print("\nšŸ“‚ Testing file system setup...") - - required_dirs = ['sources', 'responses'] - required_files = ['requirements.txt', '.env'] - - for dir_name in required_dirs: - if os.path.exists(dir_name): - print(f"āœ… Directory '{dir_name}' exists") - else: - print(f"āŒ Directory '{dir_name}' missing") - os.makedirs(dir_name) - print(f" Created '{dir_name}' directory") - - for file_name in required_files: - if os.path.exists(file_name): - print(f"āœ… File '{file_name}' exists") - else: - print(f"āŒ File '{file_name}' missing") - -def test_environment(): - """Test environment variables""" - print("\nšŸ” Testing environment variables...") - - required_vars = ['TELEGRAM_TOKEN', 'CLAUDE_KEY'] - for var in required_vars: - if os.getenv(var): - print(f"āœ… {var} is set") - if var == 'TELEGRAM_TOKEN': - print(f" Token: ...{os.getenv(var)[-10:]}") - else: - print(f" Key: ...{os.getenv(var)[-10:]}") - else: - print(f"āŒ {var} is not set") - -def main(): - parser = argparse.ArgumentParser(description='Manual CLI testing tool') - parser.add_argument('--all', action='store_true', help='Run all tests') - parser.add_argument('--claude', action='store_true', help='Test Claude API') - parser.add_argument('--telegram', action='store_true', help='Test Telegram token') - parser.add_argument('--files', action='store_true', help='Test file system') - parser.add_argument('--env', action='store_true', help='Test environment variables') - - args = parser.parse_args() - - # Load environment variables - print("šŸ”„ Loading environment variables...") - load_dotenv() - - # If no specific tests are selected, run all tests - if not (args.claude or args.telegram or args.files or args.env): - args.all = True - - if args.all or args.env: - test_environment() - - if args.all or args.files: - test_file_system() - - if args.all or args.claude: - test_claude_api() - - if args.all or args.telegram: - test_telegram_token() - -if __name__ == "__main__": - main() \ No newline at end of file diff --git a/tests/test_manual_message_check.py b/tests/test_manual_message_check.py new file mode 100644 index 0000000..f5909cf --- /dev/null +++ b/tests/test_manual_message_check.py @@ -0,0 +1,90 @@ +import os +import logging +from pathlib import Path +from telegram import Update +from telegram.ext import ApplicationBuilder, MessageHandler, filters, ContextTypes +from dotenv import load_dotenv +import anthropic + +# Load environment variables +load_dotenv() + +# Configure logging +logging.basicConfig( + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + level=logging.INFO +) +logger = logging.getLogger(__name__) + +# Configuration +TELEGRAM_TOKEN = os.getenv('TELEGRAM_TOKEN') +CLAUDE_KEY = os.getenv('CLAUDE_KEY') +GROUP_ID = -4718382612 # ApeClaudeCouncil group + +async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE): + """Handle incoming messages""" + if update.effective_chat.id != GROUP_ID: + return + + message_text = update.message.text.lower() + logger.info(f"Received message: {message_text}") + + if "what is apeworx" in message_text: + logger.info("Processing 'what is apeworx' query...") + + try: + # Get knowledge base content + kb_path = Path("knowledge-base/all.txt") + knowledge_base = kb_path.read_text(encoding='utf-8') + + # Initialize Claude + client = anthropic.Anthropic(api_key=CLAUDE_KEY) + + # Get response from Claude + response = client.messages.create( + model="claude-3-opus-20240229", + max_tokens=1000, + temperature=0, + messages=[{ + "role": "user", + "content": f"""Based on this documentation: + {knowledge_base} + + Question: What is ApeWorX? + + Please provide a concise explanation.""" + }] + ) + + # Send response back to group + await update.message.reply_text(response.content[0].text) + logger.info("Response sent successfully") + + except Exception as e: + error_msg = f"Error processing request: {str(e)}" + logger.error(error_msg) + await update.message.reply_text(error_msg) + +def main(): + """Run the manual test bot""" + logger.info("Starting manual test bot...") + + # Build application + app = ApplicationBuilder().token(TELEGRAM_TOKEN).build() + + # Add message handler + app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message)) + + logger.info("Bot is running. Send 'what is apeworx' to the ApeClaudeCouncil group...") + logger.info("Press Ctrl+C to stop") + + # Start polling + app.run_polling(allowed_updates=["message"]) + +if __name__ == '__main__': + try: + main() + except KeyboardInterrupt: + logger.info("\nBot stopped by user") + except Exception as e: + logger.error(f"Fatal error: {e}") \ No newline at end of file diff --git a/tests/test_manual_telegram.py b/tests/test_manual_telegram.py new file mode 100644 index 0000000..209ca38 --- /dev/null +++ b/tests/test_manual_telegram.py @@ -0,0 +1,131 @@ +import os +import logging +import hashlib +from pathlib import Path +from telegram import Update +from telegram.ext import ApplicationBuilder, MessageHandler, filters, ContextTypes +from dotenv import load_dotenv +import anthropic + +# Load environment variables +load_dotenv() + +# Configure logging +logging.basicConfig( + format='%(asctime)s - %(name)s - %(levelname)s - %(message)s', + level=logging.INFO +) +logger = logging.getLogger(__name__) + +# Configuration +TELEGRAM_TOKEN = os.getenv('TELEGRAM_TOKEN') +CLAUDE_KEY = os.getenv('CLAUDE_KEY') +GROUP_ID = -4718382612 # ApeClaudeCouncil group + +class KnowledgeBaseCache: + _content = None + _file_hash = None + + @classmethod + def get_content(cls): + """Get cached content or load from file if needed""" + kb_path = Path("knowledge-base/all.txt") + current_hash = hashlib.md5(kb_path.read_bytes()).hexdigest() + + if cls._content is None or current_hash != cls._file_hash: + print("Loading knowledge base from file...") + cls._content = kb_path.read_text(encoding='utf-8') + cls._file_hash = current_hash + print("Knowledge base loaded and cached") + else: + print("Using cached knowledge base") + + return cls._content + +async def handle_message(update: Update, context: ContextTypes.DEFAULT_TYPE): + """Handle incoming messages""" + if update.effective_chat.id != GROUP_ID: + return + + user_message = update.message.text + print("\n" + "="*80) + print(f"Received message: {user_message}") + print("="*80) + + try: + # Get cached knowledge base content + knowledge_base = KnowledgeBaseCache.get_content() + + # Define system and user prompts + system_prompt = """You are a technical assistant for ApeWorX, specializing in smart contract development and blockchain tooling. +Use ONLY the provided documentation to answer questions. +If the answer cannot be found in the documentation, say so clearly.""" + + user_prompt = f"""Documentation: +{knowledge_base} + +Question: {user_message} + +Please provide a clear and specific answer based solely on the documentation provided.""" + + print("\nPrompt being sent to Claude:") + print("-"*40) + print(user_prompt) + print("-"*40) + + # Initialize Claude and get response + client = anthropic.Anthropic(api_key=CLAUDE_KEY) + response = client.messages.create( + model="claude-3-opus-20240229", + max_tokens=2000, + temperature=0, + system=system_prompt, + messages=[ + {"role": "user", "content": user_prompt} + ] + ) + + claude_response = response.content[0].text + print("\nClaude's response:") + print("-"*40) + print(claude_response) + print("-"*40) + + # Send response back to group + await update.message.reply_text(claude_response) + print("\nResponse sent to Telegram group") + + except Exception as e: + error_msg = f"Error processing request: {str(e)}" + print(f"\nERROR: {error_msg}") + await update.message.reply_text(f"āŒ Error: {error_msg}") + +def main(): + """Run the manual test bot""" + print("\n" + "="*80) + print("Starting manual test bot...") + + # Pre-load knowledge base into cache + KnowledgeBaseCache.get_content() + + # Build application + app = ApplicationBuilder().token(TELEGRAM_TOKEN).build() + + # Add message handler + app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, handle_message)) + + print("\nBot is running! šŸš€") + print("- Send any question to the ApeClaudeCouncil group") + print("- Press Ctrl+C to stop") + print("="*80 + "\n") + + # Start polling + app.run_polling(allowed_updates=["message"]) + +if __name__ == '__main__': + try: + main() + except KeyboardInterrupt: + print("\nBot stopped by user") + except Exception as e: + print(f"\nFatal error: {e}") \ No newline at end of file diff --git a/tests/test_telegram_cli.py b/tests/test_telegram_cli.py index 255bf4a..6b9f8da 100644 --- a/tests/test_telegram_cli.py +++ b/tests/test_telegram_cli.py @@ -1,97 +1,65 @@ -import pytest -from telegram import Update -from telegram.ext import ContextTypes +import os import logging -import pytest_asyncio -from unittest.mock import AsyncMock, patch -import sys -from pathlib import Path +from telegram import Update +from telegram.ext import ApplicationBuilder, MessageHandler, filters, ContextTypes +from dotenv import load_dotenv -# Add project root to Python path -project_root = str(Path(__file__).parent.parent) -sys.path.append(project_root) +# Load environment variables +load_dotenv() -# Now we can import from bot -from bot import start, handle_message, add_admin +# Configuration +TELEGRAM_TOKEN = os.getenv('TELEGRAM_TOKEN') +GROUP_ID = -4718382612 # Replace with your actual group ID -# Configure logging -logging.basicConfig(level=logging.INFO) -logger = logging.getLogger(__name__) +# Setup logging for debugging +logging.basicConfig( + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", + level=logging.DEBUG +) -@pytest.mark.asyncio -class TestTelegramBot: - @pytest.fixture(autouse=True) - async def setup(self): - """Setup test fixtures""" - self.chat_id = 1978731049 - self.update = AsyncMock(spec=Update) - self.update.effective_chat.id = self.chat_id - self.update.message = AsyncMock() - self.update.message.chat.id = self.chat_id - self.context = AsyncMock(spec=ContextTypes.DEFAULT_TYPE) +async def echo_message(update: Update, context: ContextTypes.DEFAULT_TYPE): + """Echo the message back with '- claude' suffix if it's in the target group""" + chat_id = update.effective_chat.id + message = update.message.text + + # Debugging: Log chat ID and received message + logging.debug(f"Chat ID: {chat_id}, Message: {message}") + + if chat_id != GROUP_ID: + logging.info("Message not from the target group. Ignoring.") + return - async def test_start_command(self): - """Test /start command""" - self.update.message.text = "/start" - await start(self.update, self.context) - - self.update.message.reply_text.assert_awaited_once_with( - 'Hello! Ask me anything about ApeWorX!' - ) + response = f"{message} - claude" + logging.debug(f"Sending response: {response}") + + try: + await update.message.reply_text(response) + logging.info("Response sent successfully!") + except Exception as e: + logging.error(f"Failed to send response: {e}") - async def test_prompt_command(self): - """Test /p command""" - self.update.message.text = "/p What is Ape?" - # Mock group check - with patch('bot.groups', {str(self.chat_id): {'messages_today': 0, 'last_reset': '2024-12-04'}}): - await handle_message(self.update, self.context) - - assert self.update.message.reply_text.awaited +def main(): + """Run the echo bot""" + if not TELEGRAM_TOKEN: + logging.critical("TELEGRAM_TOKEN is missing. Please check your .env file.") + return - async def test_group_chat(self): - """Test group chat functionality""" - # Setup group context - self.update.message.chat.type = 'group' - self.update.message.text = "/p Test message" - - # Mock group data - with patch('bot.groups', {str(self.chat_id): {'messages_today': 0, 'last_reset': '2024-12-04'}}): - await handle_message(self.update, self.context) - assert self.update.message.reply_text.awaited + logging.info("Starting the echo bot...") + logging.info(f"Target group ID: {GROUP_ID}") + + # Create application instance + app = ApplicationBuilder().token(TELEGRAM_TOKEN).build() - @pytest.mark.parametrize("command,expected_text", [ - ("/start", "Hello! Ask me anything about ApeWorX!"), - ("/p What is Ape?", None), - ("/prompt Tell me about ApeWorX", None), - ]) - async def test_commands(self, command, expected_text): - """Test various commands""" - self.update.message.text = command - - with patch('bot.groups', {str(self.chat_id): {'messages_today': 0, 'last_reset': '2024-12-04'}}): - if command.startswith("/start"): - await start(self.update, self.context) - self.update.message.reply_text.assert_awaited_once_with(expected_text) - else: - await handle_message(self.update, self.context) - assert self.update.message.reply_text.awaited + # Add handler for text messages + app.add_handler(MessageHandler(filters.TEXT & ~filters.COMMAND, echo_message)) - async def test_message_limit(self): - """Test message limit in groups""" - # Mock group with max messages - group_data = {str(self.chat_id): {'messages_today': 10, 'last_reset': '2024-12-04'}} - - with patch('bot.groups', group_data): - await handle_message(self.update, self.context) - self.update.message.reply_text.assert_awaited_with( - 'GPT limit for this group has been reached (10 msgs a day).' - ) + logging.info("Bot is running! Send messages to the group for testing.") + try: + app.run_polling(allowed_updates=["message"]) + except KeyboardInterrupt: + logging.info("Bot stopped by user.") + except Exception as e: + logging.critical(f"Bot encountered an error: {e}") - async def test_admin_command(self): - """Test admin command""" - # Setup admin test - self.update.message.from_user.id = 67950696 # Default admin ID - self.context.args = ["12345"] - - await add_admin(self.update, self.context) - self.update.message.reply_text.assert_awaited_with('Admin added successfully.') \ No newline at end of file +if __name__ == '__main__': + main()