Public Access
730 lines
26 KiB
Python
730 lines
26 KiB
Python
"""Offline tests for the bounded Wikibooks recipe staging service."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import copy
|
||
import json
|
||
import math
|
||
import threading
|
||
import uuid
|
||
from contextlib import nullcontext
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
from queue import Empty, Queue
|
||
|
||
import pytest
|
||
from sqlalchemy import text
|
||
from sqlalchemy.dialects import postgresql
|
||
from sqlalchemy.orm import sessionmaker
|
||
|
||
from app.models import RecipeImportStaging, RecipeImportStatus
|
||
from app.services.wikibooks_import import (
|
||
CC_BY_SA_4_LICENSE_URL,
|
||
WIKIBOOKS_API_URL,
|
||
WikibooksImportError,
|
||
WikibooksImportService,
|
||
WikibooksResponseError,
|
||
_normalize_ingredient,
|
||
canonical_json_sha256,
|
||
)
|
||
|
||
|
||
FIXTURE_DIR = Path(__file__).parent / "fixtures" / "wikibooks"
|
||
|
||
|
||
class FakeResponse:
|
||
def __init__(self, payload, *, status_code=200, headers=None):
|
||
self.status_code = status_code
|
||
self.headers = {"Content-Length": str(len(json.dumps(payload).encode("utf-8")))}
|
||
self.headers.update(headers or {})
|
||
self._body = json.dumps(payload).encode("utf-8")
|
||
self.closed = False
|
||
|
||
def iter_content(self, chunk_size):
|
||
for offset in range(0, len(self._body), chunk_size):
|
||
yield self._body[offset : offset + chunk_size]
|
||
|
||
def close(self):
|
||
self.closed = True
|
||
|
||
|
||
class FakeSession:
|
||
def __init__(self, responses):
|
||
self.responses = list(responses)
|
||
self.calls = []
|
||
|
||
def get(self, url, **kwargs):
|
||
self.calls.append((url, kwargs))
|
||
return self.responses.pop(0)
|
||
|
||
|
||
class FakeStageResult:
|
||
def __init__(self, value):
|
||
self.value = value
|
||
|
||
def scalar_one(self):
|
||
return self.value
|
||
|
||
def scalars(self):
|
||
return self
|
||
|
||
def all(self):
|
||
return [] if self.value is None else [self.value]
|
||
|
||
|
||
class FakeStageDb:
|
||
def __init__(self):
|
||
self.winner = object()
|
||
self.statements = []
|
||
self.no_autoflush = nullcontext()
|
||
|
||
def execute(self, statement):
|
||
self.statements.append(statement)
|
||
return FakeStageResult(self.winner)
|
||
|
||
def flush(self):
|
||
pass
|
||
|
||
|
||
def _parse_fixture(name: str):
|
||
return json.loads((FIXTURE_DIR / name).read_text())["parse"]
|
||
|
||
|
||
def test_stage_discovered_pages_stages_each_title(monkeypatch):
|
||
service = WikibooksImportService("MealPlanner recipe test")
|
||
titles = ["Cookbook:Apple Pie I", "Cookbook:Banana Bread"]
|
||
staged = [object(), object()]
|
||
monkeypatch.setattr(service, "discover_recipe_titles", lambda limit: titles[:limit])
|
||
monkeypatch.setattr(
|
||
service,
|
||
"stage_page",
|
||
lambda db, title: staged[titles.index(title)],
|
||
)
|
||
|
||
assert service.stage_discovered_pages(object(), limit=2) == staged
|
||
|
||
|
||
def test_stage_discovered_pages_uses_feedback_queries(monkeypatch):
|
||
service = WikibooksImportService("MealPlanner recipe test")
|
||
titles = ["Cookbook:Apple Pie I"]
|
||
staged = [object()]
|
||
captured = {}
|
||
monkeypatch.setattr(
|
||
service,
|
||
"search_recipe_titles",
|
||
lambda queries, limit: captured.update(queries=queries, limit=limit) or titles,
|
||
)
|
||
monkeypatch.setattr(service, "stage_page", lambda db, title: staged[0])
|
||
|
||
assert service.stage_discovered_pages(
|
||
object(), limit=1, queries=["apple dinner"]
|
||
) == staged
|
||
assert captured == {"queries": ["apple dinner"], "limit": 1}
|
||
|
||
|
||
def _service(session=None, **kwargs):
|
||
return WikibooksImportService(
|
||
"MealPlanner-Wikibooks-Import/1.0 (tests@example.invalid)",
|
||
session=session,
|
||
min_request_interval=0,
|
||
**kwargs,
|
||
)
|
||
|
||
|
||
def test_discovery_uses_fixed_params_and_bounded_continuation():
|
||
session = FakeSession(
|
||
[
|
||
FakeResponse(
|
||
{
|
||
"continue": {"cmcontinue": "page|2", "continue": "-||"},
|
||
"query": {
|
||
"categorymembers": [
|
||
{"pageid": 1, "title": "Cookbook:First"}
|
||
]
|
||
},
|
||
}
|
||
),
|
||
FakeResponse(
|
||
{
|
||
"query": {
|
||
"categorymembers": [
|
||
{"pageid": 2, "title": "Cookbook:Second"}
|
||
]
|
||
}
|
||
}
|
||
),
|
||
]
|
||
)
|
||
|
||
assert _service(session).discover_recipe_titles(2) == ["Cookbook:First", "Cookbook:Second"]
|
||
|
||
first_url, first_kwargs = session.calls[0]
|
||
assert first_url == WIKIBOOKS_API_URL
|
||
assert first_kwargs["params"] == {
|
||
"action": "query",
|
||
"list": "categorymembers",
|
||
"cmtitle": "Category:Recipes",
|
||
"cmtype": "page",
|
||
"cmlimit": 2,
|
||
"format": "json",
|
||
"formatversion": 2,
|
||
"maxlag": 5,
|
||
}
|
||
assert first_kwargs["headers"]["User-Agent"].startswith("MealPlanner-Wikibooks-Import/")
|
||
assert first_kwargs["stream"] is True
|
||
assert first_kwargs["allow_redirects"] is False
|
||
assert session.calls[1][1]["params"]["cmcontinue"] == "page|2"
|
||
assert session.calls[1][1]["params"]["cmlimit"] == 1
|
||
|
||
|
||
def test_feedback_search_uses_cookbook_namespace_and_deduplicates():
|
||
session = FakeSession(
|
||
[
|
||
FakeResponse(
|
||
{
|
||
"query": {
|
||
"search": [
|
||
{"pageid": 1, "title": "Cookbook:Apple Pie I"},
|
||
{"pageid": 2, "title": "Cookbook:Apple Sauce"},
|
||
]
|
||
}
|
||
}
|
||
),
|
||
FakeResponse(
|
||
{
|
||
"query": {
|
||
"search": [
|
||
{"pageid": 1, "title": "Cookbook:Apple Pie I"},
|
||
{"pageid": 3, "title": "Cookbook:Pork Chops"},
|
||
]
|
||
}
|
||
}
|
||
),
|
||
]
|
||
)
|
||
|
||
assert _service(session).search_recipe_titles(
|
||
[" apple dinner ", "pork dinner"], 3
|
||
) == [
|
||
"Cookbook:Apple Pie I",
|
||
"Cookbook:Apple Sauce",
|
||
"Cookbook:Pork Chops",
|
||
]
|
||
assert session.calls[0][1]["params"] == {
|
||
"action": "query",
|
||
"list": "search",
|
||
"srsearch": "apple dinner",
|
||
"srnamespace": 102,
|
||
"srlimit": 3,
|
||
"format": "json",
|
||
"formatversion": 2,
|
||
"maxlag": 5,
|
||
}
|
||
assert session.calls[1][1]["params"]["srsearch"] == "pork dinner"
|
||
assert session.calls[1][1]["params"]["srlimit"] == 1
|
||
|
||
|
||
def test_fetch_retries_retry_after_and_mediawiki_maxlag():
|
||
parse_payload = _parse_fixture("apple_glazed_pork_chops_parse.json")
|
||
session = FakeSession(
|
||
[
|
||
FakeResponse({"error": "busy"}, status_code=429, headers={"Retry-After": "3"}),
|
||
FakeResponse({"error": {"code": "maxlag"}}),
|
||
FakeResponse({"parse": parse_payload}),
|
||
]
|
||
)
|
||
sleeps = []
|
||
|
||
assert _service(session, backoff_seconds=0.25, sleep=sleeps.append).fetch_page(
|
||
"Cookbook:Apple Glazed Pork Chops"
|
||
) == parse_payload
|
||
assert sleeps == [3.0, 0.5]
|
||
assert len(session.calls) == 3
|
||
assert session.calls[-1][1]["params"] == {
|
||
"action": "parse",
|
||
"page": "Cookbook:Apple Glazed Pork Chops",
|
||
"redirects": 1,
|
||
"prop": "text|wikitext|sections|revid|displaytitle",
|
||
"format": "json",
|
||
"formatversion": 2,
|
||
"maxlag": 5,
|
||
}
|
||
|
||
|
||
def test_fetch_rejects_redirect_responses():
|
||
response = FakeResponse({}, status_code=302)
|
||
session = FakeSession([response])
|
||
|
||
with pytest.raises(WikibooksImportError, match="redirect HTTP 302 rejected"):
|
||
_service(session).fetch_page("Cookbook:Apple Glazed Pork Chops")
|
||
|
||
assert session.calls[0][1]["allow_redirects"] is False
|
||
assert response.closed is True
|
||
|
||
|
||
def test_retry_delays_are_capped():
|
||
parse_payload = _parse_fixture("apple_glazed_pork_chops_parse.json")
|
||
retry_after_sleeps = []
|
||
retry_after_session = FakeSession(
|
||
[
|
||
FakeResponse({"error": "busy"}, status_code=429, headers={"Retry-After": "300"}),
|
||
FakeResponse({"parse": parse_payload}),
|
||
]
|
||
)
|
||
|
||
_service(
|
||
retry_after_session,
|
||
max_attempts=2,
|
||
max_retry_delay=2,
|
||
sleep=retry_after_sleeps.append,
|
||
).fetch_page("Cookbook:Apple Glazed Pork Chops")
|
||
assert retry_after_sleeps == [2.0]
|
||
|
||
backoff_sleeps = []
|
||
backoff_session = FakeSession(
|
||
[
|
||
FakeResponse({"error": {"code": "maxlag"}}),
|
||
FakeResponse({"parse": parse_payload}),
|
||
]
|
||
)
|
||
_service(
|
||
backoff_session,
|
||
backoff_seconds=10,
|
||
max_retry_delay=2,
|
||
sleep=backoff_sleeps.append,
|
||
).fetch_page("Cookbook:Apple Glazed Pork Chops")
|
||
assert backoff_sleeps == [2.0]
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
"kwargs",
|
||
[
|
||
{"timeout": None},
|
||
{"timeout": (0, 1)},
|
||
{"timeout": (1, math.inf)},
|
||
{"max_attempts": 0},
|
||
{"max_attempts": 1.5},
|
||
{"backoff_seconds": math.nan},
|
||
{"max_retry_delay": 0},
|
||
],
|
||
)
|
||
def test_constructor_rejects_invalid_timeout_and_retry_configuration(kwargs):
|
||
with pytest.raises(ValueError):
|
||
_service(**kwargs)
|
||
|
||
|
||
def test_payload_limit_boundary_and_malformed_parse_shape():
|
||
parse_payload = _parse_fixture("apple_glazed_pork_chops_parse.json")
|
||
response_payload = {"parse": parse_payload}
|
||
size = len(json.dumps(response_payload).encode("utf-8"))
|
||
|
||
assert _service(
|
||
FakeSession([FakeResponse(response_payload)]), max_payload_bytes=size
|
||
).fetch_page("Cookbook:Apple Glazed Pork Chops") == parse_payload
|
||
with pytest.raises(WikibooksResponseError, match="payload exceeds"):
|
||
_service(
|
||
FakeSession([FakeResponse(response_payload)]), max_payload_bytes=size - 1
|
||
).fetch_page("Cookbook:Apple Glazed Pork Chops")
|
||
|
||
malformed = copy.deepcopy(parse_payload)
|
||
malformed["sections"] = [{}]
|
||
with pytest.raises(WikibooksResponseError, match="sections item line"):
|
||
_service(FakeSession([FakeResponse({"parse": malformed})])).fetch_page(
|
||
"Cookbook:Apple Glazed Pork Chops"
|
||
)
|
||
|
||
|
||
def test_normalization_extracts_text_sections_times_and_fractions():
|
||
candidate = _service().normalize_page(_parse_fixture("apple_glazed_pork_chops_parse.json"))
|
||
|
||
assert candidate["name"] == "Apple Glazed Pork Chops"
|
||
assert candidate["description"] == "Sweet apples balance savory pork chops."
|
||
assert candidate["servings"] == 4
|
||
assert candidate["prep_time_minutes"] == 15
|
||
assert candidate["cook_time_minutes"] == 20
|
||
assert candidate["baking_time_minutes"] == 10
|
||
assert [item["quantity"] for item in candidate["ingredients"]] == [1.5, 0.5, 1.25, 2.0]
|
||
assert [item["quantity_text"] for item in candidate["ingredients"]] == ["1 1/2", "½", "1¼", "2"]
|
||
assert candidate["ingredients"][0] == {
|
||
"quantity": 1.5,
|
||
"quantity_text": "1 1/2",
|
||
"unit": "cups",
|
||
"name": "apple cider",
|
||
"source_text": "1 1/2 cups (360 ml) apple cider",
|
||
}
|
||
assert candidate["instructions"] == [
|
||
"Brown the pork chops in butter.",
|
||
"Add cider and cook until glazed.",
|
||
]
|
||
assert _service().validate_recipe(candidate) == {"valid": True, "errors": [], "warnings": []}
|
||
|
||
|
||
def test_validation_rejects_invalid_servings_and_ingredient_quantities():
|
||
valid_candidate = _service().normalize_page(_parse_fixture("apple_glazed_pork_chops_parse.json"))
|
||
|
||
for servings in (None, 0, -1):
|
||
invalid_servings = copy.deepcopy(valid_candidate)
|
||
invalid_servings["servings"] = servings
|
||
assert any(
|
||
error["code"] == "invalid_servings"
|
||
for error in _service().validate_recipe(invalid_servings)["errors"]
|
||
)
|
||
|
||
for quantity in (None, 0, -1):
|
||
invalid_quantity = copy.deepcopy(valid_candidate)
|
||
invalid_quantity["ingredients"][0]["quantity"] = quantity
|
||
assert any(
|
||
error["code"] == "invalid_ingredient_quantity"
|
||
for error in _service().validate_recipe(invalid_quantity)["errors"]
|
||
)
|
||
|
||
|
||
def test_validation_reports_apple_title_mismatch():
|
||
|
||
bad_candidate = _service().normalize_page(
|
||
_parse_fixture("apple_glaze_missing_apple_parse.json")
|
||
)
|
||
report = _service().validate_recipe(bad_candidate)
|
||
mismatch = next(error for error in report["errors"] if error["code"] == "title_ingredient_mismatch")
|
||
assert mismatch["missing_tokens"] == ["apple"]
|
||
assert report["valid"] is False
|
||
|
||
|
||
def test_validation_quarantines_database_integer_overflow():
|
||
candidate = _service().normalize_page(
|
||
_parse_fixture("apple_glazed_pork_chops_parse.json")
|
||
)
|
||
candidate["servings"] = 2_147_483_648
|
||
candidate["prep_time_minutes"] = 2_147_483_648
|
||
candidate["cook_time_minutes"] = 2_147_483_640
|
||
candidate["baking_time_minutes"] = 10
|
||
|
||
report = _service().validate_recipe(candidate)
|
||
assert report["valid"] is False
|
||
assert {error["code"] for error in report["errors"]} >= {
|
||
"servings_out_of_range",
|
||
"time_out_of_range",
|
||
"combined_cook_time_out_of_range",
|
||
}
|
||
|
||
|
||
def test_validation_accepts_generic_pie_form_while_rejecting_missing_apple():
|
||
candidate = _service().normalize_page(_parse_fixture("apple_glazed_pork_chops_parse.json"))
|
||
apple_pie = copy.deepcopy(candidate)
|
||
apple_pie["name"] = "Apple Pie"
|
||
apple_pie["ingredients"][0]["name"] = "apples"
|
||
assert not any(
|
||
error["code"] == "title_ingredient_mismatch"
|
||
for error in _service().validate_recipe(apple_pie)["errors"]
|
||
)
|
||
|
||
missing_apple = _service().normalize_page(
|
||
_parse_fixture("apple_glaze_missing_apple_parse.json")
|
||
)
|
||
assert any(
|
||
error["code"] == "title_ingredient_mismatch"
|
||
for error in _service().validate_recipe(missing_apple)["errors"]
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize(
|
||
("title", "ingredient_name"),
|
||
[
|
||
("Fried Apple Pies", "apples"),
|
||
("Classic American Apple Pie", "apples"),
|
||
("20-Minute Beef Stroganoff", "beef"),
|
||
("Apple Pie II", "apples"),
|
||
("8 Desserts in 1 Pan", "apples"),
|
||
("Classic American Salad Dressing Out", "apples"),
|
||
("Chocolate Chip Cookies I", "chocolate chips"),
|
||
],
|
||
)
|
||
def test_validation_ignores_generic_title_descriptors_and_variants(title, ingredient_name):
|
||
candidate = _service().normalize_page(_parse_fixture("apple_glazed_pork_chops_parse.json"))
|
||
candidate["name"] = title
|
||
candidate["ingredients"][0]["name"] = ingredient_name
|
||
|
||
report = _service().validate_recipe(candidate)
|
||
|
||
assert report["valid"] is True
|
||
assert not any(error["code"] == "title_ingredient_mismatch" for error in report["errors"])
|
||
|
||
|
||
def test_live_apple_pie_fixture_preserves_missing_quantity_failures():
|
||
payload = json.loads((FIXTURE_DIR / "apple_pie_i_live_parse.json").read_text())
|
||
candidate = _service().normalize_page(payload)
|
||
|
||
report = _service().validate_recipe(candidate)
|
||
|
||
assert candidate["name"] == "Apple Pie I"
|
||
assert sum(
|
||
error["code"] == "invalid_ingredient_quantity" for error in report["errors"]
|
||
) == 4
|
||
assert not any(
|
||
error["code"] == "title_ingredient_mismatch" for error in report["errors"]
|
||
)
|
||
assert report["valid"] is False
|
||
|
||
|
||
@pytest.mark.parametrize("source_text", ["1-2 cups apples", "1–2 cups apples", "1 — 2 cups apples"])
|
||
def test_validation_accepts_plural_title_matches_and_quarantines_quantity_ranges(source_text):
|
||
candidate = _service().normalize_page(_parse_fixture("apple_glazed_pork_chops_parse.json"))
|
||
banana_bread = copy.deepcopy(candidate)
|
||
banana_bread["name"] = "Banana Bread"
|
||
banana_bread["ingredients"][0]["name"] = "bananas"
|
||
assert not any(
|
||
error["code"] == "title_ingredient_mismatch"
|
||
for error in _service().validate_recipe(banana_bread)["errors"]
|
||
)
|
||
|
||
ranged_ingredient = _normalize_ingredient(source_text)
|
||
assert ranged_ingredient["quantity"] is None
|
||
assert ranged_ingredient["quantity_text"] is None
|
||
ranged_candidate = copy.deepcopy(candidate)
|
||
ranged_candidate["ingredients"][0] = ranged_ingredient
|
||
assert any(
|
||
error["code"] == "invalid_ingredient_quantity"
|
||
for error in _service().validate_recipe(ranged_candidate)["errors"]
|
||
)
|
||
|
||
|
||
@pytest.mark.parametrize("source_text", ["1/0 cup apples", "1/ cups apples", "1 1/0 cups apples"])
|
||
def test_validation_quarantines_invalid_fraction_prefixes_and_nonfinite_quantities(source_text):
|
||
candidate = _service().normalize_page(_parse_fixture("apple_glazed_pork_chops_parse.json"))
|
||
invalid_fraction = _normalize_ingredient(source_text)
|
||
assert invalid_fraction["quantity"] is None
|
||
assert invalid_fraction["quantity_text"] is None
|
||
|
||
invalid_candidate = copy.deepcopy(candidate)
|
||
invalid_candidate["ingredients"][0] = invalid_fraction
|
||
assert any(
|
||
error["code"] == "invalid_ingredient_quantity"
|
||
for error in _service().validate_recipe(invalid_candidate)["errors"]
|
||
)
|
||
|
||
for value in (math.nan, math.inf, -math.inf):
|
||
nonfinite_candidate = copy.deepcopy(candidate)
|
||
nonfinite_candidate["ingredients"][0]["quantity"] = value
|
||
assert any(
|
||
error["code"] == "invalid_ingredient_quantity"
|
||
for error in _service().validate_recipe(nonfinite_candidate)["errors"]
|
||
)
|
||
|
||
|
||
def test_staging_uses_atomic_insert_and_owns_payload_snapshot():
|
||
parse_payload = _parse_fixture("apple_glazed_pork_chops_parse.json")
|
||
original_text = parse_payload["text"]
|
||
db = FakeStageDb()
|
||
|
||
assert _service().stage_parse_payload(db, parse_payload) is db.winner
|
||
parse_payload["text"] = "mutated after staging"
|
||
parse_payload["sections"][0]["line"] = "mutated"
|
||
|
||
compiled = db.statements[0].compile(dialect=postgresql.dialect())
|
||
assert compiled.params["raw_payload"]["text"] == original_text
|
||
assert compiled.params["raw_payload"]["sections"][0]["line"] == "Ingredients"
|
||
assert "ON CONFLICT ON CONSTRAINT uq_recipe_import_staging_source_revision DO NOTHING" in str(
|
||
compiled
|
||
)
|
||
assert "RETURNING recipe_import_staging.id" in str(compiled)
|
||
assert len(db.statements) == 2
|
||
|
||
|
||
@pytest.mark.requires_postgres
|
||
def test_staging_checksums_deduplicates_and_quarantines_invalid_revisions(db, monkeypatch):
|
||
valid_parse = _parse_fixture("apple_glazed_pork_chops_parse.json")
|
||
invalid_parse = _parse_fixture("apple_glaze_missing_apple_parse.json")
|
||
service = _service()
|
||
monkeypatch.setattr(service, "fetch_page", lambda title: copy.deepcopy(valid_parse))
|
||
|
||
staged = service.stage_page(db, "Cookbook:Apple Glazed Pork Chops")
|
||
same_revision = service.stage_page(db, "Cookbook:Apple Glazed Pork Chops")
|
||
|
||
assert staged is same_revision
|
||
assert staged.status is RecipeImportStatus.PENDING
|
||
assert staged.raw_payload == valid_parse
|
||
assert staged.payload_sha256 == canonical_json_sha256(valid_parse)
|
||
assert staged.source_url == "https://en.wikibooks.org/wiki/Cookbook:Apple_Glazed_Pork_Chops"
|
||
assert staged.license_url == CC_BY_SA_4_LICENSE_URL
|
||
assert "Wikibooks contributors" in staged.attribution
|
||
assert "action=history" in staged.attribution
|
||
assert "normalized_recipe" not in staged.raw_payload
|
||
|
||
monkeypatch.setattr(service, "fetch_page", lambda title: copy.deepcopy(invalid_parse))
|
||
quarantined = service.stage_page(db, "Cookbook:Pan-Seared Pork Chops with Apple Glaze")
|
||
|
||
assert quarantined.status is RecipeImportStatus.QUARANTINED
|
||
assert any(
|
||
error["code"] == "title_ingredient_mismatch"
|
||
for error in quarantined.validation_report["errors"]
|
||
)
|
||
assert db.query(RecipeImportStaging).count() == 2
|
||
|
||
|
||
@pytest.mark.requires_postgres
|
||
def test_revalidation_updates_only_unreviewed_candidates_without_changing_raw_payload(db):
|
||
service = _service()
|
||
parse_payload = _parse_fixture("apple_glazed_pork_chops_parse.json")
|
||
parse_payload["pageid"] = 2_000_000_001
|
||
parse_payload["revid"] = 7_001
|
||
staged = service.stage_parse_payload(db, parse_payload)
|
||
raw_payload = copy.deepcopy(staged.raw_payload)
|
||
payload_sha256 = staged.payload_sha256
|
||
staged.status = RecipeImportStatus.QUARANTINED
|
||
staged.validation_report = {
|
||
"valid": False,
|
||
"errors": [{"code": "obsolete_validator_error"}],
|
||
"warnings": [],
|
||
}
|
||
staged.rejection_reason = "obsolete_validator_error"
|
||
|
||
tampered_checksum_payload = copy.deepcopy(parse_payload)
|
||
tampered_checksum_payload["pageid"] = 2_000_000_002
|
||
tampered_checksum_payload["revid"] = 7_002
|
||
tampered_checksum = service.stage_parse_payload(db, tampered_checksum_payload)
|
||
tampered_checksum.payload_sha256 = "0" * 64
|
||
|
||
reviewed_payload = copy.deepcopy(parse_payload)
|
||
reviewed_payload["pageid"] = 2_000_000_003
|
||
reviewed_payload["revid"] = 7_003
|
||
reviewed = service.stage_parse_payload(db, reviewed_payload)
|
||
reviewed.status = RecipeImportStatus.REJECTED
|
||
reviewed.reviewed_at = datetime.now(timezone.utc)
|
||
reviewed.reviewed_by = "reviewer@example.com"
|
||
reviewed.rejection_reason = "manual rejection"
|
||
|
||
approved_payload = copy.deepcopy(parse_payload)
|
||
approved_payload["pageid"] = 2_000_000_004
|
||
approved_payload["revid"] = 7_004
|
||
approved = service.stage_parse_payload(db, approved_payload)
|
||
approved.status = RecipeImportStatus.APPROVED
|
||
approved.reviewed_at = datetime.now(timezone.utc)
|
||
approved.reviewed_by = "reviewer@example.com"
|
||
|
||
reviewed_quarantine_payload = copy.deepcopy(parse_payload)
|
||
reviewed_quarantine_payload["pageid"] = 2_000_000_005
|
||
reviewed_quarantine_payload["revid"] = 7_005
|
||
reviewed_quarantine = service.stage_parse_payload(db, reviewed_quarantine_payload)
|
||
reviewed_quarantine.status = RecipeImportStatus.QUARANTINED
|
||
reviewed_quarantine.reviewed_at = datetime.now(timezone.utc)
|
||
reviewed_quarantine.reviewed_by = "reviewer@example.com"
|
||
reviewed_quarantine.rejection_reason = "manual quarantine"
|
||
db.flush()
|
||
|
||
result = service.revalidate_staged_imports(db, limit=10)
|
||
|
||
assert result == {"processed": 2, "pending": 1, "quarantined": 1}
|
||
assert staged.status is RecipeImportStatus.PENDING
|
||
assert staged.validation_report["valid"] is True
|
||
assert staged.rejection_reason is None
|
||
assert staged.raw_payload == raw_payload
|
||
assert staged.payload_sha256 == payload_sha256
|
||
assert tampered_checksum.status is RecipeImportStatus.QUARANTINED
|
||
assert tampered_checksum.validation_report["errors"] == [
|
||
{
|
||
"code": "payload_checksum_mismatch",
|
||
"field": "raw_payload",
|
||
"message": "Raw payload checksum does not match immutable provenance.",
|
||
}
|
||
]
|
||
assert reviewed.status is RecipeImportStatus.REJECTED
|
||
assert reviewed.rejection_reason == "manual rejection"
|
||
assert approved.status is RecipeImportStatus.APPROVED
|
||
assert reviewed_quarantine.status is RecipeImportStatus.QUARANTINED
|
||
assert reviewed_quarantine.rejection_reason == "manual quarantine"
|
||
|
||
|
||
def test_revalidation_uses_skip_locked():
|
||
db = FakeStageDb()
|
||
db.winner = None
|
||
|
||
assert _service().revalidate_staged_imports(db, limit=10) == {
|
||
"processed": 0,
|
||
"pending": 0,
|
||
"quarantined": 0,
|
||
}
|
||
compiled = db.statements[0].compile(dialect=postgresql.dialect())
|
||
assert "FOR UPDATE SKIP LOCKED" in str(compiled)
|
||
|
||
|
||
@pytest.mark.requires_postgres
|
||
def test_staging_deduplicates_two_concurrent_sessions(db):
|
||
engine = db.get_bind().engine
|
||
Session = sessionmaker(bind=engine, autoflush=False)
|
||
parse_payload = _parse_fixture("apple_glazed_pork_chops_parse.json")
|
||
external_id = str(uuid.uuid4().int % 2_000_000_000 + 1)
|
||
parse_payload["pageid"] = int(external_id)
|
||
parse_payload["revid"] = 6001
|
||
identity = {
|
||
"source": "wikibooks",
|
||
"external_id": external_id,
|
||
"source_revision": "6001",
|
||
}
|
||
start = threading.Barrier(2)
|
||
release_winner = threading.Event()
|
||
results = Queue()
|
||
errors = Queue()
|
||
service = _service()
|
||
|
||
def stage_in_session():
|
||
session = Session()
|
||
try:
|
||
session.execute(text("SET LOCAL lock_timeout = '3s'"))
|
||
session.execute(text("SET LOCAL statement_timeout = '10s'"))
|
||
start.wait(timeout=5)
|
||
staged = service.stage_parse_payload(session, copy.deepcopy(parse_payload))
|
||
results.put(staged.id)
|
||
if not release_winner.wait(timeout=5):
|
||
raise TimeoutError("winner transaction was not released")
|
||
session.commit()
|
||
except Exception as exc: # Propagate worker failures to the test thread.
|
||
session.rollback()
|
||
errors.put(exc)
|
||
finally:
|
||
session.rollback()
|
||
session.close()
|
||
|
||
threads = [threading.Thread(target=stage_in_session) for _ in range(2)]
|
||
for thread in threads:
|
||
thread.start()
|
||
|
||
def next_result(timeout: float):
|
||
try:
|
||
return results.get(timeout=timeout)
|
||
except Empty:
|
||
if not errors.empty():
|
||
raise errors.get()
|
||
pytest.fail("concurrent staging did not complete before timeout")
|
||
|
||
try:
|
||
first_id = next_result(5)
|
||
release_winner.set()
|
||
second_id = next_result(10)
|
||
for thread in threads:
|
||
thread.join(timeout=5)
|
||
assert not thread.is_alive(), "concurrent staging thread did not exit"
|
||
if not errors.empty():
|
||
raise errors.get()
|
||
|
||
verifier = Session()
|
||
try:
|
||
persisted = verifier.query(RecipeImportStaging).filter_by(**identity).all()
|
||
assert first_id == second_id == persisted[0].id
|
||
assert len(persisted) == 1
|
||
finally:
|
||
verifier.rollback()
|
||
verifier.close()
|
||
finally:
|
||
release_winner.set()
|
||
for thread in threads:
|
||
thread.join(timeout=5)
|
||
cleanup = Session()
|
||
try:
|
||
cleanup.query(RecipeImportStaging).filter_by(**identity).delete(
|
||
synchronize_session=False
|
||
)
|
||
cleanup.commit()
|
||
finally:
|
||
cleanup.rollback()
|
||
cleanup.close()
|