jacobh2/F-r-ldraledigplannerare
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity
Files
224fe7ae3e511e1caf01c6d1ef3a75261c70c043
F-r-ldraledigplannerare/backend/tests/unit/test_fk_scraper.py
T
jacobh2 e4719cb9ae Rework balances around per-parent days left and deduce the partner from FK 
Manual entry now takes one parent's sjukpenningnivå, lägstanivå and
dubbeldagar left (plus optional reserved days) and derives the rest in
build_snapshot, shared with the scraper. The scraper also reads the
family panel and exposes a partner deduction; with two parents it is
saved as a 'deduced' snapshot (migration 002). Parent count is the single
source of entitlement: one member gets 480 days and 96 savable at age 4,
two get 240 and 48 each. Workspaces are capped at two parents.

Also lands the in-flight squash of migrations 002–011 into the initial
schema and the removal of planning_mode.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
2026-09-19 12:27:29 +02:00

1322 lines
49 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 (
FamilySummary,
FKScraperService,
PartnerDeduction,
_deduce_partner,
_do_extract,
_extract_family_summary,
_SessionState,
_set_status,
_sessions,
)
from app.utils.errors import ValidationError
# ---------------------------------------------------------------------------
# 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"
FAMILY_SEL = "[data-test='forDigOchAnnanForalder']"
FAMILY_KVAR_SEL = FAMILY_SEL + " .visa-dagar__dagar-kvar"
FAMILY_DISCRETE_SEL = FAMILY_SEL + " .visa-dagar__discrete"
FAMILY_NIVA_SEL = FAMILY_SEL + " .visa-dagar__desktop .niva-statistik__dagar span"
def _family_panel_present() -> bool:
return 'data-test="forDigOchAnnanForalder"' in html
def _family_kvar() -> str | None:
m = re.search(
r'data-test="forDigOchAnnanForalder".*?'
r'class="visa-dagar__dagar-kvar">(\d+)<',
html,
re.DOTALL,
)
return m.group(1) if m else None
def _family_discrete() -> str | None:
m = re.search(
r'data-test="forDigOchAnnanForalder".*?'
r'class="visa-dagar__discrete">(.*?)<',
html,
re.DOTALL,
)
return m.group(1) if m else None
def _family_niva_all_texts() -> list[str]:
m_block = re.search(
r'data-test="forDigOchAnnanForalder"(.*?)$',
html,
re.DOTALL,
)
block = m_block.group(1) if m_block else ""
return re.findall(r"\(\d+ dagar\)", block)
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())
loc.count = AsyncMock(return_value=len(_niva_all_texts()))
elif sel == FAMILY_SEL:
loc.count = AsyncMock(return_value=1 if _family_panel_present() else 0)
elif sel == FAMILY_KVAR_SEL:
val = _family_kvar()
loc.count = AsyncMock(return_value=1 if val is not None else 0)
loc.first.inner_text = AsyncMock(return_value=val or "")
elif sel == FAMILY_DISCRETE_SEL:
val = _family_discrete()
loc.count = AsyncMock(return_value=1 if val is not None else 0)
loc.first.inner_text = AsyncMock(return_value=val or "")
elif sel == FAMILY_NIVA_SEL:
texts = _family_niva_all_texts()
loc.count = AsyncMock(return_value=len(texts))
loc.all_inner_texts = AsyncMock(return_value=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=[])
loc.count = AsyncMock(return_value=1)
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=[])
loc.count = AsyncMock(return_value=1)
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=[])
loc.count = AsyncMock(return_value=1)
else:
loc.first.inner_text = AsyncMock(return_value="")
loc.first.wait_for = AsyncMock()
loc.all_inner_texts = AsyncMock(return_value=[])
loc.count = AsyncMock(return_value=0)
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_dubbeldagar_left_raises(self) -> None:
"""Regression coverage moved here from a "negative days_left" scenario.
Derivation now lives in `app.services.leave_snapshot.build_snapshot`
(see its module docstring), which derives `days_left` from the tier
breakdown (`sjukpenning_left + lagsta_niva_left`) rather than the
page's separate "För dig, dagar kvar" text — and that breakdown is
parsed with a digits-only regex, so it can never itself go negative.
`dubbeldagar_left` is still parsed with a plain `int()`, so it's the
remaining reachable path to a negative-number guard here; the
`days_left`/tier negative-value guards themselves are unit-tested
directly against `build_snapshot` in `test_leave_snapshot.py`.
`build_snapshot` raises `ValidationError` (a `DomainError`), not a
bare `ValueError` like this function's own page-parsing code does.
"""
page = self._make_bad_page(164, 240, 119, 45, -1)
with pytest.raises(ValidationError, 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())
# ---------------------------------------------------------------------------
# Family panel -> partner deduction
# ---------------------------------------------------------------------------
_NO_FAMILY_PANEL_HTML = """
<div data-test="forDig">
<div class="visa-dagar" data-test="body">
<span class="visa-dagar__dagar-kvar">100</span>
<span class="visa-dagar__discrete">av 240 dagar kvar efter ansökan</span>
<div class="visa-dagar__desktop">
<span class="niva-statistik__dagar"><span>(60 dagar)</span></span>
<span class="niva-statistik__dagar"><span>(40 dagar)</span></span>
</div>
</div>
<div data-test="footer"></div>
</div>
<div class="visa-dagar" data-test="dubbeldag">
<span class="visa-dagar__dagar-kvar">20</span>
</div>
"""
_FAMILY_PANEL_NO_TIERS_HTML = """
<div data-test="forDig">
<div class="visa-dagar" data-test="body">
<span class="visa-dagar__dagar-kvar">100</span>
<span class="visa-dagar__discrete">av 240 dagar kvar efter ansökan</span>
<div class="visa-dagar__desktop">
<span class="niva-statistik__dagar"><span>(60 dagar)</span></span>
<span class="niva-statistik__dagar"><span>(40 dagar)</span></span>
</div>
</div>
<div data-test="footer"></div>
</div>
<div data-test="forDigOchAnnanForalder">
<div class="visa-dagar" data-test="body">
<span class="visa-dagar__dagar-kvar">250</span>
<span class="visa-dagar__discrete">av 480 dagar kvar efter ansökan</span>
</div>
</div>
<div class="visa-dagar" data-test="dubbeldag">
<span class="visa-dagar__dagar-kvar">20</span>
</div>
"""
def _state() -> _SessionState:
return _SessionState(
session_id="family-test",
workspace_id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
child_id=uuid.uuid4(),
owner_user_id=uuid.uuid4(),
)
class TestExtractFamilySummary:
"""`_extract_family_summary` parses the (assumed) family-panel DOM defensively."""
async def test_family_summary_parsed_from_fixture(self) -> None:
html = _load_fixture("fk_overview_synthetic.html")
page = _make_mock_page(html)
family = await _extract_family_summary(page, _state())
assert family == FamilySummary(
total_days=480, days_left=250, sjukpenning_left=190, lagsta_niva_left=60
)
async def test_absent_family_panel_returns_none(self) -> None:
page = _make_mock_page(_NO_FAMILY_PANEL_HTML)
family = await _extract_family_summary(page, _state())
assert family is None
async def test_absent_family_panel_does_not_affect_main_extraction(self) -> None:
"""The caller's own extraction must succeed exactly as without this feature."""
page = _make_mock_page(_NO_FAMILY_PANEL_HTML)
result = await _extract_parental_leave(page, _state())
assert result.days_left == 100
assert result.total_days == 240
async def test_tierless_family_panel_returns_summary_with_none_tiers(self) -> None:
page = _make_mock_page(_FAMILY_PANEL_NO_TIERS_HTML)
family = await _extract_family_summary(page, _state())
assert family == FamilySummary(
total_days=480, days_left=250, sjukpenning_left=None, lagsta_niva_left=None
)
class TestDeducePartner:
"""`_deduce_partner` derives the other parent's balance as family - mine."""
def _mine(self, **overrides: object) -> ParentalLeaveFetch:
now = datetime.now()
base = dict(
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=140,
days_left=100,
reserved_days_total=90,
reserved_days_left=90,
transferable_days_left=10,
dubbeldagar_left=20,
tier_data=TierData(
sjukpenning_total=195,
sjukpenning_used=135,
sjukpenning_left=60,
lagsta_niva_total=45,
lagsta_niva_used=5,
lagsta_niva_left=40,
),
created_at=now,
updated_at=now,
)
base.update(overrides)
return ParentalLeaveFetch(**base)
def test_deduction_with_family_tiers_present(self) -> None:
mine = self._mine()
family = FamilySummary(
total_days=480, days_left=250, sjukpenning_left=190, lagsta_niva_left=60
)
deduction = _deduce_partner(mine, family)
assert deduction == PartnerDeduction(
total_days=240,
days_left=150,
sjukpenning_left=130,
lagsta_niva_left=20,
dubbeldagar_left=20,
)
def test_deduction_falls_back_to_proportional_split_without_family_tiers(self) -> None:
mine = self._mine()
family = FamilySummary(
total_days=480, days_left=250, sjukpenning_left=None, lagsta_niva_left=None
)
deduction = _deduce_partner(mine, family)
# total=240, left=150; lagsta_share = 90*240 // 480 = 45
assert deduction == PartnerDeduction(
total_days=240,
days_left=150,
sjukpenning_left=105,
lagsta_niva_left=45,
dubbeldagar_left=20,
)
def test_family_left_below_mine_returns_none(self) -> None:
"""family.days_left < mine.days_left can only mean stale/inconsistent data."""
mine = self._mine(total_days=240, days_left=200)
family = FamilySummary(
total_days=480, days_left=150, sjukpenning_left=None, lagsta_niva_left=None
)
assert _deduce_partner(mine, family) is None
def test_family_total_not_greater_than_mine_returns_none(self) -> None:
mine = self._mine(total_days=480)
family = FamilySummary(
total_days=480, days_left=300, sjukpenning_left=None, lagsta_niva_left=None
)
assert _deduce_partner(mine, family) is None
def test_negative_tier_deduction_falls_back_to_proportional_split(self) -> None:
"""Family-level tiers that would go negative per-parent fall back rather
than producing a nonsensical negative PartnerDeduction field."""
mine = self._mine()
# family.sjukpenning_left (50) < mine.sjukpenning_left (60) -> negative naive diff
family = FamilySummary(
total_days=480, days_left=150, sjukpenning_left=50, lagsta_niva_left=100
)
deduction = _deduce_partner(mine, family)
assert deduction is not None
assert deduction.sjukpenning_left >= 0
assert deduction.lagsta_niva_left >= 0
# ---------------------------------------------------------------------------
# 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_done_includes_partner_deduction_when_present(self) -> None:
service = FKScraperService()
sid = str(uuid.uuid4())
owner_user_id = uuid.uuid4()
now = datetime.now()
result_fetch = 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=140,
days_left=100,
reserved_days_total=90,
reserved_days_left=90,
transferable_days_left=10,
dubbeldagar_left=20,
tier_data=TierData(195, 135, 60, 45, 5, 40),
created_at=now,
updated_at=now,
)
deduction = PartnerDeduction(
total_days=240,
days_left=150,
sjukpenning_left=130,
lagsta_niva_left=20,
dubbeldagar_left=20,
)
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="done",
result=result_fetch,
partner_deduction=deduction,
)
_sessions[sid] = state
result = await service.poll_session(sid, owner_user_id)
assert result["status"] == "done"
assert result["partner_deduction"] == {
"total_days": 240,
"days_left": 150,
"sjukpenning_left": 130,
"lagsta_niva_left": 20,
"dubbeldagar_left": 20,
}
async def test_poll_done_partner_deduction_is_none_when_absent(self) -> None:
service = FKScraperService()
sid = str(uuid.uuid4())
owner_user_id = uuid.uuid4()
now = datetime.now()
result_fetch = ParentalLeaveFetch(
id=uuid.uuid4(),
personal_data_uuid=uuid.uuid4(),
workspace_id=uuid.uuid4(),
child_id=uuid.uuid4(),
source="scraped",
fetched_at=now,
total_days=480,
days_used=0,
days_left=480,
reserved_days_total=90,
reserved_days_left=90,
transferable_days_left=390,
dubbeldagar_left=60,
tier_data=TierData(390, 0, 390, 90, 0, 90),
created_at=now,
updated_at=now,
)
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="done",
result=result_fetch,
partner_deduction=None,
)
_sessions[sid] = state
result = await service.poll_session(sid, owner_user_id)
assert result["status"] == "done"
assert result["partner_deduction"] is None
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, partner: PartnerDeduction | None) -> 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, partner: PartnerDeduction | None
) -> 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, partner: PartnerDeduction | None
) -> 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