jacobh2/F-r-ldraledigplannerare
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
02db5430eb44d8f396dc62c0b74cc341f0e8ea73
F-r-ldraledigplannerare/backend/tests/unit/test_fk_scraper.py
T
jacobh2 02db5430eb Harden the API and the data layer after a security and performance audit 
Security
- Every route now requires a login and refuses ids outside the caller's
  workspace with 404: plan, leave-wishes, day-overrides and vacation-days
  were fully anonymous; every other router trusted client-supplied
  workspace/child/parent ids. Scraper sessions are bound to their owner.
- Invitations can no longer be created for a foreign workspace, and a
  redeemer who is already a member (or a full workspace) is refused.
- Production refuses the default JWT secret; API docs are hidden in
  production; a failed migration aborts startup instead of serving.
- Login and register no longer reveal whether an email exists; inactive
  accounts are rejected; /auth/refresh is rate limited.
- The FK scraper writes debug artefacts only when explicitly enabled,
  logs no scraped values or page text, clears the BankID QR as soon as it
  is scanned and purges finished sessions after five minutes.
- Real FK page captures are removed from the tree (history rewrite
  follows); the parser tests use a synthetic fixture.

Performance and correctness
- Plan apply/sync run inside one locked transaction with batched writes
  instead of one commit per row, so a failure mid-batch leaves nothing
  half-applied and concurrent syncs no longer corrupt each other.
- SQLite runs in WAL mode with a busy timeout; 14 indexes cover the
  filters that were full scans; argon2 hashing runs off the event loop;
  responses are gzipped.
- Frontend: a 401 now refreshes the token once and retries, so sessions
  no longer die silently after 15 minutes; every data hook discards
  stale responses; logout clears state even when the server call fails.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CsXCCcDfmV4Sps4ZQSvNEY
2026-09-18 07:54:21 +02:00

966 lines
36 KiB
Python
Raw Blame History

