fix: test and knowledge

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"
-    )
+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
+    }

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"]
+
+```