Spaces:
Sleeping
Sleeping
File size: 5,061 Bytes
9f2df60 3c5b98e 9f2df60 3c5b98e 9f2df60 3c5b98e 9f2df60 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 | import pytest, os
from datetime import datetime, timedelta
from types import SimpleNamespace
from src.scraping.types import FetchResult, ScrapingStatus, UrlTimestamps
from src.scraping.url_normalizer import UrlNormalizer
from src.scraping.scraper import Scraper
from src.utils.logging import init_logging
from src.config import config
init_logging()
class TestScrapingErrors:
def _invalid_url_scrape_test(self, invalid_url):
url_filename = UrlNormalizer().url_to_filename(invalid_url)
url_html_file_path = os.path.join(config.paths.RAW_HTML_OUTPUT, url_filename + '.html')
url_txt_file_path = os.path.join(config.paths.EXTRACTED_TEXT_OUTPUT, url_filename + '.txt')
scraper = Scraper()
result = scraper.scrape_target(invalid_url)
assert not result
assert not os.path.exists(url_html_file_path)
assert not os.path.exists(url_txt_file_path)
@pytest.mark.network
@pytest.mark.integration
def test_invalid_target_url(self, caplog):
invalid_url = 'https://gugel.pl'
self._invalid_url_scrape_test(invalid_url)
assert f"Unaccessible target URL '{invalid_url}'" in caplog.text
@pytest.mark.network
@pytest.mark.integration
def test_target_url_with_no_robots(self, caplog):
invalid_url = 'https://emba.unisg.ch/programm/emba'
self._invalid_url_scrape_test(invalid_url)
assert f"Could not fetch the 'robots.txt' file for the target URL '{invalid_url}'!" in caplog.text
def test_last_scraped_priority(self, caplog):
scraper = Scraper()
scraper._url_priorities = {"high": ["h1", "h2", "h3"], "medium": ["m1", "m2", "m3"], "low": ["l1", "l2", "l3"]}
date = datetime.now()
scraper._url_timestamps = {
"h1": UrlTimestamps(last_scraped=date - timedelta(days=1)),
"h2": UrlTimestamps(last_scraped=date - timedelta(days=2)),
"h3": UrlTimestamps(last_scraped=date - timedelta(hours=23)),
"m1": UrlTimestamps(last_scraped=date - timedelta(days=1)),
"m2": UrlTimestamps(last_scraped=date - timedelta(days=6)),
"m3": UrlTimestamps(last_scraped=date - timedelta(days=7)),
"l1": UrlTimestamps(last_scraped=date - timedelta(days=7)),
"l2": UrlTimestamps(last_scraped=date - timedelta(days=29)),
"l3": UrlTimestamps(last_scraped=date - timedelta(days=30)),
}
expected = {
"h1": True,
"h2": True,
"h3": False,
"m1": False,
"m2": False,
"m3": True,
"l1": False,
"l2": False,
"l3": True,
}
for url in scraper._url_timestamps:
assert scraper._is_url_prioritized(url) == expected[url]
def test_head_timeout_falls_back_to_get(self, tmp_path, monkeypatch):
url = "https://embax.ch/contact/"
get_calls = []
scraper = Scraper.__new__(Scraper)
scraper._scrape_all = True
scraper._url_timestamps = {}
scraper._url_priorities = {}
scraper._normalizer = SimpleNamespace(
is_url_blacklisted=lambda _: False,
url_to_filename=lambda _: "embax-contact",
)
scraper._content_cleaner = SimpleNamespace(
clean_mobile_content=lambda html: html,
extract_urls=lambda document: [],
collect_repetitive_content=lambda document: None,
)
scraper._processor = SimpleNamespace(
process=lambda final_url, html: SimpleNamespace(name=final_url),
convert_to_txt=lambda document: "contact page",
)
def fail_head(*_):
raise TimeoutError("HEAD timed out")
def fetch_get(request_url, etag):
get_calls.append((request_url, etag))
return FetchResult(
final_url=request_url,
last_modified=None,
etag=None,
text="<html><body>Contact</body></html>",
page_hash="hash",
)
def no_wait_backoff(func, args=(), **_):
return {
"result": func(*args),
"retries": 0,
"last_error": None,
"status": "OK",
}
monkeypatch.setattr("src.scraping.scraper.fetch_head", fail_head)
monkeypatch.setattr("src.scraping.scraper.fetch_url", fetch_get)
monkeypatch.setattr(
"src.scraping.scraper.call_with_exponential_backoff",
no_wait_backoff,
)
monkeypatch.setattr(config.paths, "RAW_HTML_OUTPUT", str(tmp_path))
monkeypatch.setattr(config.paths, "RAW_TEXT_OUTPUT", str(tmp_path))
result = scraper._scrape_page(
url=url,
crawl_delay=0,
visited_urls=set(),
)
assert result.status == ScrapingStatus.OK
assert result.final_url == url
assert get_calls == [(url, None)]
if __name__ == "__main__":
pytest.main([__file__, '-v'])
|