154 lines
5.3 KiB
Python
154 lines
5.3 KiB
Python
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 = "<html>rendered</html>") -> None:
|
|
self.result_pages = result_pages
|
|
self.rendered_html = rendered_html
|
|
self.shell_html = "<html>shell</html>"
|
|
self.waited_selectors: list[str] = []
|
|
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:
|
|
self.waited_selectors.append(selector)
|
|
|
|
if selector in {
|
|
".card-ents .ents-name",
|
|
".card-offer .ref-offre",
|
|
".details-offer-list",
|
|
".details-post",
|
|
".details-post:has-text('Descriptif du poste')",
|
|
}:
|
|
return 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 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="<html>rendered offer</html>")
|
|
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 == "<html>rendered offer</html>"
|
|
assert len(page.waited_functions) == 1
|
|
assert ".card-ents .ents-name" in page.waited_functions[0][0]
|
|
assert page.waited_functions[0][1] == 1000
|
|
assert page.waited_selectors == [
|
|
".details-post:has-text('Descriptif du poste')",
|
|
]
|
|
|
|
|
|
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")
|