240 lines
9.5 KiB
Python
240 lines
9.5 KiB
Python
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(
|
|
[
|
|
{
|
|
"observed_product_id": "gobs_1",
|
|
"canonical_product_id": "",
|
|
"retailer": "giant",
|
|
"raw_item_name": "SB BAGGED ICE 20LB",
|
|
"normalized_item_name": "BAGGED ICE",
|
|
"upc": "",
|
|
"line_total": "3.50",
|
|
},
|
|
{
|
|
"observed_product_id": "gobs_1",
|
|
"canonical_product_id": "",
|
|
"retailer": "giant",
|
|
"raw_item_name": "SB BAG ICE CUBED 10LB",
|
|
"normalized_item_name": "BAG ICE",
|
|
"upc": "",
|
|
"line_total": "2.50",
|
|
},
|
|
],
|
|
[],
|
|
)
|
|
|
|
self.assertEqual(1, len(queue_rows))
|
|
self.assertEqual("gobs_1", queue_rows[0]["observed_product_id"])
|
|
self.assertIn("SB BAGGED ICE 20LB", queue_rows[0]["raw_item_names"])
|
|
|
|
def test_build_canonical_suggestions_prefers_upc_then_name(self):
|
|
suggestions = review_products.build_canonical_suggestions(
|
|
[
|
|
{
|
|
"normalized_item_name": "MIXED PEPPER",
|
|
"upc": "12345",
|
|
}
|
|
],
|
|
[
|
|
{
|
|
"canonical_product_id": "gcan_1",
|
|
"canonical_name": "MIXED PEPPER",
|
|
"upc": "",
|
|
},
|
|
{
|
|
"canonical_product_id": "gcan_2",
|
|
"canonical_name": "MIXED PEPPER 6 PACK",
|
|
"upc": "12345",
|
|
},
|
|
],
|
|
)
|
|
|
|
self.assertEqual("gcan_2", suggestions[0]["canonical_product_id"])
|
|
self.assertEqual("exact upc", suggestions[0]["reason"])
|
|
self.assertEqual("gcan_1", suggestions[1]["canonical_product_id"])
|
|
|
|
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) / "canonical_catalog.csv"
|
|
|
|
purchase_fields = [
|
|
"purchase_date",
|
|
"retailer",
|
|
"order_id",
|
|
"line_no",
|
|
"observed_product_id",
|
|
"canonical_product_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",
|
|
"observed_product_id": "gobs_mix",
|
|
"canonical_product_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",
|
|
"observed_product_id": "gobs_mix",
|
|
"canonical_product_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",
|
|
},
|
|
]
|
|
)
|
|
|
|
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(
|
|
{
|
|
"canonical_product_id": "gcan_mix",
|
|
"canonical_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 observed product (1/1)", result.output)
|
|
self.assertIn("Resolve this observed product group", result.output)
|
|
self.assertIn("actions: [l]ink existing [n]ew canonical e[x]clude [s]kip [q]uit", result.output)
|
|
first_item = result.output.index("2026-03-14 | 7.49")
|
|
second_item = result.output.index("2026-03-12 | 6.99")
|
|
self.assertLess(first_item, second_item)
|
|
self.assertIn("image: https://example.test/mixed-pepper.jpg", result.output)
|
|
self.assertIn("gcan_mix | MIXED PEPPER (exact normalized name)", result.output)
|
|
self.assertIn("\x1b[", result.output)
|
|
|
|
def test_review_products_creates_canonical_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) / "canonical_catalog.csv"
|
|
|
|
with purchases_csv.open("w", newline="", encoding="utf-8") as handle:
|
|
writer = csv.DictWriter(
|
|
handle,
|
|
fieldnames=[
|
|
"purchase_date",
|
|
"observed_product_id",
|
|
"canonical_product_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",
|
|
"observed_product_id": "gobs_ice",
|
|
"canonical_product_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]["canonical_name"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|