761 lines
31 KiB
Python
761 lines
31 KiB
Python
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()
|