elemes/cli/tests/test_cli_tui.py

714 lines
24 KiB
Python

"""
Unit tests for Elemes CLI and TUI modules.
Covers:
- Configuration & Environment (.env parsing, NOISE_RE filtering, Tailscale config)
- Runner abstraction & command detection
- TUI Live status container parsing (get_services_status)
- Keyboard escape sequence parser (read_key)
- Menu navigation & index wrapping logic
- StopClean command execution sequence
- CLI argument dispatcher
"""
from __future__ import annotations
import io
import os
import re
import sys
from pathlib import Path
from unittest.mock import MagicMock, patch
import pytest
# Ensure elemes directory is on sys.path
ELEMES_DIR = Path(__file__).resolve().parent.parent.parent
if str(ELEMES_DIR) not in sys.path:
sys.path.insert(0, str(ELEMES_DIR))
from cli.commands import cmd_stopclean, list_backups
from cli.config import (
NOISE_RE,
PROJECT_NAME,
elemes_services,
env_val,
generate_ts_config,
is_verbose,
read_env,
set_verbose,
toggle_verbose,
)
from cli.runner import (
_filter_stderr_stream,
detect_compose_command,
detect_container_command,
)
from cli.tui import MenuItem, get_services_status, read_key
# ─── 1. Configuration & .env Parsing Tests ────────────────────────────────────
@pytest.mark.unit
def test_read_env_parsing(tmp_path: Path):
"""Test native .env parser with various syntax styles."""
env_file = tmp_path / ".env"
env_file.write_text(
"# Comment line\n"
"APP_PORT=8000\n"
"DATABASE_URL=\"postgres://user:pass@host:5432/db\"\n"
"SINGLE_QUOTED='single_value'\n"
" SPACED_KEY = spaced_val \n"
"\n"
"# Another comment\n"
"EMPTY_VAL=\n",
encoding="utf-8",
)
parsed = read_env(env_file)
assert parsed["APP_PORT"] == "8000"
assert parsed["DATABASE_URL"] == "postgres://user:pass@host:5432/db"
assert parsed["SINGLE_QUOTED"] == "single_value"
assert parsed["SPACED_KEY"] == "spaced_val"
assert parsed["EMPTY_VAL"] == ""
assert env_val("APP_PORT", env_path=env_file) == "8000"
assert env_val("NON_EXISTENT", default="fallback", env_path=env_file) == "fallback"
@pytest.mark.unit
def test_read_env_missing_file(tmp_path: Path):
"""Test reading non-existent .env file safely returns empty dict."""
missing = tmp_path / "does_not_exist.env"
assert read_env(missing) == {}
assert env_val("KEY", default="default", env_path=missing) == "default"
@pytest.mark.unit
def test_noise_re_filtering():
"""Verify podman-compose noise lines match NOISE_RE while real errors/logs do not."""
noise_samples = [
"podman-compose version: 1.0.6",
"['podman', '--version']",
"using podman version: 4.9.3",
"podman exec -T elemes python -m alembic upgrade head",
"exit code: 0",
"exit code: 1",
]
for sample in noise_samples:
assert NOISE_RE.search(sample) is not None, f"Expected noise to match: {sample}"
legitimate_samples = [
"INFO [alembic.runtime.migration] Context impl PostgresqlImpl.",
"Error: cannot connect to Postgres database",
"Traceback (most recent call last):",
"FAILED tests/test_something.py::test_fail",
"WARNING: could not find key",
]
for sample in legitimate_samples:
assert NOISE_RE.search(sample) is None, f"Expected non-noise NOT to match: {sample}"
@pytest.mark.unit
def test_verbose_toggle():
"""Verify verbose mode state toggling and global setters."""
set_verbose(False)
assert not is_verbose()
assert toggle_verbose() is True
assert is_verbose() is True
assert toggle_verbose() is False
assert is_verbose() is False
set_verbose(True)
assert is_verbose() is True
set_verbose(False)
@pytest.mark.unit
def test_elemes_services_mode(monkeypatch):
"""Verify service list based on VELXIO_MODE."""
# Local mode
monkeypatch.setattr("cli.config.read_env", lambda *a, **k: {"VELXIO_MODE": "local"})
assert elemes_services() == []
# Remote mode
monkeypatch.setattr("cli.config.read_env", lambda *a, **k: {"VELXIO_MODE": "remote"})
svcs = elemes_services()
assert "elemes-ts" in svcs
assert "postgres" in svcs
assert "elemes" in svcs
assert "velxio" not in svcs # Velxio standalone is not in primary compose in remote mode
# ─── 2. Runner & Noise Filtering Tests ────────────────────────────────────────
@pytest.mark.unit
def test_filter_stderr_stream():
"""Verify _filter_stderr_stream strips noise unless verbose is active."""
noise_stream = io.StringIO(
"podman-compose version: 1.0.6\n"
"Real error occurred!\n"
"exit code: 0\n"
)
captured = []
# When verbose=False, noise should be filtered from output
with patch("sys.stderr.write") as mock_stderr_write:
_filter_stderr_stream(noise_stream, verbose=False, captured_lines=captured)
written = [call.args[0] for call in mock_stderr_write.call_args_list]
assert "Real error occurred!\n" in written
assert "podman-compose version: 1.0.6\n" not in written
assert "exit code: 0\n" not in written
# But captured_lines records all lines
assert len(captured) == 3
@pytest.mark.unit
def test_detect_compose_command_override(monkeypatch):
"""Verify LMS_COMPOSE_CMD environment variable overrides detection."""
monkeypatch.setenv("LMS_COMPOSE_CMD", "custom-compose --special-flag")
cmd = detect_compose_command()
assert cmd == ["custom-compose", "--special-flag"]
@pytest.mark.unit
def test_detect_container_command_override(monkeypatch):
"""Verify LMS_CONTAINER_CMD environment variable overrides detection."""
monkeypatch.setenv("LMS_CONTAINER_CMD", "custom-podman")
cmd = detect_container_command()
assert cmd == "custom-podman"
# ─── 3. TUI Container Status Parsing Tests ────────────────────────────────────
@pytest.mark.unit
def test_get_services_status_parsing(monkeypatch):
"""Test get_services_status correctly parses podman ps output without quiet=True bug."""
mock_ps_output = (
"lms-dev_elemes-ts_1|Up 2 hours|map[com.docker.compose.service:elemes-ts]\n"
"lms-dev_postgres_1|Up 2 hours (healthy)|map[com.docker.compose.service:postgres]\n"
"lms-dev_elemes_1|Up 2 hours|map[com.docker.compose.service:elemes]\n"
"lms-dev_compiler-worker_1|Up 2 hours|map[com.docker.compose.service:compiler-worker]\n"
"lms-dev_elemes-frontend_1|Up 2 hours|map[com.docker.compose.service:elemes-frontend]\n"
"lms-dev_velxio_1|Exited (0) 5 minutes ago|map[com.docker.compose.service:velxio]\n"
)
def mock_run_container_cmd(args, capture_stdout=False, **kwargs):
# Crucial check: capture_stdout must be True, and quiet must NOT suppress stdout
assert capture_stdout is True
return 0, mock_ps_output, ""
monkeypatch.setattr("cli.tui.run_container_cmd", mock_run_container_cmd)
status = get_services_status()
assert status["postgres"] == "healthy"
assert status["elemes"] == "running"
assert status["compiler-worker"] == "running"
assert status["elemes-frontend"] == "running"
assert status["velxio"] == "stopped"
assert status["flowchart"] == "stopped" # Missing from ps output = stopped
@pytest.mark.unit
def test_get_services_status_empty(monkeypatch):
"""Test get_services_status with no running containers returns stopped for all."""
monkeypatch.setattr(
"cli.tui.run_container_cmd",
lambda *a, **k: (0, "", ""),
)
status = get_services_status()
for svc in ("postgres", "elemes", "elemes-frontend", "compiler-worker", "flowchart", "velxio"):
assert status[svc] == "stopped"
# ─── 4. Keyboard Escape Sequence & Unbuffered I/O Tests ───────────────────────
@pytest.mark.unit
def test_read_key_arrow_sequences(monkeypatch):
"""
Test read_key correctly parses ANSI and SS3 arrow key sequences
without Python TextIOWrapper buffering problems.
"""
monkeypatch.setattr("sys.platform", "linux")
monkeypatch.setattr("sys.stdin.isatty", lambda: True)
monkeypatch.setattr("sys.stdin.fileno", lambda: 0)
monkeypatch.setattr("termios.tcgetattr", lambda fd: [])
monkeypatch.setattr("termios.tcsetattr", lambda fd, when, old: None)
monkeypatch.setattr("tty.setraw", lambda fd: None)
# Simulate ANSI UP arrow: \x1b followed by [A
read_queue = [b"\x1b", b"[A"]
monkeypatch.setattr("os.read", lambda fd, n: read_queue.pop(0) if read_queue else b"")
monkeypatch.setattr("select.select", lambda r, w, x, t: ([r], [], []))
assert read_key() == "UP"
# Simulate ANSI DOWN arrow: \x1b followed by [B
read_queue = [b"\x1b", b"[B"]
assert read_key() == "DOWN"
# Simulate SS3 UP arrow (application cursor mode): \x1b followed by OA
read_queue = [b"\x1b", b"OA"]
assert read_key() == "UP"
# Simulate SS3 DOWN arrow: \x1b followed by OB
read_queue = [b"\x1b", b"OB"]
assert read_key() == "DOWN"
# Simulate Enter key
read_queue = [b"\r"]
assert read_key() == "ENTER"
# Simulate Standalone Esc (select times out)
read_queue = [b"\x1b"]
monkeypatch.setattr("select.select", lambda r, w, x, t: ([], [], []))
assert read_key() == "ESC"
@pytest.mark.unit
def test_read_key_windows(monkeypatch):
"""Test read_key in Windows environment using simulated msvcrt."""
monkeypatch.setattr("sys.platform", "win32")
mock_msvcrt = MagicMock()
# Windows sends \xe0 then \x48 ('H') for UP arrow
mock_msvcrt.getch.side_effect = [b"\xe0", b"H", b"\xe0", b"P", b"\r", b"q"]
with patch.dict("sys.modules", {"msvcrt": mock_msvcrt}):
assert read_key() == "UP"
assert read_key() == "DOWN"
assert read_key() == "ENTER"
assert read_key() == "q"
# ─── 5. Menu Navigation & Index Wrapping Tests ────────────────────────────────
@pytest.mark.unit
def test_menu_item_initialization():
"""Verify MenuItem properties and defaults."""
action_called = False
def sample_action():
nonlocal action_called
action_called = True
item = MenuItem("1", "Test Item", "Test Description", sample_action)
assert item.key == "1"
assert item.label == "Test Item"
assert item.desc == "Test Description"
assert item.is_submenu is False
item.action()
assert action_called is True
@pytest.mark.unit
def test_menu_index_wrapping_logic():
"""Verify index increment and decrement wrapping calculations."""
items_count = 5
current_idx = 0
# UP wraps from 0 to items_count - 1
current_idx = (current_idx - 1) % items_count
assert current_idx == 4
# DOWN wraps from items_count - 1 to 0
current_idx = (current_idx + 1) % items_count
assert current_idx == 0
# DOWN moves forward
current_idx = (current_idx + 1) % items_count
assert current_idx == 1
@pytest.mark.unit
def test_run_menu_focused_description_rendering(monkeypatch):
"""Verify run_menu renders focused description panel below the menu list."""
from cli.tui import run_menu, console
item1 = MenuItem("1", "First Choice", "Description for First Choice")
item2 = MenuItem("2", "Second Choice", "Description for Second Choice")
items = [item1, item2]
# Simulate DOWN arrow then 'q' to exit
key_inputs = ["DOWN", "q"]
monkeypatch.setattr("cli.tui.read_key", lambda: key_inputs.pop(0) if key_inputs else "q")
printed_panels = []
original_print = console.print
def capture_print(*args, **kwargs):
for a in args:
from rich.panel import Panel
if isinstance(a, Panel):
printed_panels.append(a)
return original_print(*args, **kwargs)
monkeypatch.setattr(console, "print", capture_print)
run_menu("Test Menu", items, is_main_menu=False)
# Panels printed should include the description panels for item1 then item2
desc_panels = [p for p in printed_panels if "Deskripsi" in str(p.title)]
assert len(desc_panels) >= 2
assert "First Choice" in str(desc_panels[0].title)
assert "Description for First Choice" in str(desc_panels[0].renderable)
assert "Second Choice" in str(desc_panels[1].title)
assert "Description for Second Choice" in str(desc_panels[1].renderable)
@pytest.mark.unit
def test_run_menu_enter_on_kembali_returns(monkeypatch):
"""Verify navigating with DOWN arrow to 'Kembali' and pressing ENTER exits the submenu."""
from cli.tui import MenuItem, run_menu
items = [
MenuItem("1", "Action One", "Do action one", action=lambda: None),
MenuItem("b", "Kembali", "Kembali ke menu utama", None),
]
# Navigate DOWN to 'b', then press ENTER
key_sequence = ["DOWN", "ENTER"]
monkeypatch.setattr("cli.tui.read_key", lambda: key_sequence.pop(0) if key_sequence else "b")
# Should cleanly return from run_menu without hanging
run_menu("Submenu Test", items, is_main_menu=False)
@pytest.mark.unit
def test_run_menu_enter_on_keluar_exits(monkeypatch):
"""Verify navigating with DOWN arrow to 'Keluar' and pressing ENTER calls sys.exit(0)."""
from cli.tui import MenuItem, run_menu
items = [
MenuItem("1", "Submenu One", "Go to submenu", action=lambda: None),
MenuItem("q", "Keluar", "Keluar aplikasi", None),
]
key_sequence = ["DOWN", "ENTER"]
monkeypatch.setattr("cli.tui.read_key", lambda: key_sequence.pop(0) if key_sequence else "q")
with pytest.raises(SystemExit) as exc_info:
run_menu("Main Menu Test", items, is_main_menu=True)
assert exc_info.value.code == 0
@pytest.mark.unit
def test_run_action_screen_back_keys(monkeypatch):
"""Verify run_action_screen exits wait loop when pressing 'b', 'B', or ENTER."""
from cli.tui import run_action_screen
executed = []
def dummy_action():
executed.append(True)
return 0
# Test pressing 'b' to return
monkeypatch.setattr("cli.tui.read_key", lambda: "b")
run_action_screen("Test Action", dummy_action, "test_action")
assert len(executed) == 1
# Test pressing 'B' to return
monkeypatch.setattr("cli.tui.read_key", lambda: "B")
run_action_screen("Test Action 2", dummy_action, "test_action_2")
assert len(executed) == 2
# ─── 6. Commands & StopClean Execution Tests ──────────────────────────────────
@pytest.mark.unit
def test_cmd_stopclean_sequence(monkeypatch):
"""Verify cmd_stopclean runs compose down, container prune, and image prune in order."""
calls = []
def mock_run_compose(args, verbose=False):
calls.append(("compose", args))
return 0
def mock_run_container_cmd(args, verbose=False, **kwargs):
calls.append(("container", args))
return 0, "", ""
monkeypatch.setattr("cli.commands.run_compose", mock_run_compose)
monkeypatch.setattr("cli.commands.run_container_cmd", mock_run_container_cmd)
ret = cmd_stopclean(verbose=False)
assert ret == 0
assert len(calls) == 3
assert calls[0] == ("compose", ["down"])
assert calls[1] == ("container", ["container", "prune", "-f"])
assert calls[2] == ("container", ["image", "prune", "-f"])
@pytest.mark.unit
def test_list_backups_sorted(tmp_path: Path, monkeypatch):
"""Verify list_backups finds and sorts backups newest first."""
backups_dir = tmp_path / "backups"
backups_dir.mkdir()
f1 = backups_dir / "elemes_20260101_100000.sql"
f2 = backups_dir / "elemes_20260201_100000.sql"
f3 = backups_dir / "elemes_20260301_100000.sql"
f1.write_text("dump1")
f2.write_text("dump2")
f3.write_text("dump3")
# Set artificial mtimes: f3 newest, f1 oldest
os.utime(f1, (1000, 1000))
os.utime(f2, (2000, 2000))
os.utime(f3, (3000, 3000))
monkeypatch.setattr("cli.commands.PARENT_DIR", tmp_path)
backups = list_backups()
assert len(backups) == 3
assert backups[0].name == "elemes_20260301_100000.sql"
assert backups[1].name == "elemes_20260201_100000.sql"
assert backups[2].name == "elemes_20260101_100000.sql"
# ─── 7. Tailscale Config & CLI Main Dispatcher Tests ──────────────────────────
@pytest.mark.unit
def test_generate_ts_config(tmp_path: Path, monkeypatch):
"""Test generating lms-tail.json substitutes __VELXIO_PROXY_TARGET__ properly."""
config_dir = tmp_path / "config"
config_dir.mkdir()
template = config_dir / "lms-tail.json.template"
template.write_text('{"target": "__VELXIO_PROXY_TARGET__"}')
monkeypatch.setattr("cli.config.SCRIPT_DIR", tmp_path)
# Local mode
monkeypatch.setattr("cli.config.read_env", lambda *a, **k: {"VELXIO_MODE": "local"})
generate_ts_config()
out = (config_dir / "lms-tail.json").read_text()
assert "http://127.0.0.1:80/" in out
# Remote mode
monkeypatch.setattr(
"cli.config.read_env",
lambda *a, **k: {"VELXIO_MODE": "remote", "VELXIO_HOST": "my-velxio-host"},
)
generate_ts_config()
out = (config_dir / "lms-tail.json").read_text()
assert "http://my-velxio-host:80/" in out
@pytest.mark.unit
def test_cli_main_dispatch(monkeypatch):
"""Test CLI main() routing without running containers."""
import elemes
# 1. Help flag returns 0
monkeypatch.setattr("sys.argv", ["elemes.py", "--help"])
assert elemes.main() == 0
# 2. Unknown command returns 1
monkeypatch.setattr("sys.argv", ["elemes.py", "unknown-random-cmd"])
assert elemes.main() == 1
# 3. StopClean routes to cmd_stopclean
stopclean_called = False
def mock_stopclean(*a, **k):
nonlocal stopclean_called
stopclean_called = True
return 0
monkeypatch.setattr("elemes.cmd_stopclean", mock_stopclean)
monkeypatch.setattr("sys.argv", ["elemes.py", "stopclean"])
assert elemes.main() == 0
assert stopclean_called is True
@pytest.mark.unit
def test_all_cli_commands_dispatch_matrix(monkeypatch):
"""Verify that every single CLI command routes and executes without NameError or signature errors."""
import elemes
all_commands = [
"init",
"run",
"runbuild",
"runclearbuild",
"stop",
"stopclean",
"stop-clean",
"verify",
"dbupgrade",
"dbstatus",
"teacher",
"dbbackup",
"dbrestore",
"test",
"test-all",
"test-unit",
"test-integration",
"test-smoke",
"test-worker",
"docs-validate",
"exportall",
"importall",
"velxio-run",
"velxio-runbuild",
"velxio-stop",
"velxio-status",
"velxio-test",
"loadtest",
]
# Mock all cmd_* in elemes to record their invocation and return 0
for name in dir(elemes):
if name.startswith("cmd_") and callable(getattr(elemes, name)):
monkeypatch.setattr(elemes, name, lambda *a, **k: 0)
for cmd in all_commands:
monkeypatch.setattr("sys.argv", ["elemes.py", cmd])
ret = elemes.main()
assert ret == 0, f"Command {cmd} failed with exit code {ret}"
# Also verify with --verbose flag
for cmd in ["run", "verify", "dbstatus", "stopclean"]:
monkeypatch.setattr("sys.argv", ["elemes.py", "--verbose", cmd])
ret = elemes.main()
assert ret == 0
# ─── 8. Verbose Mode Effect on Container Menu & Commands ──────────────────────
@pytest.mark.unit
def test_container_menu_respects_verbose_toggle(monkeypatch):
"""Verify that toggling verbose mode propagates directly into container menu actions."""
import cli.tui as tui
from cli.config import set_verbose
captured_verbose = []
def mock_cmd_run(verbose=False):
captured_verbose.append(verbose)
return 0
monkeypatch.setattr("cli.tui.cmd_run", mock_cmd_run)
monkeypatch.setattr("cli.tui.run_action_screen", lambda title, func, cmd: func())
# Get menu items from menu_server_lifecycle
captured_items = []
monkeypatch.setattr("cli.tui.run_menu", lambda title, items: captured_items.extend(items))
tui.menu_server_lifecycle()
run_item = next(it for it in captured_items if it.key == "1")
# 1. When verbose is OFF
set_verbose(False)
run_item.action()
assert captured_verbose[-1] is False
# 2. When verbose is ON
set_verbose(True)
run_item.action()
assert captured_verbose[-1] is True
# Reset
set_verbose(False)
@pytest.mark.unit
def test_cmd_verify_verbose_output(capsys, monkeypatch):
"""Verify cmd_verify includes engine and project details only when verbose is True."""
from cli.commands import cmd_verify
from cli.config import set_verbose
monkeypatch.setattr("cli.commands.run_compose_out", lambda *a, **k: (0, "elemes\n", "services:\n elemes:\n image: lms:latest\n"))
monkeypatch.setattr("cli.commands.run_container_cmd", lambda *a, **k: (0, "", ""))
# Verbose OFF
set_verbose(False)
cmd_verify(verbose=False)
captured = capsys.readouterr().out
assert "Engine Compose" not in captured
assert "Project Name" not in captured
# Verbose ON
set_verbose(True)
cmd_verify(verbose=True)
captured_verbose = capsys.readouterr().out
assert "Engine Compose" in captured_verbose
assert "Project Name" in captured_verbose
set_verbose(False)
@pytest.mark.unit
def test_run_compose_verbose_vs_quiet(monkeypatch):
"""Verify run_compose executes quietly when verbose=False, and verbose=True when active."""
from cli.runner import run_compose
from cli.config import set_verbose
executed_calls = []
def mock_execute_process(cmd, quiet=False, verbose=False, **kwargs):
executed_calls.append({"cmd": cmd, "quiet": quiet, "verbose": verbose})
return 0, "", ""
monkeypatch.setattr("cli.runner.execute_process", mock_execute_process)
# 1. Verbose OFF -> quiet=True
set_verbose(False)
run_compose(["up", "-d"], verbose=False)
assert len(executed_calls) == 1
assert executed_calls[0]["quiet"] is True
# 2. Verbose ON -> quiet=False, verbose=True
set_verbose(True)
run_compose(["up", "-d"], verbose=False) # even if verbose arg is False, is_verbose() is True
assert len(executed_calls) == 2
assert executed_calls[1]["quiet"] is False
assert executed_calls[1]["verbose"] is True
set_verbose(False)
@pytest.mark.unit
def test_all_submenus_integrity_and_actions(monkeypatch):
"""
Ensure every submenu in tui.py constructs its MenuItem list properly
and all actions can be resolved and invoked without NameError (e.g. cmd_dbbackup).
"""
import cli.tui as tui
collected_submenus = []
def mock_run_menu(title, items, is_main_menu=False):
collected_submenus.append((title, items))
monkeypatch.setattr(tui, "run_menu", mock_run_menu)
monkeypatch.setattr(tui, "run_action_screen", lambda title, func, cmd: func())
monkeypatch.setattr(tui, "interactive_restore_screen", lambda: 0)
# Mock all potential cmd_* calls inside actions to return 0
from cli import commands
for attr in dir(commands):
if attr.startswith("cmd_") and callable(getattr(commands, attr)):
monkeypatch.setattr(commands, attr, lambda *a, **k: 0)
if hasattr(tui, attr):
monkeypatch.setattr(tui, attr, lambda *a, **k: 0)
# Invoke all menu builders
tui.menu_server_lifecycle()
tui.menu_database_users()
tui.menu_test_suite()
tui.menu_velxio()
tui.menu_deploy_export()
tui.menu_setup_env()
assert len(collected_submenus) == 6
# Test every item action in every submenu
for title, items in collected_submenus:
assert len(items) > 0
for item in items:
assert item.label
assert item.key
if item.action and not item.is_submenu:
# Should execute cleanly without NameError (e.g. cmd_dbbackup)
item.action()