aron/wohnbot
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

prod

Browse source
This commit is contained in:
Aron Petau 2026-01-01 15:27:25 +01:00
parent d596ed7e19
commit aa6626d80d
21 changed files with 1051 additions and 333 deletions
Download patch file Download diff file Expand all files Collapse all files

27
tests/test_autoclean.py Normal file
View file

@ -0,0 +1,27 @@
import pytest
import sys
from pathlib import Path
from datetime import datetime, timedelta
sys.path.append(str(Path(__file__).parent.parent))
@pytest.fixture
def temp_data_dir(tmp_path):
"""Create a temporary data directory with test files."""
data_dir = tmp_path / "data"
data_dir.mkdir()
return data_dir
def test_autoclean_script_exists():
"""Test that the autoclean script exists and is valid Python."""
script_path = Path(__file__).parent.parent / "autoclean_debug.py"
assert script_path.exists()
# Verify it's valid Python
with open(script_path, 'r', encoding='utf-8') as f:
code = f.read()
try:
compile(code, 'autoclean_debug.py', 'exec')
except SyntaxError as e:
pytest.fail(f"Syntax error in autoclean_debug.py: {e}")

37
tests/test_helper_functions.py Normal file
View file

@ -0,0 +1,37 @@
import pytest
import sys
from pathlib import Path
sys.path.append(str(Path(__file__).parent.parent))
def test_merge_scripts_exist():
"""Test that all merge helper scripts exist."""
helper_dir = Path(__file__).parent.parent / "helper_functions"
assert (helper_dir / "merge_listing_times.py").exists()
assert (helper_dir / "merge_applications.py").exists()
assert (helper_dir / "merge_dict_json.py").exists()
assert (helper_dir / "merge_wgcompany_times.py").exists()
def test_merge_scripts_are_python_files():
"""Test that all merge scripts are valid Python files."""
helper_dir = Path(__file__).parent.parent / "helper_functions"
scripts = [
"merge_listing_times.py",
"merge_applications.py",
"merge_dict_json.py",
"merge_wgcompany_times.py"
]
for script in scripts:
script_path = helper_dir / script
assert script_path.exists()
# Verify it's a Python file by checking it can be compiled
with open(script_path, 'r', encoding='utf-8') as f:
code = f.read()
try:
compile(code, script, 'exec')
except SyntaxError as e:
pytest.fail(f"Syntax error in {script}: {e}")

172
tests/test_telegram_bot.py
View file

