Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
async def test_paginator_interface():
bot = commands.Bot('?')
with open(__file__, 'rb') as file:
paginator = FilePaginator(file, max_size=200)
interface = PaginatorInterface(bot, paginator)
assert interface.pages == paginator.pages
assert interface.page_count == len(paginator.pages)
assert interface.page_size > 200
assert interface.page_size < interface.max_page_size
send_kwargs = interface.send_kwargs
assert isinstance(send_kwargs, dict)
assert 'content' in send_kwargs
content = send_kwargs['content']
assert isinstance(content, str)
assert len(content) <= interface.page_size
embed = send_kwargs['embed']
assert isinstance(embed, discord.Embed)
description = embed.description
assert content.startswith(description)
# check for raise on too large page size
with pytest.raises(ValueError):
PaginatorInterface(None, commands.Paginator(max_size=2000))
# check for raise on not-paginator
with pytest.raises(TypeError):
PaginatorInterface(None, 4)
paginator = commands.Paginator(max_size=100)
for _ in range(100):
paginator.add_line("test text")
# test interfacing
with utils.mock_ctx(bot) as ctx:
interface = PaginatorInterface(bot, paginator)
assert not interface.closed
await interface.send_to(ctx)
await asyncio.sleep(0.1)
await interface.add_line("test text")
)
bot.dispatch(
'raw_reaction_add',
discord.RawReactionActionEvent(payload, emoji, 'REACTION_ADD')
if discord.version_info >= (1, 3) else
discord.RawReactionActionEvent(payload, emoji)
)
await asyncio.sleep(0.1)
assert interface.closed
ctx.send.coro.return_value.delete.assert_called_once()
# test resend, no delete
with utils.mock_ctx(bot) as ctx:
interface = PaginatorInterface(bot, paginator)
assert not interface.closed
await interface.send_to(ctx)
await asyncio.sleep(0.1)
await interface.add_line("test text")
assert interface.page_count > 1
assert not interface.closed
# resend
await interface.send_to(ctx)
await asyncio.sleep(0.1)
await interface.add_line("test text")
# resend
await interface.send_to(ctx)
await asyncio.sleep(0.1)
await interface.add_line("test text")
ctx.send.coro.return_value.delete.assert_not_called()
interface.task.cancel()
await asyncio.sleep(0.1)
assert interface.closed
# test resend, delete
with utils.mock_ctx(bot) as ctx:
interface = PaginatorInterface(bot, paginator, delete_message=True)
assert not interface.closed
await interface.send_to(ctx)
await asyncio.sleep(0.1)
await interface.add_line("test text")
assert interface.page_count > 1
assert not interface.closed
# resend
await interface.send_to(ctx)
await asyncio.sleep(0.1)
await interface.add_line("test text")
def test_wrapped_paginator():
paginator = WrappedPaginator(max_size=200)
paginator.add_line("abcde " * 50)
assert len(paginator.pages) == 2
paginator = WrappedPaginator(max_size=200, include_wrapped=False)
paginator.add_line("abcde " * 50)
assert len(paginator.pages) == 2
def test_wrapped_paginator():
paginator = WrappedPaginator(max_size=200)
paginator.add_line("abcde " * 50)
assert len(paginator.pages) == 2
paginator = WrappedPaginator(max_size=200, include_wrapped=False)
paginator.add_line("abcde " * 50)
assert len(paginator.pages) == 2
async def test_reader_basic():
return_data = []
with ShellReader("echo hi") as reader:
async for result in reader:
return_data.append(result)
assert len(return_data) == 1
assert return_data[0] == "hi"
with pytest.raises(asyncio.TimeoutError):
with ShellReader("sleep 2", timeout=1) as reader:
async for result in reader:
pass
async def test_reader_basic():
return_data = []
with ShellReader("echo hi") as reader:
async for result in reader:
return_data.append(result)
assert len(return_data) == 1
assert return_data[0] == "hi"
with pytest.raises(asyncio.TimeoutError):
with ShellReader("sleep 2", timeout=1) as reader:
async for result in reader:
pass
async def test_linux():
return_data = []
with ShellReader(">&2 echo oops") as reader:
async for result in reader:
return_data.append(result)
assert len(return_data) == 1
assert return_data[0] == "[stderr] oops"
return_data = []
with ShellReader("echo one && echo two") as reader:
async for result in reader:
return_data.append(result)
assert len(return_data) == 2
assert return_data[0] == "one"
assert return_data[1] == "two"
async def test_windows():
return_data = []
with ShellReader("cmd /c \"echo one && echo two && echo three 1>&2\"") as reader:
async for result in reader:
return_data.append(result)
assert len(return_data) == 3
return_data = [x.strip() for x in return_data]
assert "one" in return_data
assert "two" in return_data
assert "[stderr] three" in return_data