678 lines
26 KiB
Python
678 lines
26 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
|
|
elif self.name in {"ACCEPTER", "Accepter tous les cookies"}:
|
|
self.page.consent_button_clicks.append(self.name)
|
|
self.page.consent_accepted = True
|
|
|
|
|
|
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 check(self, timeout: int | None = None) -> None:
|
|
if self.selector == 'input[name="cguAcceptees"]':
|
|
self.page.cgu_checkbox_checked = True
|
|
return None
|
|
|
|
raise PlaywrightTimeoutError(f"selector not found: {self.selector}")
|
|
|
|
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,
|
|
zero_result_queries: set[str] | None = None,
|
|
consent_required: bool = False,
|
|
) -> 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.zero_result_queries = zero_result_queries or set()
|
|
self.consent_required = consent_required
|
|
self.cgu_checkbox_checked = False
|
|
self.consent_button_clicks: list[str] = []
|
|
self.consent_accepted = not consent_required
|
|
self.goto_urls: list[str] = []
|
|
self.current_query = ""
|
|
self.current_page = 0
|
|
self.url = ""
|
|
self.rendered = False
|
|
self.default_timeout: int | None = None
|
|
self.closed = 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 self.current_query in self.zero_result_queries and "/detail-offre/" not in parsed_url.path:
|
|
self.url = (
|
|
f"{parsed_url.scheme}://{parsed_url.netloc}"
|
|
f"{parsed_url.path}/recherche-avancee?{parsed_url.query}&error=true"
|
|
)
|
|
|
|
if "/detail-offre/" in parsed_url.path:
|
|
self.rendered = False
|
|
|
|
def wait_for_load_state(self, state: str) -> None:
|
|
return None
|
|
|
|
def set_default_timeout(self, timeout: int) -> None:
|
|
self.default_timeout = timeout
|
|
|
|
def wait_for_function(self, function: str, polling: int | None = None, timeout: int | None = None) -> None:
|
|
if self.consent_required and not self.consent_accepted:
|
|
raise PlaywrightTimeoutError("consent not accepted")
|
|
|
|
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 and self.consent_accepted:
|
|
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, [])
|
|
|
|
def close(self) -> None:
|
|
self.closed = True
|
|
|
|
|
|
@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_search_treats_zero_results_redirect_as_usable_and_records_other_failures(monkeypatch) -> None:
|
|
page = _FakeDetailPage(
|
|
{"alpha": {0: []}, "beta": {0: []}},
|
|
zero_result_queries={"alpha"},
|
|
)
|
|
monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page))
|
|
|
|
adapter = ApecAdapter(max_listings=10)
|
|
results = adapter.search(
|
|
["alpha", "beta"],
|
|
search_filters=ApecSearchFilters(location="France", contract_type="CDI"),
|
|
)
|
|
|
|
assert results == []
|
|
assert [error.stage for error in adapter.search_errors] == ["search"]
|
|
assert "beta" in adapter.search_errors[0].url
|
|
|
|
|
|
def test_search_raises_when_every_query_renders_broken_search_shell(monkeypatch) -> None:
|
|
page = _FakeDetailPage({"alpha": {0: []}, "beta": {0: []}}, search_ready=True)
|
|
monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page))
|
|
|
|
adapter = ApecAdapter(max_listings=10)
|
|
|
|
with pytest.raises(adapter_module.ApecSearchError):
|
|
adapter.search(
|
|
["alpha", "beta"],
|
|
search_filters=ApecSearchFilters(location="France", contract_type="CDI"),
|
|
)
|
|
|
|
assert [error.stage for error in adapter.search_errors] == ["search", "search"]
|
|
|
|
|
|
def test_search_accepts_current_cgu_popin_before_waiting_for_results(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]}}, consent_required=True)
|
|
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 page.cgu_checkbox_checked is True
|
|
assert page.consent_button_clicks == ["ACCEPTER"]
|
|
assert page.consent_accepted is True
|
|
|
|
|
|
def test_search_records_pagination_navigation_failures(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: ["https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/222?motsCles=alpha&page=1&selectedIndex=0"],
|
|
},
|
|
}
|
|
)
|
|
|
|
original_goto = page.goto
|
|
|
|
def flaky_goto(url: str, wait_until: str | None = None) -> None:
|
|
if "page=1" in url:
|
|
raise RuntimeError("navigation boom")
|
|
|
|
original_goto(url, wait_until=wait_until)
|
|
|
|
monkeypatch.setattr(page, "goto", flaky_goto)
|
|
monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page))
|
|
|
|
adapter = ApecAdapter(max_listings=10)
|
|
results = adapter.search(
|
|
["alpha"],
|
|
search_filters=ApecSearchFilters(location="France", contract_type="CDI"),
|
|
)
|
|
|
|
assert [result.url for result in results] == [first_result]
|
|
assert [error.stage for error in adapter.search_errors] == ["search"]
|
|
assert "page=1" in adapter.search_errors[0].url
|
|
assert adapter.search_errors[0].message == "page 1 navigation failed"
|
|
|
|
|
|
def test_search_records_pagination_render_failures(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=alpha&page=1&selectedIndex=0"
|
|
page = _FakeDetailPage(
|
|
{
|
|
"alpha": {
|
|
0: [first_result],
|
|
1: [second_result],
|
|
},
|
|
}
|
|
)
|
|
|
|
original_wait_for_selector = page.wait_for_selector
|
|
|
|
def flaky_wait_for_selector(selector: str, timeout: int | None = None) -> None:
|
|
if selector == _RESULT_LINK_SELECTOR and page.current_page == 1:
|
|
raise PlaywrightTimeoutError(f"selector not found: {selector}")
|
|
|
|
original_wait_for_selector(selector, timeout=timeout)
|
|
|
|
monkeypatch.setattr(page, "wait_for_selector", flaky_wait_for_selector)
|
|
monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page))
|
|
|
|
adapter = ApecAdapter(max_listings=10)
|
|
results = adapter.search(
|
|
["alpha"],
|
|
search_filters=ApecSearchFilters(location="France", contract_type="CDI"),
|
|
)
|
|
|
|
assert [result.url for result in results] == [first_result]
|
|
assert [error.stage for error in adapter.search_errors] == ["search"]
|
|
assert "page=1" in adapter.search_errors[0].url
|
|
assert adapter.search_errors[0].message == "page 1 results did not render"
|
|
|
|
|
|
def test_search_records_evaluate_all_failures_and_continues_to_next_query(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=0&selectedIndex=0"
|
|
page = _FakeDetailPage(
|
|
{
|
|
"alpha": {
|
|
0: [first_result],
|
|
1: [first_result],
|
|
},
|
|
"beta": {
|
|
0: [second_result],
|
|
1: [second_result],
|
|
},
|
|
}
|
|
)
|
|
|
|
original_locator = page.locator
|
|
|
|
class _FlakyLocator:
|
|
def __init__(self, locator) -> None:
|
|
self._locator = locator
|
|
|
|
def evaluate_all(self, function: str):
|
|
if (
|
|
self._locator.selector == _RESULT_LINK_SELECTOR
|
|
and page.current_query == "alpha"
|
|
and page.current_page == 1
|
|
):
|
|
raise RuntimeError("evaluate boom")
|
|
|
|
return self._locator.evaluate_all(function)
|
|
|
|
def __getattr__(self, name: str):
|
|
return getattr(self._locator, name)
|
|
|
|
def flaky_locator(selector: str):
|
|
return _FlakyLocator(original_locator(selector))
|
|
|
|
monkeypatch.setattr(page, "locator", flaky_locator)
|
|
monkeypatch.setattr(adapter_module, "_MAX_CONSECUTIVE_NO_PROGRESS_PAGES", 1)
|
|
monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page))
|
|
|
|
adapter = ApecAdapter(max_listings=10)
|
|
results = adapter.search(
|
|
["alpha", "beta"],
|
|
search_filters=ApecSearchFilters(location="France", contract_type="CDI"),
|
|
)
|
|
|
|
assert [result.url for result in results] == [first_result, second_result]
|
|
assert [error.stage for error in adapter.search_errors] == ["search"]
|
|
assert "page=1" in adapter.search_errors[0].url
|
|
|
|
|
|
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 ".container-details-offer h1" in page.waited_functions[0][0]
|
|
assert ".ref-offre" in page.waited_functions[0][0]
|
|
assert ".details-offer-list" in page.waited_functions[0][0]
|
|
assert "Descriptif du poste" not in page.waited_functions[0][0]
|
|
assert page.waited_functions[0][1] == 1000
|
|
|
|
|
|
def test_fetch_listing_html_accepts_current_cgu_popin_before_waiting_for_detail_content(monkeypatch) -> None:
|
|
page = _FakeDetailPage({}, rendered_html="<html>rendered offer</html>", consent_required=True)
|
|
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 page.cgu_checkbox_checked is True
|
|
assert page.consent_button_clicks == ["ACCEPTER"]
|
|
assert page.consent_accepted is True
|
|
|
|
|
|
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" not in wait_script
|
|
assert ".container-details-offer h1" in wait_script
|
|
assert ".ref-offre" in wait_script
|
|
assert ".details-offer-list" in wait_script
|
|
|
|
|
|
def test_fetch_listing_html_rejects_redirected_non_apec_urls(monkeypatch) -> None:
|
|
page = _FakeDetailPage({}, rendered_html="<html>redirected</html>")
|
|
|
|
original_goto = page.goto
|
|
|
|
def redirecting_goto(url: str, wait_until: str | None = None) -> None:
|
|
original_goto(url, wait_until=wait_until)
|
|
page.url = "https://www.apec.fr/candidat/recherche-emploi.html/emploi/recherche-avancee?error=true"
|
|
|
|
monkeypatch.setattr(page, "goto", redirecting_goto)
|
|
monkeypatch.setattr(adapter_module, "_open_public_page", lambda: _fake_open_public_page(page))
|
|
|
|
with pytest.raises(ValueError, match="unexpected URL after redirects"):
|
|
ApecAdapter().fetch_listing_html(
|
|
"https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111"
|
|
)
|
|
|
|
assert len(page.waited_functions) == 1
|
|
assert page.goto_urls == ["https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111"]
|
|
|
|
|
|
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")
|
|
|
|
|
|
def test_fetch_listing_html_reuses_browser_context_across_calls(monkeypatch) -> None:
|
|
class FakePage:
|
|
def __init__(self) -> None:
|
|
self.goto_urls: list[str] = []
|
|
self.default_timeout: int | None = None
|
|
self.url = ""
|
|
|
|
def set_default_timeout(self, timeout: int) -> None:
|
|
self.default_timeout = timeout
|
|
|
|
def goto(self, url: str, wait_until: str | None = None) -> None:
|
|
self.goto_urls.append(url)
|
|
self.url = url
|
|
|
|
def wait_for_function(self, function: str, polling: int | None = None, timeout: int | None = None) -> None:
|
|
return None
|
|
|
|
def content(self) -> str:
|
|
return "<html>shared</html>"
|
|
|
|
def close(self) -> None:
|
|
return None
|
|
|
|
class FakeBrowserContext:
|
|
def __init__(self) -> None:
|
|
self.new_page_calls = 0
|
|
|
|
def new_page(self) -> FakePage:
|
|
self.new_page_calls += 1
|
|
return FakePage()
|
|
|
|
def close(self) -> None:
|
|
return None
|
|
|
|
class FakeBrowser:
|
|
def __init__(self, browser_context: FakeBrowserContext) -> None:
|
|
self.browser_context = browser_context
|
|
|
|
def new_context(self) -> FakeBrowserContext:
|
|
return self.browser_context
|
|
|
|
def close(self) -> None:
|
|
return None
|
|
|
|
class FakeChromium:
|
|
def __init__(self, browser: FakeBrowser) -> None:
|
|
self.browser = browser
|
|
self.launch_calls = 0
|
|
|
|
def launch(self, headless: bool = True) -> FakeBrowser:
|
|
self.launch_calls += 1
|
|
return self.browser
|
|
|
|
class FakePlaywright:
|
|
def __init__(self, chromium: FakeChromium) -> None:
|
|
self.chromium = chromium
|
|
|
|
class FakePlaywrightManager:
|
|
def __init__(self, chromium: FakeChromium) -> None:
|
|
self.playwright = FakePlaywright(chromium)
|
|
|
|
def __enter__(self) -> FakePlaywright:
|
|
return self.playwright
|
|
|
|
def __exit__(self, exc_type, exc, tb) -> None:
|
|
return None
|
|
|
|
browser_context = FakeBrowserContext()
|
|
browser = FakeBrowser(browser_context)
|
|
chromium = FakeChromium(browser)
|
|
|
|
monkeypatch.setattr(adapter_module, "sync_playwright", lambda: FakePlaywrightManager(chromium))
|
|
|
|
adapter = ApecAdapter()
|
|
with adapter.browser_session():
|
|
html_one = adapter.fetch_listing_html("https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/111")
|
|
html_two = adapter.fetch_listing_html("https://www.apec.fr/candidat/recherche-emploi.html/emploi/detail-offre/222")
|
|
|
|
assert html_one == "<html>shared</html>"
|
|
assert html_two == "<html>shared</html>"
|
|
assert chromium.launch_calls == 1
|
|
assert browser_context.new_page_calls == 2
|