"""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 = """
100 av 240 dagar kvar efter ansökan
(60 dagar) (40 dagar)
20
""" _FAMILY_PANEL_NO_TIERS_HTML = """
100 av 240 dagar kvar efter ansökan
(60 dagar) (40 dagar)
250 av 480 dagar kvar efter ansökan
20
""" 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="fake") 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