352 lines
14 KiB
Python
352 lines
14 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, ApecSearchFilters
|
|
|
|
|
|
_RESULT_LINK_SELECTOR = "a[href*='/candidat/recherche-emploi.html/emploi/detail-offre/']"
|
|
_SEARCH_INPUT_SELECTOR = 'input[name="keywords"]'
|
|
|
|
|
|
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>",
|
|
search_ready: bool = True,
|
|
) -> None:
|
|
self.result_pages = result_pages
|
|
self.rendered_html = rendered_html
|
|
self.shell_html = "<html>shell</html>"
|
|
self.waited_functions: list[tuple[str, int | None]] = []
|
|
self.search_ready = search_ready
|
|
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 == _SEARCH_INPUT_SELECTOR:
|
|
if self.search_ready:
|
|
return None
|
|
|
|
raise PlaywrightTimeoutError(f"selector not found: {selector}")
|
|
|
|
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"],
|
|
search_filters=ApecSearchFilters(location="France", contract_type="CDI"),
|
|
)
|
|
|
|
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_search_continues_past_duplicate_only_pages_until_a_later_hit(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=3&selectedIndex=0"
|
|
page = _FakeDetailPage(
|
|
{
|
|
"alpha": {0: [first_result], 1: [first_result], 2: [first_result], 3: [second_result], 4: []},
|
|
"beta": {0: [first_result], 1: []},
|
|
}
|
|
)
|
|
|
|
monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page))
|
|
|
|
results = ApecAdapter(max_listings=10).search(
|
|
["alpha", "beta"],
|
|
search_filters=ApecSearchFilters(location="France", contract_type="CDI"),
|
|
)
|
|
|
|
assert [result.url for result in results] == [first_result, second_result]
|
|
assert any("page=1" in url for url in page.goto_urls)
|
|
assert any("page=3" in url for url in page.goto_urls)
|
|
|
|
|
|
def test_search_continues_after_query_and_pagination_navigation_failures(monkeypatch) -> None:
|
|
first_result = "https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111?motsCles=beta&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(
|
|
{
|
|
"beta": {0: [first_result], 1: [second_result]},
|
|
}
|
|
)
|
|
|
|
original_goto = page.goto
|
|
goto_calls = 0
|
|
|
|
def flaky_goto(url: str, wait_until: str | None = None) -> None:
|
|
nonlocal goto_calls
|
|
|
|
goto_calls += 1
|
|
if goto_calls == 1:
|
|
raise RuntimeError("navigation boom")
|
|
|
|
original_goto(url, wait_until=wait_until)
|
|
|
|
original_wait_for_load_state = page.wait_for_load_state
|
|
|
|
def flaky_wait_for_load_state(state: str) -> None:
|
|
if page.current_page == 1:
|
|
raise RuntimeError("load boom")
|
|
|
|
original_wait_for_load_state(state)
|
|
|
|
monkeypatch.setattr(page, "goto", flaky_goto)
|
|
monkeypatch.setattr(page, "wait_for_load_state", flaky_wait_for_load_state)
|
|
monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page))
|
|
|
|
results = ApecAdapter(max_listings=10).search(
|
|
["alpha", "beta"],
|
|
search_filters=ApecSearchFilters(location="France", contract_type="CDI"),
|
|
)
|
|
|
|
assert [result.url for result in results] == [first_result]
|
|
assert [result.source_job_id for result in results] == ["111"]
|
|
|
|
|
|
def test_search_stops_after_max_page_count(monkeypatch) -> None:
|
|
page = _FakeDetailPage(
|
|
{
|
|
"alpha": {
|
|
0: ["https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111?motsCles=alpha&page=0&selectedIndex=0"],
|
|
1: ["https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/222?motsCles=alpha&page=1&selectedIndex=0"],
|
|
2: ["https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/333?motsCles=alpha&page=2&selectedIndex=0"],
|
|
3: ["https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/444?motsCles=alpha&page=3&selectedIndex=0"],
|
|
}
|
|
}
|
|
)
|
|
|
|
original_goto = page.goto
|
|
|
|
def bounded_goto(url: str, wait_until: str | None = None) -> None:
|
|
original_goto(url, wait_until=wait_until)
|
|
if page.current_page >= 3:
|
|
raise AssertionError("pagination should stop before page 3")
|
|
|
|
monkeypatch.setattr(page, "goto", bounded_goto)
|
|
monkeypatch.setattr(adapter_module, "_MAX_PAGES_PER_QUERY", 3)
|
|
monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page))
|
|
|
|
results = ApecAdapter(max_listings=10).search(
|
|
["alpha"],
|
|
search_filters=ApecSearchFilters(location="France", contract_type="CDI"),
|
|
)
|
|
|
|
assert [result.source_job_id for result in results] == ["111", "222", "333"]
|
|
assert not any("page=3" in url for url in page.goto_urls)
|
|
|
|
|
|
def test_search_stops_after_consecutive_no_progress_pages(monkeypatch) -> None:
|
|
page = _FakeDetailPage(
|
|
{
|
|
"alpha": {
|
|
0: ["https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111?motsCles=alpha&page=0&selectedIndex=0"],
|
|
1: ["https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111?motsCles=alpha&page=1&selectedIndex=0"],
|
|
2: ["https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111?motsCles=alpha&page=2&selectedIndex=0"],
|
|
3: ["https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/222?motsCles=alpha&page=3&selectedIndex=0"],
|
|
}
|
|
}
|
|
)
|
|
|
|
original_goto = page.goto
|
|
|
|
def bounded_goto(url: str, wait_until: str | None = None) -> None:
|
|
original_goto(url, wait_until=wait_until)
|
|
if page.current_page >= 3:
|
|
raise AssertionError("pagination should stop before page 3")
|
|
|
|
monkeypatch.setattr(page, "goto", bounded_goto)
|
|
monkeypatch.setattr(adapter_module, "_MAX_CONSECUTIVE_NO_PROGRESS_PAGES", 2)
|
|
monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page))
|
|
|
|
results = ApecAdapter(max_listings=10).search(
|
|
["alpha"],
|
|
search_filters=ApecSearchFilters(location="France", contract_type="CDI"),
|
|
)
|
|
|
|
assert [result.source_job_id for result in results] == ["111"]
|
|
assert not any("page=3" in url for url in page.goto_urls)
|
|
|
|
|
|
def test_search_stops_when_result_page_url_repeats(monkeypatch) -> None:
|
|
first_result = "https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111?motsCles=alpha&page=0&selectedIndex=0"
|
|
page = _FakeDetailPage(
|
|
{
|
|
"alpha": {0: [first_result], 1: [first_result], 2: [first_result]},
|
|
}
|
|
)
|
|
|
|
original_goto = page.goto
|
|
initial_result_page_url: str | None = None
|
|
|
|
def looping_goto(url: str, wait_until: str | None = None) -> None:
|
|
nonlocal initial_result_page_url
|
|
|
|
original_goto(url, wait_until=wait_until)
|
|
|
|
if initial_result_page_url is None and page.current_page == 0:
|
|
initial_result_page_url = page.url
|
|
elif initial_result_page_url is not None and page.current_page > 0:
|
|
page.url = initial_result_page_url
|
|
|
|
monkeypatch.setattr(page, "goto", looping_goto)
|
|
monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page))
|
|
|
|
results = ApecAdapter(max_listings=10).search(
|
|
["alpha"],
|
|
search_filters=ApecSearchFilters(location="France", contract_type="CDI"),
|
|
)
|
|
|
|
assert [result.url for result in results] == [first_result]
|
|
assert any("page=1" in url for url in page.goto_urls)
|
|
assert not any("page=2" in url for url in page.goto_urls)
|
|
|
|
|
|
def test_search_raises_when_every_query_fails_to_load_a_search_page(monkeypatch) -> None:
|
|
page = _FakeDetailPage({"alpha": {0: []}}, search_ready=False)
|
|
monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page))
|
|
|
|
with pytest.raises(adapter_module.ApecSearchError):
|
|
ApecAdapter(max_listings=10).search(
|
|
["alpha", "beta"],
|
|
search_filters=ApecSearchFilters(location="France", contract_type="CDI"),
|
|
)
|
|
|
|
|
|
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 ".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="<html>rendered offer</html>")
|
|
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")
|