192 lines
9.1 KiB
Python
192 lines
9.1 KiB
Python
import calendar
|
|
import json
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, patch
|
|
|
|
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "stempelbot"))
|
|
|
|
from playwright.sync_api import sync_playwright
|
|
import smarttime_browser as browser_jobs
|
|
import smarttime_ui as ui
|
|
|
|
|
|
# Minimal, anonymized markup matching the observed SmartTime 9 dashboard.
|
|
DASHBOARD = """
|
|
<div class="shadow-xs"><label>Test User</label>
|
|
<vaadin-icon icon="fas:phone" style="fill:green"></vaadin-icon>
|
|
<vaadin-icon icon="far:building" style="fill:COLOR"></vaadin-icon>
|
|
<vaadin-vertical-layout><label>Letzte Buchung</label><label>09:29</label></vaadin-vertical-layout>
|
|
</div>
|
|
<vaadin-vertical-layout><label>Anwesend</label><div style="background-color:green"></div></vaadin-vertical-layout>
|
|
<vaadin-vertical-layout><label>Außer Haus</label><div style="background-color:rgb(31, 161, 220)"></div></vaadin-vertical-layout>
|
|
<vaadin-vertical-layout><label>Abwesend</label><div style="background-color:red"></div></vaadin-vertical-layout>
|
|
<vaadin-vertical-layout><label>Entschuldigt</label><div style="background-color:orange"></div></vaadin-vertical-layout>
|
|
"""
|
|
HEADERS = ["Datum", "", "Kommen / Gehen", "Soll", "Ist", "TSaldo", "Saldo", "Pause", "Abwesenheit"]
|
|
|
|
|
|
class DashboardDOMTests(unittest.TestCase):
|
|
@classmethod
|
|
def setUpClass(cls):
|
|
cls.pw = sync_playwright().start()
|
|
cls.browser = cls.pw.chromium.launch(headless=True)
|
|
|
|
@classmethod
|
|
def tearDownClass(cls):
|
|
cls.browser.close()
|
|
cls.pw.stop()
|
|
|
|
def setUp(self):
|
|
self.page = self.browser.new_page()
|
|
self.addCleanup(self.page.close)
|
|
|
|
def test_status_comes_from_personal_card_not_legend_or_phone(self):
|
|
for color, status in (("green", "Anwesend"), ("red", "Abwesend"),
|
|
("rgb(31, 161, 220)", "Außer Haus"), ("orange", "Entschuldigt")):
|
|
with self.subTest(status=status):
|
|
self.page.set_content(DASHBOARD.replace("COLOR", color))
|
|
result = ui.read_dashboard(self.page)
|
|
self.assertEqual(result["last_stamp"], "09:29")
|
|
self.assertEqual(result["status_text"], status)
|
|
|
|
def test_unknown_status_is_not_assumed_present(self):
|
|
self.page.set_content(DASHBOARD.replace("COLOR", "purple"))
|
|
with self.assertRaisesRegex(ui.SmartTimeReadError, "complete personal dashboard"):
|
|
ui.read_dashboard(self.page, timeout=0)
|
|
|
|
def test_waits_for_delayed_dashboard_values(self):
|
|
self.page.set_content(DASHBOARD.replace("COLOR", "green").replace("09:29", ""))
|
|
self.page.evaluate("""() => setTimeout(() => {
|
|
document.querySelector('.shadow-xs vaadin-vertical-layout label:last-child').textContent='09:29';
|
|
}, 100)""")
|
|
self.assertEqual(ui.read_dashboard(self.page)["last_stamp"], "09:29")
|
|
|
|
def test_invalid_time_is_not_accepted(self):
|
|
self.page.set_content(DASHBOARD.replace("COLOR", "green").replace("09:29", "29:90"))
|
|
with self.assertRaises(ui.SmartTimeReadError):
|
|
ui.read_dashboard(self.page, timeout=0)
|
|
|
|
def test_empty_grid_can_finish_rendering(self):
|
|
self.page.set_content('<vaadin-grid class="Customers-grid"></vaadin-grid>')
|
|
self.page.evaluate("() => document.querySelector('vaadin-grid').attachShadow({mode:'open'})")
|
|
self.assertIsNone(self.page.evaluate(ui.GRID_JS))
|
|
|
|
def test_virtual_grid_reads_slots_and_preserves_empty_columns(self):
|
|
self.page.set_content('<vaadin-grid class="Customers-grid"></vaadin-grid>')
|
|
values = [HEADERS, ["25", "", "09:29", "08:00", "00:00", "-08:00", "22:17", "00:00", ""]]
|
|
self.page.evaluate("""values => {
|
|
const g = document.querySelector('vaadin-grid'); g.size = 30;
|
|
const root = g.attachShadow({mode:'open'});
|
|
const table = document.createElement('div'); table.id='table'; root.append(table);
|
|
values.forEach((cells, i) => {
|
|
const section = document.createElement('div'); section.id = i ? 'items' : 'header';
|
|
const row = document.createElement('tr'); section.append(row); table.append(section);
|
|
cells.forEach((text, j) => {
|
|
const slot = document.createElement('slot'); slot.name=`cell-${i}-${j}`; row.append(slot);
|
|
const content = document.createElement('span'); content.slot=slot.name; content.innerText=text; g.append(content);
|
|
});
|
|
});
|
|
}""", values)
|
|
result = self.page.evaluate(ui.GRID_JS)
|
|
self.assertEqual(result["headers"], HEADERS)
|
|
self.assertEqual(result["rows"], [values[1]])
|
|
|
|
|
|
class MonthExportTests(unittest.TestCase):
|
|
def rows(self, year=2026, month=9):
|
|
return [[f"{d:02d}", "", "", "08:00", "00:00", "-08:00", "", "00:00", ""]
|
|
for d in range(1, calendar.monthrange(year, month)[1] + 1)]
|
|
|
|
def test_preserves_all_stamps_totals_absence_and_open_pair(self):
|
|
rows = self.rows()
|
|
rows[0][2] = "09:00\n12:00\n12:30\n17:45"
|
|
rows[0][4] = "08:15"
|
|
rows[1][8] = "Urlaub"
|
|
rows[24][2] = "09:29"
|
|
result = ui.normalize_month(2026, 9, HEADERS, list(reversed(rows)))
|
|
self.assertEqual(len(result["days"]), 30)
|
|
self.assertEqual(result["days"][0]["stamps"], ["09:00", "12:00", "12:30", "17:45"])
|
|
self.assertEqual(result["days"][0]["actual"], "08:15")
|
|
self.assertEqual(result["days"][1]["absence"], "Urlaub")
|
|
self.assertTrue(result["days"][24]["incomplete_pair"])
|
|
self.assertEqual(result["days"][24]["raw_cells"], rows[24])
|
|
|
|
def test_missing_duplicate_or_changed_columns_fail_closed(self):
|
|
rows = self.rows()
|
|
for headers, values in ((HEADERS, rows[:-1]), (HEADERS, rows[:-1] + [rows[0]]),
|
|
(HEADERS[::-1], rows)):
|
|
with self.subTest(headers=headers, count=len(values)):
|
|
with self.assertRaises(ui.SmartTimeReadError):
|
|
ui.normalize_month(2026, 9, headers, values)
|
|
|
|
def test_leap_month_has_all_29_dates(self):
|
|
self.assertEqual(len(ui.normalize_month(2024, 2, HEADERS, self.rows(2024, 2))["days"]), 29)
|
|
|
|
|
|
class BrowserRecoveryTests(unittest.TestCase):
|
|
def process(self, response=None, *, error=None):
|
|
process = Mock()
|
|
process.returncode = 0
|
|
process.poll.return_value = 0
|
|
if error:
|
|
process.communicate.side_effect = error
|
|
else:
|
|
process.communicate.return_value = (json.dumps(response), "")
|
|
return process
|
|
|
|
def test_hang_starts_fresh_browser_and_recovers(self):
|
|
hung = self.process(error=subprocess.TimeoutExpired("browser", 75))
|
|
good = self.process({"result": {"last_stamp": "09:29"}})
|
|
with (patch.object(browser_jobs.subprocess, "Popen", side_effect=[hung, good]) as spawn,
|
|
patch.object(browser_jobs, "stop_process_tree") as cleanup,
|
|
patch.object(browser_jobs.time, "sleep")):
|
|
self.assertEqual(browser_jobs.run_browser_job(), {"last_stamp": "09:29"})
|
|
self.assertEqual(spawn.call_count, 2)
|
|
self.assertEqual(cleanup.call_count, 2)
|
|
hung.stdin.close.assert_called_once()
|
|
|
|
def test_permanent_hang_is_bounded_and_next_job_can_recover(self):
|
|
processes = [self.process(error=subprocess.TimeoutExpired("browser", 75)) for _ in range(3)]
|
|
with (patch.object(browser_jobs.subprocess, "Popen", side_effect=processes) as spawn,
|
|
patch.object(browser_jobs, "stop_process_tree"), patch.object(browser_jobs.time, "sleep")):
|
|
with self.assertRaisesRegex(RuntimeError, "after 3 browser reads"):
|
|
browser_jobs.run_browser_job()
|
|
self.assertEqual(spawn.call_count, 3)
|
|
|
|
def test_invalid_credentials_are_not_retried(self):
|
|
rejected = self.process({"error": "SmartTime rejected the login", "login_error": True})
|
|
rejected.returncode = 1
|
|
with (patch.object(browser_jobs.subprocess, "Popen", return_value=rejected) as spawn,
|
|
patch.object(browser_jobs, "stop_process_tree")):
|
|
with self.assertRaises(PermissionError):
|
|
browser_jobs.run_browser_job()
|
|
spawn.assert_called_once()
|
|
|
|
def test_real_hung_playwright_process_is_terminated(self):
|
|
fixture = Path(__file__).parent / "fixtures" / "hung_smarttime_browser.py"
|
|
processes = []
|
|
original_spawn = subprocess.Popen
|
|
|
|
def spawn(*args, **kwargs):
|
|
process = original_spawn(*args, **kwargs)
|
|
processes.append(process)
|
|
return process
|
|
|
|
started = time.monotonic()
|
|
with (patch.object(browser_jobs, "__file__", str(fixture)),
|
|
patch.object(browser_jobs, "STATUS_TIMEOUT_SEC", 3),
|
|
patch.object(browser_jobs, "BROWSER_ATTEMPTS", 1),
|
|
patch.object(browser_jobs.subprocess, "Popen", side_effect=spawn)):
|
|
with self.assertRaisesRegex(RuntimeError, "did not respond within 3s"):
|
|
browser_jobs.run_browser_job()
|
|
self.assertIsNotNone(processes[0].poll())
|
|
self.assertLess(time.monotonic() - started, 15)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|