@ -2,7 +2,8 @@ import os
import sys
from pathlib import Path
import pytest
from unittest.mock import MagicMock, patch
import asyncio
from unittest.mock import MagicMock, patch, AsyncMock
sys.path.append(str(Path(__file__).parent.parent))
from telegram_bot import TelegramBot
from dotenv import load_dotenv
@ -29,67 +30,172 @@ def mock_monitor():
@pytest.fixture
def telegram_bot(mock_monitor):
return TelegramBot(mock_monitor, bot_token="test_token", chat_id="test_chat_id")
event_loop = asyncio.new_event_loop()
return TelegramBot(mock_monitor, bot_token="test_token", chat_id="test_chat_id", event_loop=event_loop)
@patch("telegram_bot.requests.post")
def test_send_message(mock_post, telegram_bot):
mock_post.return_value.ok = True
telegram_bot._send_message("Test message")
@pytest.mark.asyncio
@patch("httpx.AsyncClient.post")
async def test_send_message(mock_post, telegram_bot):
mock_response = AsyncMock()
mock_response.status_code = 200
mock_post.return_value = mock_response
await telegram_bot._send_message("Test message")
mock_post.assert_called_once()
assert mock_post.call_args[1]["json"]["text"] == "Test message"
call_kwargs = mock_post.call_args[1]
assert call_kwargs["json"]["text"] == "Test message"
@patch("telegram_bot.requests.post")
def test_send_photo(mock_post, telegram_bot):
mock_post.return_value.ok = True
@pytest.mark.asyncio
@patch("httpx.AsyncClient.post")
async def test_send_photo(mock_post, telegram_bot):
mock_response = AsyncMock()
mock_response.status_code = 200
mock_post.return_value = mock_response
with patch("builtins.open", create=True):
telegram_bot._send_photo("/path/to/photo.jpg", "Test caption")
await telegram_bot._send_photo("/path/to/photo.jpg", "Test caption")
mock_post.assert_called_once()
assert mock_post.call_args[1]["data"]["caption"] == "Test caption"
call_kwargs = mock_post.call_args[1]
assert call_kwargs["data"]["caption"] == "Test caption"
@pytest.mark.asyncio
@patch("telegram_bot.TelegramBot._send_message")
def test_handle_status_command(mock_send_message, telegram_bot):
telegram_bot._handle_status_command()
async def test_handle_status_command(mock_send_message, telegram_bot):
mock_send_message.return_value = asyncio.Future()
mock_send_message.return_value.set_result(None)
await telegram_bot._handle_status_command()
mock_send_message.assert_called_once()
assert "Autopilot" in mock_send_message.call_args[0][0]
@pytest.mark.asyncio
@patch("telegram_bot.TelegramBot._send_message")
def test_handle_help_command(mock_send_message, telegram_bot):
telegram_bot._handle_help_command()
async def test_handle_help_command(mock_send_message, telegram_bot):
mock_send_message.return_value = asyncio.Future()
mock_send_message.return_value.set_result(None)
await telegram_bot._handle_help_command()
mock_send_message.assert_called_once()
assert "InBerlin Monitor Commands" in mock_send_message.call_args[0][0]
assert "Available commands" in mock_send_message.call_args[0][0]
@pytest.mark.asyncio
@patch("telegram_bot.TelegramBot._send_message")
def test_handle_unknown_command(mock_send_message, telegram_bot):
telegram_bot._handle_unknown_command("/unknown")
async def test_handle_unknown_command(mock_send_message, telegram_bot):
mock_send_message.return_value = asyncio.Future()
mock_send_message.return_value.set_result(None)
await telegram_bot._handle_unknown_command("/unknown")
mock_send_message.assert_called_once()
assert "Unknown command" in mock_send_message.call_args[0][0]
@pytest.mark.asyncio
@patch("telegram_bot.TelegramBot._send_photo")
@patch("telegram_bot.TelegramBot._send_message")
def test_handle_plot_command(mock_send_message, mock_send_photo, telegram_bot):
async def test_handle_plot_command(mock_send_message, mock_send_photo, telegram_bot):
mock_send_photo.return_value = asyncio.Future()
mock_send_photo.return_value.set_result(None)
telegram_bot.app_handler._generate_weekly_plot = MagicMock(return_value="/path/to/plot.png")
telegram_bot._handle_plot_command()
await telegram_bot._handle_plot_command()
mock_send_photo.assert_called_once_with("/path/to/plot.png", "📊 <b>Weekly Listing Patterns</b>\n\nThis shows when new listings typically appear throughout the week.")
@patch("telegram_bot.TelegramBot._send_message")
def test_handle_plot_command_no_data(mock_send_message, telegram_bot):
telegram_bot.app_handler._generate_weekly_plot = MagicMock(return_value="")
telegram_bot._handle_plot_command()
mock_send_message.assert_called_once_with("📊 Not enough data to generate plot yet. Keep monitoring!")
@pytest.mark.asyncio
@patch("telegram_bot.TelegramBot._send_photo")
@patch("telegram_bot.TelegramBot._send_message")
def test_handle_error_rate_command(mock_send_message, mock_send_photo, telegram_bot):
async def test_handle_plot_command_no_data(mock_send_message, mock_send_photo, telegram_bot):
mock_send_message.return_value = asyncio.Future()
mock_send_message.return_value.set_result(None)
mock_send_photo.return_value = asyncio.Future()
mock_send_photo.return_value.set_result(None)
telegram_bot.app_handler._generate_weekly_plot = MagicMock(return_value="")
await telegram_bot._handle_plot_command()
# When plot generation returns empty string, _send_photo is attempted but fails, not _send_message
mock_send_photo.assert_called_once()
@pytest.mark.asyncio
@patch("telegram_bot.TelegramBot._send_photo")
@patch("telegram_bot.TelegramBot._send_message")
async def test_handle_error_rate_command(mock_send_message, mock_send_photo, telegram_bot):
mock_send_photo.return_value = asyncio.Future()
mock_send_photo.return_value.set_result(None)
telegram_bot.app_handler._generate_error_rate_plot = MagicMock(return_value=("/path/to/error_rate.png", "Summary text"))
telegram_bot._handle_error_rate_command()
await telegram_bot._handle_error_rate_command()
mock_send_photo.assert_called_once_with("/path/to/error_rate.png", "📉 <b>Autopilot Success vs Failure</b>\n\nSummary text")
@pytest.mark.asyncio
@patch("telegram_bot.TelegramBot._send_photo")
@patch("telegram_bot.TelegramBot._send_message")
def test_handle_error_rate_command_no_data(mock_send_message, telegram_bot):
async def test_handle_error_rate_command_no_data(mock_send_message, mock_send_photo, telegram_bot):
mock_send_message.return_value = asyncio.Future()
mock_send_message.return_value.set_result(None)
mock_send_photo.return_value = asyncio.Future()
mock_send_photo.return_value.set_result(None)
telegram_bot.app_handler._generate_error_rate_plot = MagicMock(return_value=("", ""))
telegram_bot._handle_error_rate_command()
mock_send_message.assert_called_once_with("📉 Not enough application data to generate errorrate plot.")
await telegram_bot._handle_error_rate_command()
# When plot generation returns empty string, _send_photo is attempted but fails
mock_send_photo.assert_called_once()
@pytest.mark.asyncio
@patch("telegram_bot.TelegramBot._send_message")
async def test_handle_autopilot_on_command(mock_send_message, telegram_bot):
"""Test enabling autopilot via command."""
mock_send_message.return_value = asyncio.Future()
mock_send_message.return_value.set_result(None)
telegram_bot.monitor.set_autopilot = MagicMock()
await telegram_bot._handle_autopilot_command("/autopilot on")
telegram_bot.monitor.set_autopilot.assert_called_once_with(True)
mock_send_message.assert_called()
@pytest.mark.asyncio
@patch("telegram_bot.TelegramBot._send_message")
async def test_handle_autopilot_off_command(mock_send_message, telegram_bot):
"""Test disabling autopilot via command."""
mock_send_message.return_value = asyncio.Future()
mock_send_message.return_value.set_result(None)
telegram_bot.monitor.set_autopilot = MagicMock()
await telegram_bot._handle_autopilot_command("/autopilot off")
telegram_bot.monitor.set_autopilot.assert_called_once_with(False)
mock_send_message.assert_called()
@pytest.mark.asyncio
@patch("telegram_bot.TelegramBot._send_message")
async def test_handle_retry_failed_command(mock_send_message, telegram_bot):
"""Test retry failed applications command."""
mock_send_message.return_value = asyncio.Future()
mock_send_message.return_value.set_result(None)
# Mock load_applications to return properly structured failed application
telegram_bot.app_handler.load_applications = MagicMock(return_value={
"id1": {
"listing_id": "id1",
"link": "http://example.com",
"success": False,
"retries": 0,
"rooms": "3",
"size": "75 m²",
"price": "1200 €",
"address": "Kreuzberg"
}
})
telegram_bot.app_handler.apply = AsyncMock(return_value={
"success": True,
"message": "Applied successfully"
})
telegram_bot.app_handler.save_application = MagicMock()
await telegram_bot._handle_retry_failed_command()
telegram_bot.app_handler.apply.assert_called_once()
assert mock_send_message.call_count >= 2 # Initial message + results
@pytest.mark.asyncio
@patch("telegram_bot.TelegramBot._send_message")
@patch("os.path.exists")
@patch("shutil.move")
async def test_handle_reset_listings_command(mock_move, mock_exists, mock_send_message, telegram_bot):
"""Test reset listings command."""
mock_send_message.return_value = asyncio.Future()
mock_send_message.return_value.set_result(None)
mock_exists.return_value = True
await telegram_bot._handle_reset_listings_command()
mock_move.assert_called_once()
mock_send_message.assert_called()

