diff --git a/python/commonCrawlData/.funcignore b/python/commonCrawlData/.funcignore index dc267b1049..cf8a10746e 100644 --- a/python/commonCrawlData/.funcignore +++ b/python/commonCrawlData/.funcignore @@ -4,4 +4,5 @@ __azurite_db*__.json __blobstorage__ __queuestorage__ local.settings.json -test \ No newline at end of file +test +tests diff --git a/python/commonCrawlData/README.md b/python/commonCrawlData/README.md index 016e38d5b6..a7c215eb98 100644 --- a/python/commonCrawlData/README.md +++ b/python/commonCrawlData/README.md @@ -41,7 +41,90 @@ add overhead. Confirm sizing against representative records and the intended worker concurrency before production deployment. These limits do not bound the orchestrator's total activity or result count. -## Offline extraction tests +## Crawl counts and activity scheduling + +The HTTP starter, orchestrator, and initial index activity share the policy in +`crawl_limits.py`. The HTTP endpoint remains function-key protected. + +| Setting | Default | Allowed range or budget | +| -------------------------------- | ------------------- | ------------------------------------------------------------------------------------------------------ | +| `limit` | 1,000 | 1-1,000 accepted index records | +| `batch_size` | 50 | 1-50 input records per processing activity | +| `crawl_id` | `2025-13` | Nonempty collection identifier without whitespace, control characters, or `/`, `\`, `?`, `#`, `&`, `%` | +| `url_prefix` | Required | Nonblank search string; existing prefix patterns are supported | +| Direct `get_urls` activity limit | 100 | Same 1-1,000 range; orchestration calls pass their explicit limit | +| Outstanding activities | 4 per orchestration | Processing and batch saves use separate windows and wait before starting the next window | +| Total logical activity calls | At most 2,002 | One index call, up to 1,000 processing calls and 1,000 batch saves, and one combined save | + +For `N` accepted records and batch size `B`, there are `ceil(N / B)` batches, +with at most `2 * ceil(N / B) + 2` logical activity calls. The default 1,000/50 +crawl has 20 batches and at most 42 calls. These counts are not HTTP-attempt +budgets or exactly-once execution guarantees; the existing per-record network +retries remain unchanged. + +Missing or empty numeric HTTP query values retain their defaults. Other values +must be decimal integers within the documented range; surrounding whitespace, +an optional `+`, and leading zeros are accepted. Zero, negatives, fractions, +exponents, and malformed values are rejected. Missing direct-orchestration +fields retain defaults, but explicit null/empty/boolean/fractional values do not. +Invalid starter requests return HTTP 400 JSON, for example: + +```json +{ + "status": "error", + "message": "batch_size must be an integer between 1 and 50" +} +``` + +No orchestration is started for invalid HTTP input. Invalid direct orchestration +input returns an explicit error payload before scheduling any activity; Durable +may report the instance as **Completed**, with `status: error` in its output, +rather than **Failed**. Activity exceptions still propagate. + +The initial index request sends the validated limit and counts streamed JSON +records locally. More records than requested cause an explicit error, not silent +truncation. The orchestrator independently rejects excess/malformed index output +before processing. The index connect/read timeouts are 5/30 seconds; a read timeout +limits inactivity, not the entire download duration. The existing logged non-200 +index response still returns an empty result, so it remains indistinguishable +from no captures to the orchestrator. + +Rejected archive records may yield fewer extraction results, including empty +batches. Empty batches keep their batch saves and original numbering. The combined +save runs only when some data was extracted. `total_urls_processed` counts accepted +input records, not successful downloads; `restaurants_extracted` counts actual +results. Result formats, combined blob merging, and the starter's Durable status +response are unchanged. + +`host.json` initially limits loaded activities and orchestrators to one each per +worker. Four outstanding tasks can include queued tasks; this is not a promise of +four simultaneous executions. Do not increase activity concurrency without +reviewing worker threads/processes and the save activity's shared container state. +Neither these host settings nor per-orchestration windows are a global rate limit. +The numeric limits are conservative starting policy, not measured worker capacity. +Record counts do not impose byte limits on initial-index lines, extraction objects, +or retained workflow history. Per-record byte caps also do not establish a total +worker-memory ceiling. + +This example retains the Python v1 (`function.json`) programming model and requires +`azure-functions-durable<2`. The Durable Python SDK 2.x does not support that model; +do not upgrade the SDK major version without a separately reviewed migration. +The Requests, urllib3, and warcio pins from the per-record fix are unchanged. + +### Replay and rollout + +Output names use one recorded Durable UTC timestamp for all saves in a crawl. +Domain naming uses tldextract's bundled suffix list without network fetching or +disk caching. Its snapshot can classify rare/new suffixes differently from an +older downloaded list; verify representative domains before rollout. + +This change alters activity scheduling history. Stop new starts and drain active +instances on their old code before deployment. If draining is not possible, use +a separately versioned deployment/task hub. Do not change the policy constants, +dependencies, or bundled suffix data underneath active instances without the same +replay-compatibility review. + +## Offline tests Use Python 3.11 or later. From the repository root on Windows: @@ -50,13 +133,16 @@ Set-Location python\commonCrawlData python -m venv .venv .\.venv\Scripts\python.exe -m pip install -r requirements.txt .\.venv\Scripts\python.exe -m unittest discover -s tests -p "test_*.py" -.\.venv\Scripts\python.exe -m compileall -q process_url_batch tests +.\.venv\Scripts\python.exe -m compileall -q crawl_limits.py http_start get_urls orchestrator process_url_batch save_results tests ``` The tests use mocked HTTP adapters and small synthetic WARC/gzip records, not live Common Crawl, Azure storage, or credentials. They also exercise real HTTP session behavior and real schema extraction, check cleanup and batch continuation, -and measure bounded decoder allocation in an isolated process. +and measure bounded decoder allocation in an isolated process. The crawl tests also +cover parameter boundaries, zero scheduling on rejection, both activity windows, +total calls, partial/empty extraction, output contracts, and replay-safe naming. +Run the suites together to check the record-size and crawl-count fixes coexist. ## Trademarks diff --git a/python/commonCrawlData/crawl_limits.py b/python/commonCrawlData/crawl_limits.py new file mode 100644 index 0000000000..651eddcaad --- /dev/null +++ b/python/commonCrawlData/crawl_limits.py @@ -0,0 +1,73 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +import re + + +DEFAULT_LIMIT = 1000 +DEFAULT_BATCH_SIZE = 50 +DEFAULT_INDEX_LIMIT = 100 +DEFAULT_CRAWL_ID = "2025-13" +MAX_URLS = 1000 +MAX_BATCH_SIZE = 50 +MAX_IN_FLIGHT_ACTIVITIES = 4 +MAX_BATCHES = MAX_URLS +MAX_ACTIVITY_CALLS = 2 * MAX_BATCHES + 2 + + +class CrawlValidationError(ValueError): + """Invalid crawl input or an activity result outside the crawl budget.""" + + +def bounded_positive_integer(value, name, maximum): + message = f"{name} must be an integer between 1 and {maximum}" + if type(value) is int: + number = value + elif isinstance(value, str): + value = value.strip() + if not re.fullmatch(r"[+-]?[0-9]+", value) or value.startswith("-"): + raise CrawlValidationError(message) + digits = value.lstrip("+").lstrip("0") or "0" + ceiling = str(maximum) + if len(digits) > len(ceiling) or ( + len(digits) == len(ceiling) and digits > ceiling + ): + raise CrawlValidationError(message) + number = int(digits) + else: + raise CrawlValidationError(message) + if not 1 <= number <= maximum: + raise CrawlValidationError(message) + return number + + +def validate_search_params(params, default_limit=DEFAULT_LIMIT): + if not isinstance(params, dict): + raise CrawlValidationError("Crawl input must be an object") + limit = bounded_positive_integer( + params.get("limit", default_limit), "limit", MAX_URLS + ) + url_prefix = params.get("url_prefix") + if not isinstance(url_prefix, str) or not url_prefix.strip(): + raise CrawlValidationError("url_prefix must be a nonempty string") + crawl_id = params.get("crawl_id", DEFAULT_CRAWL_ID) + if ( + not isinstance(crawl_id, str) + or not crawl_id + or any( + not char.isprintable() or char.isspace() or char in "/\\?#&%" + for char in crawl_id + ) + ): + raise CrawlValidationError( + "crawl_id must be a nonempty string without whitespace or URL delimiters" + ) + return {"limit": limit, "crawl_id": crawl_id, "url_prefix": url_prefix} + + +def validate_crawl_params(params): + validated = validate_search_params(params) + validated["batch_size"] = bounded_positive_integer( + params.get("batch_size", DEFAULT_BATCH_SIZE), "batch_size", MAX_BATCH_SIZE + ) + return validated diff --git a/python/commonCrawlData/get_urls/__init__.py b/python/commonCrawlData/get_urls/__init__.py index ad65c56792..c87b624e0b 100644 --- a/python/commonCrawlData/get_urls/__init__.py +++ b/python/commonCrawlData/get_urls/__init__.py @@ -1,49 +1,76 @@ # Copyright (c) Microsoft Corporation and Henry Lucco. # Licensed under the MIT License. -import logging -import azure.functions as func import json +import logging + import requests +from crawl_limits import ( + DEFAULT_CRAWL_ID, + DEFAULT_INDEX_LIMIT, + CrawlValidationError, + validate_search_params, +) + + +INDEX_TIMEOUT = (5, 30) def main(params: str) -> list: """Activity function to get URLs from Common Crawl""" - if isinstance(params, str): - try: + try: + if isinstance(params, str): params = json.loads(params) - except json.JSONDecodeError: - params = {} - - limit = params.get("limit", 100) - crawl_id = params.get("crawl_id", "2025-13") - url_prefix = params.get("url_prefix", "") - - if not url_prefix: - return {"status": "error", "message": "No URL search query provided"} - - logging.info(f"Getting up to {limit} URLs from crawl {crawl_id}") - return get_urls(limit, crawl_id, url_prefix) + if not isinstance(params, dict): + raise CrawlValidationError("Index input must be an object") + except (ValueError, RecursionError) as exc: + logging.error("Invalid Common Crawl index input: expected an object") + raise CrawlValidationError( + "Index input must be an object or JSON object" + ) from exc + return get_urls( + params.get("limit", DEFAULT_INDEX_LIMIT), + params.get("crawl_id", DEFAULT_CRAWL_ID), + params.get("url_prefix", ""), + ) def get_urls(limit, crawl_id, url_prefix): """Query Common Crawl index for restaurant URLs""" - # Common Crawl index URL pattern - cc_index_url = f"https://index.commoncrawl.org/CC-MAIN-{crawl_id}-index" - query = f"{url_prefix}*" - - logging.info(f"Querying Common Crawl index: {cc_index_url}") - logging.info(f"Query: {query}") - - response = requests.get( - cc_index_url + "?" + "url=" + query + "&output=json&limit=" + str(limit) - ) - - if response.status_code != 200: - logging.error(f"Error querying index: {response.status_code}") - return [] - - cc_data = [json.loads(line) for line in response.text.strip().split("\n")] - logging.info(f"Found {len(cc_data)} matching URLs") - - return cc_data + try: + params = validate_search_params( + {"limit": limit, "crawl_id": crawl_id, "url_prefix": url_prefix} + ) + limit = params["limit"] + cc_index_url = f"https://index.commoncrawl.org/CC-MAIN-{crawl_id}-index" + with requests.get( + cc_index_url, + params={"url": f"{url_prefix}*", "output": "json", "limit": limit}, + timeout=INDEX_TIMEOUT, + stream=True, + ) as response: + if response.status_code != 200: + logging.error("Error querying index: %s", response.status_code) + return [] + records = [] + for line in response.iter_lines(): + if not line.strip(): + continue + if len(records) == limit: + raise CrawlValidationError( + f"Index returned more records than the requested limit of {limit}" + ) + try: + record = json.loads(line) + except (ValueError, RecursionError) as exc: + raise CrawlValidationError( + "Invalid JSON record in index response" + ) from exc + if not isinstance(record, dict): + raise CrawlValidationError("Index records must be objects") + records.append(record) + logging.info("Found %s matching URLs", len(records)) + return records + except (CrawlValidationError, requests.exceptions.RequestException) as exc: + logging.error("Common Crawl index request failed: %.200s", exc) + raise diff --git a/python/commonCrawlData/host.json b/python/commonCrawlData/host.json index 06d01bdaa9..6aa39486cc 100644 --- a/python/commonCrawlData/host.json +++ b/python/commonCrawlData/host.json @@ -1,5 +1,11 @@ { "version": "2.0", + "extensions": { + "durableTask": { + "maxConcurrentActivityFunctions": 1, + "maxConcurrentOrchestratorFunctions": 1 + } + }, "logging": { "applicationInsights": { "samplingSettings": { diff --git a/python/commonCrawlData/http_start/__init__.py b/python/commonCrawlData/http_start/__init__.py index e756628cbd..23fb9f2adc 100644 --- a/python/commonCrawlData/http_start/__init__.py +++ b/python/commonCrawlData/http_start/__init__.py @@ -1,45 +1,54 @@ # Copyright (c) Microsoft Corporation and Henry Lucco. # Licensed under the MIT License. +import json import logging import azure.functions as func import azure.durable_functions as df +from crawl_limits import ( + DEFAULT_BATCH_SIZE, + DEFAULT_CRAWL_ID, + DEFAULT_LIMIT, + CrawlValidationError, + validate_crawl_params, +) async def main(req: func.HttpRequest, starter: str) -> func.HttpResponse: """HTTP trigger to start the orchestration""" - client = df.DurableOrchestrationClient(starter) - - # Get parameters from the request - limit_param = req.params.get("limit") - limit = int(limit_param) if limit_param else 1000 - - crawl_id_param = req.params.get("crawl_id") - crawl_id = crawl_id_param if crawl_id_param else "2025-13" - - batch_size_param = req.params.get("batch_size") - batch_size = int(batch_size_param) if batch_size_param else 50 - - url_prefix_param = req.params.get("url_prefix") - if not url_prefix_param: - return {"status": "error", "message": "No URL search query provided"} - - url_prefix = url_prefix_param + params = { + name: req.params.get(name) + for name in ("limit", "batch_size", "crawl_id", "url_prefix") + } + for name, default in ( + ("limit", DEFAULT_LIMIT), + ("batch_size", DEFAULT_BATCH_SIZE), + ("crawl_id", DEFAULT_CRAWL_ID), + ): + if params[name] is None or params[name] == "": + params[name] = default + try: + params = validate_crawl_params(params) + except CrawlValidationError as exc: + logging.warning("Rejected Common Crawl request: %s", exc) + return func.HttpResponse( + json.dumps({"status": "error", "message": str(exc)}), + status_code=400, + mimetype="application/json", + ) logging.info( - f"Starting orchestration with limit={limit}, crawl_id={crawl_id}, batch_size={batch_size}" + "Starting orchestration with limit=%s, crawl_id=%s, batch_size=%s", + params["limit"], + params["crawl_id"], + params["batch_size"], ) - # Start the orchestration + client = df.DurableOrchestrationClient(starter) instance_id = await client.start_new( "orchestrator", None, - { - "limit": limit, - "crawl_id": crawl_id, - "batch_size": batch_size, - "url_prefix": url_prefix, - }, + params, ) logging.info(f"Started orchestration with ID: {instance_id}") diff --git a/python/commonCrawlData/orchestrator/__init__.py b/python/commonCrawlData/orchestrator/__init__.py index d08b302e89..be80772dd4 100644 --- a/python/commonCrawlData/orchestrator/__init__.py +++ b/python/commonCrawlData/orchestrator/__init__.py @@ -2,93 +2,127 @@ # Licensed under the MIT License. import logging -from urllib.parse import urlparse -import azure.functions as func import azure.durable_functions as df -from datetime import datetime import tldextract +from crawl_limits import ( + MAX_ACTIVITY_CALLS, + MAX_BATCHES, + MAX_IN_FLIGHT_ACTIVITIES, + MAX_URLS, + CrawlValidationError, + validate_crawl_params, +) -def orchestrator_function(context: df.DurableOrchestrationContext): - """Orchestrator function that coordinates the extraction of restaurant schema.org data""" - # Get parameters - params = context.get_input() - total_limit = params.get("limit", 1000) - crawl_id = params.get("crawl_id", "2025-13") - batch_size = params.get("batch_size", 50) - url_prefix = params.get("url_prefix", "") +extract_domain = tldextract.TLDExtract(suffix_list_urls=(), cache_dir=None) - if not url_prefix: - return {"status": "error", "message": "No URL search query provided"} - urls = yield context.call_activity( - "get_urls", - {"limit": total_limit, "crawl_id": crawl_id, "url_prefix": url_prefix}, - ) +def _error(message): + logging.error("Common Crawl orchestration rejected: %.200s", message) + return {"status": "error", "message": message} - if not urls or len(urls) == 0: - return {"status": "error", "message": f"No {url_prefix} URLs found"} - # Process URLs in batches - batches = [] - for i in range(0, len(urls), batch_size): - batches.append(urls[i : i + batch_size]) +def _valid_saved_urls(result): + return isinstance(result, dict) and all( + isinstance(result.get(key), str) and result[key].strip() + for key in ("json_url", "csv_url") + ) - logging.info(f"Split {len(urls)} URLs into {len(batches)} batches") - # Process batches in parallel - tasks = [] - for i, batch in enumerate(batches): - task = context.call_activity( - "process_url_batch", - { - "batch": batch, - "batch_number": i + 1, - "total_batches": len(batches), - }, - ) - tasks.append(task) +def orchestrator_function(context: df.DurableOrchestrationContext): + """Orchestrator function that coordinates the extraction of restaurant schema.org data""" + try: + params = validate_crawl_params(context.get_input()) + except CrawlValidationError as exc: + return _error(str(exc)) + batch_size = params["batch_size"] + url_prefix = params["url_prefix"] - # Wait for all batches to complete - batch_results = yield context.task_all(tasks) + urls = yield context.call_activity( + "get_urls", + { + "limit": params["limit"], + "crawl_id": params["crawl_id"], + "url_prefix": url_prefix, + }, + ) - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - parsed_url = tldextract.extract(url_prefix) + if not isinstance(urls, list): + return _error("Index activity must return a list") + if len(urls) > params["limit"] or len(urls) > MAX_URLS: + return _error("Index activity returned more records than the crawl limit") + if any(not isinstance(record, dict) for record in urls): + return _error("Index activity records must be objects") + if not urls: + return _error(f"No {url_prefix} URLs found") + + batch_count = (len(urls) + batch_size - 1) // batch_size + if batch_count > MAX_BATCHES or 2 * batch_count + 2 > MAX_ACTIVITY_CALLS: + return _error("Crawl exceeds its activity budget") + + logging.info("Split %s URLs into %s batches", len(urls), batch_count) + + batch_results = [] + extracted_count = 0 + for start in range(0, batch_count, MAX_IN_FLIGHT_ACTIVITIES): + tasks = [] + for i in range(start, min(start + MAX_IN_FLIGHT_ACTIVITIES, batch_count)): + batch = urls[i * batch_size : (i + 1) * batch_size] + tasks.append( + context.call_activity( + "process_url_batch", + { + "batch": batch, + "batch_number": i + 1, + "total_batches": batch_count, + }, + ) + ) + window_results = yield context.task_all(tasks) + for i, result in enumerate(window_results, start): + if not isinstance(result, dict) or not isinstance( + result.get("results"), list + ): + return _error("Processing activity must return a results list") + results = result["results"] + input_count = min(batch_size, len(urls) - i * batch_size) + if len(results) > input_count or any( + not isinstance(record, dict) for record in results + ): + return _error("Processing activity returned invalid or excessive results") + batch_results.append(results) + extracted_count += len(results) + + timestamp = context.current_utc_datetime.strftime("%Y%m%d_%H%M%S") + parsed_url = extract_domain(url_prefix) batches_container_name = f"{parsed_url.domain}-data-batches" combined_container_name = f"{parsed_url.domain}-data" - # split save operations to account for cases where datasets are large - save_tasks = [] - all_results = [] - for i, batch in enumerate(batch_results): - if batch and "results" in batch: - all_results.extend(batch["results"]) - - batch_json_name = f"{parsed_url.domain}_{i + 1}_{timestamp}.json" - batch_csv_name = f"{parsed_url.domain}_{i + 1}_{timestamp}.csv" - - save_task = context.call_activity( - "save_results", - { - "results": batch["results"], - "json_name": batch_json_name, - "csv_name": batch_csv_name, - "container_name": batches_container_name, - }, - ) - - save_tasks.append(save_task) - - # Wait for all batches to complete - save_results = yield context.task_all(save_tasks) batch_json_urls = [] - for save_result in save_results: - if save_result: - batch_json_urls.append(save_result.get("json_url", "")) - - logging.info(f"Saved {len(save_results)} batch results to {batches_container_name}") - - if all_results: + for start in range(0, batch_count, MAX_IN_FLIGHT_ACTIVITIES): + save_tasks = [] + for i in range(start, min(start + MAX_IN_FLIGHT_ACTIVITIES, batch_count)): + save_tasks.append( + context.call_activity( + "save_results", + { + "results": batch_results[i], + "json_name": f"{parsed_url.domain}_{i + 1}_{timestamp}.json", + "csv_name": f"{parsed_url.domain}_{i + 1}_{timestamp}.csv", + "container_name": batches_container_name, + }, + ) + ) + window_saves = yield context.task_all(save_tasks) + for result in window_saves: + if not _valid_saved_urls(result): + return _error("Save activity must return nonempty JSON and CSV URLs") + batch_json_urls.append(result["json_url"]) + + logging.info( + "Saved %s batch results to %s", len(batch_json_urls), batches_container_name + ) + if extracted_count: combined_json_name = f"{parsed_url.domain}_combined_{timestamp}.json" combined_csv_name = f"{parsed_url.domain}_combined_{timestamp}.csv" @@ -102,13 +136,15 @@ def orchestrator_function(context: df.DurableOrchestrationContext): "container_name": combined_container_name, }, ) + if not _valid_saved_urls(save_result): + return _error("Combined save must return nonempty JSON and CSV URLs") return { "status": "success", "total_urls_processed": len(urls), - "restaurants_extracted": len(all_results), - "combined_json_url": save_result.get("json_url", ""), - "combined_csv_url": save_result.get("csv_url", ""), + "restaurants_extracted": extracted_count, + "combined_json_url": save_result["json_url"], + "combined_csv_url": save_result["csv_url"], } else: return { diff --git a/python/commonCrawlData/requirements.txt b/python/commonCrawlData/requirements.txt index dc1758bdcf..e157ff1af6 100644 --- a/python/commonCrawlData/requirements.txt +++ b/python/commonCrawlData/requirements.txt @@ -1,5 +1,5 @@ azure-functions -azure-functions-durable +azure-functions-durable<2 requests==2.34.2 urllib3==2.8.0 warcio==1.8.1 diff --git a/python/commonCrawlData/tests/test_crawl_limits.py b/python/commonCrawlData/tests/test_crawl_limits.py new file mode 100644 index 0000000000..8b1f3b2223 --- /dev/null +++ b/python/commonCrawlData/tests/test_crawl_limits.py @@ -0,0 +1,88 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +import unittest + +from crawl_limits import ( + DEFAULT_BATCH_SIZE, + DEFAULT_CRAWL_ID, + DEFAULT_LIMIT, + MAX_BATCH_SIZE, + MAX_URLS, + CrawlValidationError, + bounded_positive_integer, + validate_crawl_params, + validate_search_params, +) + + +class CrawlLimitsTests(unittest.TestCase): + def test_positive_integer_boundaries(self): + for name, maximum in (("limit", MAX_URLS), ("batch_size", MAX_BATCH_SIZE)): + for value in (1, maximum, "1", str(maximum)): + with self.subTest(name=name, value=value): + self.assertEqual( + bounded_positive_integer(value, name, maximum), int(value) + ) + for value in ( + 0, -1, maximum + 1, "0", "-1", str(maximum + 1), + None, True, False, 1.0, 1.5, [], {}, + "", " ", "oops", "1.0", "1e1", "1_0", "\u0661", "+", "--1", + ): + with self.subTest(name=name, value=value): + with self.assertRaisesRegex( + CrawlValidationError, f"{name}.*1 and {maximum}" + ): + bounded_positive_integer(value, name, maximum) + + def test_decimal_whitespace_sign_and_leading_zeros(self): + for value in (" +00012 ", "\t12\n", "00012"): + self.assertEqual(bounded_positive_integer(value, "limit", MAX_URLS), 12) + + def test_long_decimal_is_checked_before_integer_conversion(self): + with self.assertRaises(CrawlValidationError) as caught: + bounded_positive_integer("9" * 10000, "limit", MAX_URLS) + self.assertLess(len(str(caught.exception)), 100) + self.assertEqual( + bounded_positive_integer("0" * 10000 + "1", "limit", MAX_URLS), 1 + ) + + def test_defaults_and_search_patterns_are_preserved(self): + for prefix in ("example.com", "*.example.com/path", "example.com/?a=1&b=2"): + self.assertEqual( + validate_crawl_params({"url_prefix": prefix}), + { + "url_prefix": prefix, + "limit": DEFAULT_LIMIT, + "batch_size": DEFAULT_BATCH_SIZE, + "crawl_id": DEFAULT_CRAWL_ID, + }, + ) + self.assertEqual( + validate_search_params({"url_prefix": "example.com"}, default_limit=100)["limit"], + 100, + ) + + def test_direct_input_does_not_default_explicit_invalid_values(self): + for field in ("limit", "batch_size", "crawl_id", "url_prefix"): + for value in (None, "", False, 1.5, [], {}): + with self.subTest(field=field, value=value): + with self.assertRaisesRegex(CrawlValidationError, field): + validate_crawl_params({"url_prefix": "example.com", field: value}) + + def test_input_shape_and_blank_prefix(self): + for params in (None, [], "{}", True, 42, {}, {"url_prefix": " \t"}): + with self.subTest(params=params): + with self.assertRaises(CrawlValidationError): + validate_crawl_params(params) + + def test_crawl_id_cannot_change_index_url_structure(self): + for value in (" ", "2025 13", "2025/13", "x\\y", "x?limit=2", "x#y", + "x&limit=2", "x%2fy", "x\n", "x\x00", "x\x7f"): + with self.subTest(value=value): + with self.assertRaisesRegex(CrawlValidationError, "crawl_id"): + validate_search_params({"url_prefix": "example.com", "crawl_id": value}) + self.assertEqual( + validate_search_params({"url_prefix": "example.com", "crawl_id": "custom-collection"})["crawl_id"], + "custom-collection", + ) diff --git a/python/commonCrawlData/tests/test_crawl_orchestration.py b/python/commonCrawlData/tests/test_crawl_orchestration.py new file mode 100644 index 0000000000..812a7e9070 --- /dev/null +++ b/python/commonCrawlData/tests/test_crawl_orchestration.py @@ -0,0 +1,564 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +from datetime import datetime, timedelta, timezone +import json +from types import SimpleNamespace +import unittest +from unittest import mock +from urllib.parse import parse_qs, urlparse + +import azure.functions as func +from azure.durable_functions.models.history import HistoryEventType +import requests +import tldextract + +from crawl_limits import ( + DEFAULT_CRAWL_ID, + MAX_ACTIVITY_CALLS, + MAX_BATCH_SIZE, + MAX_IN_FLIGHT_ACTIVITIES, + MAX_URLS, + CrawlValidationError, +) +import get_urls as index +import http_start +import orchestrator +import process_url_batch as extraction +from test_process_url_batch import ( + StreamResponse, + index_record, + make_warc, + session_for, +) + + +def url_records(count): + return [{"url": f"https://example.com/{i}"} for i in range(count)] + + +class IndexResponse: + def __init__(self, lines=(), status=200): + self.lines = lines + self.status_code = status + self.closed = False + self.consumed = 0 + + def __enter__(self): + return self + + def __exit__(self, *args): + self.closed = True + + @property + def text(self): + raise AssertionError("Index must not read the entire response") + + @property + def content(self): + raise AssertionError("Index must not read the entire response") + + def iter_lines(self): + for line in self.lines: + self.consumed += 1 + if isinstance(line, Exception): + raise line + yield line + + +class HttpStarterTests(unittest.IsolatedAsyncioTestCase): + async def test_missing_and_empty_http_values_preserve_defaults_and_status_response(self): + for method in ("GET", "POST"): + for extra in ({}, {"limit": "", "batch_size": "", "crawl_id": ""}): + with self.subTest(method=method, extra=extra): + req = func.HttpRequest( + method, "https://localhost/api/start", body=b"", + params={"url_prefix": "example.com", **extra}, + ) + with mock.patch.object(http_start.df, "DurableOrchestrationClient") as factory: + client = factory.return_value + client.start_new = mock.AsyncMock(return_value="fixture-id") + response = await http_start.main(req, "fixture-binding") + self.assertIs(response, client.create_check_status_response.return_value) + client.start_new.assert_awaited_once_with( + "orchestrator", None, + {"limit": 1000, "batch_size": 50, + "crawl_id": DEFAULT_CRAWL_ID, "url_prefix": "example.com"}, + ) + client.create_check_status_response.assert_called_once_with(req, "fixture-id") + + async def test_rejection_is_json_400_before_client_creation(self): + cases = [ + {name: value} + for name, maximum in (("limit", MAX_URLS), ("batch_size", MAX_BATCH_SIZE)) + for value in ("oops", " ", "0", "-1", "1.5", str(maximum + 1)) + ] + [{"url_prefix": ""}, {"url_prefix": " "}, {"crawl_id": "x?limit=99"}] + for extra in cases: + with self.subTest(extra=extra): + req = func.HttpRequest( + "GET", "https://localhost/api/start", body=b"", + params={"url_prefix": "example.com", **extra}, + ) + with mock.patch.object(http_start.df, "DurableOrchestrationClient") as factory: + response = await http_start.main(req, "fixture-binding") + factory.assert_not_called() + self.assertEqual(response.status_code, 400) + self.assertEqual(response.mimetype, "application/json") + error = json.loads(response.get_body()) + self.assertEqual(error["status"], "error") + self.assertIn(next(iter(extra)), error["message"]) + + async def test_valid_boundaries_and_numeric_normalization_start(self): + for limit, batch_size in (("1", "1"), (str(MAX_URLS), str(MAX_BATCH_SIZE)), (" +0002 ", "01")): + with self.subTest(limit=limit, batch_size=batch_size): + req = func.HttpRequest( + "GET", "https://localhost/api/start", body=b"", + params={"url_prefix": "example.com", "limit": limit, "batch_size": batch_size}, + ) + with mock.patch.object(http_start.df, "DurableOrchestrationClient") as factory: + factory.return_value.start_new = mock.AsyncMock(return_value="fixture-id") + await http_start.main(req, "fixture-binding") + params = factory.return_value.start_new.await_args.args[2] + self.assertEqual(params["limit"], int(limit)) + self.assertEqual(params["batch_size"], int(batch_size)) + + +class IndexTests(unittest.TestCase): + def fetch(self, response, limit=2, prefix="example.com"): + with mock.patch.object(index.requests, "get", return_value=response) as get: + result = index.get_urls(limit, DEFAULT_CRAWL_ID, prefix) + self.assertTrue(response.closed) + return result, get.call_args + + def test_empty_blank_and_exact_limit_responses(self): + for lines, expected in ( + ([], []), ([b"", b" \t"], []), + ([b"", b'{"url":"a"}', b'{"url":"b"}'], [{"url": "a"}, {"url": "b"}]), + ): + with self.subTest(lines=lines): + result, call = self.fetch(IndexResponse(lines)) + self.assertEqual(result, expected) + self.assertEqual(call.kwargs["params"]["limit"], 2) + self.assertEqual(call.kwargs["timeout"], (5, 30)) + self.assertIs(call.kwargs["stream"], True) + + def test_prefix_is_one_encoded_query_parameter(self): + prefix = "example.com/?a=1&limit=9999" + _, call = self.fetch(IndexResponse([]), prefix=prefix) + prepared = requests.Request("GET", call.args[0], params=call.kwargs["params"]).prepare() + query = parse_qs(urlparse(prepared.url).query) + self.assertEqual(query, {"url": [prefix + "*"], "output": ["json"], "limit": ["2"]}) + + def test_overflow_stops_at_sentinel_and_closes_response(self): + response = IndexResponse([b"{}"] * 10) + with mock.patch.object(index.requests, "get", return_value=response) as get: + with self.assertRaisesRegex(CrawlValidationError, "more records"): + index.get_urls(2, DEFAULT_CRAWL_ID, "example.com") + self.assertEqual(response.consumed, 3) + self.assertTrue(response.closed) + self.assertEqual(get.call_args.kwargs["params"]["limit"], 2) + + def test_requested_maximum_and_one_extra_record(self): + result, _ = self.fetch(IndexResponse([b"{}"] * MAX_URLS), limit=MAX_URLS) + self.assertEqual(len(result), MAX_URLS) + response = IndexResponse([b"{}"] * (MAX_URLS + 1)) + with mock.patch.object(index.requests, "get", return_value=response): + with self.assertRaisesRegex(CrawlValidationError, "more records"): + index.get_urls(MAX_URLS, DEFAULT_CRAWL_ID, "example.com") + self.assertTrue(response.closed) + + def test_parse_shape_and_iterator_errors_close_response(self): + for bad in (b"invalid", b"[]", b"null", b"false", b"\xff", + b"[" * 2000, requests.exceptions.ConnectionError("read failed")): + with self.subTest(bad=str(bad)[:40]): + response = IndexResponse([bad]) + with mock.patch.object(index.requests, "get", return_value=response): + with self.assertLogs(level="ERROR"): + with self.assertRaises((CrawlValidationError, requests.exceptions.ConnectionError)): + index.get_urls(2, DEFAULT_CRAWL_ID, "example.com") + self.assertTrue(response.closed) + + def test_non_200_retains_logged_empty_result_and_closes(self): + response = IndexResponse([b"must not read"], status=503) + with self.assertLogs(level="ERROR"): + result, _ = self.fetch(response) + self.assertEqual(result, []) + self.assertEqual(response.consumed, 0) + + def test_timeout_propagates_instead_of_becoming_empty(self): + with mock.patch.object(index.requests, "get", side_effect=requests.exceptions.Timeout): + with self.assertLogs(level="ERROR"): + with self.assertRaises(requests.exceptions.Timeout): + index.get_urls(2, DEFAULT_CRAWL_ID, "example.com") + + def test_wrapper_defaults_and_json_object_input(self): + for params in ({"url_prefix": "example.com"}, '{"url_prefix":"example.com"}'): + response = IndexResponse([b"{}"]) + with mock.patch.object(index.requests, "get", return_value=response) as get: + self.assertEqual(index.main(params), [{}]) + self.assertEqual(get.call_args.kwargs["params"]["limit"], 100) + self.assertTrue(response.closed) + + def test_wrapper_and_direct_helper_reject_invalid_inputs_without_http(self): + for params in ("bad JSON", "[]", "null", [], None, True, 7): + with self.subTest(params=params): + with mock.patch.object(index.requests, "get") as get: + with self.assertLogs(level="ERROR"): + with self.assertRaises(CrawlValidationError): + index.main(params) + get.assert_not_called() + for limit, crawl_id, prefix in ( + (0, DEFAULT_CRAWL_ID, "example.com"), + (MAX_URLS + 1, DEFAULT_CRAWL_ID, "example.com"), + (True, DEFAULT_CRAWL_ID, "example.com"), + (2, "x?limit=999", "example.com"), (2, DEFAULT_CRAWL_ID, " "), + ): + with self.subTest(limit=limit, crawl_id=crawl_id, prefix=prefix): + with mock.patch.object(index.requests, "get") as get: + with self.assertLogs(level="ERROR"): + with self.assertRaises(CrawlValidationError): + index.get_urls(limit, crawl_id, prefix) + get.assert_not_called() + + +class DurableDriver: + """Resolve only yielded tasks, tracking scheduled but unfinished work.""" + + def __init__(self, params, urls, process=None, save=None): + self.params = params + self.urls = urls + self.process = process + self.save = save + self.calls = [] + self.outstanding = set() + self.peak = 0 + self.groups = [] + self.current_utc_datetime = datetime(2026, 10, 22, tzinfo=timezone.utc) + + def get_input(self): + return self.params + + def call_activity(self, name, payload): + task = SimpleNamespace(id=len(self.calls), name=name, payload=payload) + self.calls.append(task) + self.outstanding.add(task.id) + self.peak = max(self.peak, len(self.outstanding)) + return task + + def task_all(self, tasks): + return tasks + + def result_for(self, task): + if task.name == "get_urls": + if isinstance(self.urls, Exception): + raise self.urls + return self.urls + if task.name == "process_url_batch": + if self.process is not None: + return self.process(task.payload) + return {"results": [{"name": item["url"]} for item in task.payload["batch"]]} + if self.save is not None: + return self.save(task.payload) + return { + "json_url": f"https://storage.test/{task.payload['json_name']}", + "csv_url": f"https://storage.test/{task.payload['csv_name']}", + } + + def resolve(self, pending): + tasks = pending if isinstance(pending, list) else [pending] + self.groups.append([task.name for task in tasks]) + results = [] + for task in tasks: + try: + results.append(self.result_for(task)) + finally: + self.outstanding.remove(task.id) + self.current_utc_datetime += timedelta(seconds=1) + return results if isinstance(pending, list) else results[0] + + def run(self): + generator = orchestrator.orchestrator_function(self) + try: + pending = next(generator) + while True: + try: + result = self.resolve(pending) + except Exception as exc: + pending = generator.throw(exc) + else: + pending = generator.send(result) + except StopIteration as done: + return done.value + + def named(self, name): + return [task.payload for task in self.calls if task.name == name] + + +class OrchestrationTests(unittest.TestCase): + def test_real_sdk_entry_point_replays_processing_and_save_windows(self): + params = {"url_prefix": "example.com", "limit": 9, "batch_size": 2} + clock = datetime(2026, 10, 22, tzinfo=timezone.utc) + + def event(kind, event_id, **extra): + return { + "EventType": kind.value, "EventId": event_id, "IsPlayed": True, + "Timestamp": clock.isoformat(), **extra, + } + + history = [ + event(HistoryEventType.ORCHESTRATOR_STARTED, -1), + event( + HistoryEventType.EXECUTION_STARTED, -1, Name="orchestrator", + Input=json.dumps(params), Version=None, + ), + ] + + def replay(): + context = { + "history": history, "instanceId": "fixture", "isReplaying": True, + "parentInstanceId": None, "input": json.dumps(params), + } + return json.loads(orchestrator.main( + SimpleNamespace(body=json.dumps(context)) + )) + + state = replay() + groups = [] + saved_batches = [] + combined = None + next_id = 0 + batch_count = (params["limit"] + params["batch_size"] - 1) // params["batch_size"] + while not state["isDone"]: + self.assertLess(next_id, 2 * batch_count + 2, "SDK replay stopped making progress") + group = state["actions"][-1] + groups.append([action["functionName"] for action in group]) + self.assertLessEqual(len(group), MAX_IN_FLIGHT_ACTIVITIES) + clock += timedelta(seconds=1) + history.append(event(HistoryEventType.ORCHESTRATOR_STARTED, -1)) + scheduled = list(enumerate(group, next_id)) + for task_id, action in scheduled: + history.append(event( + HistoryEventType.TASK_SCHEDULED, task_id, + Name=action["functionName"], Input=action["input"], + )) + if action["functionName"] == "save_results": + payload = json.loads(action["input"]) + if "results" in payload: + saved_batches.append(payload) + else: + combined = payload + for task_id, action in reversed(scheduled): + payload = json.loads(action["input"]) + if action["functionName"] == "get_urls": + result = url_records(params["limit"]) + elif action["functionName"] == "process_url_batch": + result = {"results": [{"name": row["url"]} for row in payload["batch"]]} + else: + result = { + "json_url": f"https://storage.test/{payload['json_name']}", + "csv_url": f"https://storage.test/{payload['csv_name']}", + } + history.append(event( + HistoryEventType.TASK_COMPLETED, task_id + 100, + TaskScheduledId=task_id, Result=json.dumps(result), + )) + next_id += len(group) + state = replay() + self.assertEqual(groups, [ + ["get_urls"], ["process_url_batch"] * 4, ["process_url_batch"], + ["save_results"] * 4, ["save_results"], ["save_results"], + ]) + self.assertEqual(state["output"]["total_urls_processed"], params["limit"]) + self.assertEqual(state["output"]["restaurants_extracted"], params["limit"]) + self.assertEqual( + [row["name"] for batch in saved_batches for row in batch["results"]], + [row["url"] for row in url_records(params["limit"])], + ) + self.assertEqual( + combined["json_files_urls"], + [f"https://storage.test/{batch['json_name']}" for batch in saved_batches], + ) + self.assertEqual(replay(), state) + + def test_invalid_direct_inputs_schedule_nothing(self): + cases = [None, [], "{}", {}, {"url_prefix": " "}] + cases += [ + {"url_prefix": "example.com", name: value} + for name, maximum in (("limit", MAX_URLS), ("batch_size", MAX_BATCH_SIZE)) + for value in (0, -1, None, True, 1.5, "", "oops", maximum + 1) + ] + cases += [{"url_prefix": "example.com", "crawl_id": "x?limit=999"}] + for params in cases: + with self.subTest(params=params): + driver = DurableDriver(params, []) + self.assertEqual(driver.run()["status"], "error") + self.assertEqual(driver.calls, []) + + def test_direct_defaults_reach_index_unchanged(self): + driver = DurableDriver({"url_prefix": "example.com"}, []) + self.assertEqual(driver.run()["status"], "error") + self.assertEqual(driver.named("get_urls"), [ + {"limit": 1000, "crawl_id": DEFAULT_CRAWL_ID, "url_prefix": "example.com"} + ]) + + def test_bad_or_excess_index_results_schedule_no_processing(self): + for urls in (None, False, "", {}, [None], url_records(3), url_records(MAX_URLS + 1)): + with self.subTest(kind=type(urls).__name__): + driver = DurableDriver({"url_prefix": "example.com", "limit": 2}, urls) + self.assertEqual(driver.run()["status"], "error") + self.assertEqual([task.name for task in driver.calls], ["get_urls"]) + + def test_windows_total_budget_order_and_storage_contract(self): + for count, batch_size in ((1, 50), (50, 50), (51, 50), (4, 1), (5, 1), + (9, 2), (1000, 50), (1000, 1)): + with self.subTest(count=count, batch_size=batch_size): + urls = url_records(count) + driver = DurableDriver( + {"url_prefix": "example.com", "limit": count, "batch_size": batch_size}, + urls, + ) + result = driver.run() + batches = (count + batch_size - 1) // batch_size + self.assertEqual(len(driver.calls), 2 * batches + 2) + self.assertLessEqual(len(driver.calls), MAX_ACTIVITY_CALLS) + self.assertLessEqual(driver.peak, MAX_IN_FLIGHT_ACTIVITIES) + self.assertEqual(driver.outstanding, set()) + processed = driver.named("process_url_batch") + self.assertEqual([item["batch_number"] for item in processed], list(range(1, batches + 1))) + self.assertTrue(all(item["total_batches"] == batches for item in processed)) + self.assertEqual([row for item in processed for row in item["batch"]], urls) + saves = driver.named("save_results") + self.assertEqual([row["name"] for item in saves[:-1] for row in item["results"]], + [row["url"] for row in urls]) + self.assertEqual(saves[-1]["json_files_urls"], [ + f"https://storage.test/{item['json_name']}" for item in saves[:-1] + ]) + self.assertEqual(result, { + "status": "success", "total_urls_processed": count, + "restaurants_extracted": count, + "combined_json_url": f"https://storage.test/{saves[-1]['json_name']}", + "combined_csv_url": f"https://storage.test/{saves[-1]['csv_name']}", + }) + self.assertEqual( + [group[0] for group in driver.groups], + ["get_urls"] + ["process_url_batch"] * ((batches + 3) // 4) + + ["save_results"] * ((batches + 3) // 4) + ["save_results"], + ) + + def test_invalid_first_processing_window_stops_later_windows(self): + for result in (None, {}, {"results": None}, {"results": [False]}, {"results": [{}, {}]}): + with self.subTest(result=result): + driver = DurableDriver( + {"url_prefix": "example.com", "batch_size": 1}, url_records(9), + process=lambda _: result, + ) + self.assertEqual(driver.run()["status"], "error") + self.assertEqual(len(driver.named("process_url_batch")), MAX_IN_FLIGHT_ACTIVITIES) + self.assertEqual(driver.named("save_results"), []) + + def test_invalid_save_window_or_combined_result_cannot_report_success(self): + for bad in (None, {}, {"json_url": "", "csv_url": "csv"}, + {"json_url": "json", "csv_url": " "}, + {"json_url": 123, "csv_url": "csv"}): + with self.subTest(bad=bad): + driver = DurableDriver( + {"url_prefix": "example.com", "batch_size": 1}, url_records(9), + save=lambda _: bad, + ) + self.assertEqual(driver.run()["status"], "error") + self.assertEqual(len(driver.named("save_results")), MAX_IN_FLIGHT_ACTIVITIES) + driver = DurableDriver( + {"url_prefix": "example.com"}, url_records(1), + save=lambda payload: {} if "json_files_urls" in payload else {"json_url": "json", "csv_url": "csv"}, + ) + self.assertEqual(driver.run()["status"], "error") + self.assertEqual(len(driver.named("save_results")), 2) + + def test_actual_result_lengths_override_claimed_metadata(self): + driver = DurableDriver( + {"url_prefix": "example.com", "batch_size": 1}, url_records(3), + process=lambda payload: { + "results": [] if payload["batch_number"] == 2 else [{"name": "fixture"}], + "extracted_count": 99999, + }, + ) + result = driver.run() + self.assertEqual(result["total_urls_processed"], 3) + self.assertEqual(result["restaurants_extracted"], 2) + self.assertEqual(driver.named("save_results")[1]["results"], []) + names = [item["json_name"] for item in driver.named("save_results")[:-1]] + self.assertTrue(all(f"_{i}_" in name for i, name in enumerate(names, 1))) + + def test_activity_failures_propagate_without_scheduling_later_groups(self): + for phase in ("get_urls", "process_url_batch", "save_results"): + with self.subTest(phase=phase): + failure = RuntimeError("fixture failure") + + def fail(_): + raise failure + + driver = DurableDriver( + {"url_prefix": "example.com", "batch_size": 1}, + failure if phase == "get_urls" else url_records(9), + process=fail if phase == "process_url_batch" else None, + save=fail if phase == "save_results" else None, + ) + with self.assertRaisesRegex(RuntimeError, "fixture failure"): + driver.run() + self.assertLessEqual(driver.peak, MAX_IN_FLIGHT_ACTIVITIES) + self.assertEqual(len(driver.named(phase)), 1 if phase == "get_urls" else 4) + if phase == "process_url_batch": + self.assertEqual(driver.named("save_results"), []) + + def test_replay_uses_recorded_clock_and_one_suffix_across_save_windows(self): + histories = [] + for year in (2000, 2030): + clock = mock.Mock() + clock.now.return_value = datetime(year, 1, 1) + driver = DurableDriver( + {"url_prefix": "example.com", "batch_size": 1}, url_records(9) + ) + with mock.patch.object(orchestrator, "datetime", clock, create=True): + result = driver.run() + saves = driver.named("save_results") + suffixes = {item["json_name"].split("_", 2)[-1] for item in saves[:-1]} + self.assertEqual(len(suffixes), 1) + self.assertTrue(saves[-1]["json_name"].endswith(next(iter(suffixes)))) + histories.append(([(task.name, task.payload) for task in driver.calls], result)) + self.assertEqual(histories[0], histories[1]) + + def test_domain_extraction_uses_bundled_multilabel_suffix_without_network(self): + extractor = tldextract.TLDExtract(suffix_list_urls=(), cache_dir=None) + driver = DurableDriver({"url_prefix": "www.restaurant.co.uk"}, url_records(1)) + with mock.patch.object(orchestrator, "extract_domain", extractor): + with mock.patch.object(requests.sessions.Session, "get", side_effect=AssertionError("Network in orchestrator")): + self.assertEqual(driver.run()["status"], "success") + self.assertEqual(driver.named("save_results")[0]["container_name"], "restaurant-data-batches") + + def test_real_batch_rejections_preserve_partial_and_empty_results(self): + body = make_warc() + bad = {"url": "https://example.test/bad", "length": "invalid"} + for records, expected_count in (([bad, index_record(len(body))], 1), ([bad], 0)): + with self.subTest(expected_count=expected_count): + session, adapter = session_for(StreamResponse(body)) + driver = DurableDriver( + {"url_prefix": "example.com"}, records, process=extraction.main + ) + with mock.patch.object(extraction, "_CrawlSession", return_value=session) as factory: + with mock.patch.object(extraction, "check_circuit_breaker", return_value=False): + result = driver.run() + if expected_count: + factory.assert_called_once() + self.assertTrue(adapter.closed) + self.assertEqual(result["total_urls_processed"], len(records)) + self.assertEqual(result["restaurants_extracted"], 1) + self.assertEqual(len(driver.named("save_results")), 2) + else: + factory.assert_not_called() + session.close() + self.assertEqual(result, { + "status": "warning", + "message": "No restaurant data extracted from any batch", + }) + self.assertEqual(len(driver.named("save_results")), 1) + self.assertEqual(driver.named("save_results")[0]["results"], [])