import csv import tempfile import unittest from pathlib import Path from unittest import mock from click.testing import CliRunner import enrich_costco import review_products def write_review_source_files(tmpdir, rows): giant_items_csv = Path(tmpdir) / "giant_items.csv" costco_items_csv = Path(tmpdir) / "costco_items.csv" giant_orders_csv = Path(tmpdir) / "giant_orders.csv" costco_orders_csv = Path(tmpdir) / "costco_orders.csv" fieldnames = enrich_costco.OUTPUT_FIELDS grouped_rows = {"giant": [], "costco": []} grouped_orders = {"giant": {}, "costco": {}} for index, row in enumerate(rows, start=1): retailer = row.get("retailer", "giant") normalized_row = {field: "" for field in fieldnames} normalized_row.update( { "retailer": retailer, "order_id": row.get("order_id", f"{retailer[0]}{index}"), "line_no": row.get("line_no", str(index)), "normalized_row_id": row.get( "normalized_row_id", f"{retailer}:{row.get('order_id', f'{retailer[0]}{index}')}:{row.get('line_no', str(index))}", ), "normalized_item_id": row.get("normalized_item_id", ""), "order_date": row.get("purchase_date", ""), "item_name": row.get("raw_item_name", ""), "item_name_norm": row.get("normalized_item_name", ""), "image_url": row.get("image_url", ""), "upc": row.get("upc", ""), "line_total": row.get("line_total", ""), "net_line_total": row.get("net_line_total", ""), "matched_discount_amount": row.get("matched_discount_amount", ""), "qty": row.get("qty", "1"), "unit": row.get("unit", "EA"), "normalized_quantity": row.get("normalized_quantity", ""), "normalized_quantity_unit": row.get("normalized_quantity_unit", ""), "size_value": row.get("size_value", ""), "size_unit": row.get("size_unit", ""), "pack_qty": row.get("pack_qty", ""), "measure_type": row.get("measure_type", "each"), "retailer_item_id": row.get("retailer_item_id", ""), "price_per_each": row.get("price_per_each", ""), "price_per_lb": row.get("price_per_lb", ""), "price_per_oz": row.get("price_per_oz", ""), "is_discount_line": row.get("is_discount_line", "false"), "is_coupon_line": row.get("is_coupon_line", "false"), "is_fee": row.get("is_fee", "false"), "raw_order_path": row.get("raw_order_path", ""), } ) grouped_rows[retailer].append(normalized_row) order_id = normalized_row["order_id"] grouped_orders[retailer].setdefault( order_id, { "order_id": order_id, "store_name": row.get("store_name", ""), "store_number": row.get("store_number", ""), "store_city": row.get("store_city", ""), "store_state": row.get("store_state", ""), }, ) for path, source_rows in [ (giant_items_csv, grouped_rows["giant"]), (costco_items_csv, grouped_rows["costco"]), ]: with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() writer.writerows(source_rows) order_fields = ["order_id", "store_name", "store_number", "store_city", "store_state"] for path, source_rows in [ (giant_orders_csv, grouped_orders["giant"].values()), (costco_orders_csv, grouped_orders["costco"].values()), ]: with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=order_fields) writer.writeheader() writer.writerows(source_rows) return giant_items_csv, costco_items_csv, giant_orders_csv, costco_orders_csv class ReviewWorkflowTests(unittest.TestCase): def test_build_review_queue_groups_unresolved_purchases(self): queue_rows = review_products.build_review_queue( [ { "normalized_item_id": "gnorm_1", "catalog_id": "", "retailer": "giant", "raw_item_name": "SB BAGGED ICE 20LB", "normalized_item_name": "BAGGED ICE", "upc": "", "line_total": "3.50", "is_fee": "false", "is_discount_line": "false", "is_coupon_line": "false", }, { "normalized_item_id": "gnorm_1", "catalog_id": "", "retailer": "giant", "raw_item_name": "SB BAG ICE CUBED 10LB", "normalized_item_name": "BAG ICE", "upc": "", "line_total": "2.50", "is_fee": "false", "is_discount_line": "false", "is_coupon_line": "false", }, ], [], ) self.assertEqual(1, len(queue_rows)) self.assertEqual("gnorm_1", queue_rows[0]["normalized_item_id"]) self.assertIn("SB BAGGED ICE 20LB", queue_rows[0]["raw_item_names"]) def test_build_catalog_suggestions_prefers_upc_then_name(self): suggestions = review_products.build_catalog_suggestions( [ { "normalized_item_name": "MIXED PEPPER", "upc": "12345", } ], [ { "normalized_item_id": "prior_1", "normalized_item_name": "MIXED PEPPER 6 PACK", "upc": "12345", "catalog_id": "cat_2", } ], [ { "catalog_id": "cat_1", "catalog_name": "MIXED PEPPER", }, { "catalog_id": "cat_2", "catalog_name": "MIXED PEPPER 6 PACK", }, ], ) self.assertEqual("cat_2", suggestions[0]["catalog_id"]) self.assertEqual("exact upc", suggestions[0]["reason"]) def test_search_catalog_rows_ranks_token_overlap(self): results = review_products.search_catalog_rows( "mixed pepper", [ { "catalog_id": "cat_1", "catalog_name": "MIXED PEPPER", "product_type": "pepper", "category": "produce", "variant": "", }, { "catalog_id": "cat_2", "catalog_name": "GROUND PEPPER", "product_type": "spice", "category": "baking", "variant": "", }, ], [ { "normalized_item_id": "gnorm_mix", "catalog_id": "cat_1", } ], "cnorm_mix", ) self.assertEqual("cat_1", results[0]["catalog_id"]) self.assertGreater(results[0]["score"], results[1]["score"]) def test_review_products_displays_position_items_and_suggestions(self): with tempfile.TemporaryDirectory() as tmpdir: purchases_csv = Path(tmpdir) / "purchases.csv" queue_csv = Path(tmpdir) / "review_queue.csv" resolutions_csv = Path(tmpdir) / "review_resolutions.csv" catalog_csv = Path(tmpdir) / "catalog.csv" links_csv = Path(tmpdir) / "product_links.csv" giant_items_csv, costco_items_csv, giant_orders_csv, costco_orders_csv = write_review_source_files( tmpdir, [ { "purchase_date": "2026-03-14", "retailer": "costco", "order_id": "c2", "line_no": "2", "normalized_item_id": "cnorm_mix", "raw_item_name": "MIXED PEPPER 6-PACK", "normalized_item_name": "MIXED PEPPER", "image_url": "", "upc": "", "line_total": "7.49", }, { "purchase_date": "2026-03-12", "retailer": "costco", "order_id": "c1", "line_no": "1", "normalized_item_id": "cnorm_mix", "raw_item_name": "MIXED PEPPER 6-PACK", "normalized_item_name": "MIXED PEPPER", "image_url": "https://example.test/mixed-pepper.jpg", "upc": "", "line_total": "6.99", }, { "purchase_date": "2026-03-10", "retailer": "giant", "order_id": "g1", "line_no": "1", "normalized_item_id": "gnorm_mix", "raw_item_name": "MIXED PEPPER", "normalized_item_name": "MIXED PEPPER", "image_url": "", "upc": "", "line_total": "5.99", }, ], ) with catalog_csv.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=review_products.build_purchases.CATALOG_FIELDS) writer.writeheader() writer.writerow( { "catalog_id": "cat_mix", "catalog_name": "MIXED PEPPER", "category": "produce", "product_type": "pepper", "brand": "", "variant": "", "size_value": "", "size_unit": "", "pack_qty": "", "measure_type": "", "notes": "", "created_at": "", "updated_at": "", } ) with links_csv.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=review_products.build_purchases.PRODUCT_LINK_FIELDS) writer.writeheader() writer.writerow( { "normalized_item_id": "gnorm_mix", "catalog_id": "cat_mix", "link_method": "manual_link", "link_confidence": "high", "review_status": "approved", "reviewed_by": "", "reviewed_at": "", "link_notes": "", } ) runner = CliRunner() result = runner.invoke( review_products.main, [ "--giant-items-enriched-csv", str(giant_items_csv), "--costco-items-enriched-csv", str(costco_items_csv), "--giant-orders-csv", str(giant_orders_csv), "--costco-orders-csv", str(costco_orders_csv), "--purchases-csv", str(purchases_csv), "--queue-csv", str(queue_csv), "--resolutions-csv", str(resolutions_csv), "--catalog-csv", str(catalog_csv), "--links-csv", str(links_csv), ], input="q\n", color=True, ) self.assertEqual(0, result.exit_code) self.assertIn("Review guide:", result.output) self.assertIn("Review 1/1: MIXED PEPPER", result.output) self.assertIn("2 matched items:", result.output) self.assertIn("[#] link to suggestion [f]ind [n]ew [s]kip e[x]clude [q]uit >", result.output) first_item = result.output.index("[1] MIXED PEPPER 6-PACK | costco | 2026-03-14 | 7.49 | ") second_item = result.output.index("[2] MIXED PEPPER 6-PACK | costco | 2026-03-12 | 6.99 | https://example.test/mixed-pepper.jpg") self.assertLess(first_item, second_item) self.assertIn("1 catalog_name suggestions found:", result.output) self.assertIn("[1] MIXED PEPPER, pepper, produce (1 items, 1 rows)", result.output) self.assertIn("\x1b[", result.output) def test_review_products_no_suggestions_is_informational(self): with tempfile.TemporaryDirectory() as tmpdir: purchases_csv = Path(tmpdir) / "purchases.csv" queue_csv = Path(tmpdir) / "review_queue.csv" resolutions_csv = Path(tmpdir) / "review_resolutions.csv" catalog_csv = Path(tmpdir) / "catalog.csv" links_csv = Path(tmpdir) / "product_links.csv" giant_items_csv, costco_items_csv, giant_orders_csv, costco_orders_csv = write_review_source_files( tmpdir, [ { "purchase_date": "2026-03-14", "retailer": "giant", "order_id": "g1", "line_no": "1", "normalized_item_id": "gnorm_ice", "raw_item_name": "SB BAGGED ICE 20LB", "normalized_item_name": "BAGGED ICE", "image_url": "", "upc": "", "line_total": "3.50", } ], ) with catalog_csv.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=review_products.build_purchases.CATALOG_FIELDS) writer.writeheader() result = CliRunner().invoke( review_products.main, [ "--giant-items-enriched-csv", str(giant_items_csv), "--costco-items-enriched-csv", str(costco_items_csv), "--giant-orders-csv", str(giant_orders_csv), "--costco-orders-csv", str(costco_orders_csv), "--purchases-csv", str(purchases_csv), "--queue-csv", str(queue_csv), "--resolutions-csv", str(resolutions_csv), "--catalog-csv", str(catalog_csv), "--links-csv", str(links_csv), ], input="q\n", color=True, ) self.assertEqual(0, result.exit_code) self.assertIn("no catalog_name suggestions found", result.output) def test_search_links_catalog_and_writes_link_row(self): with tempfile.TemporaryDirectory() as tmpdir: purchases_csv = Path(tmpdir) / "purchases.csv" queue_csv = Path(tmpdir) / "review_queue.csv" resolutions_csv = Path(tmpdir) / "review_resolutions.csv" catalog_csv = Path(tmpdir) / "catalog.csv" links_csv = Path(tmpdir) / "product_links.csv" giant_items_csv, costco_items_csv, giant_orders_csv, costco_orders_csv = write_review_source_files( tmpdir, [ { "purchase_date": "2026-03-14", "retailer": "costco", "order_id": "c2", "line_no": "2", "normalized_item_id": "cnorm_mix", "raw_item_name": "MIXED PEPPER 6-PACK", "normalized_item_name": "MIXED PEPPER", "image_url": "", "upc": "", "line_total": "7.49", }, { "purchase_date": "2026-03-12", "retailer": "costco", "order_id": "c1", "line_no": "1", "normalized_item_id": "cnorm_mix", "raw_item_name": "MIXED PEPPER 6-PACK", "normalized_item_name": "MIXED PEPPER", "image_url": "", "upc": "", "line_total": "6.99", }, { "purchase_date": "2026-03-10", "retailer": "giant", "order_id": "g1", "line_no": "1", "normalized_item_id": "gnorm_mix", "raw_item_name": "MIXED PEPPER", "normalized_item_name": "MIXED PEPPER", "image_url": "", "upc": "", "line_total": "5.99", }, ], ) with catalog_csv.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=review_products.build_purchases.CATALOG_FIELDS) writer.writeheader() writer.writerow( { "catalog_id": "cat_mix", "catalog_name": "MIXED PEPPER", "category": "", "product_type": "", "brand": "", "variant": "", "size_value": "", "size_unit": "", "pack_qty": "", "measure_type": "", "notes": "", "created_at": "", "updated_at": "", } ) with links_csv.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=review_products.build_purchases.PRODUCT_LINK_FIELDS) writer.writeheader() writer.writerow( { "normalized_item_id": "gnorm_mix", "catalog_id": "cat_mix", "link_method": "manual_link", "link_confidence": "high", "review_status": "approved", "reviewed_by": "", "reviewed_at": "", "link_notes": "", } ) result = CliRunner().invoke( review_products.main, [ "--giant-items-enriched-csv", str(giant_items_csv), "--costco-items-enriched-csv", str(costco_items_csv), "--giant-orders-csv", str(giant_orders_csv), "--costco-orders-csv", str(costco_orders_csv), "--purchases-csv", str(purchases_csv), "--queue-csv", str(queue_csv), "--resolutions-csv", str(resolutions_csv), "--catalog-csv", str(catalog_csv), "--links-csv", str(links_csv), "--limit", "1", ], input="f\nmixed pepper\n1\nlinked by test\n", color=True, ) self.assertEqual(0, result.exit_code) self.assertIn("1 search results found:", result.output) with resolutions_csv.open(newline="", encoding="utf-8") as handle: rows = list(csv.DictReader(handle)) with links_csv.open(newline="", encoding="utf-8") as handle: link_rows = list(csv.DictReader(handle)) self.assertEqual("cat_mix", rows[0]["catalog_id"]) self.assertEqual("link", rows[0]["resolution_action"]) self.assertEqual("cat_mix", link_rows[0]["catalog_id"]) def test_search_no_matches_allows_retry_or_return(self): with tempfile.TemporaryDirectory() as tmpdir: purchases_csv = Path(tmpdir) / "purchases.csv" queue_csv = Path(tmpdir) / "review_queue.csv" resolutions_csv = Path(tmpdir) / "review_resolutions.csv" catalog_csv = Path(tmpdir) / "catalog.csv" links_csv = Path(tmpdir) / "product_links.csv" giant_items_csv, costco_items_csv, giant_orders_csv, costco_orders_csv = write_review_source_files( tmpdir, [ { "purchase_date": "2026-03-14", "retailer": "giant", "order_id": "g1", "line_no": "1", "normalized_item_id": "gnorm_ice", "raw_item_name": "SB BAGGED ICE 20LB", "normalized_item_name": "BAGGED ICE", "image_url": "", "upc": "", "line_total": "3.50", } ], ) with catalog_csv.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=review_products.build_purchases.CATALOG_FIELDS) writer.writeheader() writer.writerow( { "catalog_id": "cat_ice", "catalog_name": "ICE", "category": "frozen", "product_type": "ice", "brand": "", "variant": "", "size_value": "", "size_unit": "", "pack_qty": "", "measure_type": "", "notes": "", "created_at": "", "updated_at": "", } ) result = CliRunner().invoke( review_products.main, [ "--giant-items-enriched-csv", str(giant_items_csv), "--costco-items-enriched-csv", str(costco_items_csv), "--giant-orders-csv", str(giant_orders_csv), "--costco-orders-csv", str(costco_orders_csv), "--purchases-csv", str(purchases_csv), "--queue-csv", str(queue_csv), "--resolutions-csv", str(resolutions_csv), "--catalog-csv", str(catalog_csv), "--links-csv", str(links_csv), ], input="f\nzzz\nq\nq\n", color=True, ) self.assertEqual(0, result.exit_code) self.assertIn("no matches found", result.output) def test_skip_remains_available_from_main_prompt(self): with tempfile.TemporaryDirectory() as tmpdir: purchases_csv = Path(tmpdir) / "purchases.csv" queue_csv = Path(tmpdir) / "review_queue.csv" resolutions_csv = Path(tmpdir) / "review_resolutions.csv" catalog_csv = Path(tmpdir) / "catalog.csv" links_csv = Path(tmpdir) / "product_links.csv" giant_items_csv, costco_items_csv, giant_orders_csv, costco_orders_csv = write_review_source_files( tmpdir, [ { "purchase_date": "2026-03-14", "retailer": "giant", "order_id": "g1", "line_no": "1", "normalized_item_id": "gnorm_skip", "raw_item_name": "TEST ITEM", "normalized_item_name": "TEST ITEM", "image_url": "", "upc": "", "line_total": "1.00", } ], ) with catalog_csv.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=review_products.build_purchases.CATALOG_FIELDS) writer.writeheader() result = CliRunner().invoke( review_products.main, [ "--giant-items-enriched-csv", str(giant_items_csv), "--costco-items-enriched-csv", str(costco_items_csv), "--giant-orders-csv", str(giant_orders_csv), "--costco-orders-csv", str(costco_orders_csv), "--purchases-csv", str(purchases_csv), "--queue-csv", str(queue_csv), "--resolutions-csv", str(resolutions_csv), "--catalog-csv", str(catalog_csv), "--links-csv", str(links_csv), "--limit", "1", ], input="s\n", color=True, ) self.assertEqual(0, result.exit_code) with resolutions_csv.open(newline="", encoding="utf-8") as handle: rows = list(csv.DictReader(handle)) self.assertEqual("skip", rows[0]["resolution_action"]) self.assertEqual("pending", rows[0]["status"]) def test_review_products_creates_catalog_and_resolution(self): with tempfile.TemporaryDirectory() as tmpdir: purchases_csv = Path(tmpdir) / "purchases.csv" queue_csv = Path(tmpdir) / "review_queue.csv" resolutions_csv = Path(tmpdir) / "review_resolutions.csv" catalog_csv = Path(tmpdir) / "catalog.csv" links_csv = Path(tmpdir) / "product_links.csv" giant_items_csv, costco_items_csv, giant_orders_csv, costco_orders_csv = write_review_source_files( tmpdir, [ { "purchase_date": "2026-03-15", "normalized_item_id": "gnorm_ice", "retailer": "giant", "raw_item_name": "SB BAGGED ICE 20LB", "normalized_item_name": "BAGGED ICE", "image_url": "", "upc": "", "line_total": "3.50", "order_id": "g1", "line_no": "1", } ], ) with mock.patch.object( review_products.click, "prompt", side_effect=["n", "ICE", "frozen", "ice", "manual merge", "q"], ): review_products.main.callback( giant_items_enriched_csv=str(giant_items_csv), costco_items_enriched_csv=str(costco_items_csv), giant_orders_csv=str(giant_orders_csv), costco_orders_csv=str(costco_orders_csv), purchases_csv=str(purchases_csv), queue_csv=str(queue_csv), resolutions_csv=str(resolutions_csv), catalog_csv=str(catalog_csv), links_csv=str(links_csv), limit=1, refresh_only=False, ) self.assertTrue(queue_csv.exists()) self.assertTrue(resolutions_csv.exists()) self.assertTrue(catalog_csv.exists()) self.assertTrue(links_csv.exists()) with queue_csv.open(newline="", encoding="utf-8") as handle: queue_rows = list(csv.DictReader(handle)) with resolutions_csv.open(newline="", encoding="utf-8") as handle: resolution_rows = list(csv.DictReader(handle)) with catalog_csv.open(newline="", encoding="utf-8") as handle: catalog_rows = list(csv.DictReader(handle)) with links_csv.open(newline="", encoding="utf-8") as handle: link_rows = list(csv.DictReader(handle)) self.assertEqual("approved", queue_rows[0]["status"]) self.assertEqual("create", queue_rows[0]["resolution_action"]) self.assertEqual("create", resolution_rows[0]["resolution_action"]) self.assertEqual("approved", resolution_rows[0]["status"]) self.assertEqual("ICE", catalog_rows[0]["catalog_name"]) self.assertEqual(catalog_rows[0]["catalog_id"], link_rows[0]["catalog_id"]) def test_build_review_queue_readds_orphaned_and_incomplete_links(self): purchase_rows = [ { "normalized_item_id": "gnorm_orphan", "catalog_id": "cat_missing", "retailer": "giant", "raw_item_name": "ORPHAN ITEM", "normalized_item_name": "ORPHAN ITEM", "upc": "", "line_total": "3.50", "is_fee": "false", "is_discount_line": "false", "is_coupon_line": "false", }, { "normalized_item_id": "gnorm_incomplete", "catalog_id": "cat_incomplete", "retailer": "giant", "raw_item_name": "INCOMPLETE ITEM", "normalized_item_name": "INCOMPLETE ITEM", "upc": "", "line_total": "4.50", "is_fee": "false", "is_discount_line": "false", "is_coupon_line": "false", }, ] link_rows = [ { "normalized_item_id": "gnorm_orphan", "catalog_id": "cat_missing", }, { "normalized_item_id": "gnorm_incomplete", "catalog_id": "cat_incomplete", }, ] catalog_rows = [ { "catalog_id": "cat_incomplete", "catalog_name": "INCOMPLETE ITEM", "product_type": "", } ] queue_rows = review_products.build_review_queue( purchase_rows, [], link_rows, catalog_rows, [], ) reasons = {row["normalized_item_id"]: row["reason_code"] for row in queue_rows} self.assertEqual("orphaned_catalog_link", reasons["gnorm_orphan"]) self.assertEqual("incomplete_catalog_link", reasons["gnorm_incomplete"]) if __name__ == "__main__": unittest.main()