roughly working again, now dev docker exists
This commit is contained in:
parent
a77a0c0393
commit
155ab39368
26 changed files with 1976 additions and 235 deletions
87
tests/test_application_handler.py
Normal file
87
tests/test_application_handler.py
Normal file
|
|
@ -0,0 +1,87 @@
|
|||
import pytest
|
||||
import json
|
||||
from pathlib import Path
|
||||
import sys
|
||||
from pathlib import Path as _Path
|
||||
sys.path.append(str(_Path(__file__).parent.parent))
|
||||
from application_handler import ApplicationHandler
|
||||
|
||||
@pytest.fixture
|
||||
def temp_applications_file(tmp_path):
|
||||
"""Fixture to create a temporary applications file."""
|
||||
file = tmp_path / "applications.json"
|
||||
file.write_text("{}", encoding="utf-8")
|
||||
return file
|
||||
|
||||
@pytest.fixture
|
||||
def application_handler(temp_applications_file, monkeypatch):
|
||||
"""Fixture to create an ApplicationHandler instance with a temporary applications file."""
|
||||
monkeypatch.setattr("application_handler.APPLICATIONS_FILE", temp_applications_file)
|
||||
return ApplicationHandler(browser_context=None, state_manager=None)
|
||||
|
||||
|
||||
def test_detect_company_domains():
|
||||
handler = ApplicationHandler(browser_context=None, state_manager=None)
|
||||
assert handler._detect_company('https://howoge.de/abc') == 'howoge'
|
||||
assert handler._detect_company('https://www.howoge.de/abc') == 'howoge'
|
||||
assert handler._detect_company('https://portal.gewobag.de/') == 'gewobag'
|
||||
assert handler._detect_company('https://degewo.de/') == 'degewo'
|
||||
assert handler._detect_company('https://gesobau.de/') == 'gesobau'
|
||||
assert handler._detect_company('https://stadtundland.de/') == 'stadtundland'
|
||||
assert handler._detect_company('https://stadt-und-land.de/') == 'stadtundland'
|
||||
assert handler._detect_company('https://wbm.de/') == 'wbm'
|
||||
|
||||
def test_detect_company_path_fallback():
|
||||
handler = ApplicationHandler(browser_context=None, state_manager=None)
|
||||
assert handler._detect_company('https://example.com/howoge/abc') == 'howoge'
|
||||
assert handler._detect_company('https://foo.bar/gewobag') == 'gewobag'
|
||||
assert handler._detect_company('https://foo.bar/degewo') == 'degewo'
|
||||
assert handler._detect_company('https://foo.bar/gesobau') == 'gesobau'
|
||||
assert handler._detect_company('https://foo.bar/stadt-und-land') == 'stadtundland'
|
||||
assert handler._detect_company('https://foo.bar/wbm') == 'wbm'
|
||||
|
||||
def test_detect_company_unknown():
|
||||
handler = ApplicationHandler(browser_context=None, state_manager=None)
|
||||
assert handler._detect_company('https://example.com/') == 'unknown'
|
||||
assert handler._detect_company('') == 'unknown'
|
||||
assert handler._detect_company(None) == 'unknown'
|
||||
|
||||
def test_load_applications_empty(application_handler):
|
||||
"""Test loading applications when the file is empty."""
|
||||
applications = application_handler.load_applications()
|
||||
assert applications == {}
|
||||
|
||||
def test_save_application(application_handler):
|
||||
"""Test saving an application."""
|
||||
result = {
|
||||
"listing_id": "12345",
|
||||
"company": "test_company",
|
||||
"link": "http://example.com",
|
||||
"timestamp": "2025-12-27T12:00:00",
|
||||
"success": True,
|
||||
"message": "Application successful",
|
||||
"address": "Test Address",
|
||||
"rooms": "3",
|
||||
"price": "1000"
|
||||
}
|
||||
application_handler.save_application(result)
|
||||
applications = application_handler.load_applications()
|
||||
assert "12345" in applications
|
||||
assert applications["12345"] == result
|
||||
|
||||
def test_has_applied(application_handler):
|
||||
"""Test checking if an application exists."""
|
||||
result = {
|
||||
"listing_id": "12345",
|
||||
"company": "test_company",
|
||||
"link": "http://example.com",
|
||||
"timestamp": "2025-12-27T12:00:00",
|
||||
"success": True,
|
||||
"message": "Application successful",
|
||||
"address": "Test Address",
|
||||
"rooms": "3",
|
||||
"price": "1000"
|
||||
}
|
||||
application_handler.save_application(result)
|
||||
assert application_handler.has_applied("12345") is True
|
||||
assert application_handler.has_applied("67890") is False
|
||||
44
tests/test_company_detection.py
Normal file
44
tests/test_company_detection.py
Normal file
|
|
@ -0,0 +1,44 @@
|
|||
import pytest
|
||||
import sys
|
||||
from pathlib import Path as _Path
|
||||
sys.path.append(str(_Path(__file__).parent.parent))
|
||||
from application_handler import ApplicationHandler
|
||||
|
||||
class DummyStateManager:
|
||||
email = None
|
||||
password = None
|
||||
logged_in = False
|
||||
def set_autopilot(self, enabled): pass
|
||||
def is_autopilot_enabled(self): return False
|
||||
|
||||
def make_handler():
|
||||
# context is not used for _detect_company
|
||||
return ApplicationHandler(browser_context=None, state_manager=DummyStateManager())
|
||||
|
||||
def test_detect_company_domains():
|
||||
handler = make_handler()
|
||||
# Domain and subdomain cases
|
||||
assert handler._detect_company('https://howoge.de/abc') == 'howoge'
|
||||
assert handler._detect_company('https://www.howoge.de/abc') == 'howoge'
|
||||
assert handler._detect_company('https://portal.gewobag.de/') == 'gewobag'
|
||||
assert handler._detect_company('https://degewo.de/') == 'degewo'
|
||||
assert handler._detect_company('https://gesobau.de/') == 'gesobau'
|
||||
assert handler._detect_company('https://stadtundland.de/') == 'stadtundland'
|
||||
assert handler._detect_company('https://stadt-und-land.de/') == 'stadtundland'
|
||||
assert handler._detect_company('https://wbm.de/') == 'wbm'
|
||||
|
||||
def test_detect_company_path_fallback():
|
||||
handler = make_handler()
|
||||
# Path/query fallback
|
||||
assert handler._detect_company('https://example.com/howoge/abc') == 'howoge'
|
||||
assert handler._detect_company('https://foo.bar/gewobag') == 'gewobag'
|
||||
assert handler._detect_company('https://foo.bar/degewo') == 'degewo'
|
||||
assert handler._detect_company('https://foo.bar/gesobau') == 'gesobau'
|
||||
assert handler._detect_company('https://foo.bar/stadt-und-land') == 'stadtundland'
|
||||
assert handler._detect_company('https://foo.bar/wbm') == 'wbm'
|
||||
|
||||
def test_detect_company_unknown():
|
||||
handler = make_handler()
|
||||
assert handler._detect_company('https://example.com/') == 'unknown'
|
||||
assert handler._detect_company('') == 'unknown'
|
||||
assert handler._detect_company(None) == 'unknown'
|
||||
|
|
@ -1,35 +1,48 @@
|
|||
|
||||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
import pytest
|
||||
from unittest.mock import patch, mock_open
|
||||
from archive.test_errorrate_runner import generate_error_rate_plot
|
||||
from unittest.mock import patch, mock_open, MagicMock
|
||||
sys.path.append(str(Path(__file__).parent.parent))
|
||||
from application_handler import ApplicationHandler
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def mock_data_dir(tmp_path):
|
||||
"""Fixture to create a temporary data directory."""
|
||||
def temp_applications_file(tmp_path):
|
||||
data_dir = tmp_path / "data"
|
||||
data_dir.mkdir()
|
||||
return data_dir
|
||||
file = data_dir / "applications.json"
|
||||
file.write_text("{}", encoding="utf-8")
|
||||
return file
|
||||
|
||||
@patch("builtins.open", new_callable=mock_open, read_data="{}")
|
||||
@patch("os.path.exists", return_value=True)
|
||||
def test_generate_error_rate_plot_no_data(mock_exists, mock_open, mock_data_dir):
|
||||
"""Test generate_error_rate_plot with no data."""
|
||||
plot_path, summary = generate_error_rate_plot(str(mock_data_dir / "applications.json"))
|
||||
assert plot_path is None
|
||||
class DummyStateManager:
|
||||
email = None
|
||||
password = None
|
||||
logged_in = False
|
||||
def set_autopilot(self, enabled): pass
|
||||
def is_autopilot_enabled(self): return False
|
||||
|
||||
|
||||
@patch("matplotlib.pyplot.savefig")
|
||||
def test_generate_error_rate_plot_no_data(mock_savefig, temp_applications_file):
|
||||
handler = ApplicationHandler(None, DummyStateManager(), applications_file=temp_applications_file)
|
||||
plot_path, summary = handler._generate_error_rate_plot()
|
||||
assert plot_path is None or plot_path == ""
|
||||
assert summary == ""
|
||||
|
||||
@patch("builtins.open", new_callable=mock_open)
|
||||
@patch("os.path.exists", return_value=True)
|
||||
|
||||
@patch("matplotlib.pyplot.savefig")
|
||||
def test_generate_error_rate_plot_with_data(mock_savefig, mock_exists, mock_open, mock_data_dir):
|
||||
"""Test generate_error_rate_plot with valid data."""
|
||||
mock_open.return_value.read.return_value = """
|
||||
def test_generate_error_rate_plot_with_data(mock_savefig, temp_applications_file):
|
||||
handler = ApplicationHandler(None, DummyStateManager(), applications_file=temp_applications_file)
|
||||
# Write valid data to the temp applications file
|
||||
temp_applications_file.write_text('''
|
||||
{
|
||||
"1": {"timestamp": "2025-12-25T12:00:00", "company": "CompanyA", "success": true},
|
||||
"2": {"timestamp": "2025-12-26T12:00:00", "company": "CompanyB", "success": false}
|
||||
}
|
||||
"""
|
||||
plot_path, summary = generate_error_rate_plot(str(mock_data_dir / "applications.json"))
|
||||
''', encoding="utf-8")
|
||||
plot_path, summary = handler._generate_error_rate_plot()
|
||||
assert plot_path is not None
|
||||
assert "Total attempts" in summary
|
||||
assert "Successes" in summary
|
||||
|
|
|
|||
|
|
@ -5,7 +5,12 @@ from handlers.degewo_handler import DegewoHandler
|
|||
from handlers.gesobau_handler import GesobauHandler
|
||||
from handlers.stadtundland_handler import StadtUndLandHandler
|
||||
from handlers.wbm_handler import WBMHandler
|
||||
from unittest.mock import AsyncMock
|
||||
from handlers.base_handler import BaseHandler
|
||||
from unittest.mock import AsyncMock, MagicMock
|
||||
|
||||
class MockBaseHandler(BaseHandler):
|
||||
async def apply(self, listing: dict, result: dict) -> dict:
|
||||
return result
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_howoge_handler():
|
||||
|
|
@ -59,4 +64,76 @@ async def test_wbm_handler():
|
|||
listing = {"link": "https://www.wbm.de/example"}
|
||||
result = {"success": False}
|
||||
await handler.apply(listing, result)
|
||||
assert "success" in result
|
||||
assert "success" in result
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_cookies():
|
||||
"""Test the handle_cookies method in BaseHandler."""
|
||||
context = AsyncMock()
|
||||
handler = MockBaseHandler(context)
|
||||
mock_page = AsyncMock()
|
||||
mock_cookie_btn = AsyncMock()
|
||||
mock_cookie_btn.is_visible = AsyncMock(return_value=True)
|
||||
mock_cookie_btn.click = AsyncMock()
|
||||
mock_page.query_selector = AsyncMock(return_value=mock_cookie_btn)
|
||||
|
||||
await handler.handle_cookies(mock_page)
|
||||
mock_cookie_btn.click.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_handle_consent():
|
||||
"""Test the handle_consent method in BaseHandler."""
|
||||
context = AsyncMock()
|
||||
handler = MockBaseHandler(context)
|
||||
mock_page = AsyncMock()
|
||||
mock_consent_btn = AsyncMock()
|
||||
mock_consent_btn.is_visible = AsyncMock(return_value=True)
|
||||
mock_consent_btn.click = AsyncMock()
|
||||
mock_page.query_selector = AsyncMock(return_value=mock_consent_btn)
|
||||
|
||||
await handler.handle_consent(mock_page)
|
||||
mock_consent_btn.click.assert_called_once()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_login():
|
||||
"""Test the login method in BaseHandler."""
|
||||
context = AsyncMock()
|
||||
handler = MockBaseHandler(context, email="test@example.com", password="password123")
|
||||
mock_page = AsyncMock()
|
||||
|
||||
# Mock the page interactions
|
||||
mock_page.goto = AsyncMock()
|
||||
mock_page.fill = AsyncMock()
|
||||
mock_page.click = AsyncMock()
|
||||
mock_page.wait_for_load_state = AsyncMock()
|
||||
mock_page.url = "https://www.inberlinwohnen.de/mein-bereich"
|
||||
mock_page.query_selector = AsyncMock(return_value=AsyncMock(is_visible=AsyncMock(return_value=True)))
|
||||
|
||||
result = await handler.login(mock_page)
|
||||
|
||||
# Assertions
|
||||
mock_page.goto.assert_called_once_with("https://www.inberlinwohnen.de/login", wait_until="networkidle")
|
||||
mock_page.fill.assert_any_call('input[name="email"], input[type="email"]', "test@example.com")
|
||||
mock_page.fill.assert_any_call('input[name="password"], input[type="password"]', "password123")
|
||||
mock_page.click.assert_called_once_with('button[type="submit"], input[type="submit"]')
|
||||
mock_page.wait_for_load_state.assert_called_once_with("networkidle")
|
||||
assert result is True
|
||||
|
||||
# Test for fetch_listings method in BaseHandler
|
||||
@pytest.mark.asyncio
|
||||
async def test_fetch_listings():
|
||||
context = AsyncMock()
|
||||
handler = MockBaseHandler(context)
|
||||
|
||||
# Mock the fetch_listings method
|
||||
handler.fetch_listings = AsyncMock(return_value=[
|
||||
{"id": "1", "title": "Listing 1", "price": 1000},
|
||||
{"id": "2", "title": "Listing 2", "price": 1200}
|
||||
])
|
||||
|
||||
listings = await handler.fetch_listings()
|
||||
|
||||
# Assertions
|
||||
assert len(listings) == 2
|
||||
assert listings[0]["id"] == "1"
|
||||
assert listings[1]["title"] == "Listing 2"
|
||||
42
tests/test_playwright_login_flow.py
Normal file
42
tests/test_playwright_login_flow.py
Normal file
|
|
@ -0,0 +1,42 @@
|
|||
|
||||
import asyncio
|
||||
import pytest
|
||||
from playwright.async_api import async_playwright
|
||||
|
||||
USER_AGENTS = [
|
||||
# Chrome on Mac
|
||||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
||||
# Chrome on Windows
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36",
|
||||
# Firefox on Mac
|
||||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10.15; rv:120.0) Gecko/20100101 Firefox/120.0",
|
||||
# Edge on Windows
|
||||
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36 Edg/120.0.0.0",
|
||||
# Safari on Mac
|
||||
"Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Safari/605.1.15",
|
||||
# iPhone Safari
|
||||
"Mozilla/5.0 (iPhone; CPU iPhone OS 17_2 like Mac OS X) AppleWebKit/605.1.15 (KHTML, like Gecko) Version/17.0 Mobile/15E148 Safari/604.1",
|
||||
]
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_inberlin_login_flow():
|
||||
async with async_playwright() as p:
|
||||
for ua in USER_AGENTS:
|
||||
print("\n==============================")
|
||||
print(f"Testing user agent: {ua}")
|
||||
browser = await p.chromium.launch(headless=True)
|
||||
context = await browser.new_context(user_agent=ua)
|
||||
page = await context.new_page()
|
||||
try:
|
||||
print("Navigating to login page...")
|
||||
login_response = await page.goto("https://www.inberlinwohnen.de/login", wait_until="networkidle")
|
||||
print(f"Login page status: {login_response.status if login_response else 'No response'}")
|
||||
print(f"Login page headers: {login_response.headers if login_response else 'No response'}")
|
||||
await asyncio.sleep(2)
|
||||
except Exception as e:
|
||||
print(f"Exception for user agent: {ua}\n{e}")
|
||||
finally:
|
||||
await browser.close()
|
||||
|
||||
if __name__ == "__main__":
|
||||
asyncio.run(test_inberlin_login_flow())
|
||||
29
tests/test_state_manager.py
Normal file
29
tests/test_state_manager.py
Normal file
|
|
@ -0,0 +1,29 @@
|
|||
import pytest
|
||||
from pathlib import Path
|
||||
from state_manager import StateManager
|
||||
import json
|
||||
|
||||
@pytest.fixture
|
||||
def state_file(tmp_path):
|
||||
return tmp_path / "state.json"
|
||||
|
||||
@pytest.fixture
|
||||
def state_manager(state_file):
|
||||
return StateManager(state_file)
|
||||
|
||||
def test_load_state_default(state_manager):
|
||||
state = state_manager.load_state()
|
||||
assert state == {"autopilot": False}
|
||||
|
||||
def test_save_state(state_manager):
|
||||
state = {"autopilot": True}
|
||||
state_manager.save_state(state)
|
||||
loaded_state = state_manager.load_state()
|
||||
assert loaded_state == state
|
||||
|
||||
def test_set_autopilot(state_manager):
|
||||
state_manager.set_autopilot(True)
|
||||
assert state_manager.is_autopilot_enabled() is True
|
||||
|
||||
state_manager.set_autopilot(False)
|
||||
assert state_manager.is_autopilot_enabled() is False
|
||||
|
|
@ -1,6 +1,9 @@
|
|||
import os
|
||||
import sys
|
||||
from pathlib import Path
|
||||
import pytest
|
||||
from unittest.mock import MagicMock, patch
|
||||
sys.path.append(str(Path(__file__).parent.parent))
|
||||
from telegram_bot import TelegramBot
|
||||
from dotenv import load_dotenv
|
||||
|
||||
|
|
@ -59,4 +62,34 @@ def test_handle_help_command(mock_send_message, telegram_bot):
|
|||
def test_handle_unknown_command(mock_send_message, telegram_bot):
|
||||
telegram_bot._handle_unknown_command("/unknown")
|
||||
mock_send_message.assert_called_once()
|
||||
assert "Unknown command" in mock_send_message.call_args[0][0]
|
||||
assert "Unknown command" in mock_send_message.call_args[0][0]
|
||||
|
||||
|
||||
@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):
|
||||
telegram_bot.app_handler._generate_weekly_plot = MagicMock(return_value="/path/to/plot.png")
|
||||
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!")
|
||||
|
||||
|
||||
@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):
|
||||
telegram_bot.app_handler._generate_error_rate_plot = MagicMock(return_value=("/path/to/error_rate.png", "Summary text"))
|
||||
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")
|
||||
|
||||
|
||||
@patch("telegram_bot.TelegramBot._send_message")
|
||||
def test_handle_error_rate_command_no_data(mock_send_message, telegram_bot):
|
||||
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.")
|
||||
Loading…
Add table
Add a link
Reference in a new issue