import csv import tempfile import unittest from pathlib import Path from unittest import mock from click.testing import CliRunner import review_products 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" purchase_fields = [ "purchase_date", "retailer", "order_id", "line_no", "normalized_item_id", "catalog_id", "raw_item_name", "normalized_item_name", "image_url", "upc", "line_total", ] with purchases_csv.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=purchase_fields) writer.writeheader() writer.writerows( [ { "purchase_date": "2026-03-14", "retailer": "costco", "order_id": "c2", "line_no": "2", "normalized_item_id": "cnorm_mix", "catalog_id": "", "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", "catalog_id": "", "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", "catalog_id": "cat_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": "", } ) runner = CliRunner() result = runner.invoke( review_products.main, [ "--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 [s]earch [n]ew 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" with purchases_csv.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter( handle, fieldnames=[ "purchase_date", "retailer", "order_id", "line_no", "normalized_item_id", "catalog_id", "raw_item_name", "normalized_item_name", "image_url", "upc", "line_total", ], ) writer.writeheader() writer.writerow( { "purchase_date": "2026-03-14", "retailer": "giant", "order_id": "g1", "line_no": "1", "normalized_item_id": "gnorm_ice", "catalog_id": "", "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, [ "--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" with purchases_csv.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter( handle, fieldnames=[ "purchase_date", "retailer", "order_id", "line_no", "normalized_item_id", "catalog_id", "raw_item_name", "normalized_item_name", "image_url", "upc", "line_total", ], ) writer.writeheader() writer.writerows( [ { "purchase_date": "2026-03-14", "retailer": "costco", "order_id": "c2", "line_no": "2", "normalized_item_id": "cnorm_mix", "catalog_id": "", "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", "catalog_id": "", "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", "catalog_id": "cat_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": "", } ) result = CliRunner().invoke( review_products.main, [ "--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\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" with purchases_csv.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter( handle, fieldnames=[ "purchase_date", "retailer", "order_id", "line_no", "normalized_item_id", "catalog_id", "raw_item_name", "normalized_item_name", "image_url", "upc", "line_total", ], ) writer.writeheader() writer.writerow( { "purchase_date": "2026-03-14", "retailer": "giant", "order_id": "g1", "line_no": "1", "normalized_item_id": "gnorm_ice", "catalog_id": "", "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, [ "--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="s\nzzz\nq\nq\n", color=True, ) self.assertEqual(0, result.exit_code) self.assertIn("no matches found", result.output) 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" with purchases_csv.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter( handle, fieldnames=[ "purchase_date", "normalized_item_id", "catalog_id", "retailer", "raw_item_name", "normalized_item_name", "image_url", "upc", "line_total", "order_id", "line_no", ], ) writer.writeheader() writer.writerow( { "purchase_date": "2026-03-15", "normalized_item_id": "gnorm_ice", "catalog_id": "", "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( 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"]) if __name__ == "__main__": unittest.main()