From 08bdc43657db4aaddb6dd49cb553f4c9242d03ff Mon Sep 17 00:00:00 2001 From: Adrien Barbaresi Date: Fri, 31 Jul 2026 17:52:57 +0200 Subject: [PATCH 1/8] fix: consolidate code and ensure consistency --- courlan/clean.py | 11 ++++++-- courlan/filters.py | 2 ++ courlan/urlstore.py | 56 ++++++++++++++++++++++++------------- docs/source/api/urlstore.md | 5 ++++ pyproject.toml | 5 ++-- tests/unit_tests.py | 25 +++++++++++++++-- tests/urlstore_tests.py | 33 +++++++++++++++++++++- 7 files changed, 109 insertions(+), 28 deletions(-) diff --git a/courlan/clean.py b/courlan/clean.py index 2664a97..ea72b6e 100644 --- a/courlan/clean.py +++ b/courlan/clean.py @@ -8,7 +8,7 @@ from .filters import is_valid_url from .settings import ALLOWED_PARAMS, LANG_PARAMS, TARGET_LANGS -from .urlutils import _parse +from .urlutils import _parse, _strip_trailing_dot LOGGER = logging.getLogger(__name__) @@ -43,7 +43,9 @@ def clean_url(url: str, language: str | None = None) -> str | None: "Helper function: chained scrubbing and normalization" try: - return normalize_url(scrub_url(url), False, language, False) + cleaned = normalize_url(scrub_url(url), False, language) + # align root URLs with scrub_url's form so a second pass is a no-op + return cleaned.rstrip("/") if cleaned.count("/") == 3 else cleaned except (AttributeError, ValueError): return None @@ -163,7 +165,10 @@ def normalize_fragment(fragment: str, language: str | None = None) -> str: def normalize_authority(parsed_url: SplitResult) -> str: "Lower-case the authority, decode punycode and strip the scheme default port." - netloc = decode_punycode(parsed_url.netloc.lower()) + # host only: preserve userinfo case, strip a trailing FQDN dot + userinfo, _, hostport = parsed_url.netloc.rpartition("@") + hostport = decode_punycode(_strip_trailing_dot(hostport.lower())) + netloc = f"{userinfo}@{hostport}" if userinfo else hostport # port: strip only the scheme's default port (80 for http, 443 for https) try: port = parsed_url.port diff --git a/courlan/filters.py b/courlan/filters.py index 0ae4d6a..5da1c16 100644 --- a/courlan/filters.py +++ b/courlan/filters.py @@ -226,6 +226,8 @@ def type_filter(url: str, strict: bool = False, with_nav: bool = False) -> bool: def validate_url(url: str | None) -> tuple[bool, SplitResult | None]: "Parse and validate the input." + if not isinstance(url, str): + return False, None try: parsed_url = urlsplit(url) except ValueError: diff --git a/courlan/urlstore.py b/courlan/urlstore.py index 565104e..bbbada8 100644 --- a/courlan/urlstore.py +++ b/courlan/urlstore.py @@ -152,8 +152,13 @@ def dump_unvisited_urls(num: Any, frame: Any) -> None: # don't use the following on Windows if verbose and not sys.platform.startswith("win"): try: - signal.signal(signal.SIGINT, dump_unvisited_urls) - signal.signal(signal.SIGTERM, dump_unvisited_urls) + for signum in (signal.SIGINT, signal.SIGTERM): + # don't overwrite handlers the host application installed + if signal.getsignal(signum) in ( + signal.SIG_DFL, + signal.default_int_handler, + ): + signal.signal(signum, dump_unvisited_urls) except ValueError: # signal handlers can only be registered in the main thread LOGGER.warning("Cannot set signal handlers outside the main thread") @@ -215,20 +220,13 @@ def _set_done(self) -> None: with self._lock: self.done = True - def _store_urls( - self, - domain: str, - to_right: deque[UrlPathTuple] | None = None, - timestamp: datetime | None = None, - to_left: deque[UrlPathTuple] | None = None, - replace: bool = False, - ) -> None: - # http/https switch + def _canonical_domain(self, domain: str) -> str: + "Merge the http/https twin of the domain, keeping a single entry." if domain.startswith("http://"): candidate = "https" + domain[4:] # switch if candidate in self.urldict: - domain = candidate + return candidate elif domain.startswith("https://"): candidate = "http" + domain[5:] # replace entry: check-and-swap must be atomic against other writers @@ -236,6 +234,17 @@ def _store_urls( if candidate in self.urldict: self.urldict[domain] = self.urldict[candidate] del self.urldict[candidate] + return domain + + def _store_urls( + self, + domain: str, + to_right: deque[UrlPathTuple] | None = None, + timestamp: datetime | None = None, + to_left: deque[UrlPathTuple] | None = None, + replace: bool = False, + ) -> None: + domain = self._canonical_domain(domain) # load URLs or create entry if domain in self.urldict and self.urldict[domain].state is State.BUSTED: @@ -250,12 +259,17 @@ def _store_urls( urls = deque() known = set() + # update "known" while extending so in-batch variants dedup too if to_right is not None: - urls.extend(t for t in to_right if not is_known_link(t.path(), known)) + for t in to_right: + if not is_known_link(t.path(), known): + urls.append(t) + known.add(t.path()) if to_left is not None: - urls.extendleft( - t for t in to_left if not is_known_link(t.path(), known) - ) + for t in to_left: + if not is_known_link(t.path(), known): + urls.appendleft(t) + known.add(t.path()) with self._lock: if self.compressed: @@ -508,6 +522,7 @@ def establish_download_schedule( def store_rules(self, website: str, rules: RobotFileParser | None) -> None: "Store crawling rules for a given website." + website = self._canonical_domain(website) if self.compressed: rules = COMPRESSOR.compress(rules) self.urldict[website].rules = rules @@ -540,7 +555,7 @@ def total_url_number(self) -> int: return sum(v.total for v in self.urldict.values()) def download_threshold_reached(self, threshold: float) -> bool: - "Find out if the download limit (in seconds) has been reached for one of the websites in store." + "Find out if the download threshold (number of retrieved URLs) has been reached for one of the websites in store." return any(v.count >= threshold for v in self.urldict.values()) def dump_urls(self) -> list[str]: @@ -571,13 +586,16 @@ def print_urls(self) -> None: # PERSISTANCE def write(self, filename: str) -> None: - "Write the URL store to disk." + "Write the URL store to disk as a pickle file, see load_store()." with open(filename, "wb") as output: pickle.dump(self, output) def load_store(filename: str) -> UrlStore: - "Load a URL store from disk." + """Load a URL store from disk. + + Warning: uses pickle, which can execute arbitrary code. + Only load files you have written yourself with UrlStore.write().""" with open(filename, "rb") as output: url_store = pickle.load(output) return url_store diff --git a/docs/source/api/urlstore.md b/docs/source/api/urlstore.md index 98bbf82..ebe6976 100644 --- a/docs/source/api/urlstore.md +++ b/docs/source/api/urlstore.md @@ -53,6 +53,11 @@ print(f"Total URLs: {store.total_url_number()}") print(f"Unvisited domains: {store.get_unvisited_domains()}") ``` +```{warning} +`write()`/`load_store()` use Python's pickle format, which can execute +arbitrary code when loading. Only load files you have written yourself. +``` + ### Statistics and reporting ```python diff --git a/pyproject.toml b/pyproject.toml index 6ddb491..6ee11ca 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -92,8 +92,9 @@ testpaths = "tests/*test*.py" target-version = "py310" [tool.ruff.lint] -# ruff defaults (E4, E7, E9, F) plus import sorting (I), -# pyupgrade (UP) and bugbear (B) +# pin the historical defaults instead of tracking ruff's own, +# plus import sorting (I), pyupgrade (UP) and bugbear (B) +select = ["E4", "E7", "E9", "F"] extend-select = ["I", "UP", "B"] [tool.mypy] diff --git a/tests/unit_tests.py b/tests/unit_tests.py index 16f04c2..cc95bc7 100644 --- a/tests/unit_tests.py +++ b/tests/unit_tests.py @@ -203,6 +203,8 @@ def test_scrub(): assert clean_url(cleaned) == cleaned # a surviving (non-tracker) query still keeps the root slash assert clean_url("http://test.org/?p=1") == "http://test.org/?p=1" + # deeper paths keep their trailing slash (only the root form is collapsed) + assert clean_url("https://example.org/path/") == "https://example.org/path/" # scrub assert scrub_url(" https://www.dwds.de") == "https://www.dwds.de" assert scrub_url("") == "https://www.dwds.de" @@ -595,6 +597,10 @@ def test_validate(): assert not is_valid_url("http://www.test[.org/test") assert is_valid_url("http://test.org/test") + # non-string input returns False instead of raising + assert not is_valid_url(None) + assert not is_valid_url(123) + assert not is_valid_url(b"http://test.org/test") # verdict no longer flips on port/userinfo/case; short valid domains accepted assert is_valid_url("http://t.co/") @@ -644,6 +650,19 @@ def test_normalization(): # non-default port preserved assert normalize_url("http://[::1]:8080/") == "http://[::1]:8080/" + # host is lowercased, userinfo case is preserved (credentials are case-sensitive) + assert ( + normalize_url("https://User:PassWord@Example.COM/Path") + == "https://User:PassWord@example.com/Path" + ) + # empty userinfo is dropped + assert normalize_url("http://@example.com/x") == "http://example.com/x" + # trailing FQDN dot is stripped from the host + assert normalize_url("http://example.com./page") == "http://example.com/page" + assert ( + normalize_url("http://example.com.:8080/page") == "http://example.com:8080/page" + ) + # punycode assert normalize_url("http://xn--Mnchen-3ya.de") == "http://münchen.de" assert normalize_url("http://Mnchen-3ya.de") == "http://mnchen-3ya.de" @@ -861,13 +880,13 @@ def test_urlcheck_domain(): # bare suffix has no registrable domain, though domain_filter alone accepts it assert check_url("https://a.ck/page") is None assert check_url("https://co.uk/page") is None - # trailing-dot FQDN URLs are valid; domain matches extract_domain stripping + # trailing-dot FQDN URLs are valid and normalized to the dotless form assert check_url("http://example.com./page") == ( - "http://example.com./page", + "http://example.com/page", "example.com", ) assert check_url("http://192.168.0.1./x") == ( - "http://192.168.0.1./x", + "http://192.168.0.1/x", "192.168.0.1", ) assert check_url("http://example.com../page") is None diff --git a/tests/urlstore_tests.py b/tests/urlstore_tests.py index 9ae8257..e811476 100644 --- a/tests/urlstore_tests.py +++ b/tests/urlstore_tests.py @@ -88,6 +88,15 @@ def test_urlstore_basics(): assert len(my_urls.urldict) == 1 and "http://example.org" not in my_urls.urldict assert len(my_urls.urldict["https://example.org"].tuples) == 2 + # slash variants in the same batch dedup like across batches + my_urls.reset() + my_urls.add_urls(["https://example.org/a", "https://example.org/a/"]) + assert len(my_urls.urldict["https://example.org"].tuples) == 1 + # same for the appendleft path + my_urls.reset() + my_urls.add_urls(appendleft=["https://example.org/b", "https://example.org/b/"]) + assert len(my_urls.urldict["https://example.org"].tuples) == 1 + def test_urlstore_rules(robots_rules): "Test storage and retrieval of crawling rules." @@ -115,6 +124,13 @@ def test_urlstore_rules(robots_rules): assert my_urls.get_rules("https://example.org").mtime() == robots_rules.mtime() my_urls.compressed = False + # http/https twin is canonicalized, no separate entry created + my_urls.store_rules("http://example.org", robots_rules) + assert "http://example.org" not in my_urls.urldict + assert my_urls.get_rules("https://example.org") == robots_rules + # no crash from twin deletion while iterating in get_download_urls + assert my_urls.get_download_urls(time_limit=0) + def test_urlstore_filters(): "Test language and strictness filters applied on insertion." @@ -419,7 +435,19 @@ def test_dbdump(capsys): sys.exit(1) captured = capsys.readouterr() assert captured.out.strip() == "" - # verbose + + # verbose: a handler installed by the host application is not overwritten + def custom_handler(num, frame): + return None + + signal.signal(signal.SIGINT, custom_handler) + signal.signal(signal.SIGTERM, custom_handler) + UrlStore(verbose=True) + assert signal.getsignal(signal.SIGINT) is custom_handler + assert signal.getsignal(signal.SIGTERM) is custom_handler + # verbose: registration only happens over default handlers + signal.signal(signal.SIGINT, signal.default_int_handler) + signal.signal(signal.SIGTERM, signal.SIG_DFL) interrupted_one = UrlStore(verbose=True) interrupted_one.add_urls(["https://www.test.org/1", "https://www.test.org/2"]) # SIGINT + SIGTERM caught @@ -428,6 +456,9 @@ def test_dbdump(capsys): os.kill(pid, signal.SIGINT) captured = capsys.readouterr() assert captured.out.strip().endswith("https://www.test.org/2") + # clean up for the rest of the test session + signal.signal(signal.SIGINT, signal.default_int_handler) + signal.signal(signal.SIGTERM, signal.SIG_DFL) def test_from_html(robots_rules): From c09851b44c7992230153beb41d3b8066ee13a241 Mon Sep 17 00:00:00 2001 From: Adrien Barbaresi Date: Tue, 4 Aug 2026 17:39:23 +0200 Subject: [PATCH 2/8] check code consistency --- .gitignore | 1 + courlan/clean.py | 8 +++----- courlan/urlstore.py | 31 ++++++++++++++-------------- courlan/urlutils.py | 7 ++----- tests/unit_tests.py | 28 ++++++++++++++++++------- tests/urlstore_tests.py | 45 ++++++++++++++++++++++++++--------------- 6 files changed, 72 insertions(+), 48 deletions(-) diff --git a/.gitignore b/.gitignore index 5c90db4..a1bc9a8 100644 --- a/.gitignore +++ b/.gitignore @@ -8,6 +8,7 @@ dist/ build/ *.egg-info/ +uv.lock # tests .cache/ diff --git a/courlan/clean.py b/courlan/clean.py index ea72b6e..3f3ad8a 100644 --- a/courlan/clean.py +++ b/courlan/clean.py @@ -43,9 +43,7 @@ def clean_url(url: str, language: str | None = None) -> str | None: "Helper function: chained scrubbing and normalization" try: - cleaned = normalize_url(scrub_url(url), False, language) - # align root URLs with scrub_url's form so a second pass is a no-op - return cleaned.rstrip("/") if cleaned.count("/") == 3 else cleaned + return normalize_url(scrub_url(url), False, language, False) except (AttributeError, ValueError): return None @@ -165,11 +163,11 @@ def normalize_fragment(fragment: str, language: str | None = None) -> str: def normalize_authority(parsed_url: SplitResult) -> str: "Lower-case the authority, decode punycode and strip the scheme default port." - # host only: preserve userinfo case, strip a trailing FQDN dot + # preserve userinfo case, strip trailing FQDN dot userinfo, _, hostport = parsed_url.netloc.rpartition("@") hostport = decode_punycode(_strip_trailing_dot(hostport.lower())) netloc = f"{userinfo}@{hostport}" if userinfo else hostport - # port: strip only the scheme's default port (80 for http, 443 for https) + # strip only the scheme's default port try: port = parsed_url.port except ValueError: diff --git a/courlan/urlstore.py b/courlan/urlstore.py index bbbada8..e61b854 100644 --- a/courlan/urlstore.py +++ b/courlan/urlstore.py @@ -149,16 +149,12 @@ def dump_unvisited_urls(num: Any, frame: Any) -> None: self.print_unvisited_urls() sys.exit(1) - # don't use the following on Windows + # opt-in only (mostly for CLI use), and not on Windows: + # existing handlers are replaced since the caller asked for it if verbose and not sys.platform.startswith("win"): try: - for signum in (signal.SIGINT, signal.SIGTERM): - # don't overwrite handlers the host application installed - if signal.getsignal(signum) in ( - signal.SIG_DFL, - signal.default_int_handler, - ): - signal.signal(signum, dump_unvisited_urls) + signal.signal(signal.SIGINT, dump_unvisited_urls) + signal.signal(signal.SIGTERM, dump_unvisited_urls) except ValueError: # signal handlers can only be registered in the main thread LOGGER.warning("Cannot set signal handlers outside the main thread") @@ -221,20 +217,24 @@ def _set_done(self) -> None: self.done = True def _canonical_domain(self, domain: str) -> str: - "Merge the http/https twin of the domain, keeping a single entry." - if domain.startswith("http://"): + "Read-only: return the key the domain is stored under, https twin included." + if domain not in self.urldict and domain.startswith("http://"): candidate = "https" + domain[4:] - # switch if candidate in self.urldict: return candidate - elif domain.startswith("https://"): + return domain + + def _merge_twin(self, domain: str) -> str: + "Canonicalize the domain and merge its http/https twin, keeping a single entry." + if domain.startswith("https://"): candidate = "http" + domain[5:] # replace entry: check-and-swap must be atomic against other writers with self._lock: if candidate in self.urldict: self.urldict[domain] = self.urldict[candidate] del self.urldict[candidate] - return domain + return domain + return self._canonical_domain(domain) def _store_urls( self, @@ -244,7 +244,7 @@ def _store_urls( to_left: deque[UrlPathTuple] | None = None, replace: bool = False, ) -> None: - domain = self._canonical_domain(domain) + domain = self._merge_twin(domain) # load URLs or create entry if domain in self.urldict and self.urldict[domain].state is State.BUSTED: @@ -522,13 +522,14 @@ def establish_download_schedule( def store_rules(self, website: str, rules: RobotFileParser | None) -> None: "Store crawling rules for a given website." - website = self._canonical_domain(website) + website = self._merge_twin(website) if self.compressed: rules = COMPRESSOR.compress(rules) self.urldict[website].rules = rules def get_rules(self, website: str) -> RobotFileParser | None: "Return the stored crawling rules for the given website." + website = self._canonical_domain(website) # rules may sit under the https twin if website not in self.urldict: return None raw = self.urldict[website].rules diff --git a/courlan/urlutils.py b/courlan/urlutils.py index 3601c3d..936762c 100644 --- a/courlan/urlutils.py +++ b/courlan/urlutils.py @@ -25,11 +25,8 @@ def get_tldinfo(url: str) -> tuple[str | None, str | None]: "Extract domain info via the public-suffix lookup, returning a ``(domain, full_domain)`` tuple." if not isinstance(url, str) or not url: return None, None - try: - parsed = _parse(url) - host = parsed.hostname - except ValueError: # e.g. unbalanced brackets in the netloc - return None, None + parsed = _parse(url) # never raises: malformed netlocs degrade to empty parts + host = parsed.hostname if host: host = _strip_trailing_dot(host) # FQDN form would defeat the IP gate below # IP literals are returned in canonical form; gates avoid the exception diff --git a/tests/unit_tests.py b/tests/unit_tests.py index cc95bc7..d3cec48 100644 --- a/tests/unit_tests.py +++ b/tests/unit_tests.py @@ -34,6 +34,7 @@ scrub_url, validate_url, ) +from courlan.clean import normalize_path from courlan.core import filter_links from courlan.filters import ( domain_filter, @@ -190,9 +191,7 @@ def test_scrub(): clean_url("https://example.org:443/file.html?p=100&abc=1#frag") == "https://example.org/file.html?abc=1&p=100#frag" ) - # clean_url must be idempotent: stripping every query parameter from a - # root path used to leave a trailing slash that a second pass removed, - # so the canonical form depended on how many times it was applied. + # clean_url is idempotent, including when all query parameters are stripped for url in ( "http://test.org/?s_cid=123&clickid=1", "http://test.org/?utm_source=&utm_medium=", @@ -203,8 +202,9 @@ def test_scrub(): assert clean_url(cleaned) == cleaned # a surviving (non-tracker) query still keeps the root slash assert clean_url("http://test.org/?p=1") == "http://test.org/?p=1" - # deeper paths keep their trailing slash (only the root form is collapsed) - assert clean_url("https://example.org/path/") == "https://example.org/path/" + # trailing slashes are stripped + assert clean_url("https://example.org/path/") == "https://example.org/path" + assert clean_url("https://example.org/path") == "https://example.org/path" # scrub assert scrub_url(" https://www.dwds.de") == "https://www.dwds.de" assert scrub_url("") == "https://www.dwds.de" @@ -597,7 +597,7 @@ def test_validate(): assert not is_valid_url("http://www.test[.org/test") assert is_valid_url("http://test.org/test") - # non-string input returns False instead of raising + # non-string input assert not is_valid_url(None) assert not is_valid_url(123) assert not is_valid_url(b"http://test.org/test") @@ -643,6 +643,20 @@ def test_normalization(): assert ( normalize_url("https://hanxiao.io//404.html") == "https://hanxiao.io/404.html" ) + # leading /../ segments are removed + assert ( + normalize_url("https://example.org/../path/page.html") + == "https://example.org/path/page.html" + ) + assert normalize_path("/../../a//b") == "/a/b" + + # pre-normalized parts passed by the caller are used as they are + assert ( + normalize_url( + "https://example.org/a?p=1", netloc="test.org", path="/b", query="" + ) + == "https://test.org/b" + ) # IPv6: default port stripped (was missed by the old \w-lookbehind regex) assert normalize_url("http://[::1]:80/") == "http://[::1]/" @@ -650,7 +664,7 @@ def test_normalization(): # non-default port preserved assert normalize_url("http://[::1]:8080/") == "http://[::1]:8080/" - # host is lowercased, userinfo case is preserved (credentials are case-sensitive) + # userinfo case is preserved (credentials are case-sensitive) assert ( normalize_url("https://User:PassWord@Example.COM/Path") == "https://User:PassWord@example.com/Path" diff --git a/tests/urlstore_tests.py b/tests/urlstore_tests.py index e811476..9780978 100644 --- a/tests/urlstore_tests.py +++ b/tests/urlstore_tests.py @@ -88,7 +88,7 @@ def test_urlstore_basics(): assert len(my_urls.urldict) == 1 and "http://example.org" not in my_urls.urldict assert len(my_urls.urldict["https://example.org"].tuples) == 2 - # slash variants in the same batch dedup like across batches + # slash variants dedup within a batch too my_urls.reset() my_urls.add_urls(["https://example.org/a", "https://example.org/a/"]) assert len(my_urls.urldict["https://example.org"].tuples) == 1 @@ -128,6 +128,9 @@ def test_urlstore_rules(robots_rules): my_urls.store_rules("http://example.org", robots_rules) assert "http://example.org" not in my_urls.urldict assert my_urls.get_rules("https://example.org") == robots_rules + # ... and reads resolve to the same entry + assert my_urls.get_rules("http://example.org") == robots_rules + assert my_urls.get_crawl_delay("http://example.org") == 5 # no crash from twin deletion while iterating in get_download_urls assert my_urls.get_download_urls(time_limit=0) @@ -436,31 +439,41 @@ def test_dbdump(capsys): captured = capsys.readouterr() assert captured.out.strip() == "" - # verbose: a handler installed by the host application is not overwritten - def custom_handler(num, frame): - return None - - signal.signal(signal.SIGINT, custom_handler) - signal.signal(signal.SIGTERM, custom_handler) + # verbose: opt-in registration, a later store takes over UrlStore(verbose=True) - assert signal.getsignal(signal.SIGINT) is custom_handler - assert signal.getsignal(signal.SIGTERM) is custom_handler - # verbose: registration only happens over default handlers - signal.signal(signal.SIGINT, signal.default_int_handler) - signal.signal(signal.SIGTERM, signal.SIG_DFL) interrupted_one = UrlStore(verbose=True) interrupted_one.add_urls(["https://www.test.org/1", "https://www.test.org/2"]) # SIGINT + SIGTERM caught pid = os.getpid() - with pytest.raises(SystemExit): - os.kill(pid, signal.SIGINT) - captured = capsys.readouterr() - assert captured.out.strip().endswith("https://www.test.org/2") + for signum in (signal.SIGINT, signal.SIGTERM): + with pytest.raises(SystemExit): + os.kill(pid, signum) + captured = capsys.readouterr() + assert captured.out.strip().endswith("https://www.test.org/2") # clean up for the rest of the test session signal.signal(signal.SIGINT, signal.default_int_handler) signal.signal(signal.SIGTERM, signal.SIG_DFL) +def test_verbose_outside_main_thread(caplog): + "Signal registration fails outside the main thread but is not fatal." + errors = [] + + def run(): + try: + UrlStore(verbose=True) + except BaseException as exc: # noqa: BLE001 + errors.append(exc) + + thread = threading.Thread(target=run) + with caplog.at_level("WARNING", logger="courlan.urlstore"): + thread.start() + thread.join() + + assert not errors + assert "outside the main thread" in caplog.text + + def test_from_html(robots_rules): "Test link extraction procedures." url_store = UrlStore() From 4e2d40642a4cdcad5bb5cb993d011e57af31e70e Mon Sep 17 00:00:00 2001 From: Adrien Barbaresi Date: Tue, 4 Aug 2026 17:48:58 +0200 Subject: [PATCH 3/8] fix windows CI test --- tests/urlstore_tests.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/urlstore_tests.py b/tests/urlstore_tests.py index 9780978..881aeed 100644 --- a/tests/urlstore_tests.py +++ b/tests/urlstore_tests.py @@ -471,7 +471,9 @@ def run(): thread.join() assert not errors - assert "outside the main thread" in caplog.text + # no registration attempt on Windows, hence no warning + if os.name != "nt": + assert "outside the main thread" in caplog.text def test_from_html(robots_rules): From df07df7d74138e45ffeef7a43500530b7f56bf08 Mon Sep 17 00:00:00 2001 From: Adrien Barbaresi Date: Tue, 11 Aug 2026 16:56:16 +0200 Subject: [PATCH 4/8] review code consistency and CI --- courlan/clean.py | 12 ++++++++---- courlan/filters.py | 7 ++----- courlan/urlstore.py | 15 ++++++++++----- pyproject.toml | 6 +++--- tests/unit_tests.py | 6 +++++- tests/urlstore_tests.py | 27 +++++++++++++++++++++++++++ 6 files changed, 55 insertions(+), 18 deletions(-) diff --git a/courlan/clean.py b/courlan/clean.py index 3f3ad8a..820c491 100644 --- a/courlan/clean.py +++ b/courlan/clean.py @@ -163,18 +163,22 @@ def normalize_fragment(fragment: str, language: str | None = None) -> str: def normalize_authority(parsed_url: SplitResult) -> str: "Lower-case the authority, decode punycode and strip the scheme default port." - # preserve userinfo case, strip trailing FQDN dot userinfo, _, hostport = parsed_url.netloc.rpartition("@") - hostport = decode_punycode(_strip_trailing_dot(hostport.lower())) + hostport = hostport.lower() + # split port before punycode decoding (e.g. "xn--n3h:8080") + host, has_port, port_str = hostport.rpartition(":") + if has_port and port_str.isdigit(): + hostport = f"{decode_punycode(_strip_trailing_dot(host))}:{port_str}" + else: + hostport = decode_punycode(_strip_trailing_dot(hostport)) netloc = f"{userinfo}@{hostport}" if userinfo else hostport # strip only the scheme's default port try: port = parsed_url.port except ValueError: - port = None # port could not be cast to integer value + port = None scheme = parsed_url.scheme.lower() if (scheme == "http" and port == 80) or (scheme == "https" and port == 443): - # strip the trailing default port (IPv6-safe) netloc = netloc.rsplit(":", 1)[0] return netloc diff --git a/courlan/filters.py b/courlan/filters.py index 5da1c16..6fe8891 100644 --- a/courlan/filters.py +++ b/courlan/filters.py @@ -209,7 +209,7 @@ def path_filter(urlpath: str, query: str) -> bool: def type_filter(url: str, strict: bool = False, with_nav: bool = False) -> bool: """Make sure the target URL is from a suitable type (HTML page with primarily text). Strict: Try to filter out other document types, spam, video and adult websites.""" - if ( + return not ( # feeds + blogspot url.endswith(("/feed", "/rss", "_archive.html")) or @@ -218,10 +218,7 @@ def type_filter(url: str, strict: bool = False, with_nav: bool = False) -> bool: or # type (also hidden in parameters), videos, adult content (strict and (FILE_TYPE.search(url) or ADULT_AND_VIDEOS.search(url))) - ): - return False - # default - return True + ) def validate_url(url: str | None) -> tuple[bool, SplitResult | None]: diff --git a/courlan/urlstore.py b/courlan/urlstore.py index e61b854..806a506 100644 --- a/courlan/urlstore.py +++ b/courlan/urlstore.py @@ -217,9 +217,14 @@ def _set_done(self) -> None: self.done = True def _canonical_domain(self, domain: str) -> str: - "Read-only: return the key the domain is stored under, https twin included." - if domain not in self.urldict and domain.startswith("http://"): - candidate = "https" + domain[4:] + "Read-only: return the key the domain is stored under, http/https twin included." + if domain not in self.urldict: + if domain.startswith("http://"): + candidate = "https" + domain[4:] + elif domain.startswith("https://"): + candidate = "http" + domain[5:] + else: + return domain if candidate in self.urldict: return candidate return domain @@ -522,7 +527,7 @@ def establish_download_schedule( def store_rules(self, website: str, rules: RobotFileParser | None) -> None: "Store crawling rules for a given website." - website = self._merge_twin(website) + website = self._canonical_domain(website) if self.compressed: rules = COMPRESSOR.compress(rules) self.urldict[website].rules = rules @@ -577,7 +582,7 @@ def print_urls(self) -> None: print( "\n".join( [ - f"{domain}{u.path()}\t{str(u.visited)}" + f"{domain}{u.path()}\t{u.visited!s}" for u in self._load_urls(domain) ] ), diff --git a/pyproject.toml b/pyproject.toml index 6ee11ca..5067f0c 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -72,9 +72,9 @@ courlan = "courlan.cli:main" # Development extras [project.optional-dependencies] dev = [ - "ruff==0.15.15", - "mypy==2.1.0", - "pytest==9.0.3", + "ruff==0.16.2", + "mypy==2.3.0", + "pytest==9.1.1", "pytest-cov==7.1.0", "pytest-httpserver==1.1.5", ] diff --git a/tests/unit_tests.py b/tests/unit_tests.py index d3cec48..c866958 100644 --- a/tests/unit_tests.py +++ b/tests/unit_tests.py @@ -681,6 +681,8 @@ def test_normalization(): assert normalize_url("http://xn--Mnchen-3ya.de") == "http://münchen.de" assert normalize_url("http://Mnchen-3ya.de") == "http://mnchen-3ya.de" assert normalize_url("http://xn--München.de") == "http://xn--münchen.de" + # punycode with non-default port + assert normalize_url("http://xn--n3h:8080/") == "http://☃:8080/" # account for particular characters assert ( @@ -1661,7 +1663,9 @@ def test_cli(tmp_path): # test for Windows and the rest assert ( subprocess.run( - [courlan_bin, "-i", inputfile, "-o", outputfile, "-p", "1"], env=env + [courlan_bin, "-i", inputfile, "-o", outputfile, "-p", "1"], + env=env, + check=False, ).returncode == 0 ) diff --git a/tests/urlstore_tests.py b/tests/urlstore_tests.py index 881aeed..1f649e1 100644 --- a/tests/urlstore_tests.py +++ b/tests/urlstore_tests.py @@ -640,3 +640,30 @@ def test_urlstore_store_non_http_domain(): store = UrlStore() store._store_urls("ftp://example.org") assert "ftp://example.org" in store.urldict + + +def test_store_rules_no_merge(): + "store_rules must not destroy independent http/https entries (uses read-only lookup)." + store = UrlStore() + # force two independent entries (simulates legacy unpickled data) + store.urldict["http://host.com"] + store.urldict["https://host.com"] + rules = RobotFileParser() + store.store_rules("https://host.com", rules) + # http entry must still exist — store_rules is read-only, not a merge + assert "http://host.com" in store.urldict + assert "https://host.com" in store.urldict + + +def test_get_rules_both_directions(): + "get_rules resolves http->https and https->http." + store = UrlStore() + rules = RobotFileParser() + # rules under http, queried via https + store.add_urls(["http://a.com/x"]) + store.store_rules("http://a.com", rules) + assert store.get_rules("https://a.com") is not None + # rules under https, queried via http + store.add_urls(["https://b.com/x"]) + store.store_rules("https://b.com", rules) + assert store.get_rules("http://b.com") is not None From 83915953cd47ea91e7ee168a3b18499516c5a368 Mon Sep 17 00:00:00 2001 From: Adrien Barbaresi Date: Tue, 11 Aug 2026 18:13:09 +0200 Subject: [PATCH 5/8] unify cleaning --- courlan/clean.py | 103 ++++++++++++++++++++++------------- courlan/core.py | 73 +++++++++++-------------- courlan/filters.py | 4 +- courlan/urlstore.py | 116 ++++++++++++++++++++++++++-------------- courlan/urlutils.py | 9 ++-- tests/unit_tests.py | 69 ++++++++++++++++-------- tests/urlstore_tests.py | 55 ++++++++++++++++++- 7 files changed, 281 insertions(+), 148 deletions(-) diff --git a/courlan/clean.py b/courlan/clean.py index 820c491..8591cb5 100644 --- a/courlan/clean.py +++ b/courlan/clean.py @@ -161,26 +161,28 @@ def normalize_fragment(fragment: str, language: str | None = None) -> str: return normalize_part(fragment) -def normalize_authority(parsed_url: SplitResult) -> str: - "Lower-case the authority, decode punycode and strip the scheme default port." +def _normalize_netloc_parts(parsed_url: SplitResult) -> tuple[str, str]: + "Return the normalized (netloc, hostport) pair, hostport without userinfo." userinfo, _, hostport = parsed_url.netloc.rpartition("@") hostport = hostport.lower() # split port before punycode decoding (e.g. "xn--n3h:8080") - host, has_port, port_str = hostport.rpartition(":") - if has_port and port_str.isdigit(): - hostport = f"{decode_punycode(_strip_trailing_dot(host))}:{port_str}" - else: - hostport = decode_punycode(_strip_trailing_dot(hostport)) - netloc = f"{userinfo}@{hostport}" if userinfo else hostport + host, sep, port = hostport.rpartition(":") + # a colon left in an unbracketed host means a malformed netloc (e.g. "x:80:80") + if not (sep and port.isascii() and port.isdecimal()) or ( + ":" in host and not host.endswith("]") + ): + host, port = hostport, "" + host = decode_punycode(_strip_trailing_dot(host)) # strip only the scheme's default port - try: - port = parsed_url.port - except ValueError: - port = None scheme = parsed_url.scheme.lower() - if (scheme == "http" and port == 80) or (scheme == "https" and port == 443): - netloc = netloc.rsplit(":", 1)[0] - return netloc + if port and ( + (scheme == "http" and int(port) == 80) + or (scheme == "https" and int(port) == 443) + ): + port = "" + hostport = f"{host}:{port}" if port else host + netloc = f"{userinfo}@{hostport}" if userinfo else hostport + return netloc, hostport def normalize_path(path: str) -> str: @@ -190,33 +192,58 @@ def normalize_path(path: str) -> str: return normalize_part(PATH2.sub("", PATH1.sub("/", path))) +def _normalized_parts( + parsed_url: SplitResult, strict: bool, language: str | None +) -> tuple[str, str, str, str, str]: + "Return (scheme, netloc, hostport, path, query), all normalized." + netloc, host = _normalize_netloc_parts(parsed_url) + return ( + parsed_url.scheme.lower(), + netloc, + host, + normalize_path(parsed_url.path), + clean_query(parsed_url.query, strict, language), + ) + + +def _rebuild_url( + scheme: str, + netloc: str, + path: str, + query: str, + fragment: str, + strict: bool, + language: str | None, + trailing_slash: bool, +) -> str: + "Assemble the final URL string from already normalized parts." + if query and not path: + path = "/" + elif path == "/" and not query: + # canonical root form has no trailing slash (matches scrub_url) + path = "" + elif not trailing_slash and not query and path.endswith("/"): + path = path.rstrip("/") + newfragment = "" if strict else normalize_fragment(fragment, language) + return urlunsplit((scheme, netloc, path, query, newfragment)) + + def normalize_url( parsed_url: SplitResult | str, strict: bool = False, language: str | None = None, trailing_slash: bool = True, - netloc: str | None = None, - path: str | None = None, - query: str | None = None, ) -> str: - """Takes a URL string or a parsed URL and returns a normalized URL string. - `netloc`, `path` and `query` skip the corresponding step when the caller has - already normalized that part (see check_url).""" + "Takes a URL string or a parsed URL and returns a normalized URL string." parsed_url = _parse(parsed_url) - # lowercase + remove fragments + normalize punycode - scheme = parsed_url.scheme.lower() - if netloc is None: - netloc = normalize_authority(parsed_url) - if path is None: - path = normalize_path(parsed_url.path) - # strip unwanted query elements - if query is None: - query = clean_query(parsed_url.query, strict, language) - if query and not path: - path = "/" - elif not trailing_slash and not query and path.endswith("/"): - path = path.rstrip("/") - # fragment - newfragment = "" if strict else normalize_fragment(parsed_url.fragment, language) - # rebuild - return urlunsplit((scheme, netloc, path, query, newfragment)) + scheme, netloc, _, path, query = _normalized_parts(parsed_url, strict, language) + return _rebuild_url( + scheme, + netloc, + path, + query, + parsed_url.fragment, + strict, + language, + trailing_slash, + ) diff --git a/courlan/core.py b/courlan/core.py index 7408e67..dd61ca6 100644 --- a/courlan/core.py +++ b/courlan/core.py @@ -7,13 +7,7 @@ import re from urllib.robotparser import RobotFileParser -from .clean import ( - clean_query, - normalize_authority, - normalize_path, - normalize_url, - scrub_url, -) +from .clean import _normalized_parts, _rebuild_url, scrub_url from .filters import ( basic_filter, domain_filter, @@ -67,8 +61,8 @@ def check_url( Nothing: invalid URLs are caught internally and None is returned. """ - # first sanity check - # use standard parsing library, validate and strip fragments, then normalize + # scrub, parse and normalize, then filter on the normalized parts + # and the final form so the output is a fixed point try: # length test if basic_filter(url) is False: @@ -82,37 +76,23 @@ def check_url( if with_redirects: url = redirection_test(url) - # spam & structural elements - if type_filter(url, strict=strict, with_nav=with_nav) is False: - LOGGER.debug("rejected, type filter: %s", url) - raise ValueError - - # internationalization and language heuristics in URL - if ( - language is not None - and lang_filter(url, language, strict, trailing_slash) is False - ): - LOGGER.debug("rejected, lang filter: %s", url) - raise ValueError - # split and validate validation_test, parsed_url = validate_url(url) if validation_test is False or parsed_url is None: LOGGER.debug("rejected, validation test: %s", url) raise ValueError - # the filters below and normalize_url() need the same normalized parts, so - # compute each one once and pass them on - path = normalize_path(parsed_url.path) + # normalize all parts once; the filters below and the rebuild share them + scheme, netloc, host, path, query = _normalized_parts( + parsed_url, strict, language + ) # content filter based on extensions if extension_filter(path) is False: LOGGER.debug("rejected, extension filter: %s", url) raise ValueError - # unsuitable domain/host name (strip userinfo; domain_filter expects host[/port]) - netloc = normalize_authority(parsed_url) - host = netloc.rsplit("@", 1)[-1] + # unsuitable domain/host name (without userinfo; domain_filter expects host[:port]) if not host or domain_filter(host) is False: LOGGER.debug("rejected, domain name: %s", url) raise ValueError @@ -120,24 +100,35 @@ def check_url( # strict content filtering: the query has to be the one that survives # normalization, else a stripped param keeps an index page that # check_url would reject on a second pass - query = None - if strict: - query = clean_query(parsed_url.query, strict, language) - if path_filter(path, query) is False: - LOGGER.debug("rejected, path filter: %s", url) - raise ValueError - - # normalize - url = normalize_url( - parsed_url, + if strict and path_filter(path, query) is False: + LOGGER.debug("rejected, path filter: %s", url) + raise ValueError + + # rebuild + url = _rebuild_url( + scheme, + netloc, + path, + query, + parsed_url.fragment, strict, language, trailing_slash, - netloc=netloc, - path=path, - query=query, ) + # spam & structural elements, filtered on the final form for idempotence + if type_filter(url, strict=strict, with_nav=with_nav) is False: + LOGGER.debug("rejected, type filter: %s", url) + raise ValueError + + # internationalization and language heuristics in URL + if ( + language is not None + and lang_filter(url, language, strict, trailing_slash) is False + ): + LOGGER.debug("rejected, lang filter: %s", url) + raise ValueError + # domain info: use blacklist in strict mode only domain = extract_domain(url, blacklist=BLACKLIST if strict else None) if domain is None: diff --git a/courlan/filters.py b/courlan/filters.py index 6fe8891..4434916 100644 --- a/courlan/filters.py +++ b/courlan/filters.py @@ -68,7 +68,7 @@ r"/([a-z]{2})([_-][a-z]{2})?(?=/|$)", re.IGNORECASE ) HOST_LANG_FILTER = re.compile( - r"https?://([a-z]{2})\.(?:[^.]{4,})\.(?:[^.]+)(?:\.[^.]+)?/", re.IGNORECASE + r"https?://([a-z]{2})\.(?:[^.]{4,})\.(?:[^.]+)(?:\.[^.]+)?(?:/|$)", re.IGNORECASE ) # navigation/crawls @@ -77,7 +77,7 @@ re.IGNORECASE, ) NOTCRAWLABLE = re.compile( - r"/([ck]onta[ck]t|datenschutzerkl.{1,2}rung|login|impressum|imprint)(\.[a-z]{3,4})?/?$|/login\?|" + r"/([ck]onta[ck]t|datenschutzerkl(?:.|%[0-9a-f]{2}){1,2}rung|login|impressum|imprint)(\.[a-z]{3,4})?/?$|/login\?|" r"/(javascript:|mailto:|tel\.?:|whatsapp:)", re.IGNORECASE, ) diff --git a/courlan/urlstore.py b/courlan/urlstore.py index 806a506..695d14f 100644 --- a/courlan/urlstore.py +++ b/courlan/urlstore.py @@ -24,7 +24,7 @@ from collections import defaultdict, deque -from collections.abc import Callable +from collections.abc import Callable, Iterable from datetime import datetime, timedelta from enum import Enum from operator import itemgetter @@ -36,7 +36,7 @@ from .core import filter_links from .filters import lang_filter, validate_url from .meta import clear_caches -from .urlutils import get_base_url, get_host_and_path, is_known_link +from .urlutils import _swap_scheme, get_base_url, get_host_and_path, is_known_link LOGGER = logging.getLogger(__name__) @@ -109,6 +109,20 @@ def path(self) -> str: return self.urlpath.decode("utf-8") +def _dedup_extend( + urls: deque[UrlPathTuple], + known: set[str], + tuples: Iterable[UrlPathTuple], + left: bool = False, +) -> None: + "Add tuples whose paths are not yet known, updating the known set as it goes." + add = urls.appendleft if left else urls.append + for utuple in tuples: + if not is_known_link(utuple.path(), known): + add(utuple) + known.add(utuple.path()) + + class UrlStore: """Defines a class to store domain-classified URLs and perform checks against it. @@ -181,22 +195,22 @@ def _buffer_urls( if validation_result is False or parsed_url is None: LOGGER.debug("Invalid URL: %s", url) raise ValueError - # filter + normalized = normalize_url( + parsed_url, + strict=self.strict, + language=self.language, + trailing_slash=self.trailing_slash, + ) + # filter on the final form, as in check_url if ( self.language is not None and lang_filter( - url, self.language, self.strict, self.trailing_slash + normalized, self.language, self.strict, self.trailing_slash ) is False ): LOGGER.debug("Wrong language: %s", url) raise ValueError - normalized = normalize_url( - parsed_url, - strict=self.strict, - language=self.language, - trailing_slash=self.trailing_slash, - ) hostinfo, urlpath = get_host_and_path(normalized) inputdict[hostinfo].append(UrlPathTuple(urlpath, visited)) except (TypeError, ValueError): @@ -218,28 +232,48 @@ def _set_done(self) -> None: def _canonical_domain(self, domain: str) -> str: "Read-only: return the key the domain is stored under, http/https twin included." - if domain not in self.urldict: - if domain.startswith("http://"): - candidate = "https" + domain[4:] - elif domain.startswith("https://"): - candidate = "http" + domain[5:] - else: - return domain + if domain not in self.urldict and domain.startswith(("http://", "https://")): + candidate = _swap_scheme(domain) if candidate in self.urldict: return candidate return domain + def _merge_entries(self, target: str, source: str) -> None: + "Merge the source entry into the target one; a discarded twin voids the domain." + tgt, src = self.urldict[target], self.urldict[source] + if State.BUSTED in (tgt.state, src.state): + self.urldict[target] = DomainEntry(state=State.BUSTED) + return + urls = self._load_urls(target) + known = {u.path() for u in urls} + _dedup_extend(urls, known, self._load_urls(source)) + tgt.tuples = COMPRESSOR.compress(urls) if self.compressed else urls + tgt.total = len(urls) + tgt.count += src.count + tgt.rules = tgt.rules if tgt.rules is not None else src.rules + tgt.state = State.ALL_VISITED if all(u.visited for u in urls) else State.OPEN + if src.timestamp and (not tgt.timestamp or src.timestamp > tgt.timestamp): + tgt.timestamp = src.timestamp + def _merge_twin(self, domain: str) -> str: - "Canonicalize the domain and merge its http/https twin, keeping a single entry." - if domain.startswith("https://"): - candidate = "http" + domain[5:] - # replace entry: check-and-swap must be atomic against other writers - with self._lock: - if candidate in self.urldict: - self.urldict[domain] = self.urldict[candidate] - del self.urldict[candidate] + """Canonicalize the domain under the https key if a twin exists, + merging coexisting entries into a single one.""" + if not domain.startswith(("http://", "https://")): return domain - return self._canonical_domain(domain) + https_key = domain if domain.startswith("https://") else _swap_scheme(domain) + http_key = _swap_scheme(https_key) + # check-and-swap must be atomic against other writers + with self._lock: + if https_key in self.urldict: + if http_key in self.urldict: + self._merge_entries(https_key, http_key) + del self.urldict[http_key] + return https_key + if domain == https_key and http_key in self.urldict: + # upgrade the existing http entry to the https key + self.urldict[https_key] = self.urldict.pop(http_key) + return https_key + return domain def _store_urls( self, @@ -264,17 +298,10 @@ def _store_urls( urls = deque() known = set() - # update "known" while extending so in-batch variants dedup too if to_right is not None: - for t in to_right: - if not is_known_link(t.path(), known): - urls.append(t) - known.add(t.path()) + _dedup_extend(urls, known, to_right) if to_left is not None: - for t in to_left: - if not is_known_link(t.path(), known): - urls.appendleft(t) - known.add(t.path()) + _dedup_extend(urls, known, to_left, left=True) with self._lock: if self.compressed: @@ -355,10 +382,11 @@ def add_from_html( self.add_urls(urls=links, appendleft=links_priority) def discard(self, domains: list[str]) -> None: - "Declare domains void and prune the store." - with self._lock: - for d in domains: - self.urldict[d] = DomainEntry(state=State.BUSTED) + "Declare domains void and prune the store, http/https twins included." + for domain in domains: + domain = self._merge_twin(domain) + with self._lock: + self.urldict[domain] = DomainEntry(state=State.BUSTED) self._set_done() num = gc.collect() LOGGER.debug("%s objects in GC after UrlStore.discard", num) @@ -428,6 +456,8 @@ def is_known(self, url: str) -> bool: def get_url(self, domain: str, as_visited: bool = True) -> str | None: "Retrieve a single URL and consider it to be visited (with corresponding timestamp)." # not fully used + # merge any twin first so the replace-store below stays on one key + domain = self._merge_twin(domain) if not self.is_exhausted_domain(domain): url_tuples = self._load_urls(domain) # get first non-seen url @@ -456,8 +486,10 @@ def get_download_urls( """Get a list of immediately downloadable URLs according to the given time limit per domain.""" urls = [] - for website, entry in self.urldict.items(): - if entry.state != State.OPEN: + # snapshot the keys: get_url may merge twin entries and mutate the dict + for website in list(self.urldict): + entry = self.urldict.get(website) + if entry is None or entry.state != State.OPEN: continue if ( not entry.timestamp @@ -485,6 +517,8 @@ def establish_download_schedule( targets: list[tuple[float, str]] = [] # iterate potential domains for domain in potential: + # merge any twin first so the replace-store below stays on one key + domain = self._merge_twin(domain) # load urls url_tuples = self._load_urls(domain) urlpaths: list[str] = [] diff --git a/courlan/urlutils.py b/courlan/urlutils.py index 936762c..56721c1 100644 --- a/courlan/urlutils.py +++ b/courlan/urlutils.py @@ -148,6 +148,11 @@ def is_external(url: str, reference: str, ignore_suffix: bool = True) -> bool: return domain != ref +def _swap_scheme(url: str) -> str: + "Switch between http and https in a URL or scheme-prefixed domain." + return "http" + url[5:] if url.startswith("https") else "https" + url[4:] + + def is_known_link(link: str, known_links: set[str]) -> bool: "Compare the link and its possible variants to the existing URL base." if not link: @@ -163,9 +168,7 @@ def is_known_link(link: str, known_links: set[str]) -> bool: # check link and variants with modified protocol if link.startswith("http"): - protocol_test = ( - "http" + link[5:] if link.startswith("https") else "https" + link[4:] - ) + protocol_test = _swap_scheme(link) slash_test = ( protocol_test.rstrip("/") if protocol_test[-1] == "/" diff --git a/tests/unit_tests.py b/tests/unit_tests.py index c866958..cf113a2 100644 --- a/tests/unit_tests.py +++ b/tests/unit_tests.py @@ -359,6 +359,10 @@ def test_path_filter(): assert check_url("http://www.case-modder.de/default/", strict=True) is None assert path_filter("/contact/", "") is False assert path_filter("/Datenschutzerklaerung", "") is False + # umlaut paths, raw and percent-quoted + assert path_filter("/datenschutzerklärung", "") is False + assert path_filter("/datenschutzerkl%C3%A4rung", "") is False + assert check_url("http://example.org/datenschutzerklärung", strict=True) is None # assert path_filter("/", "") is False # in strict mode an index page must not be kept alive by a query that @@ -619,7 +623,8 @@ def test_validate(): def test_normalization(): - assert normalize_url("HTTPS://WWW.DWDS.DE/") == "https://www.dwds.de/" + # canonical root form has no trailing slash + assert normalize_url("HTTPS://WWW.DWDS.DE/") == "https://www.dwds.de" assert ( normalize_url("http://test.net/foo.html#bar", strict=True) == "http://test.net/foo.html" @@ -650,19 +655,11 @@ def test_normalization(): ) assert normalize_path("/../../a//b") == "/a/b" - # pre-normalized parts passed by the caller are used as they are - assert ( - normalize_url( - "https://example.org/a?p=1", netloc="test.org", path="/b", query="" - ) - == "https://test.org/b" - ) - # IPv6: default port stripped (was missed by the old \w-lookbehind regex) - assert normalize_url("http://[::1]:80/") == "http://[::1]/" - assert normalize_url("https://[::1]:443/") == "https://[::1]/" + assert normalize_url("http://[::1]:80/") == "http://[::1]" + assert normalize_url("https://[::1]:443/") == "https://[::1]" # non-default port preserved - assert normalize_url("http://[::1]:8080/") == "http://[::1]:8080/" + assert normalize_url("http://[::1]:8080/") == "http://[::1]:8080" # userinfo case is preserved (credentials are case-sensitive) assert ( @@ -682,7 +679,7 @@ def test_normalization(): assert normalize_url("http://Mnchen-3ya.de") == "http://mnchen-3ya.de" assert normalize_url("http://xn--München.de") == "http://xn--münchen.de" # punycode with non-default port - assert normalize_url("http://xn--n3h:8080/") == "http://☃:8080/" + assert normalize_url("http://xn--n3h:8080/") == "http://☃:8080" # account for particular characters assert ( @@ -697,26 +694,26 @@ def test_normalization(): ) # trackers - assert normalize_url("http://test.org/?s_cid=123&clickid=1") == "http://test.org/" - assert normalize_url("http://test.org/?aftr_source=0") == "http://test.org/" - assert normalize_url("http://test.org/?fb_ref=0") == "http://test.org/" - assert normalize_url("http://test.org/?this_affiliate=0") == "http://test.org/" + assert normalize_url("http://test.org/?s_cid=123&clickid=1") == "http://test.org" + assert normalize_url("http://test.org/?aftr_source=0") == "http://test.org" + assert normalize_url("http://test.org/?fb_ref=0") == "http://test.org" + assert normalize_url("http://test.org/?this_affiliate=0") == "http://test.org" assert ( normalize_url("http://test.org/?utm_source=rss&utm_medium=rss") - == "http://test.org/" + == "http://test.org" ) assert ( normalize_url("http://test.org/?utm_source=rss&utm_medium=rss") - == "http://test.org/" + == "http://test.org" ) - assert normalize_url("http://test.org/#partnerid=123") == "http://test.org/" + assert normalize_url("http://test.org/#partnerid=123") == "http://test.org" assert ( normalize_url( "http://test.org/#mtm_campaign=documentation&mtm_keyword=demo&catpage=3" ) - == "http://test.org/#catpage=3" + == "http://test.org#catpage=3" ) - assert normalize_url("http://test.org/#page2") == "http://test.org/#page2" + assert normalize_url("http://test.org/#page2") == "http://test.org#page2" def test_qelems(): @@ -908,6 +905,34 @@ def test_urlcheck_domain(): assert check_url("http://example.com../page") is None +def test_urlcheck_fixed_point(): + "check_url is idempotent: re-checking its own output changes nothing." + urls = [ + "http://test.org/?s_cid=123&clickid=1", + "http://test.org/?utm_source=&utm_medium=", + "http://test.org/#partnerid=123", + "https://example.org/?", + "http://EXAMPLE.org:80/a//b/?page=2#frag", + "https://example.org/index.html?document=report.pdf", + "http://www.example.org/path/?utm_source=x", + "https://example.org/datenschutzerklärung", + "http://de.example.org/en/page/", + "http://xn--nxasmq6b.com:8080/feed", + ] + configs = ( + {"strict": False}, + {"strict": True}, + {"strict": False, "language": "de"}, + {"strict": True, "language": "en"}, + {"strict": True, "trailing_slash": False}, + ) + for url in urls: + for config in configs: + first = check_url(url, **config) + if first: + assert check_url(first[0], **config) == first, (url, config) + + def test_urlcheck_port(): "Test port handling through check_url." assert check_url("http://example.com:80") is not None diff --git a/tests/urlstore_tests.py b/tests/urlstore_tests.py index 1f649e1..d55453c 100644 --- a/tests/urlstore_tests.py +++ b/tests/urlstore_tests.py @@ -131,7 +131,6 @@ def test_urlstore_rules(robots_rules): # ... and reads resolve to the same entry assert my_urls.get_rules("http://example.org") == robots_rules assert my_urls.get_crawl_delay("http://example.org") == 5 - # no crash from twin deletion while iterating in get_download_urls assert my_urls.get_download_urls(time_limit=0) @@ -655,6 +654,60 @@ def test_store_rules_no_merge(): assert "https://host.com" in store.urldict +def test_twin_merge_on_download(): + "Coexisting legacy twins are merged on first write, no crash while iterating." + store = UrlStore() + store.add_urls(["http://dup.com/http-only", "http://dup.com/shared"]) + # simulate a legacy pickle with coexisting twin entries + entry = store.urldict.pop("http://dup.com") + store.add_urls(["https://dup.com/https-only", "https://dup.com/shared"]) + store.urldict["http://dup.com"] = entry + + downloaded = [] + while url := store.get_download_urls(time_limit=0): + downloaded.extend(url) + # twin deleted during iteration without RuntimeError, URLs unioned and deduped + assert "http://dup.com" not in store.urldict + assert sorted(downloaded) == [ + "https://dup.com/http-only", + "https://dup.com/https-only", + "https://dup.com/shared", + ] + assert store.urldict["https://dup.com"].total == 3 + + +def test_twin_merge_metadata(): + "Merging twins in compressed mode unions counts and keeps rules." + store = UrlStore(compressed=True) + store.add_urls(["http://m.com/a"]) + store.get_url("http://m.com") # count = 1 + entry = store.urldict.pop("http://m.com") + store.add_urls(["https://m.com/b"]) + store.store_rules("https://m.com", RobotFileParser()) + store.urldict["http://m.com"] = entry + store.add_urls(["https://m.com/c"]) # triggers the merge + merged = store.urldict["https://m.com"] + assert "http://m.com" not in store.urldict + assert merged.count == 1 and merged.total == 3 + assert store.get_rules("https://m.com") is not None + + +def test_twin_discard(): + "A discarded domain voids its http/https twin and rejects later additions." + store = UrlStore() + store.add_urls(["http://y.com/a"]) + store.discard(["https://y.com"]) + assert store.dump_urls() == [] + store.add_urls(["http://y.com/b"]) + assert store.dump_urls() == [] + assert "http://y.com" not in store.urldict + # a resurrected twin next to the BUSTED entry is voided on merge + store.urldict["http://y.com"] + store.add_urls(["http://y.com/c"]) + assert "http://y.com" not in store.urldict + assert store.dump_urls() == [] + + def test_get_rules_both_directions(): "get_rules resolves http->https and https->http." store = UrlStore() From b8cd05d77ebab96cbc0fb26a056b07ee0bdd5143 Mon Sep 17 00:00:00 2001 From: Adrien Barbaresi Date: Wed, 12 Aug 2026 17:20:02 +0200 Subject: [PATCH 6/8] prevent further bugs --- courlan/clean.py | 70 ++++++++++++++++------------ courlan/core.py | 40 ++++++++++++---- courlan/filters.py | 25 ++++++++-- courlan/urlstore.py | 27 +++++------ courlan/urlutils.py | 33 +++++++------ pyproject.toml | 4 +- tests/unit_tests.py | 101 ++++++++++++++++++++++++++++++++++++++-- tests/urlstore_tests.py | 38 +++++++++++---- 8 files changed, 255 insertions(+), 83 deletions(-) diff --git a/courlan/clean.py b/courlan/clean.py index 8591cb5..150b375 100644 --- a/courlan/clean.py +++ b/courlan/clean.py @@ -18,6 +18,9 @@ MIDDLE_URL = re.compile(r"https?://.+?(https?://.+?)(?:https?://|$)") +# netloc +DEFAULT_PORTS = {"http": 80, "https": 443} + # path PATH1 = re.compile(r"/+") PATH2 = re.compile(r"^(?:/\.\.(?![^/]))+") @@ -161,8 +164,8 @@ def normalize_fragment(fragment: str, language: str | None = None) -> str: return normalize_part(fragment) -def _normalize_netloc_parts(parsed_url: SplitResult) -> tuple[str, str]: - "Return the normalized (netloc, hostport) pair, hostport without userinfo." +def normalize_netloc_parts(parsed_url: SplitResult) -> tuple[str, str, str]: + "Return the normalized (scheme, netloc, hostport), hostport without userinfo." userinfo, _, hostport = parsed_url.netloc.rpartition("@") hostport = hostport.lower() # split port before punycode decoding (e.g. "xn--n3h:8080") @@ -173,16 +176,13 @@ def _normalize_netloc_parts(parsed_url: SplitResult) -> tuple[str, str]: ): host, port = hostport, "" host = decode_punycode(_strip_trailing_dot(host)) - # strip only the scheme's default port scheme = parsed_url.scheme.lower() - if port and ( - (scheme == "http" and int(port) == 80) - or (scheme == "https" and int(port) == 443) - ): + # strip only the scheme's default port + if port and DEFAULT_PORTS.get(scheme) == int(port): port = "" hostport = f"{host}:{port}" if port else host netloc = f"{userinfo}@{hostport}" if userinfo else hostport - return netloc, hostport + return scheme, netloc, hostport def normalize_path(path: str) -> str: @@ -192,21 +192,7 @@ def normalize_path(path: str) -> str: return normalize_part(PATH2.sub("", PATH1.sub("/", path))) -def _normalized_parts( - parsed_url: SplitResult, strict: bool, language: str | None -) -> tuple[str, str, str, str, str]: - "Return (scheme, netloc, hostport, path, query), all normalized." - netloc, host = _normalize_netloc_parts(parsed_url) - return ( - parsed_url.scheme.lower(), - netloc, - host, - normalize_path(parsed_url.path), - clean_query(parsed_url.query, strict, language), - ) - - -def _rebuild_url( +def rebuild_url( scheme: str, netloc: str, path: str, @@ -217,17 +203,43 @@ def _rebuild_url( trailing_slash: bool, ) -> str: "Assemble the final URL string from already normalized parts." + newfragment = "" if strict else normalize_fragment(fragment, language) if query and not path: path = "/" - elif path == "/" and not query: + elif path == "/" and not query and not newfragment: # canonical root form has no trailing slash (matches scrub_url) path = "" elif not trailing_slash and not query and path.endswith("/"): path = path.rstrip("/") - newfragment = "" if strict else normalize_fragment(fragment, language) return urlunsplit((scheme, netloc, path, query, newfragment)) +def normalize_and_split( + parsed_url: SplitResult, + strict: bool, + language: str | None, + trailing_slash: bool, +) -> tuple[str, str, str]: + "Return the normalized (url, base, path) triple from a single set of parts." + scheme, netloc, _ = normalize_netloc_parts(parsed_url) + url = rebuild_url( + scheme, + netloc, + normalize_path(parsed_url.path), + clean_query(parsed_url.query, strict, language), + parsed_url.fragment, + strict, + language, + trailing_slash, + ) + # without a netloc urlunsplit drops the "//" and the slice below would cut + # into the path instead of past the host + if not netloc: + raise ValueError(f"no host to split off: {url}") + base = f"{scheme}://{netloc}" + return url, base, url[len(base) :] or "/" + + def normalize_url( parsed_url: SplitResult | str, strict: bool = False, @@ -236,12 +248,12 @@ def normalize_url( ) -> str: "Takes a URL string or a parsed URL and returns a normalized URL string." parsed_url = _parse(parsed_url) - scheme, netloc, _, path, query = _normalized_parts(parsed_url, strict, language) - return _rebuild_url( + scheme, netloc, _ = normalize_netloc_parts(parsed_url) + return rebuild_url( scheme, netloc, - path, - query, + normalize_path(parsed_url.path), + clean_query(parsed_url.query, strict, language), parsed_url.fragment, strict, language, diff --git a/courlan/core.py b/courlan/core.py index dd61ca6..9250723 100644 --- a/courlan/core.py +++ b/courlan/core.py @@ -7,7 +7,13 @@ import re from urllib.robotparser import RobotFileParser -from .clean import _normalized_parts, _rebuild_url, scrub_url +from .clean import ( + clean_query, + normalize_netloc_parts, + normalize_path, + rebuild_url, + scrub_url, +) from .filters import ( basic_filter, domain_filter, @@ -52,7 +58,8 @@ def check_url( with_redirects: set to True for redirection test (per HTTP HEAD request) language: set target language (ISO 639-1 codes) with_nav: set to True to include navigation pages instead of discarding them - trailing_slash: set to False to trim trailing slashes + trailing_slash: set to False to trim trailing slashes from paths + (the root form never keeps one) Returns: A tuple consisting of canonical URL and extracted domain @@ -82,21 +89,30 @@ def check_url( LOGGER.debug("rejected, validation test: %s", url) raise ValueError - # normalize all parts once; the filters below and the rebuild share them - scheme, netloc, host, path, query = _normalized_parts( - parsed_url, strict, language - ) + # normalized parts, shared with the rebuild; the query comes last as + # it is the expensive one and the filters below may reject first + path = normalize_path(parsed_url.path) # content filter based on extensions if extension_filter(path) is False: LOGGER.debug("rejected, extension filter: %s", url) raise ValueError + scheme, netloc, host = normalize_netloc_parts(parsed_url) + # unsuitable domain/host name (without userinfo; domain_filter expects host[:port]) if not host or domain_filter(host) is False: LOGGER.debug("rejected, domain name: %s", url) raise ValueError + # spam & structural elements, also hidden in the query or the fragment + pre_normalized = url + if type_filter(url, strict=strict, with_nav=with_nav) is False: + LOGGER.debug("rejected, type filter: %s", url) + raise ValueError + + query = clean_query(parsed_url.query, strict, language) + # strict content filtering: the query has to be the one that survives # normalization, else a stripped param keeps an index page that # check_url would reject on a second pass @@ -105,7 +121,7 @@ def check_url( raise ValueError # rebuild - url = _rebuild_url( + url = rebuild_url( scheme, netloc, path, @@ -116,8 +132,11 @@ def check_url( trailing_slash, ) - # spam & structural elements, filtered on the final form for idempotence - if type_filter(url, strict=strict, with_nav=with_nav) is False: + # again if normalization changed the URL, it can create patterns of its own + if ( + url != pre_normalized + and type_filter(url, strict=strict, with_nav=with_nav) is False + ): LOGGER.debug("rejected, type filter: %s", url) raise ValueError @@ -166,7 +185,8 @@ def extract_links( no_filter: override settings and bypass checks to return all possible URLs language: set target language (ISO 639-1 codes) strict: set to True for stricter filtering - trailing_slash: set to False to trim trailing slashes + trailing_slash: set to False to trim trailing slashes from paths + (the root form never keeps one) with_nav: set to True to include navigation pages instead of discarding them redirects: set to True for redirection test (per HTTP HEAD request) reference: provide a host reference for external/internal evaluation diff --git a/courlan/filters.py b/courlan/filters.py index 4434916..43261a5 100644 --- a/courlan/filters.py +++ b/courlan/filters.py @@ -35,6 +35,8 @@ ) # content filters +# feeds + blogspot, at the end of the path and with any number of slashes +FEED_FILTER = re.compile(r"(?:/(?:feed|rss)|_archive\.html)/*$") SITE_STRUCTURE = re.compile( # wordpress r"/(?:wp-(?:admin|content|includes|json|themes)|" @@ -127,6 +129,12 @@ def _valid_port(tail: str) -> bool: return bool(tail) and tail[:1] != "0" and tail.isdigit() and int(tail) < 65536 +def _split_port(value: str) -> tuple[str, str]: + "Split a trailing valid port off a host, else return the host unchanged." + host, sep, port = value.rpartition(":") + return (host, port) if sep and _valid_port(port) else (value, "") + + def domain_filter(domain: str) -> bool: "Find invalid domain/host names." # FQDN absolute form ("example.com.") is valid; extract_domain already strips it @@ -147,16 +155,20 @@ def domain_filter(domain: str) -> bool: if all(c in IP_SET for c in domain): if _canonical_ip(domain): return True - head, sep, tail = domain.rpartition(":") - if sep and _valid_port(tail) and _canonical_ip(head): + head, port = _split_port(domain) + if port and _canonical_ip(head): return True # malformed domains: retry against the punycode form before rejecting if not VALID_DOMAIN_PORT.match(domain): + # the port is not part of the IDNA encoding + host, port = _split_port(domain) try: - ascii_domain = _idna_encode(domain) + ascii_domain = _idna_encode(host) except UnicodeError: return False + if port: + ascii_domain = f"{ascii_domain}:{port}" if not VALID_DOMAIN_PORT.match(ascii_domain): return False @@ -210,8 +222,11 @@ def type_filter(url: str, strict: bool = False, with_nav: bool = False) -> bool: """Make sure the target URL is from a suitable type (HTML page with primarily text). Strict: Try to filter out other document types, spam, video and adult websites.""" return not ( - # feeds + blogspot - url.endswith(("/feed", "/rss", "_archive.html")) + # feeds: on the path only, query and fragment cannot hold one + ( + ("feed" in url or "rss" in url or "_archive.html" in url) + and FEED_FILTER.search(url.partition("?")[0].partition("#")[0]) + ) or # website structure (SITE_STRUCTURE.search(url) and (not with_nav or not is_navigation_page(url))) diff --git a/courlan/urlstore.py b/courlan/urlstore.py index 695d14f..f98b013 100644 --- a/courlan/urlstore.py +++ b/courlan/urlstore.py @@ -32,7 +32,7 @@ from typing import Any from urllib.robotparser import RobotFileParser -from .clean import normalize_url +from .clean import normalize_and_split from .core import filter_links from .filters import lang_filter, validate_url from .meta import clear_caches @@ -118,9 +118,10 @@ def _dedup_extend( "Add tuples whose paths are not yet known, updating the known set as it goes." add = urls.appendleft if left else urls.append for utuple in tuples: - if not is_known_link(utuple.path(), known): + path = utuple.path() + if not is_known_link(path, known): add(utuple) - known.add(utuple.path()) + known.add(path) class UrlStore: @@ -195,11 +196,9 @@ def _buffer_urls( if validation_result is False or parsed_url is None: LOGGER.debug("Invalid URL: %s", url) raise ValueError - normalized = normalize_url( - parsed_url, - strict=self.strict, - language=self.language, - trailing_slash=self.trailing_slash, + # host and path come from the normalized parts, no second parse + normalized, hostinfo, urlpath = normalize_and_split( + parsed_url, self.strict, self.language, self.trailing_slash ) # filter on the final form, as in check_url if ( @@ -211,7 +210,6 @@ def _buffer_urls( ): LOGGER.debug("Wrong language: %s", url) raise ValueError - hostinfo, urlpath = get_host_and_path(normalized) inputdict[hostinfo].append(UrlPathTuple(urlpath, visited)) except (TypeError, ValueError): LOGGER.warning("Discarding URL: %s", url) @@ -222,7 +220,8 @@ def _load_urls(self, domain: str) -> deque[UrlPathTuple]: return deque() raw = self.urldict[domain].tuples if isinstance(raw, bytes): # compressed - return COMPRESSOR.decompress(raw) + urls: deque[UrlPathTuple] = COMPRESSOR.decompress(raw) + return urls return raw def _set_done(self) -> None: @@ -250,7 +249,8 @@ def _merge_entries(self, target: str, source: str) -> None: tgt.tuples = COMPRESSOR.compress(urls) if self.compressed else urls tgt.total = len(urls) tgt.count += src.count - tgt.rules = tgt.rules if tgt.rules is not None else src.rules + if tgt.rules is None: + tgt.rules = src.rules tgt.state = State.ALL_VISITED if all(u.visited for u in urls) else State.OPEN if src.timestamp and (not tgt.timestamp or src.timestamp > tgt.timestamp): tgt.timestamp = src.timestamp @@ -573,7 +573,8 @@ def get_rules(self, website: str) -> RobotFileParser | None: return None raw = self.urldict[website].rules if isinstance(raw, bytes): # compressed - return COMPRESSOR.decompress(raw) + rules: RobotFileParser = COMPRESSOR.decompress(raw) + return rules return raw def get_crawl_delay(self, website: str, default: float = 5) -> float: @@ -637,5 +638,5 @@ def load_store(filename: str) -> UrlStore: Warning: uses pickle, which can execute arbitrary code. Only load files you have written yourself with UrlStore.write().""" with open(filename, "rb") as output: - url_store = pickle.load(output) + url_store: UrlStore = pickle.load(output) return url_store diff --git a/courlan/urlutils.py b/courlan/urlutils.py index 56721c1..b455bd5 100644 --- a/courlan/urlutils.py +++ b/courlan/urlutils.py @@ -3,11 +3,11 @@ """ import re -from html import unescape from urllib.parse import SplitResult, urljoin, urlsplit, urlunsplit from .hosts import _canonical_ip, get_registrable_domain +_NUMERIC_AMP = re.compile(r"�*38;|�*26;", re.I) FEED_WHITELIST_REGEX = re.compile(r"(?:feed(?:burner|proxy))", re.I) @@ -61,7 +61,10 @@ def _parse(url: str | SplitResult) -> SplitResult: "Parse a string or use urllib.parse object directly." if isinstance(url, str): try: - parsed_url = urlsplit(unescape(url)) + # "&" and numeric ampersand entities only: full unescape() + # also expands query keys that happen to be entity names, + # turning "¶m=" into "\xb6m=" + parsed_url = urlsplit(_NUMERIC_AMP.sub("&", url.replace("&", "&"))) except ValueError: # malformed URL (e.g. bad IPv6 literal): degrade to empty parts like a hostless URL. parsed_url = SplitResult("", "", "", "", "") @@ -111,11 +114,14 @@ def fix_relative_urls(baseurl: str, url: str) -> str: if url.startswith("{"): return url - parsed_base = urlsplit(baseurl) - base_netloc = parsed_base.netloc - split_url = urlsplit(url) + # a malformed netloc passes through unchanged, like an absolute URL + try: + parsed_base = urlsplit(baseurl) + split_url = urlsplit(url) + except ValueError: + return url - if split_url.netloc not in (base_netloc, ""): + if split_url.netloc not in (parsed_base.netloc, ""): if split_url.scheme: return url return urlunsplit(split_url._replace(scheme=parsed_base.scheme or "http")) @@ -153,6 +159,11 @@ def _swap_scheme(url: str) -> str: return "http" + url[5:] if url.startswith("https") else "https" + url[4:] +def _slash_variant(url: str) -> str: + "Add a trailing slash, or drop the trailing ones if there are any." + return url.rstrip("/") if url[-1] == "/" else url + "/" + + def is_known_link(link: str, known_links: set[str]) -> bool: "Compare the link and its possible variants to the existing URL base." if not link: @@ -162,19 +173,13 @@ def is_known_link(link: str, known_links: set[str]) -> bool: return True # check link and variants with trailing slashes - slash_test = link.rstrip("/") if link[-1] == "/" else link + "/" - if slash_test in known_links: + if _slash_variant(link) in known_links: return True # check link and variants with modified protocol if link.startswith("http"): protocol_test = _swap_scheme(link) - slash_test = ( - protocol_test.rstrip("/") - if protocol_test[-1] == "/" - else protocol_test + "/" - ) - if protocol_test in known_links or slash_test in known_links: + if protocol_test in known_links or _slash_variant(protocol_test) in known_links: return True return False diff --git a/pyproject.toml b/pyproject.toml index 5067f0c..fa0ab08 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -86,7 +86,9 @@ docs = [ ] [tool.pytest.ini_options] -testpaths = "tests/*test*.py" +testpaths = ["tests"] +# the test files are named "*_tests.py"; docs/test_docs.py runs separately in CI +python_files = ["*_tests.py"] [tool.ruff] target-version = "py310" diff --git a/tests/unit_tests.py b/tests/unit_tests.py index cf113a2..43656e1 100644 --- a/tests/unit_tests.py +++ b/tests/unit_tests.py @@ -34,7 +34,7 @@ scrub_url, validate_url, ) -from courlan.clean import normalize_path +from courlan.clean import normalize_and_split, normalize_path from courlan.core import filter_links from courlan.filters import ( domain_filter, @@ -176,6 +176,12 @@ def test_fix_relative(): == "https://www.example.org/foo.html?q=bar#baz" ) assert fix_relative_urls("https://www.example.org", "{privacy}") == "{privacy}" + # a malformed netloc used to abort extract_links, discarding the whole page + assert fix_relative_urls("https://example.org", "//[zz]/x") == "//[zz]/x" + assert fix_relative_urls("https://[zz]", "/page.html") == "/page.html" + assert extract_links( + 'ab', "https://example.org" + ) == {"https://example.org/ok"} def test_scrub(): @@ -183,6 +189,11 @@ def test_scrub(): assert clean_url(5) is None assert clean_url("ø\xaa") == "%C3%B8%C2%AA" assert clean_url("https://example.org/?p=100") == "https://example.org/?p=100" + # query keys that are HTML entity names must survive intact + assert ( + clean_url("https://example.org/a?b=1¶m=2") + == "https://example.org/a?b=1¶m=2" + ) assert clean_url("https://example.org/ab'c") == "https://example.org/ab%27c" assert clean_url('https://example.org/abc"') == "https://example.org/abc" assert clean_url("https://example.org/abc<") == "https://example.org/abc" @@ -294,6 +305,18 @@ def test_spam_filter(): def test_type_filter(): assert type_filter("http://www.example.org/feed") is False + # the feed check looks at the path: trailing slashes, query and fragment + # do not change the verdict, but a feed-like param is not a feed + assert type_filter("http://www.example.org/feed/") is False + assert type_filter("http://www.example.org/feed//") is False + assert type_filter("http://www.example.org/feed/#f") is False + assert type_filter("http://www.example.org/feed?x=1") is False + assert type_filter("http://www.example.org/rss/") is False + assert type_filter("http://www.example.org/2011_archive.html/") is False + assert type_filter("http://www.example.org/feeds/posts/default") is True + assert type_filter("http://www.example.org/feed/x") is True + assert type_filter("http://www.example.org/myfeed") is True + assert type_filter("http://www.example.org/page?url=/feed") is True # wp assert type_filter("http://www.example.org/wp-admin/") is False assert type_filter("http://www.example.org/wp-includes/this") is False @@ -655,6 +678,18 @@ def test_normalization(): ) assert normalize_path("/../../a//b") == "/a/b" + # the (url, base, path) split needs a host: without one urlunsplit drops the + # "//" and slicing past the base would cut into the path + assert normalize_and_split( + urlsplit("https://ex.org/a?b=1#f"), False, None, True + ) == ( + "https://ex.org/a?b=1#f", + "https://ex.org", + "/a?b=1#f", + ) + with pytest.raises(ValueError, match="no host"): + normalize_and_split(urlsplit("https:path"), False, None, True) + # IPv6: default port stripped (was missed by the old \w-lookbehind regex) assert normalize_url("http://[::1]:80/") == "http://[::1]" assert normalize_url("https://[::1]:443/") == "https://[::1]" @@ -706,14 +741,27 @@ def test_normalization(): normalize_url("http://test.org/?utm_source=rss&utm_medium=rss") == "http://test.org" ) + # numeric ampersand entity variants: all must decode to & + assert ( + normalize_url("http://test.org/?a=1&b=2") == "http://test.org/?a=1&b=2" + ) + assert ( + normalize_url("http://test.org/?a=1&b=2") == "http://test.org/?a=1&b=2" + ) + assert ( + normalize_url("http://test.org/?a=1&b=2") == "http://test.org/?a=1&b=2" + ) + assert ( + normalize_url("http://test.org/?a=1&b=2") == "http://test.org/?a=1&b=2" + ) assert normalize_url("http://test.org/#partnerid=123") == "http://test.org" assert ( normalize_url( "http://test.org/#mtm_campaign=documentation&mtm_keyword=demo&catpage=3" ) - == "http://test.org#catpage=3" + == "http://test.org/#catpage=3" ) - assert normalize_url("http://test.org/#page2") == "http://test.org#page2" + assert normalize_url("http://test.org/#page2") == "http://test.org/#page2" def test_qelems(): @@ -785,6 +833,31 @@ def test_urlcheck_type_and_spam(): ) is not None ) + # feeds are rejected whether or not the path keeps its trailing slash + for url in ( + "http://example.org/feed/", + "http://example.org/rss/", + "http://example.org/feed/#f", + "http://example.org/feed?x=1", + ): + assert check_url(url) is None + assert check_url(url, trailing_slash=False) is None + + # patterns hidden in the query or the fragment, which normalization removes + assert check_url("https://example.org/page?file=movie.mp4", strict=True) is None + assert ( + check_url("http://example.org/page?utm_source=a&file=b.mp4", strict=True) + is None + ) + assert check_url("http://example.org/page#movie.mp4", strict=True) is None + assert check_url("https://example.org/page?p=/tags/x/") is None + assert check_url("https://example.org/page?next=/wp-admin/") is None + + # patterns normalization creates itself, seen only on the final form + assert check_url("http://example.org/tags//x/") is None + assert check_url("http://example.org//tags//x/") is None + assert check_url("http://example.org/category//123/") is None + assert check_url("http://example.org/tags/x/?") is None def test_urlcheck_language(): @@ -944,6 +1017,11 @@ def test_urlcheck_port(): # bracketed IPv6 with a port (urlsplit netloc shape) assert check_url("http://[::1]:8080/x") == ("http://[::1]:8080/x", "::1") assert check_url("http://[::1]:99999/x") is None + # IDN host with a port + assert check_url("http://xn--h1aagokeh.xn--p1ai:8888") == ( + "http://историк.рф:8888", + "историк.рф", + ) def test_domain_filter(): @@ -967,6 +1045,10 @@ def test_domain_filter(): assert domain_filter("exa-mple.co.uk") is True assert domain_filter("kräuter.de") is True assert domain_filter("xn--h1aagokeh.xn--p1ai") is True + # the port is not part of the IDNA encoding and has to be split off first + assert domain_filter("kräuter.de:8080") is True + assert domain_filter("историк.рф:8888") is True + assert domain_filter("kräuter.de:99999") is False # non-ASCII label too long to punycode -> UnicodeError -> rejected assert domain_filter("ä" * 100 + ".de") is False assert domain_filter("`$smarty.server.server_name`") is False @@ -1132,6 +1214,19 @@ def test_urlutils(): ) assert get_host_and_path("https://example.org/") == ("https://example.org", "/") assert get_host_and_path("https://example.org") == ("https://example.org", "/") + # query keys that are HTML entity names used to be turned into characters + assert get_host_and_path("https://example.org/a?b=1¶m=2") == ( + "https://example.org", + "/a?b=1¶m=2", + ) + assert get_host_and_path("https://example.org/a?b=1§=2©=3") == ( + "https://example.org", + "/a?b=1§=2©=3", + ) + assert get_host_and_path("https://example.org/a?b=1&c=2") == ( + "https://example.org", + "/a?b=1&c=2", + ) assert get_hostinfo("https://httpbun.org/") == ( "httpbun.org", "https://httpbun.org", diff --git a/tests/urlstore_tests.py b/tests/urlstore_tests.py index d55453c..9005857 100644 --- a/tests/urlstore_tests.py +++ b/tests/urlstore_tests.py @@ -71,6 +71,26 @@ def test_urlstore_basics(): assert len(my_urls.urldict["https://example.org"].tuples) == 2 firstelem = my_urls.urldict["https://example.org"].tuples[0] assert firstelem.urlpath == b"/" and firstelem.visited is False + # host and path come from the normalized parts: query and fragment are kept, + # and query keys that are HTML entity names are no longer mangled + parts = UrlStore() + parts.add_urls( + [ + "https://example.org/t?other=1¶m=2", + "https://example.org/a?b=1#frag", + "https://example.org/#frag", + "https://example.org/", + ] + ) + assert parts.dump_urls() == [ + "https://example.org/t?other=1¶m=2", + "https://example.org/a?b=1#frag", + "https://example.org/#frag", # root slash preserved when fragment present + "https://example.org/", + ] + # round-trip: is_known must find URLs that were just added + assert parts.is_known("https://example.org/#frag") + assert parts.is_known("https://example.org/a?b=1#frag") # reset num, _, _ = gc.get_count() my_urls.reset() @@ -444,14 +464,16 @@ def test_dbdump(capsys): interrupted_one.add_urls(["https://www.test.org/1", "https://www.test.org/2"]) # SIGINT + SIGTERM caught pid = os.getpid() - for signum in (signal.SIGINT, signal.SIGTERM): - with pytest.raises(SystemExit): - os.kill(pid, signum) - captured = capsys.readouterr() - assert captured.out.strip().endswith("https://www.test.org/2") - # clean up for the rest of the test session - signal.signal(signal.SIGINT, signal.default_int_handler) - signal.signal(signal.SIGTERM, signal.SIG_DFL) + try: + for signum in (signal.SIGINT, signal.SIGTERM): + with pytest.raises(SystemExit): + os.kill(pid, signum) + captured = capsys.readouterr() + assert captured.out.strip().endswith("https://www.test.org/2") + finally: + # clean up for the rest of the test session + signal.signal(signal.SIGINT, signal.default_int_handler) + signal.signal(signal.SIGTERM, signal.SIG_DFL) def test_verbose_outside_main_thread(caplog): From 866c4b93a8ccf924c01396db52a039ce49222ff5 Mon Sep 17 00:00:00 2001 From: Adrien Barbaresi Date: Sat, 15 Aug 2026 17:15:24 +0200 Subject: [PATCH 7/8] simplify code for future maintenance --- courlan/clean.py | 10 +- courlan/core.py | 19 +- courlan/urlstore.py | 53 ++-- pyproject.toml | 13 +- scripts/__init__.py | 0 setup.py | 40 --- tests/unit_tests.py | 652 ++++++++++++++++++++-------------------- tests/urlstore_tests.py | 180 ++++++++++- 8 files changed, 530 insertions(+), 437 deletions(-) create mode 100644 scripts/__init__.py delete mode 100644 setup.py diff --git a/courlan/clean.py b/courlan/clean.py index 150b375..a6a97b6 100644 --- a/courlan/clean.py +++ b/courlan/clean.py @@ -77,19 +77,14 @@ def scrub_url(url: str) -> str: match = SELECTION.match(url) if match and is_valid_url(match[1]): url = match[1] - LOGGER.debug("taking url: %s", url) else: match = MIDDLE_URL.match(url) if match and is_valid_url(match[1]): url = match[1] - LOGGER.debug("taking url: %s", url) - # too long and garbled URLs e.g. due to quotes URLs match = TRAILING_PARTS.match(url) if match: url = match[1] - if len(url) > 500: # arbitrary choice - LOGGER.debug("invalid-looking link %s of length %d", url[:50] + "…", len(url)) # trailing slashes in URLs without path or in embedded URLs if url.count("/") == 3 or url.count("://") > 1: url = url.rstrip("/") @@ -97,9 +92,7 @@ def scrub_url(url: str) -> str: return url -def clean_query( - querystring: str, strict: bool = False, language: str | None = None -) -> str: +def clean_query(querystring: str, strict: bool, language: str | None = None) -> str: "Strip unwanted query elements" if not querystring: return "" @@ -122,7 +115,6 @@ def clean_query( and teststr in LANG_PARAMS and str(qdict[qelem][0]) not in TARGET_LANGS[language] ): - LOGGER.debug("bad lang: %s %s", language, qelem) raise ValueError # insert newqdict[qelem] = qdict[qelem] diff --git a/courlan/core.py b/courlan/core.py index 9250723..18bfcc6 100644 --- a/courlan/core.py +++ b/courlan/core.py @@ -2,7 +2,6 @@ Core functions needed to make the module work. """ -# import locale import logging import re from urllib.robotparser import RobotFileParser @@ -58,8 +57,8 @@ def check_url( with_redirects: set to True for redirection test (per HTTP HEAD request) language: set target language (ISO 639-1 codes) with_nav: set to True to include navigation pages instead of discarding them - trailing_slash: set to False to trim trailing slashes from paths - (the root form never keeps one) + trailing_slash: keep trailing slashes on non-root paths (default True); + the root slash is always stripped Returns: A tuple consisting of canonical URL and extracted domain @@ -73,7 +72,6 @@ def check_url( try: # length test if basic_filter(url) is False: - LOGGER.debug("rejected, basic filter: %s", url) raise ValueError # clean @@ -86,7 +84,6 @@ def check_url( # split and validate validation_test, parsed_url = validate_url(url) if validation_test is False or parsed_url is None: - LOGGER.debug("rejected, validation test: %s", url) raise ValueError # normalized parts, shared with the rebuild; the query comes last as @@ -95,20 +92,17 @@ def check_url( # content filter based on extensions if extension_filter(path) is False: - LOGGER.debug("rejected, extension filter: %s", url) raise ValueError scheme, netloc, host = normalize_netloc_parts(parsed_url) # unsuitable domain/host name (without userinfo; domain_filter expects host[:port]) if not host or domain_filter(host) is False: - LOGGER.debug("rejected, domain name: %s", url) raise ValueError # spam & structural elements, also hidden in the query or the fragment pre_normalized = url if type_filter(url, strict=strict, with_nav=with_nav) is False: - LOGGER.debug("rejected, type filter: %s", url) raise ValueError query = clean_query(parsed_url.query, strict, language) @@ -117,7 +111,6 @@ def check_url( # normalization, else a stripped param keeps an index page that # check_url would reject on a second pass if strict and path_filter(path, query) is False: - LOGGER.debug("rejected, path filter: %s", url) raise ValueError # rebuild @@ -137,7 +130,6 @@ def check_url( url != pre_normalized and type_filter(url, strict=strict, with_nav=with_nav) is False ): - LOGGER.debug("rejected, type filter: %s", url) raise ValueError # internationalization and language heuristics in URL @@ -145,16 +137,13 @@ def check_url( language is not None and lang_filter(url, language, strict, trailing_slash) is False ): - LOGGER.debug("rejected, lang filter: %s", url) raise ValueError # domain info: use blacklist in strict mode only domain = extract_domain(url, blacklist=BLACKLIST if strict else None) if domain is None: - LOGGER.debug("rejected, domain name: %s", url) return None - # handle exceptions except (AttributeError, ValueError): LOGGER.debug("discarded URL: %s", url) return None @@ -185,8 +174,8 @@ def extract_links( no_filter: override settings and bypass checks to return all possible URLs language: set target language (ISO 639-1 codes) strict: set to True for stricter filtering - trailing_slash: set to False to trim trailing slashes from paths - (the root form never keeps one) + trailing_slash: keep trailing slashes on non-root paths (default True); + the root slash is always stripped with_nav: set to True to include navigation pages instead of discarding them redirects: set to True for redirection test (per HTTP HEAD request) reference: provide a host reference for external/internal evaluation diff --git a/courlan/urlstore.py b/courlan/urlstore.py index f98b013..42fab3a 100644 --- a/courlan/urlstore.py +++ b/courlan/urlstore.py @@ -185,7 +185,7 @@ def __setstate__(self, state: dict[str, Any]) -> None: self._lock = Lock() def _buffer_urls( - self, data: list[str], visited: bool = False + self, data: list[str], visited: bool ) -> defaultdict[str, deque[UrlPathTuple]]: inputdict: defaultdict[str, deque[UrlPathTuple]] = defaultdict(deque) for url in dict.fromkeys(data): @@ -194,7 +194,6 @@ def _buffer_urls( # validate validation_result, parsed_url = validate_url(url) if validation_result is False or parsed_url is None: - LOGGER.debug("Invalid URL: %s", url) raise ValueError # host and path come from the normalized parts, no second parse normalized, hostinfo, urlpath = normalize_and_split( @@ -208,7 +207,6 @@ def _buffer_urls( ) is False ): - LOGGER.debug("Wrong language: %s", url) raise ValueError inputdict[hostinfo].append(UrlPathTuple(urlpath, visited)) except (TypeError, ValueError): @@ -220,8 +218,8 @@ def _load_urls(self, domain: str) -> deque[UrlPathTuple]: return deque() raw = self.urldict[domain].tuples if isinstance(raw, bytes): # compressed - urls: deque[UrlPathTuple] = COMPRESSOR.decompress(raw) - return urls + result: deque[UrlPathTuple] = COMPRESSOR.decompress(raw) + return result return raw def _set_done(self) -> None: @@ -229,14 +227,6 @@ def _set_done(self) -> None: with self._lock: self.done = True - def _canonical_domain(self, domain: str) -> str: - "Read-only: return the key the domain is stored under, http/https twin included." - if domain not in self.urldict and domain.startswith(("http://", "https://")): - candidate = _swap_scheme(domain) - if candidate in self.urldict: - return candidate - return domain - def _merge_entries(self, target: str, source: str) -> None: "Merge the source entry into the target one; a discarded twin voids the domain." tgt, src = self.urldict[target], self.urldict[source] @@ -247,7 +237,6 @@ def _merge_entries(self, target: str, source: str) -> None: known = {u.path() for u in urls} _dedup_extend(urls, known, self._load_urls(source)) tgt.tuples = COMPRESSOR.compress(urls) if self.compressed else urls - tgt.total = len(urls) tgt.count += src.count if tgt.rules is None: tgt.rules = src.rules @@ -304,19 +293,17 @@ def _store_urls( _dedup_extend(urls, known, to_left, left=True) with self._lock: - if self.compressed: - self.urldict[domain].tuples = COMPRESSOR.compress(urls) - else: - self.urldict[domain].tuples = urls - self.urldict[domain].total = len(urls) + entry = self.urldict[domain] + entry.tuples = COMPRESSOR.compress(urls) if self.compressed else urls + entry.total = len(urls) if timestamp is not None: - self.urldict[domain].timestamp = timestamp + entry.timestamp = timestamp if all(u.visited for u in urls): - self.urldict[domain].state = State.ALL_VISITED + entry.state = State.ALL_VISITED else: - self.urldict[domain].state = State.OPEN + entry.state = State.OPEN if self.done: self.done = False @@ -367,7 +354,6 @@ def add_from_html( with_nav: bool = True, ) -> None: "Find links in a HTML document, filter them and add them to the data store." - # lang = lang or self.language base_url = get_base_url(url) rules = self.get_rules(base_url) links, links_priority = filter_links( @@ -415,7 +401,6 @@ def is_exhausted_domain(self, domain: str) -> bool: if domain in self.urldict: return self.urldict[domain].state != State.OPEN return False - # raise KeyError("website not in store") def unvisited_websites_number(self) -> int: "Return the number of websites for which there are still URLs to visit." @@ -455,7 +440,6 @@ def is_known(self, url: str) -> bool: def get_url(self, domain: str, as_visited: bool = True) -> str | None: "Retrieve a single URL and consider it to be visited (with corresponding timestamp)." - # not fully used # merge any twin first so the replace-store below stays on one key domain = self._merge_twin(domain) if not self.is_exhausted_domain(domain): @@ -543,8 +527,8 @@ def establish_download_schedule( ): schedule_secs = 0.0 else: - schedule_secs = time_limit - float( - f"{(now - original_timestamp).total_seconds():.2f}" + schedule_secs = time_limit - round( + (now - original_timestamp).total_seconds(), 2 ) for urlpath in urlpaths: targets.append((schedule_secs, domain + urlpath)) @@ -561,20 +545,20 @@ def establish_download_schedule( def store_rules(self, website: str, rules: RobotFileParser | None) -> None: "Store crawling rules for a given website." - website = self._canonical_domain(website) + website = self._merge_twin(website) if self.compressed: rules = COMPRESSOR.compress(rules) self.urldict[website].rules = rules def get_rules(self, website: str) -> RobotFileParser | None: "Return the stored crawling rules for the given website." - website = self._canonical_domain(website) # rules may sit under the https twin + website = self._merge_twin(website) if website not in self.urldict: return None raw = self.urldict[website].rules if isinstance(raw, bytes): # compressed - rules: RobotFileParser = COMPRESSOR.decompress(raw) - return rules + result: RobotFileParser = COMPRESSOR.decompress(raw) + return result return raw def get_crawl_delay(self, website: str, default: float = 5) -> float: @@ -601,10 +585,7 @@ def download_threshold_reached(self, threshold: float) -> bool: def dump_urls(self) -> list[str]: "Return a list of all known URLs." - urls = [] - for domain in self.urldict: - urls.extend(self.find_known_urls(domain)) - return urls + return [url for domain in self.urldict for url in self.find_known_urls(domain)] def print_unvisited_urls(self) -> None: "Print all unvisited URLs in store." @@ -624,7 +605,7 @@ def print_urls(self) -> None: flush=True, ) - # PERSISTANCE + # PERSISTENCE def write(self, filename: str) -> None: "Write the URL store to disk as a pickle file, see load_store()." diff --git a/pyproject.toml b/pyproject.toml index fa0ab08..19d51ee 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -50,10 +50,9 @@ classifiers = [ "Typing :: Typed", ] dependencies = [ - "urllib3 >= 1.26, < 3", + "urllib3 >= 2, < 3", ] -# https://setuptools.pypa.io/en/latest/userguide/pyproject_config.html [tool.setuptools] packages = ["courlan"] @@ -94,23 +93,17 @@ python_files = ["*_tests.py"] target-version = "py310" [tool.ruff.lint] -# pin the historical defaults instead of tracking ruff's own, -# plus import sorting (I), pyupgrade (UP) and bugbear (B) -select = ["E4", "E7", "E9", "F"] -extend-select = ["I", "UP", "B"] +select = ["B", "E4", "E7", "E9", "F", "I", "UP"] [tool.mypy] python_version = "3.10" +strict = true ignore_missing_imports = true -pretty = true -warn_unused_ignores = true -warn_redundant_casts = true # https://coverage.readthedocs.io/en/latest/config.html [tool.coverage.run] branch = true source = ["courlan"] -omit = ["tests/*", "setup.py", "scripts/*"] [tool.coverage.report] exclude_lines = [ diff --git a/scripts/__init__.py b/scripts/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/setup.py b/setup.py deleted file mode 100644 index b772356..0000000 --- a/setup.py +++ /dev/null @@ -1,40 +0,0 @@ -""" -URL filter and manipulation tools -https://github.com/adbar/courlan -""" - -import sys - -from setuptools import setup - - -# add argument to compile with mypyc -if len(sys.argv) > 1 and sys.argv[1] == "--use-mypyc": - sys.argv.pop(1) - USE_MYPYC = True - from mypyc.build import mypycify - - ext_modules = mypycify( - [ - "courlan/__init__.py", - "courlan/clean.py", - "courlan/core.py", - "courlan/filters.py", - "courlan/hosts.py", - "courlan/langcodes.py", - "courlan/sampling.py", - "courlan/settings.py", - "courlan/urlstore.py", - "courlan/urlutils.py", - ], - opt_level="3", - multi_file=True, - ) -else: - ext_modules = [] - - -setup( - # mypyc or not - ext_modules=ext_modules, -) diff --git a/tests/unit_tests.py b/tests/unit_tests.py index 43656e1..3f6d6fd 100644 --- a/tests/unit_tests.py +++ b/tests/unit_tests.py @@ -34,9 +34,10 @@ scrub_url, validate_url, ) -from courlan.clean import normalize_and_split, normalize_path +from courlan.clean import clean_query, normalize_and_split, normalize_path from courlan.core import filter_links from courlan.filters import ( + basic_filter, domain_filter, extension_filter, path_filter, @@ -53,7 +54,7 @@ from courlan.langcodes import langcodes_score from courlan.meta import clear_caches from courlan.network import redirection_test -from courlan.urlutils import _parse, is_known_link +from courlan.urlutils import _parse, _strip_trailing_dot, is_known_link logging.basicConfig(stream=sys.stdout, level=logging.DEBUG) RESOURCES_DIR = os.path.join(os.path.abspath(os.path.dirname(__file__)), "data") @@ -266,31 +267,25 @@ def test_scrub(): assert scrub_url(my_url) == my_url -def test_extension_filter(): - _, parsed_url = validate_url("http://www.example.org/test.js") - assert extension_filter(parsed_url.path) is False - _, parsed_url = validate_url("http://goodbasic.com/GirlInfo.aspx?Pseudo=MilfJanett") - assert extension_filter(parsed_url.path) is True - _, parsed_url = validate_url( - "https://www.familienrecht-allgaeu.de/de/vermoegensrecht.amp" - ) - assert extension_filter(parsed_url.path) is True - _, parsed_url = validate_url("http://www.example.org/test.shtml") - assert extension_filter(parsed_url.path) is True - _, parsed_url = validate_url("http://de.artsdot.com/ADC/Art.nsf/O/8EWETN") - assert extension_filter(parsed_url.path) is True - _, parsed_url = validate_url("http://de.artsdot.com/ADC/Art.nsf?param1=test") - assert extension_filter(parsed_url.path) is False - _, parsed_url = validate_url("http://www.example.org/test.xhtml?param1=this") - assert extension_filter(parsed_url.path) is True - _, parsed_url = validate_url("http://www.example.org/test.php5") - assert extension_filter(parsed_url.path) is True - _, parsed_url = validate_url("http://www.example.org/test.php6") - assert extension_filter(parsed_url.path) is True - # uppercase extensions are treated like their lowercase forms - assert extension_filter("/photo.JPG") is False - assert extension_filter("/page.HTML") is True - assert extension_filter("/index.PHP") is True +@pytest.mark.parametrize( + "path, expected", + [ + ("/test.js", False), + ("/GirlInfo.aspx", True), + ("/de/vermoegensrecht.amp", True), + ("/test.shtml", True), + ("/ADC/Art.nsf/O/8EWETN", True), + ("/ADC/Art.nsf", False), # bare .nsf (query stripped by urlsplit) + ("/test.xhtml", True), + ("/test.php5", True), + ("/test.php6", True), + ("/photo.JPG", False), # uppercase treated like lowercase + ("/page.HTML", True), + ("/index.PHP", True), + ], +) +def test_extension_filter(path, expected): + assert extension_filter(path) is expected def test_spam_filter(): @@ -440,151 +435,76 @@ def test_path_filter(): ) == ("http://www.case-modder.de/index.php", "case-modder.de") -def test_lang_filter(): - assert lang_filter("http://test.com/az", "de", trailing_slash=False) is False - assert lang_filter("http://test.com/az/", "de") is False - assert lang_filter("http://test.com/de", "de", trailing_slash=False) is True - assert lang_filter("http://test.com/de/", "de") is True - assert ( - lang_filter( - "https://www.20min.ch/fr/story/des-millions-pour-produire-de-l-energie-renouvelable-467974085377", - None, - ) - is True - ) - assert ( - lang_filter( - "https://www.20min.ch/fr/story/des-millions-pour-produire-de-l-energie-renouvelable-467974085377", +_20MIN_FR = "https://www.20min.ch/fr/story/des-millions-pour-produire-de-l-energie-renouvelable-467974085377" +_TU_BERLIN = "http://ig.cs.tu-berlin.de/oldstatic/w2000/ir1/aufgabe2/ir1-auf2-gr16.html" +_MUSCLEFOOD = "http://de.musclefood.com/neu/neue-nahrungsergaenzungsmittel.html" + + +@pytest.mark.parametrize( + "url, lang, kwargs, expected", + [ + ("http://test.com/az", "de", {"trailing_slash": False}, False), + ("http://test.com/az/", "de", {}, False), + ("http://test.com/de", "de", {"trailing_slash": False}, True), + ("http://test.com/de/", "de", {}, True), + (_20MIN_FR, None, {}, True), + (_20MIN_FR, "de", {}, False), + (_20MIN_FR, "fr", {}, True), + (_20MIN_FR, "en", {}, False), + (_20MIN_FR, "es", {}, False), + ("https://www.sitemaps.org/en_GB/protocol.html", "en", {}, True), + ("https://www.sitemaps.org/en_GB/protocol.html", "de", {}, False), + ("https://en.wikipedia.org/", "de", {"strict": True}, False), + ("https://en.wikipedia.org/", "de", {"strict": False}, True), + ("https://de.wikipedia.org/", "de", {"strict": True}, True), + (_MUSCLEFOOD, "de", {"strict": True}, True), + (_MUSCLEFOOD, "fr", {"strict": True}, False), + ("http://ch.postleitzahl.org/sankt_gallen/liste-T.html", "fr", {}, True), + ("http://ch.postleitzahl.org/sankt_gallen/liste-T.html", "de", {}, True), + # disturbing path sub-elements + ( + "http://www.uni-rostock.de/fakult/philfak/fkw/iph/thies/mythos.html", "de", - ) - is False - ) - assert ( - lang_filter( - "https://www.20min.ch/fr/story/des-millions-pour-produire-de-l-energie-renouvelable-467974085377", - "fr", - ) - is True - ) - assert ( - lang_filter( - "https://www.20min.ch/fr/story/des-millions-pour-produire-de-l-energie-renouvelable-467974085377", - "en", - ) - is False - ) - assert ( - lang_filter( - "https://www.20min.ch/fr/story/des-millions-pour-produire-de-l-energie-renouvelable-467974085377", - "es", - ) - is False - ) - assert lang_filter("https://www.sitemaps.org/en_GB/protocol.html", "en") is True - assert lang_filter("https://www.sitemaps.org/en_GB/protocol.html", "de") is False - assert lang_filter("https://en.wikipedia.org/", "de", strict=True) is False - assert lang_filter("https://en.wikipedia.org/", "de", strict=False) is True - assert lang_filter("https://de.wikipedia.org/", "de", strict=True) is True - assert ( - lang_filter( - "http://de.musclefood.com/neu/neue-nahrungsergaenzungsmittel.html", - "de", - strict=True, - ) - is True - ) - assert ( - lang_filter( - "http://de.musclefood.com/neu/neue-nahrungsergaenzungsmittel.html", - "fr", - strict=True, - ) - is False - ) - assert ( - lang_filter("http://ch.postleitzahl.org/sankt_gallen/liste-T.html", "fr") - is True - ) - assert ( - lang_filter("http://ch.postleitzahl.org/sankt_gallen/liste-T.html", "de") - is True - ) - # to complete when language mappings are more extensive - # assert lang_filter('http://ch.postleitzahl.org/sankt_gallen/liste-T.html', 'es') is False - # disturbing path sub-elements - assert ( - lang_filter( - "http://www.uni-rostock.de/fakult/philfak/fkw/iph/thies/mythos.html", "de" - ) - is True - ) - assert ( - lang_filter("http://stifter.literature.at/witiko/htm/h15-22b.html", "de") - is True - ) - assert ( - lang_filter("http://stifter.literature.at/doc/witiko/h15-22b.html", "de") - is True - ) - assert ( - lang_filter("http://stifter.literature.at/nl/witiko/h15-22b.html", "de") - is False - ) - assert ( - lang_filter("http://stifter.literature.at/de_DE/witiko/h15-22b.html", "de") - is True - ) - assert ( - lang_filter("http://stifter.literature.at/en_US/witiko/h15-22b.html", "de") - is False - ) - assert ( - lang_filter( + {}, + True, + ), + ("http://stifter.literature.at/witiko/htm/h15-22b.html", "de", {}, True), + ("http://stifter.literature.at/doc/witiko/h15-22b.html", "de", {}, True), + ("http://stifter.literature.at/nl/witiko/h15-22b.html", "de", {}, False), + ("http://stifter.literature.at/de_DE/witiko/h15-22b.html", "de", {}, True), + ("http://stifter.literature.at/en_US/witiko/h15-22b.html", "de", {}, False), + ( "http://www.stiftung.koerber.de/bg/recherche/de/beitrag.php?id=15132&refer=", "de", - ) - is True - ) - assert ( - lang_filter("http://www.solingen-internet.de/si-hgw/eiferer.htm", "de") is True - ) - assert ( - lang_filter( - "http://ig.cs.tu-berlin.de/oldstatic/w2000/ir1/aufgabe2/ir1-auf2-gr16.html", - "de", - strict=True, - ) - is True - ) - assert ( - lang_filter( - "http://ig.cs.tu-berlin.de/oldstatic/w2000/ir1/aufgabe2/ir1-auf2-gr16.html", - "de", - strict=False, - ) - is True - ) - assert ( - lang_filter("http://bz.berlin1.de/kino/050513/fans.html", "de", strict=False) - is True - ) - assert ( - lang_filter("http://bz.berlin1.de/kino/050513/fans.html", "de", strict=True) - is False - ) - # both path segments differ from target — was always True when two-occurrence branch was dead - assert lang_filter("https://x.com/fr/x/de/", "en") is False - # invalid territory (en_XY) is not a confident match, so the de segment wins (0, -1 → -1) - assert lang_filter("https://x.com/en_XY/x/de/", "en") is False - # >2 candidates: unreliable, score stays 0 - assert lang_filter("https://x.com/en/x/de/x/fr/", "en") is True - # adjacent segments: the matching /en/ must still be seen despite the shared - # slash (regression: consuming regex counted /de/en/ as one occurrence) - assert lang_filter("https://x.com/de/en/", "en") is True - assert lang_filter("https://x.com/de/fr/", "en") is False - - # /ch/ is not a language code, no rejection - assert lang_filter("http://www.verfassungen.de/ch/basel/verf03.htm", "de") is True + {}, + True, + ), + ("http://www.solingen-internet.de/si-hgw/eiferer.htm", "de", {}, True), + (_TU_BERLIN, "de", {"strict": True}, True), + (_TU_BERLIN, "de", {"strict": False}, True), + ("http://bz.berlin1.de/kino/050513/fans.html", "de", {"strict": False}, True), + ("http://bz.berlin1.de/kino/050513/fans.html", "de", {"strict": True}, False), + # both path segments differ from target + ("https://x.com/fr/x/de/", "en", {}, False), + # invalid territory: de segment wins + ("https://x.com/en_XY/x/de/", "en", {}, False), + # >2 candidates: unreliable, score stays 0 + ("https://x.com/en/x/de/x/fr/", "en", {}, True), + # adjacent segments (regression: consuming regex) + ("https://x.com/de/en/", "en", {}, True), + ("https://x.com/de/fr/", "en", {}, False), + # /ch/ is not a language code + ("http://www.verfassungen.de/ch/basel/verf03.htm", "de", {}, True), + # multiple path occurrences / strict host+path + ("http://example.com/en/page/en-US/test", "en", {"strict": True}, True), + ("http://example.com/en/page/de/test", "en", {"strict": True}, True), + ("http://de.example.com/en/page", "en", {"strict": True}, True), + ("http://fr.example.com/de/page", "en", {"strict": True}, False), + ("http://en.example.com/de/page", "en", {"strict": True}, True), + ], +) +def test_lang_filter(url, lang, kwargs, expected): + assert lang_filter(url, lang, **kwargs) is expected def test_langcodes_score(): @@ -742,18 +662,10 @@ def test_normalization(): == "http://test.org" ) # numeric ampersand entity variants: all must decode to & - assert ( - normalize_url("http://test.org/?a=1&b=2") == "http://test.org/?a=1&b=2" - ) - assert ( - normalize_url("http://test.org/?a=1&b=2") == "http://test.org/?a=1&b=2" - ) - assert ( - normalize_url("http://test.org/?a=1&b=2") == "http://test.org/?a=1&b=2" - ) - assert ( - normalize_url("http://test.org/?a=1&b=2") == "http://test.org/?a=1&b=2" - ) + assert normalize_url("http://test.org/?a=1&b=2") == "http://test.org/?a=1&b=2" + assert normalize_url("http://test.org/?a=1&b=2") == "http://test.org/?a=1&b=2" + assert normalize_url("http://test.org/?a=1&b=2") == "http://test.org/?a=1&b=2" + assert normalize_url("http://test.org/?a=1&b=2") == "http://test.org/?a=1&b=2" assert normalize_url("http://test.org/#partnerid=123") == "http://test.org" assert ( normalize_url( @@ -1024,82 +936,81 @@ def test_urlcheck_port(): ) -def test_domain_filter(): +_DNS_253 = "a." * 125 + "abc" # 253 chars — at the DNS length limit +_DNS_254 = "a." * 125 + "abcd" # 254 chars — over + + +@pytest.mark.parametrize( + "domain, expected", + [ + ("", False), + ("a" * 254 + ".com", False), # exceeds DNS length limit + (_DNS_253, True), + (_DNS_254, False), + ("too-long" + "g" * 60 + ".org", False), + ("long" + "g" * 50 + ".org", True), + ("example.-com", False), + ("example.", False), + ("-example.com", False), + ("_example.com", False), + ("example.com:", False), + ("a......b.com", False), + ("*.example.com", False), + ("exa-mple.co.uk", True), + ("kräuter.de", True), + ("xn--h1aagokeh.xn--p1ai", True), + ("kräuter.de:8080", True), # port split before IDNA + ("историк.рф:8888", True), + ("kräuter.de:99999", False), + ("ä" * 100 + ".de", False), # non-ASCII too long to punycode + ("`$smarty.server.server_name`", False), + ("$`)}if(a.tryconvertencoding)trycatch(e)const", False), + ("00x200.jpg,", False), + ("-100x100.webp", False), + ("0.gravata.html", False), + ("https:", False), + # IPv4 + ("127.0.0.1", True), + ("::1", True), + ("900.200.100.75", False), + ("111.111.111", False), + ("0127.0.0.1", False), + ("127.0.0.1:8080", True), + ("900.200.100.75:8080", False), # invalid IP, with port + ("1.2.3.4:65535", True), + ("1.2.3.4:99999", False), + ("1.2.3.4:0", False), + ("1.2.3.4:0080", False), # leading zero rejected + # bracketed IPv6 + ("[::1]", True), + ("[2001:db8::1]", True), + ("[::1]:8080", True), + ("[::1]:0", False), + ("[::1]:0080", False), + ("[::1]:99999", False), + ("[not-an-ip]", False), + ("[::1", False), + # trailing-dot FQDN + ("example.com.", True), + ("www.example.com.", True), + ("example.com.:8080", True), + ("192.168.0.1.", True), + ("192.168.0.1.:8080", True), + ("example.com..", False), # multiple trailing dots stay invalid + # hex-only strings that are not IPs + ("abc.de", True), + ("aced.de", True), + ("dead.beef", True), + # file extensions / gravatar / numeric + ("example.jpg", False), + ("example.html", False), + ("0.gravatar.com", False), + ("12345.org", False), + ], +) +def test_domain_filter(domain, expected): "Test filters related to domain and hostnames." - assert domain_filter("") is False - assert domain_filter("a" * 254 + ".com") is False # exceeds DNS length limit - d_ok = "a." * 125 + "abc" # 253 chars — at the DNS length limit - d_long = "a." * 125 + "abcd" # 254 chars — over - assert len(d_ok) == 253 and len(d_long) == 254 - assert domain_filter(d_ok) is True - assert domain_filter(d_long) is False - assert domain_filter("too-long" + "g" * 60 + ".org") is False - assert domain_filter("long" + "g" * 50 + ".org") is True - assert domain_filter("example.-com") is False - assert domain_filter("example.") is False - assert domain_filter("-example.com") is False - assert domain_filter("_example.com") is False - assert domain_filter("example.com:") is False - assert domain_filter("a......b.com") is False - assert domain_filter("*.example.com") is False - assert domain_filter("exa-mple.co.uk") is True - assert domain_filter("kräuter.de") is True - assert domain_filter("xn--h1aagokeh.xn--p1ai") is True - # the port is not part of the IDNA encoding and has to be split off first - assert domain_filter("kräuter.de:8080") is True - assert domain_filter("историк.рф:8888") is True - assert domain_filter("kräuter.de:99999") is False - # non-ASCII label too long to punycode -> UnicodeError -> rejected - assert domain_filter("ä" * 100 + ".de") is False - assert domain_filter("`$smarty.server.server_name`") is False - assert domain_filter("$`)}if(a.tryconvertencoding)trycatch(e)const") is False - assert domain_filter("00x200.jpg,") is False - assert domain_filter("-100x100.webp") is False - assert domain_filter("0.gravata.html") is False - assert domain_filter("https:") is False - - assert domain_filter("127.0.0.1") is True - assert domain_filter("::1") is True - assert domain_filter("900.200.100.75") is False - assert domain_filter("111.111.111") is False - assert domain_filter("0127.0.0.1") is False - # IPv4 with a port is accepted like the portless form (was wrongly rejected) - assert domain_filter("127.0.0.1:8080") is True - assert domain_filter("900.200.100.75:8080") is False # invalid IP, with port - # port must be in range and well-formed, like the domain path (VALID_DOMAIN_PORT) - assert domain_filter("1.2.3.4:65535") is True - assert domain_filter("1.2.3.4:99999") is False - assert domain_filter("1.2.3.4:0") is False - assert domain_filter("1.2.3.4:0080") is False # leading zero rejected - # bracketed IPv6 (urlsplit netloc), with and without port - assert domain_filter("[::1]") is True - assert domain_filter("[2001:db8::1]") is True - assert domain_filter("[::1]:8080") is True - assert domain_filter("[::1]:0") is False - assert domain_filter("[::1]:0080") is False - assert domain_filter("[::1]:99999") is False - assert domain_filter("[not-an-ip]") is False - assert domain_filter("[::1") is False - # trailing-dot FQDN (absolute DNS form) matches extract_domain, not false-reject - assert domain_filter("example.com.") is True - assert domain_filter("www.example.com.") is True - assert domain_filter("example.com.:8080") is True - assert domain_filter("192.168.0.1.") is True - assert domain_filter("192.168.0.1.:8080") is True - assert ( - domain_filter("example.com..") is False - ) # multiple trailing dots stay invalid - - # hex-only strings that are not IPs must still be validated as domains - assert domain_filter("abc.de") is True - assert domain_filter("aced.de") is True - assert domain_filter("dead.beef") is True - - assert domain_filter("example.jpg") is False - assert domain_filter("example.html") is False - assert domain_filter("0.gravatar.com") is False - assert domain_filter("12345.org") is False - # assert domain_filter("test.invalidtld") is False + assert domain_filter(domain) is expected def test_urlcheck_redirects(): @@ -1258,72 +1169,75 @@ def test_urlutils(): assert len(filter_urls(["https://feedburner.google.com/aabb"], None)) == 1 -def test_tld(): - "Test get_registrable_domain; inputs mirror urlsplit().hostname's pre-cleaned shape." - cases = { +@pytest.mark.parametrize( + "host, expected", + [ # standard cases - "www.bbc.co.uk": ("bbc", "bbc.co.uk"), - "example.com": ("example", "example.com"), - "a.b.example.com": ("example", "example.com"), - "foo.ne.jp": ("foo", "foo.ne.jp"), - "shop.example.org.au": ("example", "example.org.au"), - # IPv4 / numeric final label rejected (even with non-numeric labels) - "192.168.0.1": (None, None), - "www.example.42": (None, None), - # only ASCII digits count as numeric (WHATWG); fullwidth is an ordinary label - "example.42": ("example", "example.42"), - # hex IPv4 (browser-resolvable) rejected like decimal - "0xc0.0xa8.0x0.0x1": (None, None), - "foo.0x1": (None, None), - "foo.0x": (None, None), - # IPv6 (brackets already stripped) carries colons, rejected - "2001:db8::1": (None, None), - "::ffff:192.0.2.128": (None, None), - # trailing-dot FQDN is normalized - "www.example.com.": ("example", "example.com"), + ("www.bbc.co.uk", ("bbc", "bbc.co.uk")), + ("example.com", ("example", "example.com")), + ("a.b.example.com", ("example", "example.com")), + ("foo.ne.jp", ("foo", "foo.ne.jp")), + ("shop.example.org.au", ("example", "example.org.au")), + # IPv4 / numeric final label rejected + ("192.168.0.1", (None, None)), + ("www.example.42", (None, None)), + ("example.42", ("example", "example.42")), # fullwidth digits are ordinary + # hex IPv4 rejected like decimal + ("0xc0.0xa8.0x0.0x1", (None, None)), + ("foo.0x1", (None, None)), + ("foo.0x", (None, None)), + # IPv6 (colons) rejected + ("2001:db8::1", (None, None)), + ("::ffff:192.0.2.128", (None, None)), + # trailing-dot FQDN normalized + ("www.example.com.", ("example", "example.com")), # malformed / edge cases - "": (None, None), - None: (None, None), - "localhost": (None, None), - "a..b.com": (None, None), - ".uk": (None, None), - ".foo.ck": (None, None), # leading dot + bare wildcard suffix - # suffix-set boundary: last two labels not a known compound suffix - "sub.unknown.xyz": ("unknown", "unknown.xyz"), - "blog.ax": ("blog", "blog.ax"), - # a bare public suffix has no registrable domain - "co.uk": (None, None), - # www IS the registrable label here (old CLEAN_FLD_REGEX wrongly stripped it) - "www.co.uk": ("www", "www.co.uk"), - "www.gov.uk": ("www", "www.gov.uk"), - # matching is case-insensitive, original case kept in the result - "BBC.CO.UK": ("BBC", "BBC.CO.UK"), - "Example.COM": ("Example", "Example.COM"), - "FOO.CK": (None, None), - "X.CITY.KOBE.JP": ("CITY", "CITY.KOBE.JP"), - "FOO.LØDINGEN.NO": ("FOO", "FOO.LØDINGEN.NO"), - # full-PSL coverage: ccTLDs outside the old ~208-entry curated set - "x.co.tz": ("x", "x.co.tz"), - # 3-label suffix resolved via longest-match (was mis-split under the - # old fixed 2-label span logic) - "school.act.edu.au": ("school", "school.act.edu.au"), - # both Unicode and xn-- forms of the suffix match (old tld missed xn--) - "foo.lødingen.no": ("foo", "foo.lødingen.no"), - "foo.xn--ldingen-q1a.no": ("foo", "foo.xn--ldingen-q1a.no"), + ("", (None, None)), + (None, (None, None)), + ("localhost", (None, None)), + ("a..b.com", (None, None)), + (".uk", (None, None)), + (".foo.ck", (None, None)), # leading dot + bare wildcard suffix + # suffix-set boundary + ("sub.unknown.xyz", ("unknown", "unknown.xyz")), + ("blog.ax", ("blog", "blog.ax")), + # bare public suffix + ("co.uk", (None, None)), + # www IS the registrable label + ("www.co.uk", ("www", "www.co.uk")), + ("www.gov.uk", ("www", "www.gov.uk")), + # case-insensitive matching, original case kept + ("BBC.CO.UK", ("BBC", "BBC.CO.UK")), + ("Example.COM", ("Example", "Example.COM")), + ("FOO.CK", (None, None)), + ("X.CITY.KOBE.JP", ("CITY", "CITY.KOBE.JP")), + ("FOO.LØDINGEN.NO", ("FOO", "FOO.LØDINGEN.NO")), + # full-PSL coverage + ("x.co.tz", ("x", "x.co.tz")), + # 3-label suffix via longest-match + ("school.act.edu.au", ("school", "school.act.edu.au")), + # Unicode and xn-- suffix forms + ("foo.lødingen.no", ("foo", "foo.lødingen.no")), + ("foo.xn--ldingen-q1a.no", ("foo", "foo.xn--ldingen-q1a.no")), # ICANN-only: private-suffix hosts resolve as ordinary domains - "user.github.io": ("github", "github.io"), + ("user.github.io", ("github", "github.io")), # wildcard (*.) and exception (!) PSL rules - "www.foo.ck": ("www", "www.foo.ck"), # *.ck -> foo.ck is the suffix - "foo.ck": (None, None), # bare wildcard suffix - "www.ck": ("www", "www.ck"), # !www.ck exception - "a.b.kobe.jp": ("a", "a.b.kobe.jp"), # *.kobe.jp - "x.city.kobe.jp": ("city", "city.kobe.jp"), # !city.kobe.jp exception - "foo.sch.uk": (None, None), # bare *.sch.uk suffix - "bar.foo.sch.uk": ("bar", "bar.foo.sch.uk"), # *.sch.uk - } - for host, expected in cases.items(): - assert get_registrable_domain(host) == expected, host - # overlong non-ASCII label falls back instead of raising + ("www.foo.ck", ("www", "www.foo.ck")), + ("foo.ck", (None, None)), + ("www.ck", ("www", "www.ck")), # !www.ck exception + ("a.b.kobe.jp", ("a", "a.b.kobe.jp")), + ("x.city.kobe.jp", ("city", "city.kobe.jp")), # !city.kobe.jp exception + ("foo.sch.uk", (None, None)), + ("bar.foo.sch.uk", ("bar", "bar.foo.sch.uk")), + ], +) +def test_tld(host, expected): + "Test get_registrable_domain." + assert get_registrable_domain(host) == expected + + +def test_tld_idna_fallback(): + "Overlong non-ASCII label falls back instead of raising." overlong_cjk = "".join(chr(0x4E00 + i) for i in range(50)) assert _idna_label(overlong_cjk) == overlong_cjk assert get_registrable_domain(f"{overlong_cjk}.example.com") == ( @@ -1935,3 +1849,105 @@ def test_meta(): assert get_registrable_domain.cache_info().currsize == 0 if has_urlsplit_cache: assert urlsplit.cache_info().currsize == 0 + + +# --- mutation-gap tests --- + + +def test_extract_links_loop_resilience(): + "Filtered/invalid/duplicate links must not prevent extraction of later valid links." + # nofollow before valid link + html = '' + assert "https://test.com/keep" in extract_links(html, "https://test.com/", False) + # invalid link before valid + html = '' + result = extract_links(html, "https://test.com/", False) + assert "https://test.com/ok" in result + # duplicate before unique + html = '' + result = extract_links(html, "https://test.com/", False) + assert "https://test.com/a" in result and "https://test.com/b" in result + + +def test_clean_url_protocol_detection(): + "Double-URL detection must not fire on single-protocol URLs." + # single protocol — must survive unchanged + assert clean_url("http://example.org/page") == "http://example.org/page" + # double protocol — inner URL extracted + result = clean_url("http://redirect.com/?url=http://real.com/page") + assert "real.com" in result + + +def test_scrub_url_trailing_slash(): + "Trailing-slash strip fires on root URLs (3 slashes) but not on paths." + # path with trailing slash preserved + assert scrub_url("http://example.org/path/") == "http://example.org/path/" + # root URL (exactly 3 slashes) gets stripped + assert scrub_url("http://example.org/") == "http://example.org" + + +def test_clean_query_param_ordering(): + "Strict: disallowed param must not prevent later allowed params. Tracker likewise." + # strict: disallowed 'badparam' then allowed 'page' + result = clean_query("badparam=1&page=2", strict=True) + assert "page=2" in result + assert "badparam" not in result + # tracker then non-tracker + result = clean_query("utm_source=twitter&title=hello", strict=False) + assert "title=hello" in result + assert "utm_source" not in result + + +def test_normalize_path_leading_dotdot(): + "Leading /../ segments are collapsed." + assert normalize_path("/../foo/bar") == "/foo/bar" + assert normalize_path("/../../a/b") == "/a/b" + assert normalize_path("/a/../b") == "/a/../b" # only leading ones + + +def test_basic_filter_boundaries(): + "basic_filter boundary: 10-char URL passes, 9-char fails; non-http rejected." + assert basic_filter("http://x.y") is True # exactly 10 + assert basic_filter("http://x.") is False # 9 chars + assert basic_filter("ftp://example.org/page") is False # non-http, >10 chars + assert basic_filter("httpx://example.org") is True # starts with "http" + + +def test_notcrawlable_schemes(): + "NOTCRAWLABLE catches javascript:, mailto:, tel:, whatsapp: in path." + assert path_filter("/javascript:void(0)", "") is False + assert path_filter("/mailto:user@example.com", "") is False + assert path_filter("/tel:+1234567890", "") is False + assert path_filter("/whatsapp:send", "") is False + assert path_filter("/normal-page", "") is True + + +def test_strip_trailing_dot_edge_cases(): + "Double trailing dots stay invalid; userinfo+port handled." + assert _strip_trailing_dot("example.com.") == "example.com" + assert _strip_trailing_dot("example.com..") == "example.com.." + assert _strip_trailing_dot("example.com.:8080") == "example.com:8080" + # with userinfo — the host:port branch should not fire + assert _strip_trailing_dot("user@example.com.:80") == "user@example.com.:80" + + +def test_get_tldinfo_ipv6(): + "IPv6 addresses through get_tldinfo." + from courlan.urlutils import get_tldinfo + + # bracketed IPv6 + assert get_tldinfo("https://[2001:db8::1]/path") == ("2001:db8::1", "2001:db8::1") + # unbracketed IPv6 (2 colons minimum) + assert get_tldinfo("http://2001:db8::1/path") == ("2001:db8::1", "2001:db8::1") + + +def test_hosts_hex_labels(): + "Hex numeric labels (0xAB) detected as numeric, preventing domain registration." + from courlan.hosts import _is_numeric_label + + assert _is_numeric_label("0xAB") is True + assert _is_numeric_label("0xA") is True + assert _is_numeric_label("0x") is True # empty hex digits + assert _is_numeric_label("0xGG") is False + assert _is_numeric_label("123") is True + assert _is_numeric_label("abc") is False diff --git a/tests/urlstore_tests.py b/tests/urlstore_tests.py index 9005857..16978c3 100644 --- a/tests/urlstore_tests.py +++ b/tests/urlstore_tests.py @@ -117,6 +117,21 @@ def test_urlstore_basics(): my_urls.add_urls(appendleft=["https://example.org/b", "https://example.org/b/"]) assert len(my_urls.urldict["https://example.org"].tuples) == 1 + # visited flag defaults + v = UrlStore() + v.add_urls(["http://a.com/x"]) + assert v.find_unvisited_urls("http://a.com") == ["http://a.com/x"] + assert not v.has_been_visited("http://a.com/x") + v2 = UrlStore() + v2.add_urls(["http://a.com/y"], visited=True) + assert v2.find_unvisited_urls("http://a.com") == [] + assert v2.has_been_visited("http://a.com/y") + + # invalid URLs rejected + inv = UrlStore() + inv.add_urls(["not-a-url", "http://valid.org/ok"]) + assert inv.dump_urls() == ["http://valid.org/ok"] + def test_urlstore_rules(robots_rules): "Test storage and retrieval of crawling rules." @@ -181,19 +196,19 @@ def test_urlstore_compression(): urls = example_urls + test_urls # test loading - url_buffer = UrlStore()._buffer_urls(urls) + url_buffer = UrlStore()._buffer_urls(visited=False, data=urls) assert sum(len(v) for _, v in url_buffer.items()) == len(urls) # compression 1 my_urls = UrlStore(compressed=True) - url_buffer = UrlStore()._buffer_urls(example_urls) + url_buffer = UrlStore()._buffer_urls(visited=False, data=example_urls) my_urls.add_urls(example_urls) assert my_urls.total_url_number() == len(example_urls) assert len(pickle.dumps(my_urls)) < len(pickle.dumps(url_buffer)) assert my_urls.is_known(f"{example_domain}/100") is True # compression 2 my_urls = UrlStore(compressed=True) - url_buffer = UrlStore()._buffer_urls(test_urls) + url_buffer = UrlStore()._buffer_urls(visited=False, data=test_urls) my_urls.add_urls(test_urls) assert my_urls.total_url_number() == len(test_urls) assert len(pickle.dumps(my_urls)) < len(pickle.dumps(url_buffer)) @@ -447,6 +462,14 @@ def test_dbdump(capsys): captured = capsys.readouterr() assert captured.out.strip() == "http://test.org/this\tFalse" + # print unvisited (direct call, works on all platforms) + unvisited_store = UrlStore() + unvisited_store.add_urls(["https://www.test.org/a", "https://www.test.org/b"]) + unvisited_store.print_unvisited_urls() + captured = capsys.readouterr() + lines = sorted(captured.out.strip().splitlines()) + assert lines == ["https://www.test.org/a", "https://www.test.org/b"] + # dump unvisited, don't test it on Windows if os.name != "nt": # standard @@ -466,8 +489,9 @@ def test_dbdump(capsys): pid = os.getpid() try: for signum in (signal.SIGINT, signal.SIGTERM): - with pytest.raises(SystemExit): + with pytest.raises(SystemExit) as exc_info: os.kill(pid, signum) + assert exc_info.value.code == 1 captured = capsys.readouterr() assert captured.out.strip().endswith("https://www.test.org/2") finally: @@ -663,17 +687,18 @@ def test_urlstore_store_non_http_domain(): assert "ftp://example.org" in store.urldict -def test_store_rules_no_merge(): - "store_rules must not destroy independent http/https entries (uses read-only lookup)." +def test_store_rules_merges_twin(): + "store_rules canonicalizes under the https key, merging twins." store = UrlStore() # force two independent entries (simulates legacy unpickled data) store.urldict["http://host.com"] store.urldict["https://host.com"] rules = RobotFileParser() - store.store_rules("https://host.com", rules) - # http entry must still exist — store_rules is read-only, not a merge - assert "http://host.com" in store.urldict + store.store_rules("http://host.com", rules) + # twins merged under https key + assert "http://host.com" not in store.urldict assert "https://host.com" in store.urldict + assert store.get_rules("https://host.com") is rules def test_twin_merge_on_download(): @@ -730,6 +755,143 @@ def test_twin_discard(): assert store.dump_urls() == [] +def test_done_lifecycle(): + "done starts False, flips True when exhausted, reverts on new URLs." + s = UrlStore() + assert s.done is False + s.add_urls(["http://a.com/1"]) + assert s.done is False + assert s.get_url("http://a.com") is not None + assert s.get_url("http://a.com") is None + assert s.done is True + s.add_urls(["http://a.com/2"]) + assert s.done is False + # done stays False while any domain is still OPEN + s2 = UrlStore() + s2.add_urls(["http://a.com/1", "http://b.com/1"]) + s2.get_url("http://a.com") + s2.get_url("http://a.com") + assert s2.done is False + s2.get_url("http://b.com") + s2.get_url("http://b.com") + assert s2.done is True + + +def test_merge_adds_counts(): + "Twin merge sums counts from both entries." + s = UrlStore(compressed=True) + s.add_urls(["http://m.com/a"]) + s.get_url("http://m.com") + entry = s.urldict.pop("http://m.com") + s.add_urls(["https://m.com/b"]) + s.get_url("https://m.com") + s.urldict["http://m.com"] = entry + s.add_urls(["https://m.com/c"]) + merged = s.urldict["https://m.com"] + assert "http://m.com" not in s.urldict + assert merged.count == 2 + + +def test_merge_copies_src_rules(): + "Twin merge copies rules from src when tgt has none." + s = UrlStore() + s.add_urls(["http://r.com/a"]) + s.store_rules("http://r.com", RobotFileParser()) + entry = s.urldict.pop("http://r.com") + s.add_urls(["https://r.com/b"]) + s.urldict["http://r.com"] = entry + s.add_urls(["https://r.com/c"]) + assert "http://r.com" not in s.urldict + assert s.get_rules("https://r.com") is not None + + +def test_filter_unknown_same_domain(): + "filter_unknown_urls works correctly for many same-domain URLs." + s = UrlStore() + s.add_urls([f"http://z.com/{i:04d}" for i in range(50)]) + assert s.filter_unknown_urls([f"http://z.com/{i:04d}" for i in range(50)]) == [] + + +def test_add_from_html_with_and_without_rules(robots_rules): + "add_from_html passes stored rules to filter_links." + html = '' + for with_rules in (True, False): + s = UrlStore() + s.add_urls(["https://example.org/seed"]) + if with_rules: + s.store_rules("https://example.org", robots_rules) + s.add_from_html(html, "https://example.org/seed") + assert s.is_known("https://example.org/page1") + + +def test_get_download_urls_skips_exhausted(): + "get_download_urls skips non-OPEN domains and continues to OPEN ones." + s = UrlStore() + s.add_urls(["http://a.com/1", "http://b.com/1"]) + s.get_url("http://a.com") + s.get_url("http://a.com") + assert s.urldict["http://a.com"].state != State.OPEN + urls = s.get_download_urls(time_limit=0) + assert len(urls) == 1 + assert "b.com" in urls[0] + + +def test_schedule_integer_division(): + "establish_download_schedule uses floor division for per_domain." + s = UrlStore() + for d in ("http://a.com", "http://b.com", "http://c.com"): + s.add_urls([f"{d}/{i}" for i in range(5)]) + schedule = s.establish_download_schedule(max_urls=7, time_limit=10) + # per_domain = 7 // 3 = 2 → 2 × 3 = 6 + assert len(schedule) == 6 + from collections import Counter + + counts = Counter(url.rsplit("/", 1)[0] for _, url in schedule) + for c in counts.values(): + assert c == 2 + + +def test_schedule_per_domain_fallback(): + "per_domain falls back to 1 when max_urls < num_domains, spreading across domains." + s = UrlStore() + for d in ("http://d1.com", "http://d2.com", "http://d3.com"): + s.add_urls([f"{d}/{i}" for i in range(3)]) + schedule = s.establish_download_schedule(max_urls=2, time_limit=10) + assert len(schedule) == 2 + domains_hit = {url.split("/")[2] for _, url in schedule} + assert len(domains_hit) == 2 + + +def test_schedule_per_domain_cap(): + "establish_download_schedule caps each domain at per_domain URLs." + s = UrlStore() + s.add_urls([f"http://a.com/{i}" for i in range(10)]) + schedule = s.establish_download_schedule(max_urls=3, time_limit=10) + assert len(schedule) == 3 + + +def test_schedule_starts_at_zero(): + "First URL in a fresh domain starts at schedule_secs=0.0." + s = UrlStore() + s.add_urls(["http://fresh.com/a"]) + schedule = s.establish_download_schedule(max_urls=10, time_limit=10) + assert len(schedule) == 1 + assert schedule[0][0] == 0.0 + + +def test_schedule_stores_future_timestamp(): + "establish_download_schedule sets a future timestamp on the domain." + s = UrlStore() + s.add_urls(["http://t.com/a", "http://t.com/b"]) + before = datetime.now() + schedule = s.establish_download_schedule(max_urls=10, time_limit=10) + assert len(schedule) == 2 + key = next(k for k in s.urldict if "t.com" in k) + ts = s.urldict[key].timestamp + assert ts is not None + assert ts >= before + + def test_get_rules_both_directions(): "get_rules resolves http->https and https->http." store = UrlStore() From 69920af3a8130274fa158ed7364aa39e006f890f Mon Sep 17 00:00:00 2001 From: Adrien Barbaresi Date: Sat, 15 Aug 2026 17:30:17 +0200 Subject: [PATCH 8/8] fix coverage and CodeQL warnings --- .github/workflows/tests.yml | 2 +- tests/unit_tests.py | 2 +- tests/urlstore_tests.py | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/.github/workflows/tests.yml b/.github/workflows/tests.yml index 0952ce2..fa8f1a7 100644 --- a/.github/workflows/tests.yml +++ b/.github/workflows/tests.yml @@ -57,7 +57,7 @@ jobs: - name: Test with pytest run: | - python -m pytest --cov=./ --cov-report=xml + python -m pytest --cov=courlan --cov-report=xml - name: Test docs # version matches .readthedocs.yml diff --git a/tests/unit_tests.py b/tests/unit_tests.py index 3f6d6fd..432ff9a 100644 --- a/tests/unit_tests.py +++ b/tests/unit_tests.py @@ -1875,7 +1875,7 @@ def test_clean_url_protocol_detection(): assert clean_url("http://example.org/page") == "http://example.org/page" # double protocol — inner URL extracted result = clean_url("http://redirect.com/?url=http://real.com/page") - assert "real.com" in result + assert result == "http://real.com/page" def test_scrub_url_trailing_slash(): diff --git a/tests/urlstore_tests.py b/tests/urlstore_tests.py index 16978c3..b3577c5 100644 --- a/tests/urlstore_tests.py +++ b/tests/urlstore_tests.py @@ -833,7 +833,7 @@ def test_get_download_urls_skips_exhausted(): assert s.urldict["http://a.com"].state != State.OPEN urls = s.get_download_urls(time_limit=0) assert len(urls) == 1 - assert "b.com" in urls[0] + assert urls[0] == "http://b.com/1" def test_schedule_integer_division():