"""render(): the browser sidecar door, under the same rules as fetch().
Two sources draw their prices with JavaScript, and the Seoul retailer answers
every product page with a Cloudflare challenge. The rules that keep the
rendered fetch honest are pure logic and are pinned here: the egress
allowlist in the sidecar, the refusal decision on content rather than status,
the floor delay, the per-run budget, and robots before any request.
"""
import re
import pytest
from app.services.collectors import fetch as fetch_module
from app.services.collectors.fetch import (
RENDER_FLOOR_DELAY,
disallow_regexes,
USER_AGENT,
FetchError,
RenderBudgetExhausted,
RenderRefused,
SourceBlocked,
render,
render_budget,
render_refusal,
render_wait,
)
from app.services.collectors.robots import BOT_NAME, parse
from browser.server import egress_allowed, host_permitted, is_json, path_denied
# What Shilla's product pages returned to the text fetch on 2026-09-04, and
# what a browser that cannot pass the challenge draws.
CHALLENGE_HEADERS = {"cf-mitigated": "challenge", "server": "cloudflare"}
CHALLENGE_HTML = "Just a moment...Checking your browser
"
# The Singapore shell as served: 62 KB of markup, 71 characters of text, no price.
SHELL_HTML = "iShopChangi"
PRODUCT_HTML = "Whisky 70cl" + "Whisky 70cl SGD 89.00 in stock " * 8 + ""
class TestEgressAllowlist:
"""The sidecar aborts anything the source did not declare (trackers included)."""
@pytest.mark.parametrize("host", ["www.shilla.example", "cdn.shilla.example"])
def test_origin_and_subdomains_pass(self, host):
assert host_permitted(host, ["shilla.example"])
@pytest.mark.parametrize("host", ["shilla.example.evil.net", "notshilla.example", "google-analytics.com"])
def test_other_hosts_are_aborted(self, host):
assert not host_permitted(host, ["shilla.example"])
@pytest.mark.parametrize(
"host", ["172.17.0.1", "127.0.0.1", "10.0.0.5", "169.254.169.254", "[::1]", "localhost", "db.internal", "printer.local"]
)
def test_addresses_are_never_permitted_even_when_listed(self, host):
assert not host_permitted(host, [host, "shilla.example"])
def test_route_decision_by_scheme(self):
allow = ["shilla.example"]
assert egress_allowed("https://www.shilla.example/p/1", allow)
assert egress_allowed("data:image/png;base64,AAAA", allow)
assert not egress_allowed("https://172.17.0.1:8000/", allow)
assert not egress_allowed("ws://tracker.example/socket", allow)
assert not egress_allowed("ftp://www.shilla.example/x", allow)
@pytest.mark.parametrize(
"pattern, hit, miss",
[
("/medias/", "/medias/sys_master/x.jpg", "/estore/kr/en/p/1"),
("/estore/_ui/", "/estore/_ui/desktop/app.js?v=3", "/estore/kr/en/p/1"),
("/*/search/", "/en/global/search/x", "/en/global/searching"),
("/*?", "/en/p/1?page=2", "/en/p/1"),
("/cart$", "/cart", "/cart/items"),
],
)
def test_the_hosts_disallows_bind_the_pages_own_requests(self, pattern, hit, miss):
"""Measured 2026-09-05: the Seoul host disallows /files/, /medias/ and
/estore/_ui/ for *, which is where a product page's assets live. The
regex sent to the sidecar must agree with the robots matcher."""
robots = parse(f"User-agent: *\nDisallow: {pattern}\n", host="https://h")
deny = [re.compile(rx) for rx in disallow_regexes(robots)]
assert path_denied(f"https://h{hit}", "h", deny) is (not robots.allows(hit))
assert path_denied(f"https://h{miss}", "h", deny) is (not robots.allows(miss))
assert not robots.allows(hit) and robots.allows(miss)
def test_disallows_bind_only_the_origin_host(self):
deny = [re.compile(rx) for rx in disallow_regexes(parse("User-agent: *\nDisallow: /medias/"))]
assert not path_denied("https://cdn.h/medias/x.jpg", "h", deny)
def test_json_detection(self):
assert is_json("application/json; charset=utf-8")
assert is_json("application/vnd.api+json")
assert not is_json("text/html")
class TestRefusalOnContent:
def test_cloudflare_challenge_is_a_refusal(self):
assert render_refusal(403, CHALLENGE_HEADERS, "Just a moment...", CHALLENGE_HTML, 20)
def test_a_challenge_served_with_200_is_still_a_refusal(self):
assert render_refusal(200, {}, "Just a moment...", CHALLENGE_HTML, 300)
def test_an_empty_shell_is_a_refusal_not_a_catalogue(self):
"""A 62 KB shell with 71 characters of text was once a 'successful' fetch."""
assert "empty render" in render_refusal(200, {}, "iShopChangi", SHELL_HTML, 0)
def test_a_drawn_page_passes(self):
assert render_refusal(200, {"server": "nginx"}, "Whisky 70cl", PRODUCT_HTML, 240) is None
def test_a_forbidden_status_with_real_text_is_still_a_refusal(self):
assert render_refusal(403, {}, "Forbidden", PRODUCT_HTML, 240) == "HTTP 403"
class TestFloorDelay:
def test_a_host_without_crawl_delay_gets_the_floor(self):
assert render_wait(1.0, None) == RENDER_FLOOR_DELAY
def test_a_host_with_a_slower_rate_wins(self):
assert render_wait(1.0, 30.0) == 30.0
def test_our_slower_rate_wins_over_theirs(self):
assert render_wait(20.0, 5.0) == 20.0
class TestSourceFloor:
"""A source may declare its own pace (`render_floor_seconds`): the first
Singapore run drew 18 pages at the general floor and was answered 403 on
the 19th, and a slower retry must be the source's declaration, not a flag
someone remembers. Nothing changes for a source that declares none."""
def test_the_source_floor_is_the_slowest_of_ours(self):
assert render_wait(1.0, None, floor=30.0) == 30.0
assert render_wait(1.0, 5.0, floor=30.0) == 30.0 # over the host's published rate too
assert render_wait(40.0, 5.0, floor=30.0) == 40.0 # the run's slower delay still wins
def test_no_source_floor_is_the_old_reading(self):
assert render_wait(1.0, 5.0, floor=None) == render_wait(1.0, 5.0) == 5.0
assert render_wait(1.0, None, floor=0) == RENDER_FLOOR_DELAY
def test_the_rendering_collectors_declare_no_floor_until_rian_decides(self):
"""decide-singapore-answered-403 is open: Singapore's pace stays the
general floor until he answers; Seoul had no refusal."""
from app.services.collectors.changi import Changi
from app.services.collectors.shilla import Shilla
assert Changi.render_floor_seconds is None
assert Shilla.render_floor_seconds is None
def test_a_collector_hands_its_floor_to_render(self, monkeypatch):
from app.services.collectors import changi, shilla
seen: list[dict] = []
def fake_render(url, **kwargs):
seen.append(kwargs)
raise FetchError("stop here")
monkeypatch.setattr(changi, "render", fake_render)
monkeypatch.setattr(shilla, "render", fake_render)
sin, icn = changi.Changi(), shilla.Shilla()
sin.render_floor_seconds = 30.0
for collector, call in ((sin, lambda: sin._render_listings("https://h/p", None, 1.0)),
(icn, lambda: icn._render_page("https://h/p", None, 1.0))):
try:
call()
except FetchError:
pass
assert [k["floor"] for k in seen] == [30.0, None]
assert [k["delay"] for k in seen] == [1.0, 1.0]
class TestBeforeAnyRequest:
"""Robots and the budget are decided before the sidecar is asked anything."""
def test_identity_sent_is_the_identity_honoured(self):
assert BOT_NAME in USER_AGENT
def test_a_disallowed_path_never_reaches_the_sidecar(self, monkeypatch):
monkeypatch.setattr(fetch_module, "_sidecar_render", lambda *a, **k: pytest.fail("rendered"))
robots = parse("User-agent: *\nDisallow: /p/\n", host="https://www.shilla.example")
with pytest.raises(SourceBlocked):
render("https://www.shilla.example/p/1", robots=robots, delay=0)
def test_the_budget_stops_the_run(self, monkeypatch):
calls = []
monkeypatch.setattr(fetch_module, "_sidecar_render", lambda payload, base_url: calls.append(payload) or {
"final_url": payload["url"], "status": 200, "html": PRODUCT_HTML, "text_length": 240, "title": "Whisky 70cl",
})
monkeypatch.setattr(fetch_module, "_last_render_at", {})
monkeypatch.setattr(fetch_module, "RENDER_FLOOR_DELAY", 0.0)
robots = parse("", host="https://www.shilla.example")
with render_budget(2, "test") as budget:
render("https://www.shilla.example/p/1", robots=robots, delay=0, sidecar_url="http://x")
render("https://www.shilla.example/p/2", robots=robots, delay=0, sidecar_url="http://x")
with pytest.raises(RenderBudgetExhausted):
render("https://www.shilla.example/p/3", robots=robots, delay=0, sidecar_url="http://x")
assert budget.count == 2 and len(calls) == 2
assert calls[0]["user_agent"] == USER_AGENT
assert calls[0]["wait_until"] == "load" and calls[0]["deny_regex"] == []
def test_a_refusal_carries_what_arrived(self, monkeypatch):
"""Measured 2026-09-05: the Singapore shell drew 0 characters at load while
the page's own API calls had already answered; a probe needs those."""
monkeypatch.setattr(fetch_module, "_sidecar_render", lambda payload, base_url: {
"final_url": payload["url"], "status": 200, "html": SHELL_HTML, "text_length": 0,
"title": "iShopChangi", "api_responses": [{"url": "https://h/bin/x/products/search", "status": 200, "body": "{\"a\": 1}"}],
})
monkeypatch.setattr(fetch_module, "_last_render_at", {})
monkeypatch.setattr(fetch_module, "RENDER_FLOOR_DELAY", 0.0)
robots = parse("", host="https://h")
with render_budget(1), pytest.raises(RenderRefused) as info:
render("https://h/p/1", robots=robots, delay=0, sidecar_url="http://x")
assert isinstance(info.value, SourceBlocked)
assert list(info.value.rendered.api_json("products/search")) == [{"a": 1}]
def test_no_sidecar_is_an_error_never_a_text_fallback(self, monkeypatch):
monkeypatch.setattr(fetch_module, "_last_render_at", {})
robots = parse("", host="https://www.shilla.example")
with render_budget(5), pytest.raises(FetchError):
render("https://www.shilla.example/p/1", robots=robots, delay=0, sidecar_url="")