"""CSV parsing and menu matching for the Redefine cart filler.""" import importlib.util from pathlib import Path from playwright.sync_api import Error as PlaywrightError from scraper.fill_cart import build_plan, cart_lines, error_text, quantity_for from scraper.parse_menu import extract_catalog_products _SCRIPT = Path(__file__).resolve().parents[1] / "scripts" / "fill_redefine_cart.py" _spec = importlib.util.spec_from_file_location("fill_redefine_cart", _SCRIPT) fill_redefine_cart = importlib.util.module_from_spec(_spec) assert _spec.loader is not None _spec.loader.exec_module(fill_redefine_cart) PRODUCTS = [ { "name": "Korean Steak Bowl", "uuid": "korean-id", "available": True, }, { "name": "Sale Bowl", "uuid": "sale-id", "available": True, }, { "name": "Hidden Bowl", "uuid": "hidden-id", "available": False, }, { "name": "Twin Bowl", "uuid": "twin-a", "available": True, }, { "name": "twin bowl", "uuid": "twin-b", "available": True, }, { "name": "No Id Bowl", "uuid": "", "available": True, }, ] def test_build_plan_matches_names_and_skips_the_rest(): csv_text = "\n".join( [ "Item,Quantity", "korean steak bowl,4", "Sale Bowl,2", "Sale Bowl,3", "Hidden Bowl,1", "Twin Bowl,8", "Missing Bowl,1", "No Id Bowl,1", "=cmd,1", "Zero Bowl,0", "Bad Qty,nope", ] ) # The admin exporter prefixes formula-like names with an apostrophe. csv_text = csv_text.replace("=cmd", "'=cmd") plan = build_plan(csv_text, PRODUCTS) assert [(item.name, item.quantity, item.uuid) for item in plan.adds] == [ ("Korean Steak Bowl", 4, "korean-id"), ("Sale Bowl", 5, "sale-id"), ] assert [(line.name, line.reason) for line in plan.skipped] == [ ("Bad Qty", "invalid quantity 'nope'"), ("Hidden Bowl", "unavailable on the menu"), ("Twin Bowl", "matches more than one menu item"), ("Missing Bowl", "not on the menu"), ("No Id Bowl", "not on the menu"), ("=cmd", "not on the menu"), ] assert plan.ignored_zero == 1 def test_build_plan_reads_quoted_fields_and_a_byte_order_mark(): csv_text = '\ufeffItem,Quantity\r\n"Bowl, Large",2\r\n' products = [{"name": "Bowl, Large", "uuid": "bowl", "available": True}] plan = build_plan(csv_text, products) assert plan.adds[0].name == "Bowl, Large" assert plan.adds[0].quantity == 2 assert plan.skipped == () def test_build_plan_requires_the_admin_columns(): try: build_plan("Meal,Qty\nSoup,1\n", PRODUCTS) except ValueError as exc: assert "Item and Quantity" in str(exc) else: raise AssertionError("expected a column error") def test_extract_catalog_products_reads_the_menu_attribute(): page = """ """ assert extract_catalog_products(page)[0]["uuid"] == "korean-id" def test_cart_lines_use_the_product_uuid_and_quantity(): cart = { "items": [ { "uuid": "line-1", "quantity": "4", "product": {"name": "Korean Steak Bowl", "uuid": "korean-id"}, } ] } lines = cart_lines(cart) assert lines[0].name == "Korean Steak Bowl" assert lines[0].product_uuid == "korean-id" assert quantity_for(lines, "korean-id") == 4 assert quantity_for(lines, "line-1") == 0 class _CartPage: def __init__(self, responses): self._responses = list(responses) def goto(self, url, **kwargs): return None def evaluate(self, script, payload): response = self._responses.pop(0) if isinstance(response, Exception): raise response return response def test_main_rejects_a_non_utf8_csv(tmp_path, capsys): path = tmp_path / "order.csv" path.write_bytes(b"Item,Quantity\nCaf\xe9 Bowl,1\n") code = fill_redefine_cart.main([str(path)]) error = capsys.readouterr().err assert code == 1 assert error.strip() == "CSV must be UTF-8." assert "Traceback" not in error def test_fill_browser_prints_a_summary_when_the_page_cannot_open(capsys): plan = build_plan("Item,Quantity\nKorean Steak Bowl,1\n", PRODUCTS) class ClosedBrowser: def __init__(self): self.closed = False def new_page(self): raise PlaywrightError("browser has been closed") def close(self): self.closed = True browser = ClosedBrowser() code = fill_redefine_cart._fill_browser( browser, plan, "https://www.redefinemeals.com/menu" ) output = capsys.readouterr().out assert code == 1 assert browser.closed assert "Traceback" not in output assert "Added 0. Failed 1. Skipped 0." in output def test_fill_page_prints_a_summary_when_the_window_closes_mid_add(capsys): plan = build_plan("Item,Quantity\nKorean Steak Bowl,1\n", PRODUCTS) page = _CartPage( [ {"ok": True, "status": 200, "data": {"items": []}}, PlaywrightError("Target page, context or browser has been closed"), ] ) code = fill_redefine_cart._fill_page( page, plan, "https://www.redefinemeals.com/menu" ) output = capsys.readouterr().out assert code == 1 assert "Traceback" not in output assert "Added 0. Failed 1. Skipped 0." in output assert "Cart is ready" not in output def test_fill_page_counts_items_left_when_the_window_closes(capsys): plan = build_plan( "Item,Quantity\nKorean Steak Bowl,1\nSale Bowl,2\n", PRODUCTS, ) page = _CartPage( [ {"ok": True, "status": 200, "data": {"items": []}}, PlaywrightError("Target page, context or browser has been closed"), ] ) code = fill_redefine_cart._fill_page( page, plan, "https://www.redefinemeals.com/menu" ) output = capsys.readouterr().out assert code == 1 assert "Korean Steak Bowl:" in output assert "Sale Bowl: not attempted" in output assert "Added 0. Failed 2. Skipped 0." in output def test_fill_page_prints_a_summary_when_the_first_cart_read_fails(capsys): plan = build_plan("Item,Quantity\nKorean Steak Bowl,1\n", PRODUCTS) page = _CartPage( [ { "ok": False, "status": 500, "data": {"message": "cart unavailable", "trace": [{"file": "x"}]}, } ] ) code = fill_redefine_cart._fill_page( page, plan, "https://www.redefinemeals.com/menu" ) output = capsys.readouterr().out assert code == 1 assert "HTTP 500: cart unavailable" in output assert "x" not in output.split("Failed:", 1)[-1] assert "Added 0. Failed 1. Skipped 0." in output def test_error_text_keeps_the_message_and_drops_the_trace(): body = { "message": "Product is not available.", "exception": "HttpException", "file": "/home/app/secret.php", "trace": [{"file": "/home/app/secret.php"}], } text = error_text(422, body) assert text == "HTTP 422: Product is not available." assert "secret.php" not in text