70 lines
2.6 KiB
Python
70 lines
2.6 KiB
Python
import pytest
|
|
|
|
from bot_handlers import CommandRouter
|
|
from conftest import CHANNEL_ID, EXAMPLE_CAPTION, build_queue, run
|
|
from models import State
|
|
|
|
ADMIN, STRANGER = 111, 999
|
|
|
|
COMMANDS = ["/start", "/help", "/status", "/run", "/pause", "/resume", "/stop", "/retry", "/failed",
|
|
"/progress", "/folders", "/check 1", "/sync"]
|
|
|
|
|
|
@pytest.fixture
|
|
def router(config, db, source, drive):
|
|
source.add(1, EXAMPLE_CAPTION)
|
|
queue = build_queue(config, db, source, drive)
|
|
return CommandRouter(config, db, queue)
|
|
|
|
|
|
@pytest.mark.parametrize("command", COMMANDS)
|
|
def test_non_admin_is_rejected(router, command):
|
|
reply = run(router.handle(STRANGER, command))
|
|
assert reply == "You are not authorized to control this bot."
|
|
assert router.queue.running is False
|
|
assert router.db.counts()["PENDING"] == 0 # /sync did not run
|
|
|
|
|
|
def test_none_sender_is_rejected(router):
|
|
assert "not authorized" in run(router.handle(None, "/status"))
|
|
|
|
|
|
def test_non_commands_are_ignored(router):
|
|
assert run(router.handle(ADMIN, "hello")) is None
|
|
|
|
|
|
def test_admin_flow(router, drive):
|
|
async def flow():
|
|
replies = {}
|
|
replies["help"] = await router.handle(ADMIN, "/help")
|
|
replies["sync"] = await router.handle(ADMIN, "/sync@MyBot")
|
|
replies["check"] = await router.handle(ADMIN, "/check 1")
|
|
replies["run"] = await router.handle(ADMIN, "/run")
|
|
await router.queue.wait()
|
|
replies["progress"] = await router.handle(ADMIN, "/progress")
|
|
replies["folders"] = await router.handle(ADMIN, "/folders")
|
|
replies["status"] = await router.handle(ADMIN, "/status")
|
|
replies["failed"] = await router.handle(ADMIN, "/failed")
|
|
replies["unknown"] = await router.handle(ADMIN, "/nope")
|
|
replies["badcheck"] = await router.handle(ADMIN, "/check abc")
|
|
return replies
|
|
|
|
r = run(flow())
|
|
assert "/status" in r["help"]
|
|
assert "New videos queued: 1" in r["sync"]
|
|
assert "Class 01 Aman sir.mp4" in r["check"] and "001, 002" in r["check"]
|
|
assert r["run"].startswith("Processing started")
|
|
assert "Processed (uploaded & verified): 1" in r["progress"]
|
|
assert "Misc• Grammar Practice: 1" in r["folders"]
|
|
assert "Queue: idle" in r["status"]
|
|
assert r["failed"] == "No failed items."
|
|
assert "Unknown command" in r["unknown"]
|
|
assert r["badcheck"].startswith("Usage")
|
|
|
|
|
|
def test_failed_listing_hides_paths(router, db):
|
|
run(router.handle(ADMIN, "/sync"))
|
|
db.transition((CHANNEL_ID, 1), State.FAILED, last_error="cannot open /Users/someone/secret/dir/file.part")
|
|
reply = run(router.handle(ADMIN, "/failed"))
|
|
assert "/Users/someone" not in reply and "file.part" in reply
|