107 lines
4.9 KiB
Python
107 lines
4.9 KiB
Python
import json
|
|
from datetime import datetime
|
|
|
|
import pytest
|
|
|
|
from app.strategies import (
|
|
_events_from_html,
|
|
_events_from_json,
|
|
execute_strategy,
|
|
json_path,
|
|
parse_strategy,
|
|
)
|
|
|
|
|
|
def test_parse_strategy_deep_copies_dict():
|
|
original = {"mode": "html", "body": {"page": 1}}
|
|
parsed = parse_strategy(original)
|
|
parsed["body"]["page"] = 2
|
|
assert original["body"]["page"] == 1
|
|
assert parse_strategy(json.dumps(original)) == original
|
|
assert parse_strategy("not json") == {}
|
|
|
|
|
|
def test_json_path_handles_dicts_lists_and_defaults():
|
|
value = {"results": [{"title": "One"}]}
|
|
assert json_path(value, "results.0.title") == "One"
|
|
assert json_path(value, "results.2.title", "missing") == "missing"
|
|
assert json_path(value, "") is value
|
|
|
|
|
|
def test_html_strategy_extracts_text_attributes_and_links(source, html_cards):
|
|
strategy = {
|
|
"item_selector": ".event",
|
|
"title_selector": ".title",
|
|
"date_selector": "time",
|
|
"date_attribute": "datetime",
|
|
"link_selector": ".title",
|
|
"link_attribute": "href",
|
|
"description_selector": ".description",
|
|
"location_selector": ".venue",
|
|
}
|
|
events = _events_from_html(html_cards, strategy, source)
|
|
assert [event.title for event in events] == ["Event One", "Event Two"]
|
|
assert events[0].starts_at.hour == 19
|
|
assert events[0].url == "https://example.org/events/one"
|
|
assert events[0].description == "First description"
|
|
assert events[0].address == "Town Hall"
|
|
|
|
|
|
def test_html_scope_attribute_and_duration(source):
|
|
html = '<article class="event" data-date="2026-09-07T19:00:00+10:00"><h2>One</h2></article>'
|
|
strategy = {"item_selector": ".event", "title_selector": "h2", "date_selector": ":scope", "date_attribute": "data-date", "duration_minutes": 90}
|
|
event = _events_from_html(html, strategy, source)[0]
|
|
assert (event.ends_at - event.starts_at).total_seconds() == 5400
|
|
|
|
|
|
def test_html_range_populates_end(source):
|
|
html = '<article class="event"><h2>Festival</h2><time>2-13 Sep 2026</time></article>'
|
|
event = _events_from_html(html, {"item_selector": ".event", "title_selector": "h2", "date_selector": "time"}, source)[0]
|
|
assert event.starts_at.day == 2
|
|
assert event.ends_at.day == 13
|
|
|
|
|
|
def test_json_strategy_extracts_nested_fields(source):
|
|
payload = {"results": [{"title": "Talk", "start": "2026-09-10T18:00:00+10:00", "end": "2026-09-10T20:00:00+10:00", "venue": {"address": "Library"}, "url": "/talk"}]}
|
|
strategy = {"items_path": "results", "title_path": "title", "date_path": "start", "end_path": "end", "location_path": "venue.address", "link_path": "url"}
|
|
event = _events_from_json(payload, strategy, source)[0]
|
|
assert event.title == "Talk"
|
|
assert event.ends_at.hour == 20
|
|
assert event.address == "Library"
|
|
|
|
|
|
def test_json_join_supplies_metadata_and_duration(source):
|
|
payload = [{"movieId": "M1", "utcDate": "2026-09-11T10:00:00+00:00", "link": "/book"}]
|
|
strategy = {"items_path": "", "date_path": "utcDate", "link_path": "link", "join_item_key": "movieId", "join_title_path": "name", "join_description_path": "summary", "join_duration_path": "duration"}
|
|
event = _events_from_json(payload, strategy, source, {"M1": {"name": "Movie", "summary": "Plot", "duration": 120}})[0]
|
|
assert event.title == "Movie"
|
|
assert event.description == "Plot"
|
|
assert (event.ends_at - event.starts_at).total_seconds() == 7200
|
|
|
|
|
|
def test_disabled_json_item_is_skipped(source):
|
|
payload = [{"title": "Hidden", "start": "2026-09-10", "disabled": True}]
|
|
strategy = {"items_path": "", "title_path": "title", "date_path": "start", "disabled_path": "disabled"}
|
|
assert _events_from_json(payload, strategy, source) == []
|
|
|
|
|
|
def test_html_fragment_paginates(source, fake_response, recording_client, monkeypatch):
|
|
import app.strategies as module
|
|
monkeypatch.setattr(module, "validate_public_url", lambda url: None)
|
|
responses = [
|
|
fake_response(payload={"html": '<article class="event"><h2>One</h2><time>7 Sep 2026</time></article>', "pages": 2}),
|
|
fake_response(payload={"html": '<article class="event"><h2>Two</h2><time>8 Sep 2026</time></article>', "pages": 2}),
|
|
]
|
|
client = recording_client(responses)
|
|
source = {**source, "strategy_json": json.dumps({"mode": "html_fragment", "request_url": "https://example.org/ajax", "method": "POST", "body": {"action": "events"}, "html_path": "html", "item_selector": ".event", "title_selector": "h2", "date_selector": "time", "page_parameter": "page", "max_pages": 5, "total_pages_path": "pages"})}
|
|
events = execute_strategy(source, client)
|
|
assert [event.title for event in events] == ["One", "Two"]
|
|
assert client.calls[0][2]["data"]["page"] == "1"
|
|
assert client.calls[1][2]["data"]["page"] == "2"
|
|
|
|
|
|
def test_unsupported_mode_raises(source, recording_client):
|
|
source = {**source, "strategy_json": '{"mode":"magic"}'}
|
|
with pytest.raises(Exception, match="Unsupported extraction mode"):
|
|
execute_strategy(source, recording_client([]))
|