From 5675dfd20b24f1e0eef06b11d9ec41a894d3a5bf Mon Sep 17 00:00:00 2001 From: salmonik Date: Tue, 28 Apr 2026 20:05:36 +0200 Subject: multitude of changes. but primarly, use db for all data. Signed-off-by: salmonik --- tests/cogs/test_uploader.py | 226 ++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 226 insertions(+) create mode 100644 tests/cogs/test_uploader.py (limited to 'tests/cogs/test_uploader.py') diff --git a/tests/cogs/test_uploader.py b/tests/cogs/test_uploader.py new file mode 100644 index 0000000..addf87e --- /dev/null +++ b/tests/cogs/test_uploader.py @@ -0,0 +1,226 @@ +import pytest +from unittest.mock import AsyncMock, patch, MagicMock +import os +import discord +from tests.conftest import MockApplicationContext, MockMessage, MockUser + + +@pytest.fixture +def uploader_cog(bot): + with patch('cogs.utils.temp.get_temp_dir', return_value='/tmp/test'): + from cogs.uploader import Uploader + cog = Uploader(bot) + return cog + + +@pytest.mark.asyncio +async def test_file_autocomplete(uploader_cog): + mock_autocomplete_ctx = MagicMock() + mock_autocomplete_ctx.value = "test" + mock_interaction = MagicMock() + mock_user = MockUser(id=12345) + mock_interaction.user = mock_user + mock_autocomplete_ctx.interaction = mock_interaction + + with patch('cogs.utils.file_search.UniversalSearcher.search_copyparty_autocomplete', new_callable=AsyncMock) as mock_search: + mock_search.return_value = [ + "https://copyparty.local/testfile1.mp3", + "https://copyparty.local/testfile2.mp3" + ] + result = await uploader_cog._file_autocomplete(mock_autocomplete_ctx) + assert len(result) == 2 + assert result[0].name == "testfile1.mp3" + + +@pytest.mark.asyncio +async def test_file_autocomplete_empty_query(uploader_cog): + mock_autocomplete_ctx = MagicMock() + mock_autocomplete_ctx.value = "" + mock_interaction = MagicMock() + mock_interaction.user = MockUser(id=12345) + mock_autocomplete_ctx.interaction = mock_interaction + + with patch('cogs.utils.file_search.UniversalSearcher.search_copyparty_autocomplete', new_callable=AsyncMock) as mock_search: + result = await uploader_cog._file_autocomplete(mock_autocomplete_ctx) + mock_search.assert_called_once_with("", max_results=25) + + +@pytest.mark.asyncio +async def test_download_from_url_success(uploader_cog): + with patch('requests.get') as mock_get: + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.content = b"test file content" + mock_get.return_value = mock_response + + mock_file = MagicMock() + mock_file.write = AsyncMock() + + mock_cm = AsyncMock() + mock_cm.__aenter__ = AsyncMock(return_value=mock_file) + mock_cm.__aexit__ = AsyncMock(return_value=None) + + with patch('cogs.uploader.aiofiles.open', return_value=mock_cm): + result = await uploader_cog._download_from_url("https://copyparty.local/file.txt") + + assert result is not None + assert "file.txt" in result + mock_file.write.assert_called_once_with(b"test file content") + + +@pytest.mark.asyncio +async def test_download_from_url_failure(uploader_cog): + with patch('requests.get') as mock_get: + mock_response = MagicMock() + mock_response.status_code = 404 + mock_get.return_value = mock_response + + result = await uploader_cog._download_from_url("https://copyparty.local/notfound.txt") + + assert result is None + + +@pytest.mark.asyncio +async def test_upload_to_copyparty_success(uploader_cog): + mock_file = MagicMock() + mock_file.read = AsyncMock(return_value=b"test content") + + mock_cm = AsyncMock() + mock_cm.__aenter__ = AsyncMock(return_value=mock_file) + mock_cm.__aexit__ = AsyncMock(return_value=None) + + with patch('cogs.uploader.aiofiles.open', return_value=mock_cm): + with patch('requests.post') as mock_post: + mock_response = MagicMock() + mock_response.status_code = 200 + mock_response.json.return_value = {"fileurl": "https://copyparty.local/uploaded.txt"} + mock_post.return_value = mock_response + + with patch.dict(os.environ, {"COPYPARTY_PASSWORD": "testpass"}): + result = await uploader_cog._upload_to_copyparty("/tmp/testfile.txt") + + assert result == "https://copyparty.local/uploaded.txt" + + +@pytest.mark.asyncio +async def test_upload_to_copyparty_failure(uploader_cog): + mock_file = MagicMock() + mock_file.read = AsyncMock(return_value=b"test content") + + mock_cm = AsyncMock() + mock_cm.__aenter__ = AsyncMock(return_value=mock_file) + mock_cm.__aexit__ = AsyncMock(return_value=None) + + with patch('cogs.uploader.aiofiles.open', return_value=mock_cm): + with patch('requests.post') as mock_post: + mock_response = MagicMock() + mock_response.status_code = 500 + mock_post.return_value = mock_response + + result = await uploader_cog._upload_to_copyparty("/tmp/testfile.txt") + + assert result is None + + +@pytest.mark.asyncio +async def test_upload_file_no_url(uploader_cog): + uploader_cog._uploadurl = None + + result = await uploader_cog._upload_file("/tmp/test.txt") + + assert result is None + + +@pytest.mark.asyncio +async def test_message_upload_no_attachments(uploader_cog, mock_channel, mock_guild, bot): + ctx = MockApplicationContext(bot=bot, channel=mock_channel, guild=mock_guild) + + mock_message = MagicMock() + mock_message.attachments = [] + + with patch.object(ctx, 'respond', new_callable=AsyncMock) as mock_respond: + await uploader_cog.message_upload.callback(uploader_cog, ctx, message=mock_message) + mock_respond.assert_called_once_with("Brak załączników w wiadomości", ephemeral=True) + + +@pytest.mark.asyncio +async def test_message_upload_success(uploader_cog, mock_channel, mock_guild, bot): + ctx = MockApplicationContext(bot=bot, channel=mock_channel, guild=mock_guild) + + mock_attachment = MagicMock() + mock_attachment.filename = "test.txt" + mock_attachment.save = AsyncMock() + + mock_message = MagicMock() + mock_message.attachments = [mock_attachment] + + with patch.object(uploader_cog, '_upload_file', new_callable=AsyncMock) as mock_upload: + mock_upload.return_value = "https://copyparty.local/uploaded.txt" + + with patch('os.path.exists', return_value=True): + with patch('os.remove'): + with patch.object(ctx, 'defer', new_callable=AsyncMock): + with patch.object(ctx, 'respond', new_callable=AsyncMock) as mock_respond: + await uploader_cog.message_upload.callback(uploader_cog, ctx, message=mock_message) + mock_respond.assert_called_once() + assert "Wysłano" in str(mock_respond.call_args) + + +@pytest.mark.asyncio +async def test_message_upload_all_failed(uploader_cog, mock_channel, mock_guild, bot): + ctx = MockApplicationContext(bot=bot, channel=mock_channel, guild=mock_guild) + + mock_attachment = MagicMock() + mock_attachment.filename = "test.txt" + mock_attachment.save = AsyncMock() + + mock_message = MagicMock() + mock_message.attachments = [mock_attachment] + + with patch.object(uploader_cog, '_upload_file', new_callable=AsyncMock) as mock_upload: + mock_upload.return_value = None + + with patch('os.path.exists', return_value=True): + with patch('os.remove'): + with patch.object(ctx, 'defer', new_callable=AsyncMock): + with patch.object(ctx, 'respond', new_callable=AsyncMock) as mock_respond: + await uploader_cog.message_upload.callback(uploader_cog, ctx, message=mock_message) + mock_respond.assert_called_once_with("Nie udało się wysłać żadnych plików", ephemeral=True) + + +@pytest.mark.asyncio +async def test_upload_slash_command_success(uploader_cog, mock_channel, mock_guild, bot): + ctx = MockApplicationContext(bot=bot, channel=mock_channel, guild=mock_guild) + + mock_attachment = MagicMock() + mock_attachment.filename = "test.txt" + mock_attachment.save = AsyncMock() + + with patch.object(uploader_cog, '_upload_file', new_callable=AsyncMock) as mock_upload: + mock_upload.return_value = "https://copyparty.local/uploaded.txt" + + with patch('os.path.exists', return_value=True): + with patch('os.remove'): + with patch.object(ctx, 'defer', new_callable=AsyncMock): + with patch.object(ctx, 'respond', new_callable=AsyncMock) as mock_respond: + await uploader_cog.upload.callback(uploader_cog, ctx, zalacznik=mock_attachment) + mock_respond.assert_called_once_with("https://copyparty.local/uploaded.txt") + + +@pytest.mark.asyncio +async def test_upload_slash_command_failure(uploader_cog, mock_channel, mock_guild, bot): + ctx = MockApplicationContext(bot=bot, channel=mock_channel, guild=mock_guild) + + mock_attachment = MagicMock() + mock_attachment.filename = "test.txt" + mock_attachment.save = AsyncMock() + + with patch.object(uploader_cog, '_upload_file', new_callable=AsyncMock) as mock_upload: + mock_upload.return_value = None + + with patch('os.path.exists', return_value=True): + with patch('os.remove'): + with patch.object(ctx, 'defer', new_callable=AsyncMock): + with patch.object(ctx, 'respond', new_callable=AsyncMock) as mock_respond: + await uploader_cog.upload.callback(uploader_cog, ctx, zalacznik=mock_attachment) + mock_respond.assert_called_once_with("Nie udało się wysłać pliku", ephemeral=True) \ No newline at end of file -- cgit v1.2.3