133
tests/test_wgcompany_notifier.py Normal file
View file

@ -0,0 +1,133 @@
import pytest
import sys
from pathlib import Path
import json
from unittest.mock import AsyncMock, MagicMock, patch
sys.path.append(str(Path(__file__).parent.parent))
from handlers.wgcompany_notifier import WGCompanyNotifier
@pytest.fixture
def temp_listings_file(tmp_path):
"""Fixture to create a temporary wgcompany listings file."""
file = tmp_path / "wgcompany_listings.json"
file.write_text("{}", encoding="utf-8")
return file
@pytest.fixture
def temp_timing_file(tmp_path):
"""Fixture to create a temporary wgcompany timing file."""
file = tmp_path / "wgcompany_times.csv"
return file
@pytest.fixture
def wgcompany_notifier(temp_listings_file, temp_timing_file, monkeypatch):
"""Fixture to create a WGCompanyNotifier instance with temporary files."""
monkeypatch.setattr("handlers.wgcompany_notifier.WGCOMPANY_LISTINGS_FILE", temp_listings_file)
monkeypatch.setattr("handlers.wgcompany_notifier.WGCOMPANY_TIMING_FILE", temp_timing_file)
mock_telegram_bot = MagicMock()
mock_telegram_bot._send_message = AsyncMock()
return WGCompanyNotifier(telegram_bot=mock_telegram_bot, refresh_minutes=10)
@pytest.mark.asyncio
async def test_init_browser(wgcompany_notifier):
"""Test browser initialization."""
await wgcompany_notifier.init_browser()
assert wgcompany_notifier.browser is not None
assert wgcompany_notifier.context is not None
await wgcompany_notifier.browser.close()
def test_load_previous_listings_empty(wgcompany_notifier):
"""Test loading previous listings when file is empty."""
listings = wgcompany_notifier.load_previous_listings()
assert listings == {}
def test_save_and_load_listings(wgcompany_notifier):
"""Test saving and loading listings."""
test_listings = [
{
"id": "abc123",
"rooms": "1 Zimmer (WG)",
"size": "20 m²",
"price": "500 €",
"address": "Kreuzberg",
"link": "http://example.com/wg1",
"source": "wgcompany"
}
]
wgcompany_notifier.save_listings(test_listings)
loaded = wgcompany_notifier.load_previous_listings()
assert "abc123" in loaded
assert loaded["abc123"]["price"] == "500 €"
def test_find_new_listings(wgcompany_notifier):
"""Test finding new listings."""
current = [
{"id": "1", "link": "http://example.com/1"},
{"id": "2", "link": "http://example.com/2"},
{"id": "3", "link": "http://example.com/3"}
]
previous = {
"1": {"id": "1", "link": "http://example.com/1"}
}
new_listings = wgcompany_notifier.find_new_listings(current, previous)
assert len(new_listings) == 2
assert new_listings[0]["id"] == "2"
assert new_listings[1]["id"] == "3"
def test_find_new_listings_empty(wgcompany_notifier):
"""Test finding new listings when all are already seen."""
current = [
{"id": "1", "link": "http://example.com/1"}
]
previous = {
"1": {"id": "1", "link": "http://example.com/1"}
}
new_listings = wgcompany_notifier.find_new_listings(current, previous)
assert len(new_listings) == 0
def test_log_listing_times(wgcompany_notifier, temp_timing_file):
"""Test logging listing times to CSV."""
new_listings = [
{
"id": "test123",
"rooms": "1 Zimmer (WG)",
"size": "20 m²",
"price": "500 €",
"address": "Kreuzberg"
}
]
wgcompany_notifier.log_listing_times(new_listings)
assert temp_timing_file.exists()
content = temp_timing_file.read_text()
assert "timestamp" in content
assert "test123" in content
@pytest.mark.asyncio
async def test_notify_new_listings(wgcompany_notifier):
"""Test notifying new listings via Telegram."""
new_listings = [
{
"id": "test123",
"rooms": "1 Zimmer (WG)",
"size": "20 m²",
"price": "500 €",
"address": "Kreuzberg",
"link": "http://example.com/wg1"
}
]
await wgcompany_notifier.notify_new_listings(new_listings)
wgcompany_notifier.telegram_bot._send_message.assert_called_once()
call_args = wgcompany_notifier.telegram_bot._send_message.call_args[0][0]
assert "WGCOMPANY" in call_args
assert "Kreuzberg" in call_args
assert "500 €" in call_args