465 lines
19 KiB
Python
465 lines
19 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(
|
|
[
|
|
{
|
|
"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"
|
|
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", 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_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"
|
|
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="1\nlinked by test\n",
|
|
color=True,
|
|
)
|
|
|
|
self.assertEqual(0, result.exit_code)
|
|
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_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()
|