645 lines
24 KiB
Python
645 lines
24 KiB
Python
"""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 date, 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
|
|
# ---------------------------------------------------------------------------
|
|
|
|
CAPTURES_DIR = Path(__file__).parents[3] / "captures"
|
|
|
|
|
|
def _load_fixture(name: str) -> str:
|
|
"""Read a page capture from captures/, skipping if not present."""
|
|
path = CAPTURES_DIR / name
|
|
if not path.exists():
|
|
pytest.skip(f"Fixture {name} not found — run the captures step first")
|
|
return path.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.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.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.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.all_inner_texts = AsyncMock(return_value=[])
|
|
else:
|
|
loc.first.inner_text = AsyncMock(return_value="")
|
|
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 captured overview HTML and assert correct field values."""
|
|
|
|
async def test_days_left_and_total_parsed(self) -> None:
|
|
html = _load_fixture("05_foraldrapenning_overview.html")
|
|
state = _SessionState(
|
|
session_id="test-1",
|
|
workspace_id=uuid.uuid4(),
|
|
personal_data_uuid=uuid.uuid4(),
|
|
child_id=uuid.uuid4(),
|
|
)
|
|
page = _make_mock_page(html)
|
|
result = await _extract_parental_leave(page, state)
|
|
# From the capture: 164 av 240 dagar
|
|
assert result.days_left == 164
|
|
assert result.total_days == 240
|
|
assert result.days_used == 76
|
|
|
|
async def test_tier_values_parsed(self) -> None:
|
|
html = _load_fixture("05_foraldrapenning_overview.html")
|
|
state = _SessionState(
|
|
session_id="test-2",
|
|
workspace_id=uuid.uuid4(),
|
|
personal_data_uuid=uuid.uuid4(),
|
|
child_id=uuid.uuid4(),
|
|
)
|
|
page = _make_mock_page(html)
|
|
result = await _extract_parental_leave(page, state)
|
|
# From the capture: sjukpenning 119, lägstanivå 45
|
|
assert result.tier_data.sjukpenning_left == 119
|
|
assert result.tier_data.lagsta_niva_left == 45
|
|
|
|
async def test_dubbeldagar_parsed(self) -> None:
|
|
html = _load_fixture("05_foraldrapenning_overview.html")
|
|
state = _SessionState(
|
|
session_id="test-3",
|
|
workspace_id=uuid.uuid4(),
|
|
personal_data_uuid=uuid.uuid4(),
|
|
child_id=uuid.uuid4(),
|
|
)
|
|
page = _make_mock_page(html)
|
|
result = await _extract_parental_leave(page, state)
|
|
# From the capture: 39 av 60 dubbeldagar
|
|
assert result.dubbeldagar_left == 39
|
|
|
|
async def test_source_is_scraped(self) -> None:
|
|
html = _load_fixture("05_foraldrapenning_overview.html")
|
|
state = _SessionState(
|
|
session_id="test-4",
|
|
workspace_id=uuid.uuid4(),
|
|
personal_data_uuid=uuid.uuid4(),
|
|
child_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("05_foraldrapenning_overview.html")
|
|
state = _SessionState(
|
|
session_id="test-5",
|
|
workspace_id=uuid.uuid4(),
|
|
personal_data_uuid=uuid.uuid4(),
|
|
child_id=uuid.uuid4(),
|
|
)
|
|
page = _make_mock_page(html)
|
|
result = await _extract_parental_leave(page, state)
|
|
# 76 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 - 76 # 74
|
|
|
|
async def test_tier_totals_sum_to_total_days(self) -> None:
|
|
html = _load_fixture("05_foraldrapenning_overview.html")
|
|
state = _SessionState(
|
|
session_id="test-6",
|
|
workspace_id=uuid.uuid4(),
|
|
personal_data_uuid=uuid.uuid4(),
|
|
child_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("05_foraldrapenning_overview.html")
|
|
state = _SessionState(
|
|
session_id="test-7",
|
|
workspace_id=uuid.uuid4(),
|
|
personal_data_uuid=uuid.uuid4(),
|
|
child_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.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.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.all_inner_texts = AsyncMock(return_value=[])
|
|
elif "dubbeldag" in sel:
|
|
loc.first.inner_text = AsyncMock(return_value=str(dubbel))
|
|
loc.all_inner_texts = AsyncMock(return_value=[])
|
|
else:
|
|
loc.first.inner_text = AsyncMock(return_value="")
|
|
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(),
|
|
)
|
|
|
|
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.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.all_inner_texts = AsyncMock(return_value=[])
|
|
elif "discrete" in sel and "forDig" in sel:
|
|
loc.first.inner_text = AsyncMock(return_value="INVALID TEXT")
|
|
loc.all_inner_texts = AsyncMock(return_value=[])
|
|
elif "dubbeldag" in sel:
|
|
loc.first.inner_text = AsyncMock(return_value="39")
|
|
loc.all_inner_texts = AsyncMock(return_value=[])
|
|
else:
|
|
loc.first.inner_text = AsyncMock(return_value="0")
|
|
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.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.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.all_inner_texts = AsyncMock(return_value=[])
|
|
elif "dubbeldag" in sel:
|
|
loc.first.inner_text = AsyncMock(return_value="39")
|
|
loc.all_inner_texts = AsyncMock(return_value=[])
|
|
else:
|
|
loc.first.inner_text = AsyncMock(return_value="0")
|
|
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())
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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()
|
|
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(),
|
|
)
|
|
assert sid in _sessions
|
|
assert _sessions[sid].status == "launching"
|
|
|
|
async def test_poll_unknown_session_returns_error(self) -> None:
|
|
service = FKScraperService()
|
|
result = await service.poll_session("nonexistent")
|
|
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())
|
|
state = _SessionState(
|
|
session_id=sid,
|
|
workspace_id=uuid.uuid4(),
|
|
personal_data_uuid=uuid.uuid4(),
|
|
child_id=uuid.uuid4(),
|
|
status="awaiting_scan",
|
|
qr_data=b"PNG_BYTES",
|
|
)
|
|
_sessions[sid] = state
|
|
|
|
result = await service.poll_session(sid)
|
|
assert result["status"] == "awaiting_scan"
|
|
assert "qr_data" in result
|
|
import base64
|
|
assert base64.b64decode(result["qr_data"]) == b"PNG_BYTES"
|
|
|
|
async def test_cancel_sets_status(self) -> None:
|
|
service = FKScraperService()
|
|
sid = str(uuid.uuid4())
|
|
state = _SessionState(
|
|
session_id=sid,
|
|
workspace_id=uuid.uuid4(),
|
|
personal_data_uuid=uuid.uuid4(),
|
|
child_id=uuid.uuid4(),
|
|
status="awaiting_scan",
|
|
)
|
|
_sessions[sid] = state
|
|
|
|
await service.cancel_session(sid)
|
|
assert _sessions[sid].status == "cancelled"
|
|
|
|
async def test_cancel_nonexistent_session_is_safe(self) -> None:
|
|
service = FKScraperService()
|
|
await service.cancel_session("does-not-exist") # 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(),
|
|
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_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(),
|
|
)
|
|
_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(),
|
|
persist_callback=mock_persist,
|
|
)
|
|
_sessions[sid] = state
|
|
|
|
page = MagicMock()
|
|
page.wait_for_selector = 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(),
|
|
persist_callback=failing_persist,
|
|
)
|
|
_sessions[sid] = state
|
|
|
|
page = MagicMock()
|
|
page.wait_for_selector = 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()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# 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."""
|
|
from app.repositories_impl.sqlite.parental_leave_repository import (
|
|
SQLiteParentalLeaveRepository,
|
|
)
|
|
|
|
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,
|
|
persist_callback=persist_callback,
|
|
)
|
|
_sessions[sid] = state
|
|
|
|
page = MagicMock()
|
|
page.wait_for_selector = 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
|