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_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" 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), ], input="q\n", color=True, ) self.assertEqual(0, result.exit_code) self.assertIn("Review 1/1: Resolve normalized_item MIXED PEPPER to catalog_name [__]?", result.output) self.assertIn("2 matched items:", result.output) self.assertIn("[l]ink existing [n]ew catalog e[x]clude [s]kip [q]uit:", result.output) first_item = result.output.index("[1] 2026-03-14 | 7.49") second_item = result.output.index("[2] 2026-03-12 | 6.99") self.assertLess(first_item, second_item) self.assertIn("https://example.test/mixed-pepper.jpg", result.output) self.assertIn("1 catalog_name suggestions found:", result.output) self.assertIn("[1] MIXED PEPPER", 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" 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), ], input="q\n", color=True, ) self.assertEqual(0, result.exit_code) self.assertIn("no catalog_name suggestions found", result.output) def test_link_existing_uses_numbered_selection_and_confirmation(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" 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), "--limit", "1", ], input="l\n1\ny\nlinked by test\n", color=True, ) self.assertEqual(0, result.exit_code) self.assertIn("Select the catalog_name to associate 2 items with:", result.output) self.assertIn("[1] MIXED PEPPER | cat_mix", result.output) self.assertIn('2 "MIXED PEPPER" items and future matches will be associated with "MIXED PEPPER".', result.output) self.assertIn("actions: [y]es [n]o [b]ack [s]kip [q]uit", result.output) with resolutions_csv.open(newline="", encoding="utf-8") as handle: rows = list(csv.DictReader(handle)) self.assertEqual("cat_mix", rows[0]["catalog_id"]) self.assertEqual("link", rows[0]["resolution_action"]) 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" 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), limit=1, refresh_only=False, ) self.assertTrue(queue_csv.exists()) self.assertTrue(resolutions_csv.exists()) self.assertTrue(catalog_csv.exists()) 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)) self.assertEqual("create", resolution_rows[0]["resolution_action"]) self.assertEqual("approved", resolution_rows[0]["status"]) self.assertEqual("ICE", catalog_rows[0]["catalog_name"]) if __name__ == "__main__": unittest.main()