"""Tests for the FK scraper: extractor, validation, and session state machine.
Three independent test concerns:
1. _extract_parental_leave — parses the FK #/valjdagar HTML fixture and
produces a correct ParentalLeaveFetch. No network, no BankID.
2. Validation helpers — out-of-bounds numbers raise rather than persist.
3. Session state machine — start/poll/cancel transitions, TTL auto-cancel,
double-cancel safety. Browser is mocked; no headless Chromium needed.
"""
from __future__ import annotations
import re
import uuid
from datetime import datetime
from pathlib import Path
from typing import Any
from unittest.mock import AsyncMock, MagicMock, patch
import pytest
from app.entities.parental_leave import ParentalLeaveFetch, TierData
from app.scrapers.fk_scraper import (
FKScraperService,
_SessionState,
_do_extract,
_set_status,
_sessions,
)
# ---------------------------------------------------------------------------
# Helpers for the HTML-fixture tests
# ---------------------------------------------------------------------------
FIXTURES_DIR = Path(__file__).parents[1] / "fixtures"
def _load_fixture(name: str) -> str:
"""Read a synthetic HTML fixture from tests/fixtures/.
Fixtures here are hand-built (never real FK page captures — see F1 of
the scraper/config/ops security audit) and always present in the repo.
"""
return (FIXTURES_DIR / name).read_text(encoding="utf-8")
# ---------------------------------------------------------------------------
# Minimal mock of a Playwright Page that answers our CSS queries from HTML
# ---------------------------------------------------------------------------
def _make_mock_page(html: str):
"""
Build a minimal async mock page that serves inner_text / all_inner_texts
for the selectors used by _extract_parental_leave.
Uses basic regex over the fixture HTML rather than a real browser — fast
and dependency-free.
"""
NIVA_SEL = "[data-test='forDig'] .visa-dagar__desktop .niva-statistik__dagar span"
def _for_dig_kvar() -> str:
m = re.search(
r'data-test="forDig".*?'
r'class="visa-dagar__dagar-kvar">(\d+)<',
html,
re.DOTALL,
)
if m:
return m.group(1)
raise ValueError("No forDig dagar-kvar found in fixture")
def _for_dig_discrete() -> str:
m = re.search(
r'data-test="forDig".*?'
r'class="visa-dagar__discrete">(.*?)<',
html,
re.DOTALL,
)
if m:
return m.group(1)
raise ValueError("No forDig discrete found in fixture")
def _dubbeldag_kvar() -> str:
m = re.search(
r'data-test="dubbeldag".*?'
r'class="visa-dagar__dagar-kvar">(\d+)<',
html,
re.DOTALL,
)
if m:
return m.group(1)
raise ValueError("No dubbeldag dagar-kvar found in fixture")
def _niva_all_texts() -> list[str]:
m_block = re.search(
r'data-test="forDig"(.*?)data-test="footer"',
html,
re.DOTALL,
)
block = m_block.group(1) if m_block else ""
desktop_m = re.search(r"visa-dagar__desktop(.*?)visa-dagar__mobile", block, re.DOTALL)
target = desktop_m.group(1) if desktop_m else block
return re.findall(r"\(\d+ dagar\)", target)
page = MagicMock()
def locator_factory(sel: str) -> MagicMock:
loc = MagicMock()
if sel == NIVA_SEL:
loc.first.inner_text = AsyncMock(return_value="")
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=_niva_all_texts())
elif "visa-dagar__dagar-kvar" in sel and "forDig" in sel:
loc.first.inner_text = AsyncMock(return_value=_for_dig_kvar())
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
elif "visa-dagar__discrete" in sel and "forDig" in sel:
loc.first.inner_text = AsyncMock(return_value=_for_dig_discrete())
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
elif "dubbeldag" in sel and "visa-dagar__dagar-kvar" in sel:
loc.first.inner_text = AsyncMock(return_value=_dubbeldag_kvar())
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
else:
loc.first.inner_text = AsyncMock(return_value="")
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
return loc
page.locator.side_effect = locator_factory
return page
# ---------------------------------------------------------------------------
# Extract from real HTML fixture
# ---------------------------------------------------------------------------
# Import the extractor directly (no browser needed)
from app.scrapers.fk_scraper import _extract_parental_leave # noqa: E402
class TestExtractFromFixture:
"""Parse the synthetic overview HTML fixture and assert correct field values."""
async def test_days_left_and_total_parsed(self) -> None:
html = _load_fixture("fk_overview_synthetic.html")
state = _SessionState(
session_id="test-1",
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
)
page = _make_mock_page(html)
result = await _extract_parental_leave(page, state)
# From the synthetic fixture: 100 av 240 dagar
assert result.days_left == 100
assert result.total_days == 240
assert result.days_used == 140
async def test_tier_values_parsed(self) -> None:
html = _load_fixture("fk_overview_synthetic.html")
state = _SessionState(
session_id="test-2",
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
)
page = _make_mock_page(html)
result = await _extract_parental_leave(page, state)
# From the synthetic fixture: sjukpenning 60, lägstanivå 40
assert result.tier_data.sjukpenning_left == 60
assert result.tier_data.lagsta_niva_left == 40
async def test_dubbeldagar_parsed(self) -> None:
html = _load_fixture("fk_overview_synthetic.html")
state = _SessionState(
session_id="test-3",
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
)
page = _make_mock_page(html)
result = await _extract_parental_leave(page, state)
# From the synthetic fixture: 20 av 60 dubbeldagar
assert result.dubbeldagar_left == 20
async def test_source_is_scraped(self) -> None:
html = _load_fixture("fk_overview_synthetic.html")
state = _SessionState(
session_id="test-4",
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
)
page = _make_mock_page(html)
result = await _extract_parental_leave(page, state)
assert result.source == "scraped"
async def test_reserved_days_derived(self) -> None:
html = _load_fixture("fk_overview_synthetic.html")
state = _SessionState(
session_id="test-5",
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
)
page = _make_mock_page(html)
result = await _extract_parental_leave(page, state)
# 140 days used < 150 transferable → all reserved days remain
assert result.reserved_days_total == 90
assert result.reserved_days_left == 90
assert result.transferable_days_left == 240 - 90 - 140 # 10
async def test_tier_totals_sum_to_total_days(self) -> None:
html = _load_fixture("fk_overview_synthetic.html")
state = _SessionState(
session_id="test-6",
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
)
page = _make_mock_page(html)
result = await _extract_parental_leave(page, state)
assert (
result.tier_data.sjukpenning_total + result.tier_data.lagsta_niva_total
== result.total_days
)
async def test_tier_used_nonnegative(self) -> None:
html = _load_fixture("fk_overview_synthetic.html")
state = _SessionState(
session_id="test-7",
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
)
page = _make_mock_page(html)
result = await _extract_parental_leave(page, state)
assert result.tier_data.sjukpenning_used >= 0
assert result.tier_data.lagsta_niva_used >= 0
# ---------------------------------------------------------------------------
# Validation: bad numbers raise rather than persist
# ---------------------------------------------------------------------------
class TestValidation:
"""Feed impossible numbers and assert the extractor raises."""
def _make_bad_page(
self, days_left: int, total_days: int, sjukpenning: int, lagsta: int, dubbel: int
) -> Any:
NIVA_SEL = "[data-test='forDig'] .visa-dagar__desktop .niva-statistik__dagar span"
page = MagicMock()
def locator_factory(sel: str) -> MagicMock:
loc = MagicMock()
if sel == NIVA_SEL:
loc.first.inner_text = AsyncMock(return_value="")
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(
return_value=[f"({sjukpenning} dagar)", f"({lagsta} dagar)"]
)
elif "visa-dagar__dagar-kvar" in sel and "forDig" in sel:
loc.first.inner_text = AsyncMock(return_value=str(days_left))
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
elif "visa-dagar__discrete" in sel and "forDig" in sel:
loc.first.inner_text = AsyncMock(
return_value=f"av {total_days} dagar kvar efter ansökan"
)
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
elif "dubbeldag" in sel:
loc.first.inner_text = AsyncMock(return_value=str(dubbel))
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
else:
loc.first.inner_text = AsyncMock(return_value="")
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
return loc
page.locator.side_effect = locator_factory
return page
def _state(self) -> _SessionState:
return _SessionState(
session_id="val-test",
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
)
async def test_negative_days_left_raises(self) -> None:
page = self._make_bad_page(-1, 240, 100, 45, 39)
with pytest.raises(ValueError, match="negative"):
await _extract_parental_leave(page, self._state())
async def test_total_days_parse_failure_raises(self) -> None:
NIVA_SEL = "[data-test='forDig'] .visa-dagar__desktop .niva-statistik__dagar span"
page = MagicMock()
def locator_factory(sel: str) -> MagicMock:
loc = MagicMock()
if sel == NIVA_SEL:
loc.first.inner_text = AsyncMock(return_value="")
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=["(80 dagar)", "(20 dagar)"])
elif "dagar-kvar" in sel and "forDig" in sel:
loc.first.inner_text = AsyncMock(return_value="100")
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
elif "discrete" in sel and "forDig" in sel:
loc.first.inner_text = AsyncMock(return_value="INVALID TEXT")
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
elif "dubbeldag" in sel:
loc.first.inner_text = AsyncMock(return_value="39")
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
else:
loc.first.inner_text = AsyncMock(return_value="0")
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
return loc
page.locator.side_effect = locator_factory
with pytest.raises(ValueError, match="parse total days"):
await _extract_parental_leave(page, self._state())
async def test_fewer_than_two_niva_values_raises(self) -> None:
NIVA_SEL = "[data-test='forDig'] .visa-dagar__desktop .niva-statistik__dagar span"
page = MagicMock()
def locator_factory(sel: str) -> MagicMock:
loc = MagicMock()
if sel == NIVA_SEL:
loc.first.inner_text = AsyncMock(return_value="")
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=["(119 dagar)"]) # only one
elif "dagar-kvar" in sel and "forDig" in sel:
loc.first.inner_text = AsyncMock(return_value="164")
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
elif "discrete" in sel and "forDig" in sel:
loc.first.inner_text = AsyncMock(return_value="av 240 dagar kvar efter ansökan")
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
elif "dubbeldag" in sel:
loc.first.inner_text = AsyncMock(return_value="39")
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
else:
loc.first.inner_text = AsyncMock(return_value="0")
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
return loc
page.locator.side_effect = locator_factory
with pytest.raises(ValueError, match="Expected 2 tier values"):
await _extract_parental_leave(page, self._state())
async def test_zero_total_days_raises(self) -> None:
"""Guard against placeholder zero data: total_days=0 is impossible and indicates hydration failure."""
page = self._make_bad_page(164, 0, 119, 45, 39)
with pytest.raises(ValueError, match="total_days=0.*had not hydrated"):
await _extract_parental_leave(page, self._state())
# ---------------------------------------------------------------------------
# Session state machine
# ---------------------------------------------------------------------------
class TestSessionStateMachine:
"""Session transitions, TTL watchdog, and double-cancel safety."""
def setup_method(self) -> None:
_sessions.clear()
def teardown_method(self) -> None:
_sessions.clear()
async def test_start_session_creates_entry(self) -> None:
service = FKScraperService()
owner_user_id = uuid.uuid4()
with patch("app.scrapers.fk_scraper._run_session", new_callable=AsyncMock) as mock_run:
mock_run.return_value = None
sid = await service.start_session(
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=owner_user_id,
)
assert sid in _sessions
assert _sessions[sid].status == "launching"
assert _sessions[sid].owner_user_id == owner_user_id
async def test_poll_unknown_session_returns_error(self) -> None:
service = FKScraperService()
result = await service.poll_session("nonexistent", uuid.uuid4())
assert result["status"] == "error"
assert "Unknown" in result["message"]
async def test_poll_awaiting_scan_returns_qr_data(self) -> None:
service = FKScraperService()
sid = str(uuid.uuid4())
owner_user_id = uuid.uuid4()
state = _SessionState(
session_id=sid,
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=owner_user_id,
status="awaiting_scan",
qr_data=b"PNG_BYTES",
)
_sessions[sid] = state
result = await service.poll_session(sid, owner_user_id)
assert result["status"] == "awaiting_scan"
assert "qr_data" in result
import base64
assert base64.b64decode(result["qr_data"]) == b"PNG_BYTES"
async def test_poll_with_wrong_owner_is_treated_as_unknown_session(self) -> None:
"""F4 (T1 authz audit): a session_id known to a caller who did not
start it must behave EXACTLY like a made-up session_id — never leak
that the session exists (no QR data, no distinct error)."""
service = FKScraperService()
sid = str(uuid.uuid4())
state = _SessionState(
session_id=sid,
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
status="awaiting_scan",
qr_data=b"PNG_BYTES",
)
_sessions[sid] = state
result = await service.poll_session(sid, uuid.uuid4()) # different owner
assert result == {"status": "error", "message": "Unknown session_id"}
async def test_cancel_sets_status(self) -> None:
service = FKScraperService()
sid = str(uuid.uuid4())
owner_user_id = uuid.uuid4()
state = _SessionState(
session_id=sid,
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=owner_user_id,
status="awaiting_scan",
)
_sessions[sid] = state
await service.cancel_session(sid, owner_user_id)
assert _sessions[sid].status == "cancelled"
async def test_cancel_clears_qr_data(self) -> None:
"""F5: cancelling a session must not leave the last QR frame in RAM."""
service = FKScraperService()
sid = str(uuid.uuid4())
owner_user_id = uuid.uuid4()
state = _SessionState(
session_id=sid,
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=owner_user_id,
status="awaiting_scan",
qr_data=b"PNG_BYTES",
)
_sessions[sid] = state
await service.cancel_session(sid, owner_user_id)
assert _sessions[sid].qr_data is None
async def test_cancel_with_wrong_owner_is_a_no_op(self) -> None:
"""F4: cancelling with the wrong owner must not affect the session —
exactly like cancelling an unknown session_id."""
service = FKScraperService()
sid = str(uuid.uuid4())
state = _SessionState(
session_id=sid,
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
status="awaiting_scan",
)
_sessions[sid] = state
await service.cancel_session(sid, uuid.uuid4()) # different owner
assert _sessions[sid].status == "awaiting_scan"
async def test_cancel_nonexistent_session_is_safe(self) -> None:
service = FKScraperService()
await service.cancel_session("does-not-exist", uuid.uuid4()) # should not raise
async def test_ttl_watchdog_cancels_stale_session(self) -> None:
from app.scrapers.fk_scraper import _ttl_watchdog
with patch("app.scrapers.fk_scraper.settings") as mock_settings:
mock_settings.scraper_session_ttl_seconds = 0 # immediate
sid = str(uuid.uuid4())
state = _SessionState(
session_id=sid,
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
status="awaiting_scan",
)
_sessions[sid] = state
await _ttl_watchdog(sid)
assert _sessions[sid].status == "error"
assert "timed out" in (_sessions[sid].message or "")
async def test_ttl_watchdog_purges_already_terminal_session(self) -> None:
"""F5: if the session already finished by the time the watchdog
fires, it's a safety net that purges it immediately rather than
leaving it for `_purge_later`."""
from app.scrapers.fk_scraper import _ttl_watchdog
with patch("app.scrapers.fk_scraper.settings") as mock_settings:
mock_settings.scraper_session_ttl_seconds = 0 # immediate
sid = str(uuid.uuid4())
state = _SessionState(
session_id=sid,
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
status="done",
)
_sessions[sid] = state
await _ttl_watchdog(sid)
assert sid not in _sessions
async def test_purge_later_removes_session_after_retention_window(self) -> None:
"""F5: a finished session is removed from `_sessions` after
`scraper_result_retention_seconds`."""
from app.scrapers.fk_scraper import _purge_later
with patch("app.scrapers.fk_scraper.settings") as mock_settings:
mock_settings.scraper_result_retention_seconds = 0 # immediate
sid = str(uuid.uuid4())
state = _SessionState(
session_id=sid,
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
status="done",
result=None,
)
_sessions[sid] = state
await _purge_later(sid)
assert sid not in _sessions
async def test_purge_later_is_a_no_op_if_already_purged(self) -> None:
from app.scrapers.fk_scraper import _purge_later
with patch("app.scrapers.fk_scraper.settings") as mock_settings:
mock_settings.scraper_result_retention_seconds = 0
await _purge_later("does-not-exist") # should not raise
async def test_start_session_cancels_existing_session_for_same_owner(self) -> None:
"""F5: at most one live session per owner — starting a new one
cancels any non-terminal session already in flight for that owner."""
service = FKScraperService()
owner_user_id = uuid.uuid4()
with patch("app.scrapers.fk_scraper._run_session", new_callable=AsyncMock):
first_sid = await service.start_session(
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=owner_user_id,
)
_sessions[first_sid].status = "awaiting_scan"
second_sid = await service.start_session(
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=owner_user_id,
)
assert _sessions[first_sid].status == "cancelled"
assert _sessions[second_sid].status == "launching"
assert first_sid != second_sid
async def test_start_session_does_not_cancel_other_owners_session(self) -> None:
service = FKScraperService()
owner_a = uuid.uuid4()
owner_b = uuid.uuid4()
with patch("app.scrapers.fk_scraper._run_session", new_callable=AsyncMock):
sid_a = await service.start_session(
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=owner_a,
)
_sessions[sid_a].status = "awaiting_scan"
await service.start_session(
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=owner_b,
)
assert _sessions[sid_a].status == "awaiting_scan"
async def test_set_status_updates_message(self) -> None:
sid = str(uuid.uuid4())
state = _SessionState(
session_id=sid,
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
)
_set_status(state, "error", "Something went wrong")
assert state.status == "error"
assert state.message == "Something went wrong"
async def test_do_extract_calls_persist_callback_on_success(self) -> None:
"""_do_extract invokes the persist_callback with the extracted result."""
persisted: list[ParentalLeaveFetch] = []
async def mock_persist(fetch: ParentalLeaveFetch) -> None:
persisted.append(fetch)
now = datetime.now()
expected = ParentalLeaveFetch(
id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
workspace_id=uuid.uuid4(),
child_id=uuid.uuid4(),
source="scraped",
fetched_at=now,
total_days=240,
days_used=76,
days_left=164,
reserved_days_total=90,
reserved_days_left=90,
transferable_days_left=74,
dubbeldagar_left=39,
tier_data=TierData(
sjukpenning_total=195,
sjukpenning_used=76,
sjukpenning_left=119,
lagsta_niva_total=45,
lagsta_niva_used=0,
lagsta_niva_left=45,
),
created_at=now,
updated_at=now,
)
sid = str(uuid.uuid4())
state = _SessionState(
session_id=sid,
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
persist_callback=mock_persist,
)
_sessions[sid] = state
page = MagicMock()
page.wait_for_selector = AsyncMock()
page.wait_for_function = AsyncMock()
def locator_factory(sel: str) -> MagicMock:
loc = MagicMock()
loc.first.click = AsyncMock()
return loc
page.locator.side_effect = locator_factory
with patch(
"app.scrapers.fk_scraper._extract_parental_leave", new_callable=AsyncMock
) as mock_extract:
mock_extract.return_value = expected
await _do_extract(page, state)
assert state.status == "done"
assert state.result is expected
assert len(persisted) == 1
assert persisted[0] is expected
async def test_do_extract_sets_error_when_persist_callback_fails(self) -> None:
"""_do_extract sets status=error if the persist_callback raises."""
async def failing_persist(fetch: ParentalLeaveFetch) -> None:
raise RuntimeError("DB unavailable")
now = datetime.now()
dummy = ParentalLeaveFetch(
id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
workspace_id=uuid.uuid4(),
child_id=uuid.uuid4(),
source="scraped",
fetched_at=now,
total_days=240,
days_used=0,
days_left=240,
reserved_days_total=90,
reserved_days_left=90,
transferable_days_left=150,
dubbeldagar_left=60,
tier_data=TierData(195, 0, 195, 45, 0, 45),
created_at=now,
updated_at=now,
)
sid = str(uuid.uuid4())
state = _SessionState(
session_id=sid,
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
persist_callback=failing_persist,
)
_sessions[sid] = state
page = MagicMock()
page.wait_for_selector = AsyncMock()
page.wait_for_function = AsyncMock()
def locator_factory(sel: str) -> MagicMock:
loc = MagicMock()
loc.first.click = AsyncMock()
return loc
page.locator.side_effect = locator_factory
with patch(
"app.scrapers.fk_scraper._extract_parental_leave", new_callable=AsyncMock
) as mock_extract:
mock_extract.return_value = dummy
await _do_extract(page, state)
assert state.status == "error"
assert "persist" in (state.message or "").lower()
async def test_do_extract_waits_for_hydration_before_extracting(self) -> None:
"""_do_extract calls wait_for_function to ensure data hydrates before extraction."""
now = datetime.now()
expected = ParentalLeaveFetch(
id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
workspace_id=uuid.uuid4(),
child_id=uuid.uuid4(),
source="scraped",
fetched_at=now,
total_days=240,
days_used=0,
days_left=240,
reserved_days_total=90,
reserved_days_left=90,
transferable_days_left=150,
dubbeldagar_left=60,
tier_data=TierData(195, 0, 195, 45, 0, 45),
created_at=now,
updated_at=now,
)
sid = str(uuid.uuid4())
state = _SessionState(
session_id=sid,
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
)
_sessions[sid] = state
page = MagicMock()
page.wait_for_selector = AsyncMock()
page.wait_for_function = AsyncMock()
def locator_factory(sel: str) -> MagicMock:
loc = MagicMock()
loc.first.click = AsyncMock()
return loc
page.locator.side_effect = locator_factory
with patch(
"app.scrapers.fk_scraper._extract_parental_leave", new_callable=AsyncMock
) as mock_extract:
mock_extract.return_value = expected
await _do_extract(page, state)
# Verify wait_for_function was called with the correct selector
page.wait_for_function.assert_called_once()
call_args = page.wait_for_function.call_args
assert call_args[1]["arg"] == "[data-test='forDig'] .visa-dagar__discrete"
assert call_args[1]["timeout"] == 30_000
# ---------------------------------------------------------------------------
# F3: debug artifacts are opt-in (contain personal data when written)
# ---------------------------------------------------------------------------
class TestSaveDebugArtifacts:
def _make_page(self) -> MagicMock:
async def _fake_screenshot(path: str, full_page: bool = True) -> None:
# Real Playwright writes the PNG to `path` itself; mirror that so
# the "both files got written" assertion is meaningful.
Path(path).write_bytes(b"fake-png-bytes")
page = MagicMock()
page.content = AsyncMock(return_value="<html>fake</html>")
page.screenshot = AsyncMock(side_effect=_fake_screenshot)
return page
async def test_noop_when_debug_disabled_even_with_dir_set(self, tmp_path) -> None:
from app.scrapers.fk_scraper import _save_debug_artifacts
with patch("app.scrapers.fk_scraper.settings") as mock_settings:
mock_settings.scraper_debug_dir = str(tmp_path)
mock_settings.scraper_debug_enabled = False
await _save_debug_artifacts(self._make_page(), "sid-1", ["[log] hi"])
assert list(tmp_path.iterdir()) == []
async def test_noop_when_dir_unset_even_if_enabled(self, tmp_path) -> None:
from app.scrapers.fk_scraper import _save_debug_artifacts
with patch("app.scrapers.fk_scraper.settings") as mock_settings:
mock_settings.scraper_debug_dir = None
mock_settings.scraper_debug_enabled = True
await _save_debug_artifacts(self._make_page(), "sid-2", ["[log] hi"])
assert list(tmp_path.iterdir()) == []
async def test_writes_artifacts_when_both_enabled_and_dir_set(self, tmp_path) -> None:
from app.scrapers.fk_scraper import _save_debug_artifacts
with patch("app.scrapers.fk_scraper.settings") as mock_settings:
mock_settings.scraper_debug_dir = str(tmp_path)
mock_settings.scraper_debug_enabled = True
await _save_debug_artifacts(self._make_page(), "sid-3", ["[log] hi"])
written = {p.name for p in tmp_path.iterdir()}
assert written == {"sid-3.html", "sid-3.png", "sid-3.console.txt"}
# ---------------------------------------------------------------------------
# Persistence integration: real SQLite repo round-trip
# ---------------------------------------------------------------------------
class TestPersistenceIntegration:
"""Run _do_extract with a real SQLiteParentalLeaveRepository and verify the row."""
def setup_method(self) -> None:
_sessions.clear()
def teardown_method(self) -> None:
_sessions.clear()
async def test_scraped_row_is_readable_after_session_done(
self, parental_leave_repo, seeded_ids
) -> None:
"""After _do_extract completes, repo.get_latest_by_parent returns the scraped row."""
async def persist_callback(fetch: ParentalLeaveFetch) -> None:
await parental_leave_repo.create(fetch)
now = datetime.now()
scraped = ParentalLeaveFetch(
id=uuid.uuid4(),
personal_data_uuid=seeded_ids.personal_data_uuid,
workspace_id=seeded_ids.workspace_id,
child_id=seeded_ids.child_id,
source="scraped",
fetched_at=now,
total_days=240,
days_used=76,
days_left=164,
reserved_days_total=90,
reserved_days_left=90,
transferable_days_left=74,
dubbeldagar_left=39,
tier_data=TierData(195, 76, 119, 45, 0, 45),
created_at=now,
updated_at=now,
)
sid = str(uuid.uuid4())
state = _SessionState(
session_id=sid,
workspace_id=seeded_ids.workspace_id,
personal_data_uuid=seeded_ids.personal_data_uuid,
child_id=seeded_ids.child_id,
owner_user_id=uuid.uuid4(),
persist_callback=persist_callback,
)
_sessions[sid] = state
page = MagicMock()
page.wait_for_selector = AsyncMock()
page.wait_for_function = AsyncMock()
def locator_factory(sel: str) -> MagicMock:
loc = MagicMock()
loc.first.click = AsyncMock()
return loc
page.locator.side_effect = locator_factory
with patch(
"app.scrapers.fk_scraper._extract_parental_leave", new_callable=AsyncMock
) as mock_extract:
mock_extract.return_value = scraped
await _do_extract(page, state)
assert state.status == "done"
result = await parental_leave_repo.get_latest_by_parent(
seeded_ids.personal_data_uuid, seeded_ids.child_id
)
assert result is not None
assert result.source == "scraped"
assert result.days_left == 164
assert result.total_days == 240
Reference in New Issue View Git Blame Copy Permalink