patx/micropie
added development version under `unstable` for easy public access/testing. websockets support currently being worked on here, along with testing.
Commit 3c0e59f · patx · 2025-05-28T22:53:08-04:00
Comments
No comments yet.
Diff
diff --git a/examples/chatroom/__pycache__/app.cpython-312.pyc b/examples/chatroom/__pycache__/app.cpython-312.pyc
new file mode 100644
index 0000000..4ef8034
Binary files /dev/null and b/examples/chatroom/__pycache__/app.cpython-312.pyc differ
diff --git a/examples/chatroom/__pycache__/kenobi.cpython-312.pyc b/examples/chatroom/__pycache__/kenobi.cpython-312.pyc
new file mode 100644
index 0000000..92a11ac
Binary files /dev/null and b/examples/chatroom/__pycache__/kenobi.cpython-312.pyc differ
diff --git a/examples/chatroom/app.py b/examples/chatroom/app.py
new file mode 100644
index 0000000..67ba3d5
--- /dev/null
+++ b/examples/chatroom/app.py
@@ -0,0 +1,174 @@
+import socketio
+from MicroPie import App
+from kenobi import KenobiDB
+from datetime import datetime
+import asyncio
+import re
+from urllib.parse import unquote
+
+# KenobiDB setup
+db = KenobiDB("chat.db")
+
+# Create a Socket.IO server with CORS support
+sio = socketio.AsyncServer(async_mode="asgi", cors_allowed_origins="*")
+
+# Store connected users by channel {channel: {sid: username}}
+connected_users = {}
+
+# Store channel passwords {channel: password or None}
+channel_passwords = {}
+
+# Create the MicroPie server
+class MyApp(App):
+ async def index(self):
+ """Render channel creation page"""
+ return await self._render_template("chat.html", is_index=True)
+
+ async def channel(self, channel_name):
+ """Render specific channel page"""
+ channel_name = unquote(channel_name)
+ # Validate channel name
+ if not re.match(r'^[a-zA-Z0-9_-]{1,30}$', channel_name):
+ return {"error": "Invalid channel name"}, 400
+ return await self._render_template("chat.html", channel=channel_name, is_index=False)
+
+# Socket.IO event handlers
[email protected]
+async def connect(sid, environ):
+ print(f"Client connected: {sid}")
+
[email protected]
+async def disconnect(sid):
+ print(f"Client disconnected: {sid}")
+ # Remove user from all channels
+ for channel in connected_users:
+ if sid in connected_users[channel]:
+ del connected_users[channel][sid]
+ await update_user_list(channel)
+
[email protected]
+async def join_channel(sid, data):
+ """Handle joining a channel with optional password"""
+ channel = data.get('channel', '').strip()
+ password = data.get('password', '')
+ username = data.get('username', '').strip()
+
+ if not channel or not re.match(r'^[a-zA-Z0-9_-]{1,30}$', channel):
+ await sio.emit('error', {'message': 'Invalid channel name'}, room=sid)
+ return
+
+ if not username or len(username) > 20:
+ await sio.emit('error', {'message': 'Invalid username'}, room=sid)
+ return
+
+ # Initialize channel if it doesn't exist
+ if channel not in connected_users:
+ connected_users[channel] = {}
+
+ # Check password if required
+ if channel in channel_passwords and channel_passwords[channel] and password != channel_passwords[channel]:
+ await sio.emit('error', {'message': 'Incorrect password'}, room=sid)
+ return
+
+ # Check if username is taken in this channel
+ if username in connected_users[channel].values():
+ await sio.emit('error', {'message': 'Username already taken'}, room=sid)
+ return
+
+ # Join the socket.io room for this channel
+ await sio.enter_room(sid, channel)
+
+ # Add user to channel
+ connected_users[channel][sid] = username
+ print(f"User {sid} joined channel {channel} as {username}")
+
+ # Send recent messages for this channel
+ try:
+ # Get all messages with type="message"
+ all_messages = db.search("type", "message", limit=1000) # Use a high limit to ensure we get all messages
+ # Filter messages for the specific channel
+ recent_messages = [msg for msg in all_messages if msg.get('channel') == channel][:50] # Limit to 50
+ print(f"Retrieved {len(recent_messages)} recent messages for channel {channel}")
+ for msg in recent_messages:
+ await sio.emit('message', {
+ 'username': msg['username'],
+ 'message': msg['message'],
+ 'timestamp': msg['timestamp']
+ }, room=sid)
+ except Exception as e:
+ print(f"Error retrieving messages for channel {channel}: {str(e)}")
+ await sio.emit('error', {'message': 'Failed to load recent messages'}, room=sid)
+
+ # Update user list for this channel
+ await update_user_list(channel)
+
+ # Confirm successful join to the client
+ await sio.emit('join_success', room=sid)
+
[email protected]
+async def create_channel(sid, data):
+ """Handle channel creation"""
+ channel = data.get('channel', '').strip()
+ password = data.get('password', '').strip() or None
+
+ if not channel or not re.match(r'^[a-zA-Z0-9_-]{1,30}$', channel):
+ await sio.emit('error', {'message': 'Invalid channel name'}, room=sid)
+ return
+
+ if channel in connected_users:
+ await sio.emit('error', {'message': 'Channel already exists'}, room=sid)
+ return
+
+ # Create new channel
+ connected_users[channel] = {}
+ channel_passwords[channel] = password
+ await sio.emit('channel_created', {'channel': channel}, room=sid)
+
[email protected]
+async def message(sid, data):
+ """Handle and store messages"""
+ channel = data.get('channel', '').strip()
+ message = data.get('message', '').strip()
+
+ if not channel or not message:
+ return
+
+ username = None
+ for chan, users in connected_users.items():
+ if sid in users:
+ username = users[sid]
+ break
+
+ if not username:
+ return
+
+ # Store message in KenobiDB
+ message_doc = {
+ 'type': 'message',
+ 'channel': channel,
+ 'username': username,
+ 'message': message,
+ 'timestamp': datetime.utcnow().isoformat()
+ }
+ db.insert(message_doc)
+
+ # Broadcast message to channel
+ await sio.emit('message', {
+ 'username': username,
+ 'message': message,
+ 'timestamp': message_doc['timestamp']
+ }, room=channel)
+
+async def update_user_list(channel):
+ """Broadcast updated user list to channel"""
+ if channel in connected_users:
+ print(f"Broadcasting user list for channel {channel}: {list(connected_users[channel].values())}")
+ await sio.emit('user_list', list(connected_users[channel].values()), room=channel)
+
+# Attach Socket.IO to the ASGI app
+asgi_app = MyApp()
+app = socketio.ASGIApp(sio, asgi_app)
+
+# Ensure database is closed properly on shutdown
+import atexit
+atexit.register(db.close)
diff --git a/examples/chatroom/chat.db b/examples/chatroom/chat.db
new file mode 100644
index 0000000..58ba8ee
Binary files /dev/null and b/examples/chatroom/chat.db differ
diff --git a/examples/chatroom/chat.db-shm b/examples/chatroom/chat.db-shm
new file mode 100644
index 0000000..92450e0
Binary files /dev/null and b/examples/chatroom/chat.db-shm differ
diff --git a/examples/chatroom/chat.db-wal b/examples/chatroom/chat.db-wal
new file mode 100644
index 0000000..f0890e1
Binary files /dev/null and b/examples/chatroom/chat.db-wal differ
diff --git a/examples/chatroom/kenobi.py b/examples/chatroom/kenobi.py
new file mode 100644
index 0000000..9548a50
--- /dev/null
+++ b/examples/chatroom/kenobi.py
@@ -0,0 +1,378 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+KenobiDB is a small document-based DB, supporting simple usage including
+insertion, removal, and basic search.
+Written by Harrison Erd (https://patx.github.io/)
+https://patx.github.io/kenobi/
+"""
+# Copyright Harrison Erd
+#
+# Redistribution and use in source and binary forms, with or without
+# modification, are permitted provided that the following conditions are met:
+#
+# 1. Redistributions of source code must retain the above copyright notice,
+# this list of conditions and the following disclaimer.
+# 2. Redistributions in binary form must reproduce the above copyright notice,
+# this list of conditions and the following disclaimer in the documentation
+# and/or other materials provided with the distribution.
+# 3. Neither the name of the copyright holder nor the names of its contributors
+# may be used to endorse or promote products derived from this software
+# without specific prior written permission.
+#
+# THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS"
+# AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE
+# IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE
+# ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE
+# LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR
+# CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF
+# SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS
+# INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN
+# CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE)
+# ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE, EVEN IF ADVISED OF THE
+# POSSIBILITY OF SUCH DAMAGE.
+
+import os
+import json
+import sqlite3
+from threading import RLock
+from concurrent.futures import ThreadPoolExecutor
+import re
+
+class KenobiDB:
+ """
+ A lightweight document-based database built on SQLite. Supports basic
+ operations such as insert, remove, search, update, and asynchronous
+ execution.
+ """
+
+ def __init__(self, file):
+ """
+ Initialize the KenobiDB instance.
+
+ Args:
+ file (str): Path to the SQLite file. If it does not exist,
+ it will be created.
+ """
+ self.file = os.path.expanduser(file)
+ self._lock = RLock()
+ self.executor = ThreadPoolExecutor(max_workers=5)
+ self._regexp_connections = set() # Track connections with REGEXP added
+ self._connection = sqlite3.connect(self.file, check_same_thread=False)
+ self._add_regexp_support(self._connection) # Add REGEXP support lazily
+ self._initialize_db()
+
+ def _initialize_db(self):
+ """
+ Create the table and index if they do not exist, and set
+ journal mode to WAL.
+ """
+ with self._lock:
+ self._connection.execute("""
+ CREATE TABLE IF NOT EXISTS documents (
+ id INTEGER PRIMARY KEY AUTOINCREMENT,
+ data TEXT NOT NULL
+ )
+ """)
+ self._connection.execute("""
+ CREATE INDEX IF NOT EXISTS idx_key
+ ON documents (
+ json_extract(data, '$.key')
+ )
+ """)
+ self._connection.execute("PRAGMA journal_mode=WAL;")
+
+ @staticmethod
+ def _add_regexp_support(conn):
+ """
+ Add REGEXP function support to the SQLite connection.
+ """
+ def regexp(pattern, value):
+ return re.search(pattern, value) is not None
+ conn.create_function("REGEXP", 2, regexp)
+
+ def _get_connection(self):
+ """
+ Return the active SQLite connection.
+ """
+ return self._connection
+
+ def insert(self, document):
+ """
+ Insert a single document (dict) into the database.
+
+ Args:
+ document (dict): The document to insert.
+
+ Returns:
+ bool: True upon successful insertion.
+
+ Raises:
+ TypeError: If the provided document is not a dictionary.
+ """
+ if not isinstance(document, dict):
+ raise TypeError("Must insert a dict")
+ with self._lock:
+ self._connection.execute(
+ "INSERT INTO documents (data) VALUES (?)",
+ (json.dumps(document),)
+ )
+ self._connection.commit()
+ return True
+
+ def insert_many(self, document_list):
+ """
+ Insert multiple documents (list of dicts) into the database.
+
+ Args:
+ document_list (list): The list of documents to insert.
+
+ Returns:
+ bool: True upon successful insertion.
+
+ Raises:
+ TypeError: If the provided object is not a list of dicts.
+ """
+ if (
+ not isinstance(document_list, list)
+ or not all(isinstance(doc, dict) for doc in document_list)
+ ):
+ raise TypeError("Must insert a list of dicts")
+ with self._lock:
+ self._connection.executemany(
+ "INSERT INTO documents (data) VALUES (?)",
+ [(json.dumps(doc),) for doc in document_list]
+ )
+ self._connection.commit()
+ return True
+
+ def remove(self, key, value):
+ """
+ Remove all documents where the given key matches the specified value.
+
+ Args:
+ key (str): The field name to match.
+ value (Any): The value to match.
+
+ Returns:
+ int: Number of documents removed.
+
+ Raises:
+ ValueError: If 'key' is empty or 'value' is None.
+ """
+ if not key or not isinstance(key, str):
+ raise ValueError("key must be a non-empty string")
+ if value is None:
+ raise ValueError("value cannot be None")
+ query = (
+ "DELETE FROM documents "
+ "WHERE json_extract(data, '$.' || ?) = ?"
+ )
+ with self._lock:
+ result = self._connection.execute(query, (key, value))
+ self._connection.commit()
+ return result.rowcount
+
+ def update(self, id_key, id_value, new_dict):
+ """
+ Update documents that match (id_key == id_value) by merging new_dict.
+
+ Args:
+ id_key (str): The field name to match.
+ id_value (Any): The value to match.
+ new_dict (dict): A dictionary of changes to apply.
+
+ Returns:
+ bool: True if at least one document was updated, False otherwise.
+
+ Raises:
+ TypeError: If new_dict is not a dict.
+ ValueError: If id_key is invalid or id_value is None.
+ """
+ if not isinstance(new_dict, dict):
+ raise TypeError("new_dict must be a dictionary")
+ if not id_key or not isinstance(id_key, str):
+ raise ValueError("id_key must be a non-empty string")
+ if id_value is None:
+ raise ValueError("id_value cannot be None")
+
+ select_query = (
+ "SELECT data FROM documents "
+ "WHERE json_extract(data, '$.' || ?) = ?"
+ )
+ update_query = (
+ "UPDATE documents "
+ "SET data = ? "
+ "WHERE json_extract(data, '$.' || ?) = ?"
+ )
+ with self._lock:
+ cursor = self._connection.execute(select_query, (id_key, id_value))
+ documents = cursor.fetchall()
+ if not documents:
+ return False
+ for row in documents:
+ document = json.loads(row[0])
+ if not isinstance(document, dict):
+ continue
+ document.update(new_dict)
+ self._connection.execute(
+ update_query,
+ (json.dumps(document), id_key, id_value)
+ )
+ self._connection.commit()
+ return True
+
+ def purge(self):
+ """
+ Remove all documents from the database.
+
+ Returns:
+ bool: True upon successful purge.
+ """
+ with self._lock:
+ self._connection.execute("DELETE FROM documents")
+ self._connection.commit()
+ return True
+
+ def all(self, limit=100, offset=0):
+ """
+ Return a paginated list of all documents.
+
+ Args:
+ limit (int): The maximum number of documents to return.
+ offset (int): The starting point for retrieval.
+
+ Returns:
+ list: A list of all documents (dicts).
+ """
+ query = "SELECT data FROM documents LIMIT ? OFFSET ?"
+ with self._lock:
+ cursor = self._connection.execute(query, (limit, offset))
+ return [json.loads(row[0]) for row in cursor.fetchall()]
+
+ def search(self, key, value, limit=100, offset=0):
+ """
+ Return a list of documents matching (key == value).
+
+ Args:
+ key (str): The document field to match on.
+ value (Any): The value for which to search.
+ limit (int): The maximum number of documents to return.
+ offset (int): The starting point for retrieval.
+
+ Returns:
+ list: A list of matching documents (dicts).
+ """
+ if not key or not isinstance(key, str):
+ raise ValueError("Key must be a non-empty string")
+
+ query = (
+ "SELECT data FROM documents "
+ "WHERE json_extract(data, '$.' || ?) = ? "
+ "LIMIT ? OFFSET ?"
+ )
+ with self._lock:
+ cursor = self._connection.execute(query, (key, value, limit, offset))
+ return [json.loads(row[0]) for row in cursor.fetchall()]
+
+ def search_pattern(self, key, pattern, limit=100, offset=0):
+ """
+ Search documents matching a regex pattern.
+
+ Args:
+ key (str): The document field to match on.
+ pattern (str): The regex pattern to match.
+ limit (int): The maximum number of documents to return.
+ offset (int): The starting point for retrieval.
+
+ Returns:
+ list: A list of matching documents (dicts).
+
+ Raises:
+ ValueError: If the key or pattern is invalid.
+ """
+ if not key or not isinstance(key, str):
+ raise ValueError("key must be a non-empty string")
+ if not pattern or not isinstance(pattern, str):
+ raise ValueError("pattern must be a non-empty string")
+
+ query = """
+ SELECT data FROM documents
+ WHERE json_extract(data, '$.' || ?) REGEXP ?
+ LIMIT ? OFFSET ?
+ """
+ with self._lock:
+ cursor = self._connection.execute(query, (key, pattern, limit, offset))
+ return [json.loads(row[0]) for row in cursor.fetchall()]
+
+ def find_any(self, key, value_list):
+ """
+ Return documents where key matches any value in value_list.
+
+ Args:
+ key (str): The document field to match on.
+ value_list (list): A list of possible values.
+
+ Returns:
+ list: A list of matching documents.
+ """
+ placeholders = ", ".join(["?"] * len(value_list))
+ query = f"""
+ SELECT DISTINCT documents.data
+ FROM documents, json_each(documents.data, '$.' || ?)
+ WHERE json_each.value IN ({placeholders})
+ """
+ with self._lock:
+ cursor = self._connection.execute(query, [key] + value_list)
+ return [json.loads(row[0]) for row in cursor.fetchall()]
+
+ def find_all(self, key, value_list):
+ """
+ Return documents where the key contains all values in value_list.
+
+ Args:
+ key (str): The field to match.
+ value_list (list): The required values to match.
+
+ Returns:
+ list: A list of matching documents.
+ """
+ placeholders = ", ".join(["?"] * len(value_list))
+ query = f"""
+ SELECT documents.data
+ FROM documents
+ WHERE (
+ SELECT COUNT(DISTINCT value)
+ FROM json_each(documents.data, '$.' || ?)
+ WHERE value IN ({placeholders})
+ ) = ?
+ """
+ with self._lock:
+ cursor = self._connection.execute(
+ query, [key] + value_list + [len(value_list)]
+ )
+ return [json.loads(row[0]) for row in cursor.fetchall()]
+
+ def execute_async(self, func, *args, **kwargs):
+ """
+ Execute a function asynchronously using a thread pool.
+
+ Args:
+ func (callable): The function to execute.
+ *args: Arguments for the function.
+ **kwargs: Keyword arguments for the function.
+
+ Returns:
+ concurrent.futures.Future: A Future object representing
+ the execution.
+ """
+ return self.executor.submit(func, *args, **kwargs)
+
+ def close(self):
+ """
+ Shutdown the thread pool executor and close the database connection.
+ """
+ self.executor.shutdown()
+ with self._lock:
+ self._connection.close()
+
diff --git a/examples/chatroom/templates/chat.html b/examples/chatroom/templates/chat.html
new file mode 100644
index 0000000..5b62609
--- /dev/null
+++ b/examples/chatroom/templates/chat.html
@@ -0,0 +1,207 @@
+<!DOCTYPE html>
+<html lang="en">
+<head>
+ <meta charset="UTF-8">
+ <meta name="viewport" content="width=device-width, initial-scale=1.0">
+ <title>Chat App</title>
+ <!-- Import Google Fonts: Permanent Marker and Montserrat for the logo -->
+ <link href="https://fonts.googleapis.com/css2?family=Permanent+Marker&family=Montserrat:wght@700&display=swap" rel="stylesheet">
+ <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.7.5/socket.io.min.js"></script>
+ <script src="https://cdn.tailwindcss.com"></script>
+ <style>
+ /* Custom animations */
+ @keyframes fadeIn {
+ from { opacity: 0; transform: translateY(10px); }
+ to { opacity: 1; transform: translateY(0); }
+ }
+ .message-bubble {
+ animation: fadeIn 0.3s ease-out;
+ }
+ /* Hide scrollbar but keep functionality */
+ #messages::-webkit-scrollbar, #user-list::-webkit-scrollbar {
+ display: none;
+ }
+ #messages, #user-list {
+ -ms-overflow-style: none;
+ scrollbar-width: none;
+ }
+ /* Mobile layout */
+ @media (max-width: 1023px) {
+ #chat-container {
+ min-height: calc(100vh - env(safe-area-inset-top) - env(safe-area-inset-bottom));
+ margin-top: env(safe-area-inset-top);
+ margin-bottom: env(safe-area-inset-bottom);
+ border-radius: 0;
+ }
+ #chat-area {
+ padding-bottom: calc(1rem + env(safe-area-inset-bottom));
+ }
+ #user-list {
+ max-height: 150px;
+ flex-shrink: 0;
+ }
+ }
+ /* Logo styling */
+ .app-name-vibe {
+ font-family: 'Permanent Marker', cursive;
+ font-size: 100px;
+ color: #2c3e50;
+ text-shadow: 2px 2px 4px rgba(0, 0, 0, 0.2);
+ }
+ .app-name-chat {
+ font-family: 'Montserrat', sans-serif;
+ font-size: 50px;
+ font-weight: 700;
+ color: #2c3e50;
+ text-shadow: 2px 2px 4px rgba(0, 0, 0, 0.2);
+ }
+ </style>
+</head>
+<body class="bg-gray-100 min-h-screen flex flex-col items-center justify-start pt-4 md:pt-8">
+ <!-- VibeChat logo at the top center with mixed fonts -->
+ <div class="w-full text-center mb-4 md:mb-6">
+ <h1 class="inline-flex items-baseline">
+ <span class="app-name-vibe">Vibechat</span>
+ </h1>
+ </div>
+
+ {% if is_index %}
+ <div id="create-channel-form" class="bg-white rounded-2xl shadow-lg p-6 w-full max-w-md m-4">
+ <h2 class="text-2xl font-semibold text-gray-800 mb-4">Create New Channel</h2>
+ <div id="error-create" class="text-red-500 text-sm mb-4"></div>
+ <input type="text" id="channel-name-create" placeholder="Channel name" class="w-full p-3 mb-4 rounded-lg border border-gray-300 focus:outline-none focus:ring-2 focus:ring-blue-500">
+ <input type="password" id="channel-password" placeholder="Password (optional)" class="w-full p-3 mb-4 rounded-lg border border-gray-300 focus:outline-none focus:ring-2 focus:ring-blue-500">
+ <button onclick="createChannel()" class="w-full bg-blue-500 text-white p-3 rounded-lg hover:bg-blue-600 transition">Create Channel</button>
+ </div>
+ {% else %}
+ <div id="join-form" class="bg-white rounded-2xl shadow-lg p-6 w-full max-w-md m-4">
+ <h2 class="text-2xl font-semibold text-gray-800 mb-4">Join Channel: {{ channel }}</h2>
+ <div id="error-join" class="text-red-500 text-sm mb-4"></div>
+ <input type="text" id="username" placeholder="Your username" class="w-full p-3 mb-4 rounded-lg border border-gray-300 focus:outline-none focus:ring-2 focus:ring-blue-500">
+ <input type="password" id="password" placeholder="Channel password (if required)" class="w-full p-3 mb-4 rounded-lg border border-gray-300 focus:outline-none focus:ring-2 focus:ring-blue-500">
+ <button onclick="joinChannel()" class="w-full bg-blue-500 text-white p-3 rounded-lg hover:bg-blue-600 transition">Join</button>
+ </div>
+ <div id="chat-container" style="display: none;" class="bg-white shadow-lg w-full flex flex-col lg:flex-row h-[calc(100vh-2rem)] md:rounded-2xl">
+ <div id="chat-area" class="flex-1 flex flex-col p-4 lg:p-6">
+ <div id="messages" class="flex-1 overflow-y-auto mb-4 space-y-2"></div>
+ <div id="input-container" class="flex items-center px-2">
+ <input type="text" id="message-input" placeholder="Type a message..." class="flex-1 p-3 rounded-full border border-gray-300 focus:outline-none focus:ring-2 focus:ring-blue-500 text-sm">
+ <button onclick="sendMessage()" class="ml-2 bg-blue-500 text-white p-2 rounded-full hover:bg-blue-600 transition w-10 h-10 flex items-center justify-center">
+ <svg xmlns="http://www.w3.org/2000/svg" fill="none" viewBox="0 0 24 24" stroke="currentColor" class="w-5 h-5">
+ <path stroke-linecap="round" stroke-linejoin="round" stroke-width="2" d="M12 19l9 2-9-18-9 18 9-2zm0 0v-8" />
+ </svg>
+ </button>
+ </div>
+ </div>
+ <div id="user-list" class="w-full lg:w-80 bg-gray-50 p-4 lg:p-6 border-t lg:border-t-0 lg:border-l border-gray-200 overflow-y-auto">
+ <h3 class="text-base font-semibold text-gray-800 mb-3">Connected Users</h3>
+ <div id="users" class="text-gray-600 text-sm"></div>
+ </div>
+ </div>
+ {% endif %}
+
+ <script>
+ const socket = io();
+ let currentChannel = {% if channel %}'{{ channel }}'{% else %}null{% endif %};
+ let currentUsername = null;
+
+ // Pre-fill username input with stored username
+ const storedUsername = localStorage.getItem('chatUsername');
+ if (storedUsername && document.getElementById('username')) {
+ document.getElementById('username').value = storedUsername;
+ }
+
+ socket.on('connect', () => {
+ console.log('Connected to server');
+ });
+
+ socket.on('error', (data) => {
+ const errorDiv = document.getElementById(currentChannel ? 'error-join' : 'error-create');
+ errorDiv.textContent = data.message;
+ setTimeout(() => errorDiv.textContent = '', 3000);
+ });
+
+ socket.on('message', (data) => {
+ const messages = document.getElementById('messages');
+ const messageDiv = document.createElement('div');
+ const isOwnMessage = currentUsername && data.username === currentUsername;
+ messageDiv.className = `message-bubble max-w-[80%] p-3 rounded-2xl ${isOwnMessage ? 'bg-blue-500 text-white ml-auto' : 'bg-gray-200 text-gray-800 mr-auto'}`;
+ messageDiv.innerHTML = `
+ <div class="text-sm">${data.username}: ${data.message}</div>
+ <div class="text-xs ${isOwnMessage ? 'text-blue-200' : 'text-gray-500'} mt-1">${new Date(data.timestamp).toLocaleTimeString()}</div>
+ `;
+ messages.appendChild(messageDiv);
+ messages.scrollTop = messages.scrollHeight;
+ });
+
+ socket.on('user_list', (users) => {
+ const usersDiv = document.getElementById('users');
+ if (users && Array.isArray(users)) {
+ usersDiv.innerHTML = users.length > 0
+ ? users.map(user => `<div class="py-1">${user}</div>`).join('')
+ : '<div class="text-gray-500">No users connected</div>';
+ } else {
+ usersDiv.innerHTML = '<div class="text-gray-500">No users connected</div>';
+ }
+ console.log('Updated user list:', users);
+ });
+
+ socket.on('channel_created', (data) => {
+ window.location.href = `/channel/${encodeURIComponent(data.channel)}`;
+ });
+
+ socket.on('join_success', () => {
+ currentUsername = document.getElementById('username').value.trim();
+ localStorage.setItem('chatUsername', currentUsername);
+ document.getElementById('join-form').style.display = 'none';
+ document.getElementById('chat-container').style.display = 'flex';
+ console.log('Successfully joined channel:', currentChannel, 'as', currentUsername);
+ });
+
+ function createChannel() {
+ const channel = document.getElementById('channel-name-create').value.trim();
+ const password = document.getElementById('channel-password').value;
+ if (channel) {
+ socket.emit('create_channel', { channel, password });
+ } else {
+ const errorDiv = document.getElementById('error-create');
+ errorDiv.textContent = 'Channel name cannot be empty';
+ setTimeout(() => errorDiv.textContent = '', 3000);
+ }
+ }
+
+ function joinChannel() {
+ const username = document.getElementById('username').value.trim();
+ const password = document.getElementById('password').value;
+ if (!username) {
+ const errorDiv = document.getElementById('error-join');
+ errorDiv.textContent = 'Username cannot be empty';
+ setTimeout(() => errorDiv.textContent = '', 3000);
+ return;
+ }
+ currentUsername = username; // Set username immediately for past messages
+ socket.emit('join_channel', {
+ channel: currentChannel,
+ username,
+ password
+ });
+ }
+
+ function sendMessage() {
+ const messageInput = document.getElementById('message-input');
+ const message = messageInput.value.trim();
+ if (message && currentChannel) {
+ socket.emit('message', { channel: currentChannel, message });
+ messageInput.value = '';
+ }
+ }
+
+ // Allow sending message with Enter key
+ document.getElementById('message-input')?.addEventListener('keypress', (e) => {
+ if (e.key === 'Enter') {
+ sendMessage();
+ }
+ });
+ </script>
+</body>
+</html>
diff --git a/examples/socketio/chatroom.py b/examples/socketio/chatroom.py
index d22672c..0503c22 100644
--- a/examples/socketio/chatroom.py
+++ b/examples/socketio/chatroom.py
@@ -1,8 +1,17 @@
import socketio
from MicroPie import App
+from kenobi import KenobiDB
+from datetime import datetime
+import asyncio
+
+# KenobiDB setup
+db = KenobiDB("chat.db")
# Create a Socket.IO server with CORS support
-sio = socketio.AsyncServer(async_mode="asgi", cors_allowed_origins="*") # Allow all origins
+sio = socketio.AsyncServer(async_mode="asgi", cors_allowed_origins="*")
+
+# Store connected users
+connected_users = {}
# Create the MicroPie server
class MyApp(App):
@@ -13,19 +22,71 @@ class MyApp(App):
@sio.event
async def connect(sid, environ):
print(f"Client connected: {sid}")
+ # Add user with temporary username
+ connected_users[sid] = f"User_{sid[:4]}"
+ await update_user_list()
+
+ # Send recent messages to the newly connected client
+ recent_messages = db.search("type", "message", limit=50)
+ for msg in recent_messages:
+ await sio.emit('message', {
+ 'username': msg['username'],
+ 'message': msg['message'],
+ 'timestamp': msg['timestamp']
+ }, room=sid)
@sio.event
async def disconnect(sid):
print(f"Client disconnected: {sid}")
+ # Remove user from connected users
+ if sid in connected_users:
+ del connected_users[sid]
+ await update_user_list()
@sio.event
-async def message(sid, data):
- print(f"Received message from {sid}: {data}")
- # Broadcast the message to all connected clients
- await sio.emit("message", f"User: {data}", room=None)
+async def set_username(sid, data):
+ """Handle username setting"""
+ username = data.get('username', '').strip()
+ if username and len(username) <= 20: # Basic validation
+ if username in connected_users.values():
+ await sio.emit('error', {'message': 'Invalid username'}, room=sid)
+ connected_users[sid] = username
+ print(f"User {sid} set username to {username}")
+ await update_user_list()
+ else:
+ await sio.emit('error', {'message': 'Invalid username'}, room=sid)
[email protected]
+async def message(sid, data):
+ """Handle and store messages"""
+ username = connected_users.get(sid, f"User_{sid[:4]}")
+ message = data.get('message', '').strip()
+
+ if message:
+ # Store message in KenobiDB
+ message_doc = {
+ 'type': 'message',
+ 'username': username,
+ 'message': message,
+ 'timestamp': datetime.utcnow().isoformat()
+ }
+ db.insert(message_doc)
+
+ # Broadcast message to all clients
+ await sio.emit('message', {
+ 'username': username,
+ 'message': message,
+ 'timestamp': message_doc['timestamp']
+ }, room=None)
+async def update_user_list():
+ """Broadcast updated user list to all clients"""
+ await sio.emit('user_list', list(connected_users.values()), room=None)
# Attach Socket.IO to the ASGI app
asgi_app = MyApp()
app = socketio.ASGIApp(sio, asgi_app)
+
+# Ensure database is closed properly on shutdown
+import atexit
+atexit.register(db.close)
diff --git a/examples/socketio/templates/chat.html b/examples/socketio/templates/chat.html
index ecb11f6..4922578 100644
--- a/examples/socketio/templates/chat.html
+++ b/examples/socketio/templates/chat.html
@@ -1,28 +1,122 @@
+<!DOCTYPE html>
<html>
<head>
-<script src="https://cdn.socket.io/4.5.4/socket.io.min.js"></script>
-<script>
-var socket = io("http://localhost:8000");
-socket.on("connect", function() {
- console.log("Connected to Socket.IO server");
-});
-socket.on("message", function(data) {
- document.getElementById("output").innerHTML += data + "<br>";
-});
-function sendMessage() {
- var message = document.getElementById("message").value;
- socket.send(message);
- document.getElementById("message").value = ""; // Clear input after sending
-}
-window.onbeforeunload = function() {
- socket.disconnect();
-};
-</script>
+ <title>Socket.IO Chat</title>
+ <script src="https://cdn.socket.io/4.7.5/socket.io.min.js"></script>
+ <style>
+ body {
+ font-family: Arial, sans-serif;
+ max-width: 800px;
+ margin: 0 auto;
+ padding: 20px;
+ }
+ #chat-container {
+ border: 1px solid #ccc;
+ padding: 10px;
+ height: 400px;
+ overflow-y: scroll;
+ margin-bottom: 10px;
+ }
+ #user-list {
+ width: 200px;
+ float: right;
+ border: 1px solid #ccc;
+ padding: 10px;
+ margin-left: 10px;
+ }
+ #message-form {
+ display: flex;
+ gap: 10px;
+ }
+ #message-input {
+ flex-grow: 1;
+ padding: 5px;
+ }
+ #username-container {
+ margin-bottom: 10px;
+ }
+ .message {
+ margin: 5px 0;
+ }
+ .error {
+ color: red;
+ }
+ </style>
</head>
- <body>
- <h1>Socket.IO Chat</h1>
- <input type="text" id="message" placeholder="Type a message">
- <button onclick="sendMessage()">Send</button>
- <div id="output"></div>
- </body>
+<body>
+ <div id="username-container">
+ <input type="text" id="username-input" placeholder="Enter your username">
+ <button onclick="setUsername()">Set Username</button>
+ </div>
+ <div id="user-list">
+ <h3>Connected Users</h3>
+ <ul id="users"></ul>
+ </div>
+ <div id="chat-container"></div>
+ <div id="message-form">
+ <input type="text" id="message-input" placeholder="Type a message...">
+ <button onclick="sendMessage()">Send</button>
+ </div>
+
+ <script>
+ const socket = io();
+
+ socket.on('connect', () => {
+ console.log('Connected to server');
+ });
+
+ socket.on('message', (data) => {
+ const chatContainer = document.getElementById('chat-container');
+ const messageElement = document.createElement('div');
+ messageElement.className = 'message';
+ messageElement.innerHTML = `<strong>${data.username}</strong> (${new Date(data.timestamp).toLocaleTimeString()}): ${data.message}`;
+ chatContainer.appendChild(messageElement);
+ chatContainer.scrollTop = chatContainer.scrollHeight;
+ });
+
+ socket.on('user_list', (users) => {
+ const userList = document.getElementById('users');
+ userList.innerHTML = '';
+ users.forEach(user => {
+ const li = document.createElement('li');
+ li.textContent = user;
+ userList.appendChild(li);
+ });
+ });
+
+ socket.on('error', (data) => {
+ const chatContainer = document.getElementById('chat-container');
+ const errorElement = document.createElement('div');
+ errorElement.className = 'error';
+ errorElement.textContent = data.message;
+ chatContainer.appendChild(errorElement);
+ });
+
+ function setUsername() {
+ const usernameInput = document.getElementById('username-input');
+ const username = usernameInput.value.trim();
+ if (username) {
+ socket.emit('set_username', { username });
+ usernameInput.disabled = true;
+ document.querySelector('#username-container button').disabled = true;
+ }
+ }
+
+ function sendMessage() {
+ const messageInput = document.getElementById('message-input');
+ const message = messageInput.value.trim();
+ if (message) {
+ socket.emit('message', { message });
+ messageInput.value = '';
+ }
+ }
+
+ // Allow sending message with Enter key
+ document.getElementById('message-input').addEventListener('keypress', (e) => {
+ if (e.key === 'Enter') {
+ sendMessage();
+ }
+ });
+ </script>
+</body>
</html>
diff --git a/unstable/MicroPie.py b/unstable/MicroPie.py
new file mode 100644
index 0000000..4a0dff0
--- /dev/null
+++ b/unstable/MicroPie.py
@@ -0,0 +1,658 @@
+"""
+MicroPie: A simple Python ultra-micro web framework with ASGI
+support. https://patx.github.io/micropie
+
+Copyright 2025 Harrison Erd
+
+Redistribution and use in source and binary forms, with or without
+modification, are permitted provided that the following conditions are met:
+
+1. Redistributions of source code must retain the above copyright notice,
+ this list of conditions and the following disclaimer.
+
+2. Redistributions in binary form must reproduce the above copyright notice,
+ this list of conditions and the following disclaimer in the documentation
+ and/or other materials provided with the distribution.
+
+3. Neither the name of the copyright holder nor the names of its
+ contributors may be used to endorse or promote products derived from this
+ software without specific prior written permission.
+
+THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS
+IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO,
+THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR
+PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR
+CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL,
+EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT LIMITED TO,
+PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, DATA, OR PROFITS;
+OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY THEORY OF LIABILITY,
+WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT (INCLUDING NEGLIGENCE OR
+OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS SOFTWARE,
+EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+"""
+
+import asyncio
+import contextvars
+import inspect
+import json
+import os
+import re
+import time
+import uuid
+from abc import ABC, abstractmethod
+from typing import Any, Awaitable, Callable, Dict, List, Optional, Tuple
+from urllib.parse import parse_qs
+
+try:
+ from jinja2 import Environment, FileSystemLoader, select_autoescape
+ JINJA_INSTALLED = True
+except ImportError:
+ JINJA_INSTALLED = False
+
+try:
+ import aiofiles, aiofiles.os
+ from multipart import PushMultipartParser, MultipartSegment
+ MULTIPART_INSTALLED = True
+except ImportError:
+ MULTIPART_INSTALLED = False
+
+
+# -----------------------------
+# Session Backend Abstraction
+# -----------------------------
+SESSION_TIMEOUT: int = 8 * 3600 # Default 8 hours
+
+class SessionBackend(ABC):
+ @abstractmethod
+ async def load(self, session_id: str) -> Dict[str, Any]:
+ """
+ Load session data given a session ID.
+
+ Args:
+ session_id: str
+ """
+ pass
+
+ @abstractmethod
+ async def save(self, session_id: str, data: Dict[str, Any], timeout: int) -> None:
+ """
+ Save session data.
+
+ Args:
+ session_id: str
+ data: Dict
+ timeout: int (in seconds)
+ """
+ pass
+
+class InMemorySessionBackend(SessionBackend):
+ def __init__(self):
+ self.sessions: Dict[str, Dict[str, Any]] = {}
+ self.last_access: Dict[str, float] = {}
+
+ async def load(self, session_id: str) -> Dict[str, Any]:
+ now = time.time()
+ if session_id in self.sessions and (now - self.last_access.get(session_id, now)) < SESSION_TIMEOUT:
+ self.last_access[session_id] = now
+ return self.sessions[session_id]
+ return {}
+
+ async def save(self, session_id: str, data: Dict[str, Any], timeout: int) -> None:
+ self.sessions[session_id] = data
+ self.last_access[session_id] = time.time()
+
+
+# -----------------------------
+# Request Object
+# -----------------------------
+current_request: contextvars.ContextVar[Any] = contextvars.ContextVar("current_request")
+
+class Request:
+ """Represents an HTTP request in the MicroPie framework."""
+ def __init__(self, scope: Dict[str, Any]) -> None:
+ """
+ Initialize a new Request instance.
+
+ Args:
+ scope: The ASGI scope dictionary for the request.
+ """
+ self.scope: Dict[str, Any] = scope
+ self.method: str = scope["method"]
+ self.path_params: List[str] = []
+ self.query_params: Dict[str, List[str]] = {}
+ self.body_params: Dict[str, List[str]] = {}
+ self.get_json: Any = {}
+ self.session: Dict[str, Any] = {}
+ self.files: Dict[str, Any] = {}
+ self.headers: Dict[str, str] = {
+ k.decode("utf-8", errors="replace").lower(): v.decode("utf-8", errors="replace")
+ for k, v in scope.get("headers", [])
+ }
+
+
+# -----------------------------
+# WebSocket Object
+# -----------------------------
+class WebSocket:
+ """Represents a WebSocket connection in the MicroPie framework."""
+ def __init__(self, scope: Dict[str, Any], receive: Callable[[], Awaitable[Dict[str, Any]]], send: Callable[[Dict[str, Any]], Awaitable[None]]) -> None:
+ """
+ Initialize a new WebSocket instance.
+
+ Args:
+ scope: The ASGI scope dictionary for the WebSocket connection.
+ receive: The callable to receive ASGI events.
+ send: The callable to send ASGI events.
+ """
+ self.scope = scope
+ self.receive = receive
+ self.send = send
+ self.query_params = parse_qs(scope.get("query_string", b"").decode("utf-8", "ignore"))
+ self.headers = {
+ k.decode("utf-8", errors="replace").lower(): v.decode("utf-8", errors="replace")
+ for k, v in scope.get("headers", [])
+ }
+
+ async def accept(self) -> None:
+ """Accept the WebSocket connection."""
+ await self.send({
+ "type": "websocket.accept"
+ })
+
+ async def send_text(self, data: str) -> None:
+ """Send text data over the WebSocket."""
+ await self.send({
+ "type": "websocket.send",
+ "text": data
+ })
+
+ async def send_json(self, data: Any) -> None:
+ """Send JSON data over the WebSocket."""
+ await self.send({
+ "type": "websocket.send",
+ "text": json.dumps(data)
+ })
+
+ async def receive_text(self) -> str:
+ """Receive text data from the WebSocket."""
+ message = await self.receive()
+ if message["type"] == "websocket.disconnect":
+ raise ConnectionError("WebSocket disconnected")
+ return message.get("text", "")
+
+ async def receive_json(self) -> Any:
+ """Receive JSON data from the WebSocket."""
+ text = await self.receive_text()
+ return json.loads(text)
+
+ async def close(self, code: int = 1000) -> None:
+ """Close the WebSocket connection."""
+ await self.send({
+ "type": "websocket.close",
+ "code": code
+ })
+
+
+# -----------------------------
+# Middleware Abstraction
+# -----------------------------
+class HttpMiddleware(ABC):
+ """
+ Pluggable middleware class that allows hooking into the request lifecycle.
+ """
+
+ @abstractmethod
+ async def before_request(self, request: Request) -> None:
+ """
+ Called before the request is processed.
+ """
+ pass
+
+ @abstractmethod
+ async def after_request(
+ self,
+ request: Request,
+ status_code: int,
+ response_body: Any,
+ extra_headers: List[Tuple[str, str]]
+ ) -> None:
+ """
+ Called after the request is processed, but before the final response
+ is sent to the client. You may alter the status_code, response_body,
+ or extra_headers if needed.
+ """
+ pass
+
+
+# -----------------------------
+# Application Base
+# -----------------------------
+class App:
+ """
+ ASGI application for handling HTTP and WebSocket requests in MicroPie.
+ It supports pluggable session backends via the 'session_backend' attribute
+ and pluggable middlewares via the 'middlewares' list.
+ """
+
+ def __init__(self, session_backend: Optional[SessionBackend] = None) -> None:
+ if JINJA_INSTALLED:
+ self.env = Environment(
+ loader=FileSystemLoader("templates"),
+ autoescape=select_autoescape(["html", "xml"]),
+ enable_async=True
+ )
+ else:
+ self.env = None
+ self.session_backend: SessionBackend = session_backend or InMemorySessionBackend()
+ self.middlewares: List[HttpMiddleware] = []
+
+ @property
+ def request(self) -> Request:
+ """
+ Retrieve the current request from the context variable.
+
+ Returns: The current Request instance.
+ """
+ return current_request.get()
+
+ async def __call__(
+ self,
+ scope: Dict[str, Any],
+ receive: Callable[[], Awaitable[Dict[str, Any]]],
+ send: Callable[[Dict[str, Any]], Awaitable[None]]
+ ) -> None:
+ """
+ ASGI callable interface for the server.
+
+ Args:
+ scope: The ASGI scope dictionary.
+ receive: The callable to receive ASGI events.
+ send: The callable to send ASGI events.
+ """
+ if scope["type"] == "http":
+ await self._asgi_app_http(scope, receive, send)
+ elif scope["type"] == "websocket":
+ await self._handle_websocket(scope, receive, send)
+ else:
+ pass # Handle lifespan and other scope types in the future.
+
+ async def _asgi_app_http(
+ self,
+ scope: Dict[str, Any],
+ receive: Callable[[], Awaitable[Dict[str, Any]]],
+ send: Callable[[Dict[str, Any]], Awaitable[None]]
+ ) -> None:
+ """
+ ASGI application entry point for handling HTTP requests.
+
+ Args:
+ scope: The ASGI scope dictionary.
+ receive: The callable to receive ASGI events.
+ send: The callable to send ASGI events.
+ """
+ request: Request = Request(scope)
+ token = current_request.set(request)
+ status_code: int = 200
+ response_body: Any = ""
+ extra_headers: List[Tuple[str, str]] = []
+ try:
+ # Middleware: before request
+ for mw in self.middlewares:
+ if result := await mw.before_request(request):
+ status_code, response_body, extra_headers = (
+ result["status_code"],
+ result["body"],
+ result.get("headers", []),
+ )
+ await self._send_response(send, status_code, response_body, extra_headers)
+ return
+
+ # Parse path and find handler
+ path: str = scope["path"].lstrip("/")
+ parts: List[str] = path.split("/") if path else []
+ func_name: str = parts[0] if parts else "index"
+ if func_name.startswith("_"):
+ await self._send_response(send, 404, "404 Not Found")
+ return
+
+ request.path_params = parts[1:] if len(parts) > 1 else []
+ handler = getattr(self, func_name, None) or getattr(self, "index", None)
+ if not handler:
+ await self._send_response(send, 404, "404 Not Found")
+ return
+
+ # Parse request details
+ request.query_params = parse_qs(scope.get("query_string", b"").decode("utf-8", "ignore"))
+ cookies = self._parse_cookies(request.headers.get("cookie", ""))
+ request.session = await self.session_backend.load(cookies.get("session_id", "")) or {}
+
+ # Parse body parameters.
+ if request.method in ("POST", "PUT", "PATCH"):
+ body_data = bytearray()
+ while True:
+ msg: Dict[str, Any] = await receive()
+ body_data += msg.get("body", b"")
+ if not msg.get("more_body"):
+ break
+ content_type = request.headers.get("content-type", "")
+ if "application/json" in content_type:
+ try:
+ request.get_json = json.loads(body_data.decode("utf-8"))
+ if isinstance(request.get_json, dict):
+ request.body_params = {k: [str(v)] for k, v in request.get_json.items()}
+ except Exception as e:
+ print(f"Request error: {e}")
+ await self._send_response(send, 400, "400 Bad Request: Bad JSON")
+ return
+ elif "multipart/form-data" in content_type:
+ if boundary := re.search(r"boundary=([^;]+)", content_type):
+ reader = asyncio.StreamReader()
+ reader.feed_data(body_data)
+ reader.feed_eof()
+ request.body_params, request.files = await self._parse_multipart(reader, boundary.group(1).encode("utf-8"))
+ else:
+ await self._send_response(send, 400, "400 Bad Request: Missing boundary")
+ return
+ else:
+ request.body_params = parse_qs(body_data.decode("utf-8", "ignore"))
+
+ # Build function arguments from path, query, body, files, and session values.
+ sig = inspect.signature(handler)
+ func_args: List[Any] = []
+ for param in sig.parameters.values():
+ param_value = None
+ if request.path_params:
+ param_value = request.path_params.pop(0)
+ elif param.name in request.query_params:
+ param_value = request.query_params[param.name][0]
+ elif param.name in request.body_params:
+ param_value = request.body_params[param.name][0] if request.body_params[param.name] else ""
+ elif param.name in request.files:
+ param_value = request.files[param.name]
+ elif param.name in request.session:
+ param_value = request.session[param.name]
+ elif param.default is not param.empty:
+ param_value = param.default
+ else:
+ status_code = 400
+ response_body = f"400 Bad Request: Missing required parameter '{param.name}'"
+ await self._send_response(send, status_code, response_body)
+ return
+ func_args.append(param_value)
+
+ if handler == getattr(self, "index", None) and not func_args and path:
+ await self._send_response(send, 404, "404 Not Found")
+ return
+
+ # Execute handler
+ try:
+ result = await handler(*func_args) if inspect.iscoroutinefunction(handler) else handler(*func_args)
+ except Exception as e:
+ print(f"Request error: {e}")
+ await self._send_response(send, 500, "500 Internal Server Error")
+ return
+
+ # Normalize response
+ if isinstance(result, tuple):
+ status_code, response_body = result[0], result[1]
+ extra_headers = result[2] if len(result) > 2 else []
+ else:
+ response_body = result
+ if isinstance(response_body, (dict, list)):
+ response_body = json.dumps(response_body)
+ extra_headers.append(("Content-Type", "application/json"))
+
+ # Save session
+ if request.session:
+ session_id = cookies.get("session_id") or str(uuid.uuid4())
+ await self.session_backend.save(session_id, request.session, SESSION_TIMEOUT)
+ if not cookies.get("session_id"):
+ extra_headers.append(("Set-Cookie", f"session_id={session_id}; Path=/; SameSite=Lax; HttpOnly; Secure;"))
+
+ # Middleware: after request
+ for mw in self.middlewares:
+ if result := await mw.after_request(request, status_code, response_body, extra_headers):
+ status_code, response_body, extra_headers = (
+ result.get("status_code", status_code),
+ result.get("body", response_body),
+ result.get("headers", extra_headers)
+ )
+
+ await self._send_response(send, status_code, response_body, extra_headers)
+
+ finally:
+ current_request.reset(token)
+
+ async def _handle_websocket(
+ self,
+ scope: Dict[str, Any],
+ receive: Callable[[], Awaitable[Dict[str, Any]]],
+ send: Callable[[Dict[str, Any]], Awaitable[None]]
+ ) -> None:
+ """
+ Handle WebSocket connections.
+
+ Args:
+ scope: The ASGI scope dictionary for the WebSocket connection.
+ receive: The callable to receive ASGI events.
+ send: The callable to send ASGI events.
+ """
+ websocket = WebSocket(scope, receive, send)
+ path: str = scope["path"].lstrip("/")
+ parts: List[str] = path.split("/") if path else []
+ func_name: str = parts[0] if parts else "ws_index"
+ if func_name.startswith("_"):
+ await websocket.close(1008) # Policy violation
+ return
+
+ handler_name = f"ws_{func_name}"
+ handler = getattr(self, handler_name, None) or getattr(self, "ws_index", None)
+ if not handler:
+ await websocket.close(1008) # Policy violation
+ return
+
+ closed = False
+ try:
+ await handler(websocket, parts[1:] if len(parts) > 1 else [])
+ except Exception as e:
+ print(f"WebSocket error: {e}")
+ await websocket.close(1011) # Internal error
+ closed = True
+ finally:
+ if not closed:
+ await websocket.close()
+
+ def _parse_cookies(self, cookie_header: str) -> Dict[str, str]:
+ """
+ Parse the Cookie header and return a dictionary of cookie names and values.
+
+ Args:
+ cookie_header: The raw Cookie header string.
+
+ Returns:
+ A dictionary mapping cookie names to their corresponding values.
+ """
+ cookies: Dict[str, str] = {}
+ if not cookie_header:
+ return cookies
+ for cookie in cookie_header.split(";"):
+ if "=" in cookie:
+ k, v = cookie.strip().split("=", 1)
+ cookies[k] = v
+ return cookies
+
+ async def _parse_multipart(self, reader: asyncio.StreamReader, boundary: bytes):
+ """
+ Asynchronously parses a multipart form-data request.
+
+ This method processes incoming multipart form-data, handling
+ both text fields and file uploads. It reads data from the provided
+ asyncio stream reader and extracts form values and files,
+ saving uploaded files to a designated directory.
+
+ Args:
+ reader (asyncio.StreamReader): The stream reader from which
+ to read the multipart data.
+ boundary (bytes): The boundary string used to separate form
+ fields in the multipart request.
+
+ Returns:
+ tuple[dict, dict]: A tuple containing form_data & files.
+ """
+ if not MULTIPART_INSTALLED:
+ print("For multipart form data support install 'multipart' and 'aiofiles'.")
+ await self._send_response(send, 500, "500 Internal Server Error")
+ return
+
+ with PushMultipartParser(boundary) as parser:
+ form_data: dict = {}
+ files: dict = {}
+ current_field_name: Optional[str] = None
+ current_filename: Optional[str] = None
+ current_content_type: Optional[str] = None
+ current_file: Optional[aiofiles.threadpool.binary.AsyncBufferedIOBase] = None
+ form_value: str = ""
+ upload_directory: str = "uploads"
+ await aiofiles.os.makedirs(upload_directory, exist_ok=True)
+ while not parser.closed:
+ chunk: bytes = await reader.read(65536)
+ if not chunk:
+ break
+ for result in parser.parse(chunk):
+ if isinstance(result, MultipartSegment):
+ current_field_name = result.name
+ current_filename = result.filename
+ current_content_type = None
+ form_value = ""
+ for header, value in result.headerlist:
+ if header.lower() == "content-type":
+ current_content_type = value
+
+ if current_filename:
+ safe_filename: str = f"{uuid.uuid4()}_{current_filename}"
+ safe_filename = re.sub(r"[^a-zA-Z0-9_.-]", "_", safe_filename)
+ file_path: str = os.path.join(upload_directory, safe_filename)
+ current_file = await aiofiles.open(file_path, "wb")
+ else:
+ # Initialize form_data with an empty list for text fields
+ if current_field_name not in form_data:
+ form_data[current_field_name] = []
+ elif result:
+ if current_file:
+ await current_file.write(result)
+ else:
+ form_value += result.decode("utf-8", "ignore")
+ else:
+ if current_file:
+ await current_file.close()
+ current_file = None
+ files[current_field_name] = {
+ "filename": current_filename,
+ "content_type": current_content_type or "application/octet-stream",
+ "saved_path": os.path.join(upload_directory, safe_filename),
+ }
+ else:
+ # Append form_value to form_data, even if empty
+ if current_field_name:
+ form_data[current_field_name].append(form_value or "")
+ form_value = ""
+ # Ensure any remaining form_value is appended
+ if current_field_name and not current_filename:
+ form_data[current_field_name].append(form_value or "")
+ return form_data, files
+
+ async def _send_response(
+ self,
+ send: Callable[[Dict[str, Any]], Awaitable[None]],
+ status_code: int,
+ body: Any,
+ extra_headers: Optional[List[Tuple[str, str]]] = None
+ ) -> None:
+ """
+ Send an HTTP response using the ASGI send callable.
+
+ Args:
+ send: The ASGI send callable.
+ status_code: The HTTP status code for the response.
+ body: The response body, which may be a string, bytes, or
+ generator.
+ extra_headers: Optional list of extra header tuples.
+ """
+ if extra_headers is None:
+ extra_headers = []
+ sanitized_headers: List[Tuple[str, str]] = []
+ for k, v in extra_headers:
+ if "\n" in k or "\r" in k or "\n" in v or "\r" in v:
+ print(f"Header injection attempt detected: {k}: {v}")
+ continue
+ sanitized_headers.append((k, v))
+ if not any(h[0].lower() == "content-type" for h in sanitized_headers):
+ sanitized_headers.append(("Content-Type", "text/html; charset=utf-8"))
+ await send({
+ "type": "http.response.start",
+ "status": status_code,
+ "headers": [(k.encode("latin-1"), v.encode("latin-1")) for k, v in sanitized_headers],
+ })
+ if hasattr(body, "__aiter__"):
+ async for chunk in body:
+ if isinstance(chunk, str):
+ chunk = chunk.encode("utf-8")
+ await send({
+ "type": "http.response.body",
+ "body": chunk,
+ "more_body": True
+ })
+ await send({"type": "http.response.body", "body": b"", "more_body": False})
+ return
+ if hasattr(body, "__iter__") and not isinstance(body, (bytes, str)):
+ for chunk in body:
+ if isinstance(chunk, str):
+ chunk = chunk.encode("utf-8")
+ await send({
+ "type": "http.response.body",
+ "body": chunk,
+ "more_body": True
+ })
+ await send({"type": "http.response.body", "body": b"", "more_body": False})
+ return
+ response_body = (body if isinstance(body, bytes)
+ else str(body).encode("utf-8"))
+ await send({
+ "type": "http.response.body",
+ "body": response_body,
+ "more_body": False
+ })
+
+ def _redirect(self, location: str, extra_headers: list = None) -> Tuple[int, str]:
+ """
+ Generate an HTTP redirect response.
+
+ Args:
+ location: The URL to redirect to.
+ extra_headers: Optional list of tuples (header_name, header_value) to include in the response.
+
+ Returns:
+ A tuple containing the HTTP status code, the HTML body, and headers list.
+ """
+ headers = [("Location", location)]
+ if extra_headers:
+ headers.extend(extra_headers)
+ return 302, "", headers
+
+ async def _render_template(self, name: str, **kwargs: Any) -> str:
+ """
+ Render a template asynchronously using Jinja2.
+
+ Args:
+ name: The name of the template file.
+ **kwargs: Additional keyword arguments for the template.
+
+ Returns:
+ The rendered template as a string.
+ """
+ if not JINJA_INSTALLED:
+ print("To use the `_render_template` method install 'jinja2'.")
+ return 500, "500 Internal Server Error"
+ assert self.env is not None
+ template = await asyncio.to_thread(self.env.get_template, name)
+ return await template.render_async(**kwargs)
diff --git a/unstable/tests.py b/unstable/tests.py
new file mode 100644
index 0000000..7080ed3
--- /dev/null
+++ b/unstable/tests.py
@@ -0,0 +1,990 @@
+import asyncio
+import os
+import shutil
+import time
+import uuid
+import pytest
+from MicroPie import App, Request, WebSocket, HttpMiddleware, InMemorySessionBackend
+from urllib.parse import parse_qs
+
+# Mock MULTIPART_INSTALLED and JINJA_INSTALLED for testing optional dependencies
+MULTIPART_INSTALLED = True
+JINJA_INSTALLED = True
+
+# Import optional dependencies safely
+try:
+ import aiofiles
+ from multipart import PushMultipartParser, MultipartSegment
+except ImportError:
+ pass
+
+try:
+ from jinja2 import Environment
+except ImportError:
+ pass
+
+# Setup fixture for uploads directory
[email protected](autouse=True)
+def setup_uploads():
+ upload_dir = "uploads"
+ if os.path.exists(upload_dir):
+ shutil.rmtree(upload_dir)
+ yield
+ if os.path.exists(upload_dir):
+ shutil.rmtree(upload_dir)
+
+# Setup fixture for templates directory
[email protected]
+def setup_templates():
+ os.makedirs("templates", exist_ok=True)
+ yield
+ if os.path.exists("templates"):
+ shutil.rmtree("templates")
+
+# Test 1: Basic HTTP GET Request
[email protected]
+async def test_basic_get_request():
+ class TestApp(App):
+ async def index(self):
+ return "Hello, World!"
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert len(sent_messages) == 2
+ assert sent_messages[0]["type"] == "http.response.start"
+ assert sent_messages[0]["status"] == 200
+ assert sent_messages[1]["type"] == "http.response.body"
+ assert sent_messages[1]["body"] == b"Hello, World!"
+
+# Test 2: HTTP GET with Path Parameters
[email protected]
+async def test_get_with_path_params():
+ class TestApp(App):
+ async def user(self, user_id):
+ return f"User {user_id}"
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/user/123",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert len(sent_messages) == 2
+ assert sent_messages[0]["status"] == 200
+ assert sent_messages[1]["body"] == b"User 123"
+
+# Test 3: HTTP GET with Query Parameters
[email protected]
+async def test_get_with_query_params():
+ class TestApp(App):
+ async def search(self, query):
+ return f"Search for {query}"
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/search",
+ "headers": [],
+ "query_string": b"query=python",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert len(sent_messages) == 2
+ assert sent_messages[0]["status"] == 200
+ assert sent_messages[1]["body"] == b"Search for python"
+
+# Test 4: HTTP POST with Form Data
[email protected]
+async def test_post_with_form_data():
+ class TestApp(App):
+ async def login(self, username, password):
+ return "Login successful" if username == "admin" and password == "secret" else ("Invalid credentials", 401)
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "POST",
+ "path": "/login",
+ "headers": [(b"content-type", b"application/x-www-form-urlencoded")],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"username=admin&password=secret", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert len(sent_messages) == 2
+ assert sent_messages[0]["status"] == 200
+ assert sent_messages[1]["body"] == b"Login successful"
+
+# Test 5: HTTP POST with JSON Data
[email protected]
+async def test_post_with_json_data():
+ class TestApp(App):
+ async def create_user(self, name):
+ return f"User {name} created"
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "POST",
+ "path": "/create_user",
+ "headers": [(b"content-type", b"application/json")],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b'{"name": "Alice"}', "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert len(sent_messages) == 2
+ assert sent_messages[0]["status"] == 200
+ assert sent_messages[1]["body"] == b"User Alice created"
+
+# Test 6: HTTP POST with Multipart File Upload
[email protected]
+async def test_post_with_multipart_file_upload():
+ class TestApp(App):
+ async def upload(self, file):
+ return f"File {file['filename']} uploaded to {file['saved_path']}"
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "POST",
+ "path": "/upload",
+ "headers": [(b"content-type", b"multipart/form-data; boundary=boundary")],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ body = (
+ b"--boundary\r\n"
+ b'Content-Disposition: form-data; name="file"; filename="test.txt"\r\n'
+ b"Content-Type: text/plain\r\n"
+ b"\r\n"
+ b"Hello, World!\r\n"
+ b"--boundary--\r\n"
+ )
+
+ async def mock_receive():
+ return {"type": "http.request", "body": body, "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert len(sent_messages) == 2
+ assert sent_messages[0]["status"] == 200
+ assert b"File test.txt uploaded to" in sent_messages[1]["body"]
+ files = os.listdir("uploads")
+ assert len(files) == 1
+ with open(os.path.join("uploads", files[0]), "rb") as f:
+ assert f.read() == b"Hello, World!"
+
+# Test 7: Session Management
[email protected]
+async def test_session_management():
+ class TestApp(App):
+ async def login(self, username):
+ request = self.request
+ request.session["username"] = username
+ return "Logged in"
+
+ async def profile(self):
+ request = self.request
+ return f"Welcome, {request.session.get('username', 'Guest')}"
+
+ app = TestApp()
+ scope_login = {
+ "type": "http",
+ "method": "POST",
+ "path": "/login",
+ "headers": [],
+ "query_string": b"username=alice",
+ }
+ sent_messages_login = []
+
+ async def mock_send_login(message):
+ sent_messages_login.append(message)
+
+ async def mock_receive_login():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope_login, mock_receive_login, mock_send_login)
+ assert sent_messages_login[0]["status"] == 200
+
+ session_id = [h[1].decode().split(";")[0].split("=")[1] for h in sent_messages_login[0]["headers"] if h[0] == b"Set-Cookie"][0]
+ scope_profile = {
+ "type": "http",
+ "method": "GET",
+ "path": "/profile",
+ "headers": [(b"cookie", f"session_id={session_id}".encode())],
+ "query_string": b"",
+ }
+ sent_messages_profile = []
+
+ async def mock_send_profile(message):
+ sent_messages_profile.append(message)
+
+ async def mock_receive_profile():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope_profile, mock_receive_profile, mock_send_profile)
+ assert sent_messages_profile[0]["status"] == 200
+ assert sent_messages_profile[1]["body"] == b"Welcome, alice"
+
+# Test 8: WebSocket Connection
[email protected]
+async def test_websocket_connection():
+ class TestApp(App):
+ async def ws_echo(self, websocket, path_params):
+ await websocket.accept()
+ message = await websocket.receive_text()
+ await websocket.send_text(f"Echo: {message}")
+ await websocket.close()
+
+ app = TestApp()
+ scope = {
+ "type": "websocket",
+ "path": "/echo",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+ received_messages = [
+ {"type": "websocket.connect"},
+ {"type": "websocket.receive", "text": "Hello"},
+ {"type": "websocket.disconnect", "code": 1000},
+ ]
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return received_messages.pop(0)
+
+ await app(scope, mock_receive, mock_send)
+ assert len(sent_messages) == 3
+ assert sent_messages[0]["type"] == "websocket.accept"
+ assert sent_messages[1]["type"] == "websocket.send"
+ assert sent_messages[1]["text"] == "Echo: Hello"
+ assert sent_messages[2]["type"] == "websocket.close"
+ assert sent_messages[2]["code"] == 1000
+
+# Test 9: HTTP Middleware
[email protected]
+async def test_http_middleware():
+ class CustomHeaderMiddleware(HttpMiddleware):
+ async def before_request(self, request):
+ pass
+
+ async def after_request(self, request, status_code, response_body, extra_headers):
+ extra_headers.append(("X-Custom-Header", "Test"))
+ return {"headers": extra_headers}
+
+ class TestApp(App):
+ async def index(self):
+ return "Hello"
+
+ app = TestApp()
+ app.middlewares.append(CustomHeaderMiddleware())
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert len(sent_messages) == 2
+ assert any(h[0] == b"X-Custom-Header" and h[1] == b"Test" for h in sent_messages[0]["headers"])
+
+# Test 10: Template Rendering
[email protected]
+async def test_template_rendering(setup_templates):
+ with open("templates/hello.html", "w") as f:
+ f.write("Hello, {{ name }}!")
+
+ class TestApp(App):
+ async def index(self):
+ return await self._render_template("hello.html", name="World")
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert sent_messages[0]["status"] == 200
+ assert sent_messages[1]["body"] == b"Hello, World!"
+
+# Test 11: 404 Not Found
[email protected]
+async def test_404_not_found():
+ class TestApp(App):
+ pass
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/nonexistent",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert sent_messages[0]["status"] == 404
+ assert sent_messages[1]["body"] == b"404 Not Found"
+
+# Test 12: 400 Bad Request (Missing Parameter)
[email protected]
+async def test_400_missing_parameter():
+ class TestApp(App):
+ async def index(self, required_param):
+ return "Should not reach here"
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert sent_messages[0]["status"] == 400
+ assert b"Missing required parameter" in sent_messages[1]["body"]
+
+# Test 13: 500 Internal Server Error
[email protected]
+async def test_500_internal_server_error():
+ class TestApp(App):
+ async def index(self):
+ raise Exception("Test error")
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert sent_messages[0]["status"] == 500
+ assert sent_messages[1]["body"] == b"500 Internal Server Error"
+
+# Test 14: WebSocket Error Handling
[email protected]
+async def test_websocket_error_handling():
+ class TestApp(App):
+ async def ws_index(self, websocket, path_params):
+ raise Exception("Test error")
+
+ app = TestApp()
+ scope = {
+ "type": "websocket",
+ "path": "/",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "websocket.connect"}
+
+ await app(scope, mock_receive, mock_send)
+ assert len(sent_messages) == 1
+ assert sent_messages[0]["type"] == "websocket.close"
+ assert sent_messages[0]["code"] == 1011
+
+# Test 15: Parse Cookies
+def test_parse_cookies():
+ app = App()
+ cookies = app._parse_cookies("session_id=abc123; user=alice")
+ assert cookies == {"session_id": "abc123", "user": "alice"}
+
+# Test 16: Redirect
[email protected]
+async def test_redirect():
+ class TestApp(App):
+ async def index(self):
+ return self._redirect("/new_location")
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert sent_messages[0]["status"] == 302
+ assert any(h[0] == b"Location" and h[1] == b"/new_location" for h in sent_messages[0]["headers"])
+
+# Test 17: In-Memory Session Backend
[email protected]
+async def test_in_memory_session_backend():
+ backend = InMemorySessionBackend()
+ session_id = "test_session"
+ data = {"key": "value"}
+ await backend.save(session_id, data, 3600)
+ loaded_data = await backend.load(session_id)
+ assert loaded_data == data
+ # Simulate session timeout
+ backend.last_access[session_id] = time.time() - 8 * 3600 - 1
+ assert await backend.load(session_id) == {}
+
+# Test 18: Synchronous Handler
[email protected]
+async def test_synchronous_handler():
+ class TestApp(App):
+ def index(self):
+ return "Sync Hello"
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert sent_messages[0]["status"] == 200
+ assert sent_messages[1]["body"] == b"Sync Hello"
+
+# Test 19: Asynchronous Streaming Response
[email protected]
+async def test_async_streaming_response():
+ class TestApp(App):
+ async def stream(self):
+ async def generate():
+ yield "Chunk 1"
+ await asyncio.sleep(0.1)
+ yield "Chunk 2"
+ return generate()
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/stream",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert len(sent_messages) == 4
+ assert sent_messages[1]["body"] == b"Chunk 1"
+ assert sent_messages[2]["body"] == b"Chunk 2"
+ assert sent_messages[3]["body"] == b""
+
+# Test 20: Synchronous Generator Response
[email protected]
+async def test_sync_generator_response():
+ class TestApp(App):
+ def stream(self):
+ def generate():
+ yield "Chunk 1"
+ yield "Chunk 2"
+ return generate()
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/stream",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert len(sent_messages) == 4
+ assert sent_messages[1]["body"] == b"Chunk 1"
+ assert sent_messages[2]["body"] == b"Chunk 2"
+
+# Test 21: JSON Response
[email protected]
+async def test_json_response():
+ class TestApp(App):
+ async def data(self):
+ return {"key": "value"}
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/data",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert any(h[0] == b"Content-Type" and h[1] == b"application/json" for h in sent_messages[0]["headers"])
+ assert sent_messages[1]["body"] == b'{"key": "value"}'
+
+# Test 22: Protected Path (Starting with '_')
[email protected]
+async def test_protected_path():
+ class TestApp(App):
+ async def _hidden(self):
+ return "Should not reach here"
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/_hidden",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert sent_messages[0]["status"] == 404
+
+# Test 23: WebSocket Protected Path
[email protected]
+async def test_websocket_protected_path():
+ class TestApp(App):
+ async def _ws_hidden(self, websocket, path_params):
+ pass
+
+ app = TestApp()
+ scope = {
+ "type": "websocket",
+ "path": "/_hidden",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "websocket.connect"}
+
+ await app(scope, mock_receive, mock_send)
+ assert sent_messages[0]["type"] == "websocket.close"
+ assert sent_messages[0]["code"] == 1008
+
+# Test 24: Invalid JSON
[email protected]
+async def test_invalid_json():
+ class TestApp(App):
+ async def index(self):
+ pass
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "POST",
+ "path": "/index",
+ "headers": [(b"content-type", b"application/json")],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"{invalid}", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert sent_messages[0]["status"] == 400
+ assert sent_messages[1]["body"] == b"400 Bad Request: Bad JSON"
+
+# Test 25: Header Injection Prevention
[email protected]
+async def test_header_injection_prevention():
+ class TestApp(App):
+ async def index(self):
+ return "Hello", 200, [("X-Test", "Value\nInjection")]
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert not any(b"\n" in h[1] for h in sent_messages[0]["headers"])
+
+# Test 26: Missing Jinja2 Dependency
[email protected]
+async def test_missing_jinja2(monkeypatch):
+ monkeypatch.setattr("MicroPie.JINJA_INSTALLED", False)
+ class TestApp(App):
+ async def index(self):
+ return await self._render_template("hello.html", name="World")
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert sent_messages[0]["status"] == 500
+ assert sent_messages[1]["body"] == b"500 Internal Server Error"
+
+# Test 27: Missing Multipart Dependency
[email protected]
+async def test_missing_multipart(monkeypatch):
+ monkeypatch.setattr("MicroPie.MULTIPART_INSTALLED", False)
+ class TestApp(App):
+ async def upload(self, file):
+ return "Should not reach here"
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "POST",
+ "path": "/upload",
+ "headers": [(b"content-type", b"multipart/form-data; boundary=boundary")],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert sent_messages[0]["status"] == 500
+ assert sent_messages[1]["body"] == b"500 Internal Server Error"
+
+# Test 28: WebSocket Send JSON
[email protected]
+async def test_websocket_send_json():
+ class TestApp(App):
+ async def ws_json(self, websocket, path_params):
+ await websocket.accept()
+ await websocket.send_json({"message": "Hello"})
+ await websocket.close()
+
+ app = TestApp()
+ scope = {
+ "type": "websocket",
+ "path": "/json",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "websocket.connect"}
+
+ await app(scope, mock_receive, mock_send)
+ assert len(sent_messages) == 3
+ assert sent_messages[0]["type"] == "websocket.accept"
+ assert sent_messages[1]["type"] == "websocket.send"
+ assert sent_messages[1]["text"] == '{"message": "Hello"}'
+ assert sent_messages[2]["type"] == "websocket.close"
+
+# Test 29: WebSocket Receive JSON
[email protected]
+async def test_websocket_receive_json():
+ class TestApp(App):
+ async def ws_json(self, websocket, path_params):
+ await websocket.accept()
+ data = await websocket.receive_json()
+ await websocket.send_text(f"Received: {data['message']}")
+ await websocket.close()
+
+ app = TestApp()
+ scope = {
+ "type": "websocket",
+ "path": "/json",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+ received_messages = [
+ {"type": "websocket.connect"},
+ {"type": "websocket.receive", "text": '{"message": "Hello"}'},
+ {"type": "websocket.disconnect", "code": 1000},
+ ]
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return received_messages.pop(0)
+
+ await app(scope, mock_receive, mock_send)
+ assert len(sent_messages) == 3
+ assert sent_messages[0]["type"] == "websocket.accept"
+ assert sent_messages[1]["type"] == "websocket.send"
+ assert sent_messages[1]["text"] == "Received: Hello"
+ assert sent_messages[2]["type"] == "websocket.close"
+
+# Test 30: WebSocket Disconnect
[email protected]
+async def test_websocket_disconnect():
+ class TestApp(App):
+ async def ws_disconnect(self, websocket, path_params):
+ await websocket.accept()
+ await websocket.receive_text() # Should raise ConnectionError
+ await websocket.close()
+
+ app = TestApp()
+ scope = {
+ "type": "websocket",
+ "path": "/disconnect",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+ received_messages = [
+ {"type": "websocket.connect"},
+ {"type": "websocket.disconnect", "code": 1000},
+ ]
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return received_messages.pop(0)
+
+ await app(scope, mock_receive, mock_send)
+ assert len(sent_messages) == 2
+ assert sent_messages[0]["type"] == "websocket.accept"
+ assert sent_messages[1]["type"] == "websocket.close"
+
+# Test 31: Middleware Before Request Early Exit
[email protected]
+async def test_middleware_before_request_early_exit():
+ class EarlyExitMiddleware(HttpMiddleware):
+ async def before_request(self, request):
+ return {"status_code": 403, "body": "Forbidden"}
+
+ async def after_request(self, request, status_code, response_body, extra_headers):
+ pass
+
+ class TestApp(App):
+ async def index(self):
+ return "Should not reach here"
+
+ app = TestApp()
+ app.middlewares.append(EarlyExitMiddleware())
+ scope = {
+ "type": "http",
+ "method": "GET",
+ "path": "/",
+ "headers": [],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ async def mock_receive():
+ return {"type": "http.request", "body": b"", "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert sent_messages[0]["status"] == 403
+ assert sent_messages[1]["body"] == b"Forbidden"
+
+# Test 32: Empty Cookie Header
+def test_empty_cookie_header():
+ app = App()
+ cookies = app._parse_cookies("")
+ assert cookies == {}
+
+# Test 33: Multipart Form Data Without File
[email protected]
+async def test_multipart_form_data_without_file():
+ class TestApp(App):
+ async def form(self, field):
+ return f"Field: {field[0]}"
+
+ app = TestApp()
+ scope = {
+ "type": "http",
+ "method": "POST",
+ "path": "/form",
+ "headers": [(b"content-type", b"multipart/form-data; boundary=boundary")],
+ "query_string": b"",
+ }
+ sent_messages = []
+
+ async def mock_send(message):
+ sent_messages.append(message)
+
+ body = (
+ b"--boundary\r\n"
+ b'Content-Disposition: form-data; name="field"\r\n'
+ b"\r\n"
+ b"test_value\r\n"
+ b"--boundary--\r\n"
+ )
+
+ async def mock_receive():
+ return {"type": "http.request", "body": body, "more_body": False}
+
+ await app(scope, mock_receive, mock_send)
+ assert sent_messages[0]["status"] == 200
+ assert sent_messages[1]["body"] == b"Field: test_value"