import csv import tempfile import unittest from pathlib import Path import build_purchases import enrich_costco class PurchaseLogTests(unittest.TestCase): def test_derive_metrics_prefers_picked_weight_and_pack_count(self): metrics = build_purchases.derive_metrics( { "line_total": "4.00", "qty": "1", "pack_qty": "4", "size_value": "", "size_unit": "", "picked_weight": "2", "price_per_each": "", "price_per_lb": "", "price_per_oz": "", } ) self.assertEqual("4", metrics["price_per_each"]) self.assertEqual("1", metrics["price_per_count"]) self.assertEqual("2", metrics["price_per_lb"]) self.assertEqual("0.125", metrics["price_per_oz"]) self.assertEqual("picked_weight_lb", metrics["price_per_lb_basis"]) def test_build_purchase_rows_maps_canonical_ids(self): fieldnames = enrich_costco.OUTPUT_FIELDS giant_row = {field: "" for field in fieldnames} giant_row.update( { "retailer": "giant", "order_id": "g1", "line_no": "1", "observed_item_key": "giant:g1:1", "order_date": "2026-03-01", "item_name": "FRESH BANANA", "item_name_norm": "BANANA", "retailer_item_id": "100", "upc": "4011", "qty": "1", "unit": "LB", "line_total": "1.29", "unit_price": "1.29", "measure_type": "weight", "price_per_lb": "1.29", "raw_order_path": "giant_output/raw/g1.json", "is_discount_line": "false", "is_coupon_line": "false", "is_fee": "false", } ) costco_row = {field: "" for field in fieldnames} costco_row.update( { "retailer": "costco", "order_id": "c1", "line_no": "1", "observed_item_key": "costco:c1:1", "order_date": "2026-03-12", "item_name": "BANANAS 3 LB / 1.36 KG", "item_name_norm": "BANANA", "retailer_item_id": "30669", "qty": "1", "unit": "E", "line_total": "2.98", "unit_price": "2.98", "size_value": "3", "size_unit": "lb", "measure_type": "weight", "price_per_lb": "0.9933", "raw_order_path": "costco_output/raw/c1.json", "is_discount_line": "false", "is_coupon_line": "false", "is_fee": "false", } ) giant_orders = [ { "order_id": "g1", "store_name": "Giant", "store_number": "42", "store_city": "Springfield", "store_state": "VA", } ] costco_orders = [ { "order_id": "c1", "store_name": "MT VERNON", "store_number": "1115", "store_city": "ALEXANDRIA", "store_state": "VA", } ] <<<<<<< HEAD rows, _observed, _canon, _links = build_purchases.build_purchase_rows( ======= rows = build_purchases.build_purchase_rows( >>>>>>> be1bf63 (Build pivot-ready purchase log) [giant_row], [costco_row], giant_orders, costco_orders, <<<<<<< HEAD [], ======= >>>>>>> be1bf63 (Build pivot-ready purchase log) ) self.assertEqual(2, len(rows)) self.assertTrue(all(row["canonical_product_id"] for row in rows)) self.assertEqual({"giant", "costco"}, {row["retailer"] for row in rows}) def test_main_writes_purchase_and_example_csvs(self): with tempfile.TemporaryDirectory() as tmpdir: giant_items = Path(tmpdir) / "giant_items.csv" costco_items = Path(tmpdir) / "costco_items.csv" giant_orders = Path(tmpdir) / "giant_orders.csv" costco_orders = Path(tmpdir) / "costco_orders.csv" purchases_csv = Path(tmpdir) / "combined" / "purchases.csv" examples_csv = Path(tmpdir) / "combined" / "comparison_examples.csv" fieldnames = enrich_costco.OUTPUT_FIELDS rows = [] giant_row = {field: "" for field in fieldnames} giant_row.update( { "retailer": "giant", "order_id": "g1", "line_no": "1", "observed_item_key": "giant:g1:1", "order_date": "2026-03-01", "item_name": "FRESH BANANA", "item_name_norm": "BANANA", "retailer_item_id": "100", "upc": "4011", "qty": "1", "unit": "LB", "line_total": "1.29", "unit_price": "1.29", "measure_type": "weight", "price_per_lb": "1.29", "raw_order_path": "giant_output/raw/g1.json", "is_discount_line": "false", "is_coupon_line": "false", "is_fee": "false", } ) costco_row = {field: "" for field in fieldnames} costco_row.update( { "retailer": "costco", "order_id": "c1", "line_no": "1", "observed_item_key": "costco:c1:1", "order_date": "2026-03-12", "item_name": "BANANAS 3 LB / 1.36 KG", "item_name_norm": "BANANA", "retailer_item_id": "30669", "qty": "1", "unit": "E", "line_total": "2.98", "unit_price": "2.98", "size_value": "3", "size_unit": "lb", "measure_type": "weight", "price_per_lb": "0.9933", "raw_order_path": "costco_output/raw/c1.json", "is_discount_line": "false", "is_coupon_line": "false", "is_fee": "false", } ) rows.extend([giant_row, costco_row]) for path, source_rows in [ (giant_items, [giant_row]), (costco_items, [costco_row]), ]: with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=fieldnames) writer.writeheader() writer.writerows(source_rows) for path, source_rows in [ (giant_orders, [{"order_id": "g1", "store_name": "Giant", "store_number": "42", "store_city": "Springfield", "store_state": "VA"}]), (costco_orders, [{"order_id": "c1", "store_name": "MT VERNON", "store_number": "1115", "store_city": "ALEXANDRIA", "store_state": "VA"}]), ]: with path.open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=["order_id", "store_name", "store_number", "store_city", "store_state"]) writer.writeheader() writer.writerows(source_rows) build_purchases.main.callback( giant_items_enriched_csv=str(giant_items), costco_items_enriched_csv=str(costco_items), giant_orders_csv=str(giant_orders), costco_orders_csv=str(costco_orders), <<<<<<< HEAD resolutions_csv=str(Path(tmpdir) / "review_resolutions.csv"), catalog_csv=str(Path(tmpdir) / "canonical_catalog.csv"), links_csv=str(Path(tmpdir) / "product_links.csv"), ======= >>>>>>> be1bf63 (Build pivot-ready purchase log) output_csv=str(purchases_csv), examples_csv=str(examples_csv), ) self.assertTrue(purchases_csv.exists()) self.assertTrue(examples_csv.exists()) with purchases_csv.open(newline="", encoding="utf-8") as handle: purchase_rows = list(csv.DictReader(handle)) with examples_csv.open(newline="", encoding="utf-8") as handle: example_rows = list(csv.DictReader(handle)) self.assertEqual(2, len(purchase_rows)) self.assertEqual(1, len(example_rows)) <<<<<<< HEAD def test_build_purchase_rows_applies_manual_resolution(self): fieldnames = enrich_costco.OUTPUT_FIELDS giant_row = {field: "" for field in fieldnames} giant_row.update( { "retailer": "giant", "order_id": "g1", "line_no": "1", "observed_item_key": "giant:g1:1", "order_date": "2026-03-01", "item_name": "SB BAGGED ICE 20LB", "item_name_norm": "BAGGED ICE", "retailer_item_id": "100", "upc": "", "qty": "1", "unit": "EA", "line_total": "3.50", "unit_price": "3.50", "measure_type": "each", "raw_order_path": "giant_output/raw/g1.json", "is_discount_line": "false", "is_coupon_line": "false", "is_fee": "false", } ) observed_rows, _canonical_rows, _link_rows, _observed_id_by_key, _canonical_by_observed = ( build_purchases.build_link_state([giant_row]) ) observed_product_id = observed_rows[0]["observed_product_id"] rows, _observed, _canon, _links = build_purchases.build_purchase_rows( [giant_row], [], [{"order_id": "g1", "store_name": "Giant", "store_number": "42", "store_city": "Springfield", "store_state": "VA"}], [], [ { "observed_product_id": observed_product_id, "canonical_product_id": "gcan_manual_ice", "resolution_action": "create", "status": "approved", "resolution_notes": "manual ice merge", "reviewed_at": "2026-03-16", } ], ) self.assertEqual("gcan_manual_ice", rows[0]["canonical_product_id"]) self.assertEqual("approved", rows[0]["review_status"]) self.assertEqual("create", rows[0]["resolution_action"]) ======= >>>>>>> be1bf63 (Build pivot-ready purchase log) if __name__ == "__main__": unittest.main()