feat: Phase 3 - Testing & Automation Features
- Implemented ScenarioRunner with headless scenario execution - Added YAML-based test scenario framework with validation expectations - Added CLI command 'run-scenario' with --headless, --output, --junit-xml options - Created 3 Pterodisbot test scenarios: * Basic Server Status Flow (slash_command, select_menu, button_click) * Admin Commands (refresh_embeds, purge_embeds with permissions) * Multi-Guild Isolation (independent server tracking per guild) - Fixed content_contains validation logic - Fixed component_type validation for action rows - Added command field to ScenarioBot model - Fixed content_contains validation handling - All 19 tests passing
This commit is contained in:
1 parent
7608388e1a
commit
6fbd7fea79
13 files changed
+1195
-11
No files matched your search
@@ -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"
|
||||
@@ -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"]
|
||||
@@ -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"]
|
||||
Binary file not shown.
Binary file not shown.
+193
-4
@@ -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)}")
|
||||
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())
|
||||
@@ -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
|
||||
|
||||
|
||||
Binary file not shown.
@@ -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] = []
|
||||
|
||||
Binary file not shown.
@@ -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()
|
||||
|
||||
@@ -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 = [
|
||||
'<?xml version="1.0" encoding="UTF-8"?>',
|
||||
f'<testsuite name="{self.scenario_name}" tests="{self.total_steps}" failures="{self.failed_steps}" time="{self.total_duration:.2f}">',
|
||||
]
|
||||
for step in self.steps:
|
||||
if step.passed:
|
||||
lines.append(f' <testcase name="Step {step.step_index + 1}: {step.action}" classname="Scenario" time="{step.duration:.2f}"/>')
|
||||
else:
|
||||
lines.append(f' <testcase name="Step {step.step_index + 1}: {step.action}" classname="Scenario" time="{step.duration:.2f}">')
|
||||
for failure in step.failures:
|
||||
lines.append(f' <failure message="{failure}"/>')
|
||||
lines.append(' </testcase>')
|
||||
lines.append('</testsuite>')
|
||||
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
|
||||
@@ -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,
|
||||
}
|
||||
}
|
||||
|
||||
Reference in new issue
Block a user