from contextlib import contextmanager from urllib.parse import parse_qs, quote_plus, urlparse import pytest from playwright.sync_api import TimeoutError as PlaywrightTimeoutError from job_research.apec import adapter as adapter_module from job_research.apec.adapter import ApecAdapter _RESULT_LINK_SELECTOR = "a[href*='/candidat/recherche-emploi.html/emploi/detail-offre/']" class _FakeResultButton: def __init__(self, page, name: str) -> None: self.page = page self.name = name def click(self, timeout: int | None = None) -> None: if self.name == "Rechercher": self.page.url = ( "https://www.apec.fr/candidat/recherche-emploi.html/emploi" f"?motsCles={quote_plus(self.page.current_query)}&page=0" ) self.page.current_page = 0 class _FakeLocator: def __init__(self, page, selector: str) -> None: self.page = page self.selector = selector def fill(self, value: str) -> None: self.page.current_query = value def evaluate_all(self, function: str): if self.selector == _RESULT_LINK_SELECTOR: return list(self.page.current_results()) return [] class _FakeDetailPage: def __init__(self, result_pages: dict[str, dict[int, list[str]]], *, rendered_html: str = "rendered") -> None: self.result_pages = result_pages self.rendered_html = rendered_html self.shell_html = "shell" self.waited_functions: list[tuple[str, int | None]] = [] self.goto_urls: list[str] = [] self.current_query = "" self.current_page = 0 self.url = "" self.rendered = False def goto(self, url: str, wait_until: str | None = None) -> None: self.goto_urls.append(url) self.url = url parsed_url = urlparse(url) params = parse_qs(parsed_url.query) if "motsCles" in params: self.current_query = params["motsCles"][0] if "page" in params: self.current_page = int(params["page"][0]) if "/detail-offre/" in parsed_url.path: self.rendered = False def wait_for_load_state(self, state: str) -> None: return None def wait_for_function(self, function: str, polling: int | None = None, timeout: int | None = None) -> None: self.waited_functions.append((function, polling)) self.rendered = True return None def wait_for_selector(self, selector: str, timeout: int | None = None) -> None: if selector == _RESULT_LINK_SELECTOR and self.current_results(): return None raise PlaywrightTimeoutError(f"selector not found: {selector}") def get_by_role(self, role: str, name: str): return _FakeResultButton(self, name) def locator(self, selector: str): return _FakeLocator(self, selector) def content(self) -> str: return self.rendered_html if self.rendered else self.shell_html def current_results(self) -> list[str]: return self.result_pages.get(self.current_query, {}).get(self.current_page, []) @contextmanager def _fake_open_public_page(page: _FakeDetailPage): yield page def test_search_continues_past_duplicate_only_pages(monkeypatch) -> None: first_result = "https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111?motsCles=alpha&page=0&selectedIndex=0" second_result = "https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/222?motsCles=beta&page=1&selectedIndex=0" page = _FakeDetailPage( { "alpha": {0: [first_result], 1: []}, "beta": {0: [first_result], 1: [second_result], 2: []}, } ) monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page)) results = ApecAdapter(max_listings=10).search(["alpha", "beta"]) assert [result.url for result in results] == [first_result, second_result] assert [result.source_job_id for result in results] == ["111", "222"] assert "motsCles=alpha" in page.goto_urls[0] assert "lieux=799" in page.goto_urls[0] assert "typesContrat=101888" in page.goto_urls[0] assert any( "motsCles=beta" in url and "lieux=799" in url and "typesContrat=101888" in url and "page=1" in url for url in page.goto_urls ) assert any("page=1" in url for url in page.goto_urls) def test_fetch_listing_html_waits_for_rendered_offer_content(monkeypatch) -> None: page = _FakeDetailPage({}, rendered_html="rendered offer") monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page)) html = ApecAdapter().fetch_listing_html( "https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111" ) assert html == "rendered offer" assert len(page.waited_functions) == 1 assert ".details-offer-list" in page.waited_functions[0][0] assert "Descriptif du poste" in page.waited_functions[0][0] assert page.waited_functions[0][1] == 1000 def test_fetch_listing_html_uses_explicit_company_fallback_chain(monkeypatch) -> None: page = _FakeDetailPage({}, rendered_html="rendered offer") monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page)) ApecAdapter().fetch_listing_html( "https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111" ) wait_script = page.waited_functions[0][0] assert "companySelectors" in wait_script assert ".card-ents .ents-name" in wait_script assert ".card-ents-quote" in wait_script assert ".details-offer-list li:first-of-type" in wait_script def test_fetch_listing_html_rejects_non_apec_hosts() -> None: adapter = ApecAdapter() with pytest.raises(ValueError): adapter.fetch_listing_html("https://evilapec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111")