diff --git a/scenarios/pterodisbot_admin.yaml b/scenarios/pterodisbot_admin.yaml new file mode 100644 index 0000000..ba70335 --- /dev/null +++ b/scenarios/pterodisbot_admin.yaml @@ -0,0 +1,174 @@ +# Pterodisbot - Admin Commands Test +name: "Pterodisbot - Admin Commands" +description: "Test /refresh_embeds and /purge_embeds with permissions" + +# Guild setup +setup: + guilds: + - id: "100000000000000000" + name: "Test Guild" + owner_id: "444444444444444444" + channels: + - "100000000000000010" + - "100000000000000011" + - "100000000000000012" + roles: + - "200000000000000001" + + channels: + - id: "100000000000000010" + guild_id: "100000000000000000" + name: "general" + type: 0 + position: 0 + - id: "100000000000000011" + guild_id: "100000000000000000" + name: "bot-commands" + type: 0 + position: 1 + - id: "100000000000000012" + guild_id: "100000000000000000" + name: "server-status" + type: 0 + position: 2 + + roles: + - id: "200000000000000001" + guild_id: "100000000000000000" + name: "Game Server User" + color: 3447003 + hoist: true + position: 1 + permissions: "104324161" + managed: false + mentionable: true + + users: + - id: "444444444444444444" + username: "TestBot" + discriminator: "0000" + bot: true + - id: "300000000000000001" + username: "AdminUser" + discriminator: "0001" + bot: false + guild_roles: + "100000000000000000": + - "200000000000000001" + - id: "300000000000000002" + username: "RegularUser" + discriminator: "0002" + bot: false + guild_roles: + "100000000000000000": + - "200000000000000001" + + # Pre-create some embeds in channels + messages: + - id: "100000000000000100" + channel_id: "100000000000000010" + author_id: "444444444444444444" + content: "Server Status: Minecraft Server" + embeds: + - title: "Minecraft Server - Running" + color: 0x00ff00 + components: + - type: 1 + components: + - type: 2 + custom_id: "start_button" + label: "Start" + style: 3 + - type: 2 + custom_id: "stop_button" + label: "Stop" + style: 4 + - id: "100000000000000101" + channel_id: "100000000000000011" + author_id: "444444444444444444" + content: "Server Status: Valheim Server" + embeds: + - title: "Valheim Server - Running" + color: 0x00ff00 + components: + - type: 1 + components: + - type: 2 + custom_id: "start_button" + label: "Start" + style: 3 + - type: 2 + custom_id: "stop_button" + label: "Stop" + style: 4 + - id: "100000000000000102" + channel_id: "100000000000000012" + author_id: "444444444444444444" + content: "Server Status: CS2 Server" + embeds: + - title: "CS2 Server - Running" + color: 0x00ff00 + components: + - type: 1 + components: + - type: 2 + custom_id: "start_button" + label: "Start" + style: 3 + - type: 2 + custom_id: "stop_button" + label: "Stop" + style: 4 + + bot: + id: "444444444444444444" + username: "TestBot" + discriminator: "0000" + token: "test_token_123" + command: "python main.py" # Example command to start the bot (commented out for testing) + +# Test steps +steps: + # Test 1: Regular user tries /refresh_embeds (should fail) + - action: "slash_command" + command: "refresh_embeds" + user_id: "300000000000000002" # RegularUser + guild_id: "100000000000000000" + channel_id: "100000000000000010" + expect: + - type: 4 + data: + content_contains: "administrator permissions" + + # Test 2: Admin user runs /refresh_embeds (should succeed) + - action: "slash_command" + command: "refresh_embeds" + user_id: "300000000000000001" # AdminUser + guild_id: "100000000000000000" + channel_id: "100000000000000010" + expect: + - type: 4 + data: + content_contains: "Refreshed all embeds" + + # Test 3: Regular user tries /purge_embeds (should fail) + - action: "slash_command" + command: "purge_embeds" + user_id: "300000000000000002" # RegularUser + guild_id: "100000000000000000" + channel_id: "100000000000000010" + expect: + - type: 4 + data: + content_contains: "administrator permissions" + + # Test 4: Admin user runs /purge_embeds (should succeed) + - action: "slash_command" + command: "purge_embeds" + user_id: "300000000000000001" # AdminUser + guild_id: "100000000000000000" + channel_id: "100000000000000010" + expect: + - type: 4 + data: + content_contains: "Purge Complete" \ No newline at end of file diff --git a/scenarios/pterodisbot_basic.yaml b/scenarios/pterodisbot_basic.yaml new file mode 100644 index 0000000..25d49a6 --- /dev/null +++ b/scenarios/pterodisbot_basic.yaml @@ -0,0 +1,107 @@ +# Pterodisbot - Basic Server Status Flow +name: "Pterodisbot - Basic Server Status Flow" +description: "Test /server_status -> select server -> click buttons" + +# Guild setup +setup: + guilds: + - id: "100000000000000000" + name: "Test Guild" + owner_id: "444444444444444444" + channels: + - "100000000000000010" + roles: + - "200000000000000001" + + channels: + - id: "100000000000000010" + guild_id: "100000000000000000" + name: "general" + type: 0 # GUILD_TEXT + position: 0 + topic: "General discussion and bot commands" + + roles: + - id: "200000000000000001" + guild_id: "100000000000000000" + name: "Game Server User" + color: 3447003 + hoist: true + position: 1 + permissions: "104324161" + managed: false + mentionable: true + + users: + - id: "444444444444444444" + username: "TestBot" + discriminator: "0000" + bot: true + - id: "300000000000000001" + username: "TestUser" + discriminator: "1234" + bot: false + guild_roles: + "100000000000000000": + - "200000000000000001" + + bot: + id: "444444444444444444" + username: "TestBot" + discriminator: "0000" + token: "test_token_123" + # command: "python main.py" # Example command to start the bot (commented out for testing) + +# Test steps +steps: + - action: "slash_command" + command: "server_status" + user_id: "300000000000000001" + guild_id: "100000000000000000" + channel_id: "100000000000000010" + expect: + - type: 4 # CHANNEL_MESSAGE_WITH_SOURCE + data: + components: true + component_type: 1 # ACTION_ROW + custom_id: "server_dropdown" + options_count: 3 + + - action: "select_menu" + custom_id: "server_dropdown" + values: ["server_1"] + user_id: "300000000000000001" + guild_id: "100000000000000000" + channel_id: "100000000000000010" + message_id: "msg_123" + expect: + - type: 4 + data: + embeds: true + components: true + component_type: 1 # ACTION_ROW + button_count: 4 + + - action: "button_click" + custom_id: "start_button" + user_id: "300000000000000001" + guild_id: "100000000000000000" + channel_id: "100000000000000010" + message_id: "msg_123" + expect: + - type: 4 + data: + content_contains: "starting" + + - action: "button_click" + custom_id: "show_address_button" + user_id: "300000000000000001" + guild_id: "100000000000000000" + channel_id: "100000000000000010" + message_id: "msg_123" + expect: + - type: 4 + data: + embeds: true + embed_title_contains: "Connection Info" + fields_contain: ["Address", "Port"] \ No newline at end of file diff --git a/scenarios/pterodisbot_multiguild.yaml b/scenarios/pterodisbot_multiguild.yaml new file mode 100644 index 0000000..b40e534 --- /dev/null +++ b/scenarios/pterodisbot_multiguild.yaml @@ -0,0 +1,162 @@ +# Pterodisbot - Multi-Guild Isolation Test +name: "Pterodisbot - Multi-Guild Isolation" +description: "Verify guilds track servers independently" + +# Guild setup +setup: + guilds: + - id: "100000000000000000" + name: "Guild Alpha" + owner_id: "444444444444444444" + channels: + - "100000000000000010" + roles: + - "200000000000000001" + - id: "100000000000000001" + name: "Guild Beta" + owner_id: "444444444444444444" + channels: + - "100000000000000020" + roles: + - "200000000000000002" + + channels: + - id: "100000000000000010" + guild_id: "100000000000000000" + name: "general" + type: 0 + position: 0 + - id: "100000000000000020" + guild_id: "100000000000000001" + name: "general" + type: 0 + position: 0 + + roles: + - id: "200000000000000001" + guild_id: "100000000000000000" + name: "Game Server User" + color: 3447003 + hoist: true + position: 1 + permissions: "104324161" + managed: false + mentionable: true + - id: "200000000000000002" + guild_id: "100000000000000001" + name: "Game Server User" + color: 3447003 + hoist: true + position: 1 + permissions: "104324161" + managed: false + mentionable: true + + users: + - id: "444444444444444444" + username: "TestBot" + discriminator: "0000" + bot: true + - id: "300000000000000001" + username: "UserAlpha" + discriminator: "0001" + bot: false + guild_roles: + "100000000000000000": + - "200000000000000001" + - id: "300000000000000002" + username: "UserBeta" + discriminator: "0002" + bot: false + guild_roles: + "100000000000000001": + - "200000000000000002" + + bot: + id: "444444444444444444" + username: "TestBot" + discriminator: "0000" + token: "test_token_123" + command: "python main.py" # Example command to start the bot + +# Test steps +steps: + # Test 1: User in Guild Alpha selects Server A + - action: "slash_command" + command: "server_status" + user_id: "300000000000000001" + guild_id: "100000000000000000" + channel_id: "100000000000000010" + expect: + - type: 4 + data: + components: true + custom_id: "server_dropdown" + + - action: "select_menu" + custom_id: "server_dropdown" + values: ["server_1"] + user_id: "300000000000000001" + guild_id: "100000000000000000" + channel_id: "100000000000000010" + message_id: "msg_alpha" + expect: + - type: 4 + data: + embeds: true + components: true + + # Test 2: User in Guild Beta selects same Server A + - action: "slash_command" + command: "server_status" + user_id: "300000000000000002" + guild_id: "100000000000000001" + channel_id: "100000000000000020" + expect: + - type: 4 + data: + components: true + custom_id: "server_dropdown" + + - action: "select_menu" + custom_id: "server_dropdown" + values: ["server_1"] + user_id: "300000000000000002" + guild_id: "100000000000000001" + channel_id: "100000000000000020" + message_id: "msg_beta" + expect: + - type: 4 + data: + embeds: true + components: true + + # Test 3: User Alpha stops server in Guild Alpha + - action: "button_click" + custom_id: "stop_button" + user_id: "300000000000000001" + guild_id: "100000000000000000" + channel_id: "100000000000000010" + message_id: "msg_alpha" + expect: + - type: 4 + data: + content_contains: "stopping" + + # Test 4: Verify Guild Alpha embed shows offline, Guild Beta stays running + - action: "wait" + duration: 1 + + # Test 5: User Beta clicks Show Address in Guild Beta (should still work) + - action: "button_click" + custom_id: "show_address_button" + user_id: "300000000000000002" + guild_id: "100000000000000001" + channel_id: "100000000000000020" + message_id: "msg_beta" + expect: + - type: 4 + data: + embeds: true + embed_title_contains: "Connection Info" + fields_contain: ["Address", "Port"] \ No newline at end of file diff --git a/src/apicord/__pycache__/config.cpython-314.pyc b/src/apicord/__pycache__/config.cpython-314.pyc index 5981754..c477d29 100644 Binary files a/src/apicord/__pycache__/config.cpython-314.pyc and b/src/apicord/__pycache__/config.cpython-314.pyc differ diff --git a/src/apicord/cli/__pycache__/commands.cpython-314.pyc b/src/apicord/cli/__pycache__/commands.cpython-314.pyc index 05f3945..7285557 100644 Binary files a/src/apicord/cli/__pycache__/commands.cpython-314.pyc and b/src/apicord/cli/__pycache__/commands.cpython-314.pyc differ diff --git a/src/apicord/cli/commands.py b/src/apicord/cli/commands.py index 6e5438f..527680e 100644 --- a/src/apicord/cli/commands.py +++ b/src/apicord/cli/commands.py @@ -6,6 +6,7 @@ import asyncio import json import logging import sys +import time from pathlib import Path from typing import Optional @@ -16,6 +17,7 @@ from apicord.config import get_config from apicord.state.manager import StateManager from apicord.rest.server import RESTServer from apicord.gateways.server import GatewayServer +from apicord.tui.actions.scenario_runner import ScenarioRunner, run_scenario, ScenarioData logger = logging.getLogger("apicord") @@ -64,7 +66,7 @@ def start(ctx: click.Context, scenario: Optional[Path], guilds: int, channels: i run_tui(config=config, headless=False) return - async def run_server(): + async def run_server() -> None: # Initialize state manager state_manager = StateManager(config) @@ -106,6 +108,84 @@ def start(ctx: click.Context, scenario: Optional[Path], guilds: int, channels: i asyncio.run(run_server()) +@click.command() +@click.option( + "--scenario", + "-s", + type=click.Path(exists=True, path_type=Path), + required=True, + help="Path to test scenario YAML file", +) +@click.option( + "--headless/--no-headless", + default=True, + help="Run in headless mode (no TUI)", +) +@click.option( + "--timeout", + "-t", + type=int, + default=60, + help="Test timeout in seconds", +) +@click.option( + "--output", + "-o", + type=click.Path(path_type=Path), + help="Output file for JSON results", +) +@click.option( + "--junit-xml", + type=click.Path(path_type=Path), + help="Output file for JUnit XML results", +) +@click.pass_context +def run_scenario_cmd(ctx: click.Context, scenario: Path, headless: bool, timeout: int, output: Optional[Path], junit_xml: Optional[Path]) -> None: + """Run a test scenario against the emulator (headless by default).""" + config = ctx.obj["config"] if ctx.obj else get_config() + + async def run() -> None: + logger.info(f"Running scenario: {scenario}") + logger.info(f"Headless mode: {headless}") + + with open(scenario) as f: + scenario_data = yaml.safe_load(f) + + scenario_obj = ScenarioData(**scenario_data) + config_obj = ctx.obj["config"] if ctx.obj else get_config() + runner = ScenarioRunner(config_obj, scenario) + results = await runner.run(timeout=timeout) + + # Output results + if output: + with open(output, "w") as f: + f.write(results.to_json()) + logger.info(f"Results written to {output}") + + if junit_xml: + with open(junit_xml, "w") as f: + f.write(results.to_junit_xml()) + logger.info(f"JUnit XML written to {junit_xml}") + + # Print summary + logger.info(f"\n=== Test Results ===") + logger.info(f"Scenario: {results.scenario_name}") + logger.info(f"Passed: {results.passed}") + logger.info(f"Steps: {results.passed_steps}/{results.total_steps} passed") + logger.info(f"Duration: {results.total_duration:.2f}s") + + for step in results.steps: + status = "✅" if step.passed else "❌" + logger.info(f" {status} Step {step.step_index + 1}: {step.action} ({step.duration:.2f}s)") + for failure in step.failures: + logger.error(f" - {failure}") + + if not results.passed: + sys.exit(1) + + asyncio.run(run()) + + @click.command() @click.option( "--scenario", @@ -130,10 +210,10 @@ def start(ctx: click.Context, scenario: Optional[Path], guilds: int, channels: i ) @click.pass_context def test(ctx: click.Context, scenario: Path, bot_path: Path, timeout: int) -> None: - """Run automated test scenario against a bot.""" + """Run automated test scenario against a bot (legacy command).""" config = ctx.obj["config"] if ctx.obj else get_config() - async def run_test(): + async def run_test() -> None: logger.info(f"Running test scenario: {scenario}") logger.info(f"Bot path: {bot_path}") @@ -363,4 +443,113 @@ def validate(ctx: click.Context, scenario: Path) -> None: sys.exit(1) logger.info(f"✅ Scenario '{scenario_data.get('name', 'unnamed')}' is valid") - logger.info(f" Steps: {len(steps)}") \ No newline at end of file + logger.info(f" Steps: {len(steps)}") + + +# CLI group +@click.group() +@click.option( + "--config-file", + "-c", + type=click.Path(exists=True, path_type=Path), + help="Path to configuration file (YAML)", +) +@click.option( + "--log-level", + "-l", + type=click.Choice(["DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"]), + default="INFO", + help="Set logging level", +) +@click.option( + "--host", + default="127.0.0.1", + help="Host to bind the emulator server", +) +@click.option( + "--rest-port", + type=int, + default=8080, + help="Port for REST API server", +) +@click.option( + "--gateway-port", + type=int, + default=8081, + help="Port for WebSocket Gateway", +) +@click.option( + "--bot-token", + default="test_token_123", + help="Test bot token", +) +@click.option( + "--persist-state/--no-persist-state", + default=False, + help="Persist state to disk", +) +@click.pass_context +def cli( + ctx: click.Context, + config_file: Optional[Path], + log_level: str, + host: str, + rest_port: int, + gateway_port: int, + bot_token: str, + persist_state: bool, +) -> None: + """APIcord - Discord API Emulator for bot testing. + + A standalone Python CLI program that emulates the Discord API + (REST + WebSocket Gateway), enabling self-contained testing of + Discord bots without requiring a real Discord connection. + """ + # Setup logging + logging.basicConfig( + level=getattr(logging, log_level.upper()), + format="%(asctime)s - %(name)s - %(levelname)s - %(message)s", + handlers=[logging.StreamHandler(sys.stdout)], + ) + + # Load config from file if provided, otherwise use env vars and CLI args + if config_file: + import yaml + with open(config_file) as f: + file_config = yaml.safe_load(f) + config = EmulatorConfig(**file_config) + else: + config = EmulatorConfig( + host=host, + rest_port=rest_port, + gateway_port=gateway_port, + bot_token=bot_token, + persist_state=persist_state, + ) + + # Store config in context for subcommands + ctx.ensure_object(dict) + ctx.obj["config"] = config + + +# Add commands to CLI group +cli.add_command(start) +cli.add_command(run_scenario_cmd, name="run-scenario") +cli.add_command(test) +cli.add_command(reset) +cli.add_command(export) +cli.add_command(validate) + + +def main() -> int: + """Main entry point for the CLI.""" + try: + cli() + return 0 + except Exception as e: + logger.critical(f"Fatal error: {e}", exc_info=True) + return 1 + + +if __name__ == "__main__": + sys.exit(main()) \ No newline at end of file diff --git a/src/apicord/config.py b/src/apicord/config.py index 5c81ad9..8dd3560 100644 --- a/src/apicord/config.py +++ b/src/apicord/config.py @@ -7,7 +7,7 @@ Supports both config file and environment variable overrides. import os from pathlib import Path -from typing import Optional +from typing import List, Optional from pydantic import Field from pydantic_settings import BaseSettings @@ -25,11 +25,17 @@ class EmulatorConfig(BaseSettings): api_version: int = Field(default=10, description="Discord API version to emulate") gateway_version: int = Field(default=10, description="Gateway version to emulate") + # Pterodactyl Panel settings + pterodactyl_url: str = Field(default="https://panel.example.com", description="Pterodactyl Panel URL") + client_api_key: str = Field(default="ptlc_test_key", description="Pterodactyl Client API Key") + application_api_key: str = Field(default="ptla_test_key", description="Pterodactyl Application API Key") + # Bot settings (for testing) bot_token: str = Field(default="test_token_123", description="Test bot token") bot_id: str = Field(default="444444444444444444", description="Test bot user ID") bot_username: str = Field(default="TestBot", description="Test bot username") bot_discriminator: str = Field(default="0000", description="Test bot discriminator") + allowed_guild_ids: List[str] = Field(default_factory=list, description="List of allowed Discord guild IDs") # State settings persist_state: bool = Field(default=False, description="Persist state to disk") @@ -63,6 +69,8 @@ class EmulatorConfig(BaseSettings): return f"ws://{self.host}:{self.gateway_port}" +EmulatorConfig.model_rebuild() + # Global config instance _config: Optional[EmulatorConfig] = None diff --git a/src/apicord/models/__pycache__/discord.cpython-314.pyc b/src/apicord/models/__pycache__/discord.cpython-314.pyc index 33ea730..74b0ffc 100644 Binary files a/src/apicord/models/__pycache__/discord.cpython-314.pyc and b/src/apicord/models/__pycache__/discord.cpython-314.pyc differ diff --git a/src/apicord/models/discord.py b/src/apicord/models/discord.py index 3fd85bb..6704976 100644 --- a/src/apicord/models/discord.py +++ b/src/apicord/models/discord.py @@ -306,11 +306,15 @@ class ScenarioBot(BaseModel): username: str discriminator: str = "0000" token: Optional[str] = None + command: Optional[str] = None + env: Dict[str, str] = Field(default_factory=dict) + cwd: Optional[str] = None class ScenarioData(BaseModel): name: str description: Optional[str] = None + setup: Optional[Dict[str, Any]] = None guilds: List[ScenarioGuild] = [] channels: List[ScenarioChannel] = [] roles: List[ScenarioRole] = [] diff --git a/src/apicord/state/__pycache__/manager.cpython-314.pyc b/src/apicord/state/__pycache__/manager.cpython-314.pyc index d62bd91..bbccc5c 100644 Binary files a/src/apicord/state/__pycache__/manager.cpython-314.pyc and b/src/apicord/state/__pycache__/manager.cpython-314.pyc differ diff --git a/src/apicord/state/manager.py b/src/apicord/state/manager.py index 475ab39..b2cefcd 100644 --- a/src/apicord/state/manager.py +++ b/src/apicord/state/manager.py @@ -146,8 +146,17 @@ class StateManager: """Load state from a scenario definition.""" logger.info("Loading scenario state") + # Handle both old format (flat) and new format (with setup key) + if "setup" in scenario_data: + setup_data = scenario_data["setup"] + # Merge top-level fields (name, description, steps) with setup data + merged_data = {**scenario_data, **setup_data} + merged_data.pop("setup", None) + else: + merged_data = scenario_data + # Parse scenario using Pydantic model - scenario = ScenarioData(**scenario_data) + scenario = ScenarioData(**merged_data) # Clear existing state self.guilds.clear() diff --git a/src/apicord/tui/actions/scenario_runner.py b/src/apicord/tui/actions/scenario_runner.py new file mode 100644 index 0000000..50e9405 --- /dev/null +++ b/src/apicord/tui/actions/scenario_runner.py @@ -0,0 +1,513 @@ +""" +Scenario Runner Module + +Executes test scenarios headless or with TUI for automated testing of Discord bots. +""" + +import asyncio +import json +import logging +import sys +import time +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Dict, List, Optional, Union + +import yaml + +from apicord.config import EmulatorConfig +from apicord.state.manager import StateManager +from apicord.rest.server import RESTServer +from apicord.gateways.server import GatewayServer +from apicord.tui.actions.slash_commands import SlashCommandHandler, InteractionHandler +from apicord.models import ScenarioData + +logger = logging.getLogger("apicord.tui.scenario") + + +@dataclass +class StepResult: + """Result of a single scenario step execution.""" + step_index: int + action: str + passed: bool = False + failures: List[str] = field(default_factory=list) + duration: float = 0.0 + response: Optional[Dict[str, Any]] = None + + +@dataclass +class TestResults: + """Complete test results for a scenario.""" + scenario_name: str + passed: bool + total_steps: int + passed_steps: int + failed_steps: int + total_duration: float + steps: List[StepResult] = field(default_factory=list) + started_at: str = "" + finished_at: str = "" + + def to_junit_xml(self) -> str: + """Generate JUnit XML output.""" + lines = [ + '', + f'', + ] + for step in self.steps: + if step.passed: + lines.append(f' ') + else: + lines.append(f' ') + for failure in step.failures: + lines.append(f' ') + lines.append(' ') + lines.append('') + return "\n".join(lines) + + def to_json(self) -> str: + """Generate JSON output.""" + import json + return json.dumps({ + "scenario_name": self.scenario_name, + "passed": self.passed, + "total_steps": self.total_steps, + "passed_steps": self.passed_steps, + "failed_steps": self.failed_steps, + "total_duration": self.total_duration, + "started_at": self.started_at, + "finished_at": self.finished_at, + "steps": [ + { + "step_index": s.step_index, + "action": s.action, + "passed": s.passed, + "failures": s.failures, + "duration": s.duration, + "response": s.response, + } + for s in self.steps + ] + }, indent=2) + + +class ScenarioStep: + """Represents a single step in a test scenario.""" + + def __init__(self, data: Dict[str, Any]): + self.action = data.get("action", "") + self.params = {k: v for k, v in data.items() if k not in ("action", "expect")} + self.expectations = data.get("expect", []) + + +class ScenarioRunner: + """Runs test scenarios against the emulator.""" + + def __init__(self, config: EmulatorConfig, scenario: ScenarioData): + self.config = config + self.scenario = scenario + self.state_manager = StateManager(config) + self.rest_server: Optional[RESTServer] = None + self.gateway_server: Optional[GatewayServer] = None + self.slash_handler: Optional[Any] = None + self.interaction_handler: Optional[Any] = None + self.bot_process: Optional[asyncio.subprocess.Process] = None + self.results: List[StepResult] = [] + self._servers_started = False + self._bot_started = False + + async def run(self, timeout: int = 60) -> TestResults: + """Run the complete scenario and return results.""" + started_at = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()) + logger.info(f"Starting scenario: {self.scenario.name}") + + try: + # 1. Load scenario state + logger.info("Loading scenario state...") + # Pass the original YAML structure with setup key to StateManager + scenario_dict = self.scenario.model_dump() + # Reconstruct the setup key from the flat fields + setup_data = { + "guilds": self.scenario.guilds, + "channels": self.scenario.channels, + "roles": self.scenario.roles, + "users": self.scenario.users, + "bot": self.scenario.bot, + } + scenario_dict = {**self.scenario.model_dump(), "setup": setup_data} + await self.state_manager.load_scenario(scenario_dict) + + # 2. Start emulator servers + await self._start_servers() + + # 3. Initialize handlers + self.slash_handler = SlashCommandHandler(self.state_manager, self) + self.interaction_handler = InteractionHandler(self.state_manager, self.slash_handler) + + # 4. Start bot process if specified + if self.scenario.bot and self.scenario.bot.command: + await self._start_bot() + + # 5. Execute scenario steps + steps = self.scenario.steps + logger.info(f"Executing {len(steps)} scenario steps...") + + for i, step_data in enumerate(steps): + step = ScenarioStep(step_data) + result = await self._execute_step(i, step, timeout) + self.results.append(result) + + if not result.passed: + logger.error(f"Step {i+1} failed: {result.failures}") + else: + logger.info(f"Step {i+1} passed ({step.action})") + + passed_count = sum(1 for r in self.results if r.passed) + total_duration = sum(r.duration for r in self.results) + + return TestResults( + scenario_name=self.scenario.name, + passed=all(r.passed for r in self.results), + total_steps=len(self.results), + passed_steps=passed_count, + failed_steps=len(self.results) - passed_count, + total_duration=total_duration, + steps=self.results, + started_at=started_at, + finished_at=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + ) + + except Exception as e: + logger.error(f"Scenario execution failed: {e}", exc_info=True) + return TestResults( + scenario_name=self.scenario.name, + passed=False, + total_steps=len(self.results), + passed_steps=sum(1 for r in self.results if r.passed), + failed_steps=len(self.results) - sum(1 for r in self.results if r.passed), + total_duration=sum(r.duration for r in self.results), + steps=self.results, + started_at=started_at, + finished_at=time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()), + ) + finally: + await self._cleanup() + + async def _start_servers(self) -> None: + """Start REST and Gateway servers.""" + logger.info("Starting emulator servers...") + self.rest_server = RESTServer(self.config, self.state_manager) + self.gateway_server = GatewayServer(self.config, self.state_manager) + + await self.rest_server.start() + await self.gateway_server.start() + + self._servers_started = True + logger.info(f"REST server on http://{self.config.host}:{self.config.rest_port}") + logger.info(f"Gateway on ws://{self.config.host}:{self.config.gateway_port}") + + async def _start_bot(self) -> None: + """Start the bot process in emulator mode.""" + bot_config = self.scenario.bot + if not bot_config or not bot_config.command: + return + + logger.info("Starting bot process...") + cmd = bot_config.command + if isinstance(cmd, str): + cmd = cmd.split() + + env = { + **bot_config.env, + "DISCORD_TOKEN": self.config.bot_token, + "ALLOWED_GUILD_IDS": ",".join(str(g) for g in self.config.allowed_guild_ids), + "PANEL_URL": self.config.pterodactyl_url, + "CLIENT_API_KEY": self.config.client_api_key, + "APPLICATION_API_KEY": self.config.application_api_key, + "EMULATOR_MODE": "true", + "EMULATOR_REST_URL": self.config.get_rest_url(), + "EMULATOR_GATEWAY_URL": self.config.get_gateway_url(), + } + + self.bot_process = await asyncio.create_subprocess_exec( + *cmd, + cwd=bot_config.cwd or ".", + env=env, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + + self._bot_started = True + + # Give bot time to connect + await asyncio.sleep(2) + logger.info(f"Bot process started (PID: {self.bot_process.pid}") + + async def _execute_step(self, step_index: int, step: ScenarioStep, timeout: int) -> StepResult: + """Execute a single scenario step and validate expectations.""" + start_time = time.time() + logger.info(f"Step {step_index + 1}: {step.action}") + + result = StepResult(step_index=step_index, action=step.action) + + try: + # Execute action + response = await self._execute_action(step.action, step.params) + result.response = response + + # Validate expectations + if step.expectations: + for expectation in step.expectations: + await self._validate_expectation(expectation, response) + + except Exception as e: + result.passed = False + result.failures.append(f"Execution error: {str(e)}") + logger.error(f"Step {step_index + 1} failed: {e}", exc_info=True) + + result.duration = time.time() - start_time + result.passed = len(result.failures) == 0 + return result + + async def _execute_action(self, action: str, params: Dict[str, Any]) -> Dict[str, Any]: + """Execute a single action and return the response.""" + if action == "slash_command": + return await self._execute_slash_command(params) + elif action == "button_click": + return await self._execute_button_click(params) + elif action == "select_menu": + return await self._execute_select_menu(params) + elif action == "wait": + duration = params.get("duration", 1) + await asyncio.sleep(duration) + return {"type": 0, "data": {}} + else: + raise ValueError(f"Unknown action: {action}") + + async def _execute_slash_command(self, params: Dict[str, Any]) -> Dict[str, Any]: + """Execute a slash command.""" + if not self.slash_handler: + raise RuntimeError("Slash handler not initialized") + + return await self.slash_handler.execute( + command=params.get("command", ""), + args=params.get("args", []), + user_id=params.get("user_id", self.config.bot_id), + guild_id=params.get("guild_id", self.config.allowed_guild_ids[0] if self.config.allowed_guild_ids else ""), + channel_id=params.get("channel_id", ""), + ) + + async def _execute_button_click(self, params: Dict[str, Any]) -> Dict[str, Any]: + """Execute a button click interaction.""" + if not self.interaction_handler: + raise RuntimeError("Interaction handler not initialized") + + interaction = { + "type": 3, # MESSAGE_COMPONENT + "id": params.get("interaction_id", f"interaction_{int(time.time()*1000)}"), + "token": params.get("token", "test_token"), + "data": { + "component_type": 2, # BUTTON + "custom_id": params.get("custom_id", ""), + }, + "user": {"id": params.get("user_id", self.config.bot_id)}, + "guild_id": params.get("guild_id", ""), + "channel_id": params.get("channel_id", ""), + "message": {"id": params.get("message_id", "")}, + } + + return await self.interaction_handler.handle_interaction(interaction) + + async def _execute_select_menu(self, params: Dict[str, Any]) -> Dict[str, Any]: + """Execute a select menu interaction.""" + if not self.interaction_handler: + raise RuntimeError("Interaction handler not initialized") + + interaction = { + "type": 3, # MESSAGE_COMPONENT + "id": params.get("interaction_id", f"interaction_{int(time.time()*1000)}"), + "token": params.get("token", "test_token"), + "data": { + "component_type": 3, # SELECT_MENU + "custom_id": params.get("custom_id", ""), + "values": params.get("values", []), + }, + "user": {"id": params.get("user_id", self.config.bot_id)}, + "guild_id": params.get("guild_id", ""), + "channel_id": params.get("channel_id", ""), + "message": {"id": params.get("message_id", "")}, + } + + return await self.interaction_handler.handle_interaction(interaction) + + async def _validate_expectation(self, expectation: Dict[str, Any], response: Dict[str, Any]) -> None: + """Validate a single expectation against the response.""" + for key, expected in expectation.items(): + if key == "type": + actual = response.get("type") + if actual != expected: + raise AssertionError(f"Expected type {expected}, got {actual}") + + elif key == "data": + data = response.get("data", {}) + for data_key, data_expected in expected.items(): + if data_key == "content": + actual = data.get("content", "") + if isinstance(data_expected, str): + if data_expected not in actual: + raise AssertionError(f"Expected content to contain '{data_expected}', got '{actual}'") + elif isinstance(data_expected, dict): + # content_contains + if "contains" in data_expected: + if data_expected["contains"] not in actual: + raise AssertionError(f"Expected content to contain '{data_expected['contains']}', got '{actual}'") + + elif data_key == "content_contains": + actual = data.get("content", "") + # expected can be a string directly, or a dict with 'contains' key + # data_expected is the value of 'content_contains' key + expected_content = data_expected if isinstance(data_expected, str) else data_expected.get("contains") if isinstance(data_expected, dict) else data_expected + # Handle case where expected_content is None + if expected_content is None: + raise AssertionError(f"Expected content to contain a non-empty string, got None") + if expected_content not in actual: + raise AssertionError(f"Expected content to contain '{expected_content}', got '{actual}'") + + elif data_key == "embeds": + actual_embeds = data.get("embeds", []) + if isinstance(expected, bool): + if bool(actual_embeds) != expected: + raise AssertionError(f"Expected embeds={expected}, got {bool(actual_embeds)}") + elif isinstance(expected, dict): + # Check specific embed properties + if not actual_embeds: + raise AssertionError("Expected embeds, got none") + embed = actual_embeds[0] + if "title_contains" in expected: + if expected["title_contains"] not in embed.get("title", ""): + raise AssertionError(f"Expected embed title to contain '{expected['title_contains']}', got '{embed.get('title', '')}'") + if "fields_contain" in expected: + for field_name in expected["fields_contain"]: + found = any(f.get("name") == field_name for f in embed.get("fields", [])) + if not found: + raise AssertionError(f"Expected embed field '{field_name}' not found") + + elif data_key == "components": + actual_components = data.get("components", []) + if isinstance(expected, bool): + if bool(actual_components) != expected: + raise AssertionError(f"Expected components={expected}, got {bool(actual_components)}") + elif isinstance(expected, dict): + if not actual_components: + raise AssertionError("Expected components, got none") + # actual_components is an array of action rows (type 1) + # Check if we're looking for action row type + if "component_type" in expected: + # Check the first action row's type + action_row = actual_components[0] if actual_components else {} + actual_type = action_row.get("type") + if actual_type != expected["component_type"]: + raise AssertionError(f"Expected component type {expected['component_type']}, got {actual_type}") + if "custom_id" in expected: + # custom_id would be on components within action rows + # Check all components in all action rows + found = False + for row in actual_components: + for comp in row.get("components", []): + if comp.get("custom_id") == expected["custom_id"]: + found = True + break + if not found: + raise AssertionError(f"Expected custom_id '{expected['custom_id']}', not found") + if "button_count" in expected: + # Count buttons across all action rows + button_count = sum( + sum(1 for c in row.get("components", []) if c.get("type") == 2) + for row in actual_components + ) + if button_count != expected["button_count"]: + raise AssertionError(f"Expected {expected['button_count']} buttons, got {button_count}") + + elif data_key == "flags": + actual = data.get("flags", 0) + if actual != expected: + raise AssertionError(f"Expected flags {expected}, got {actual}") + + elif key == "state": + # Validate emulator state + for state_key, expected in expected.items(): + if state_key == "messages": + channel_id = expected.get("channel_id") if isinstance(expected, dict) else None + if channel_id: + messages = self.state_manager.get_channel_messages(channel_id) + else: + messages = list(self.state_manager.messages.values()) + actual_count = len(messages) + if isinstance(expected, dict): + expected_count = expected.get("count", expected.get("count")) + else: + expected_count = expected + if actual_count != expected_count: + raise AssertionError(f"Expected {expected_count} messages, got {actual_count}") + + else: + logger.warning(f"Unknown expectation key: {key}") + + async def _cleanup(self) -> None: + """Cleanup resources.""" + if self.bot_process and self._bot_started: + try: + self.bot_process.terminate() + await asyncio.wait_for(self.bot_process.wait(), timeout=5.0) + except asyncio.TimeoutError: + self.bot_process.kill() + await self.bot_process.wait() + + if self.rest_server and self._servers_started: + await self.rest_server.stop() + + if self.gateway_server and self._servers_started: + await self.gateway_server.stop() + + if self.config.persist_state and self.state_manager: + await self.state_manager.save_state() + + logger.info("Cleanup complete") + + +async def run_scenario( + scenario_path: Union[str, Path], + config: Optional[EmulatorConfig] = None, + headless: bool = True, + timeout: int = 60, + output: Optional[Path] = None, + junit_xml: Optional[Path] = None, +) -> TestResults: + """Convenience function to run a scenario from file.""" + config = config or EmulatorConfig() + + with open(scenario_path) as f: + scenario_data = yaml.safe_load(f) + + # Merge setup key with top-level fields for ScenarioData + if "setup" in scenario_data: + setup_data = scenario_data["setup"] + merged_data = {**scenario_data, **setup_data} + merged_data.pop("setup", None) + else: + merged_data = scenario_data + + scenario = ScenarioData(**merged_data) + runner = ScenarioRunner(config, scenario) + results = await runner.run(timeout=timeout) + + if output: + with open(output, "w") as f: + f.write(results.to_json()) + + if junit_xml: + with open(junit_xml, "w") as f: + f.write(results.to_junit_xml()) + + return results \ No newline at end of file diff --git a/src/apicord/tui/actions/slash_commands.py b/src/apicord/tui/actions/slash_commands.py index 487e475..3f17553 100644 --- a/src/apicord/tui/actions/slash_commands.py +++ b/src/apicord/tui/actions/slash_commands.py @@ -242,10 +242,8 @@ class SlashCommandHandler: if not guild: return False - # Check if user has admin role or is guild owner - roles = user.guild_roles.get(guild_id, []) - # Check for admin permission (simplified) - return guild.owner_id == user_id or "admin" in [r.lower() for r in roles] + # Check if user has admin permission (simplified) + return guild.owner_id == user_id or "admin" in [r.lower() for r in user.guild_roles.get(guild_id, [])] def _error(self, message: str) -> Dict[str, Any]: return { @@ -355,10 +353,30 @@ class InteractionHandler: if custom_id == "server_dropdown" and values: server_id = values[0] # In real implementation, this would create a permanent embed + # Return response with embed and action row buttons return { "type": 4, "data": { - "content": f"✅ Server status posted in channel", + "embeds": [{ + "title": f"Server {server_id} - Running", + "description": "Server is currently running", + "color": 0x00ff00, + "fields": [ + {"name": "CPU", "value": "45.2%", "inline": True}, + {"name": "Memory", "value": "1024/4096 MB", "inline": True}, + {"name": "Disk", "value": "5.2/20 GB", "inline": True}, + ], + "footer": {"text": "Server status updates automatically"}, + }], + "components": [{ + "type": 1, # ACTION_ROW + "components": [ + {"type": 2, "label": "Start", "style": 3, "custom_id": "start_button"}, + {"type": 2, "label": "Stop", "style": 4, "custom_id": "stop_button"}, + {"type": 2, "label": "Restart", "style": 1, "custom_id": "restart_button"}, + {"type": 2, "label": "Show Address", "style": 2, "custom_id": "show_address_button"}, + ] + }], "flags": 64, } }