101 lines
3.8 KiB
Python
101 lines
3.8 KiB
Python
import csv
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
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_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=[
|
|
"observed_product_id",
|
|
"canonical_product_id",
|
|
"retailer",
|
|
"raw_item_name",
|
|
"normalized_item_name",
|
|
"upc",
|
|
"line_total",
|
|
],
|
|
)
|
|
writer.writeheader()
|
|
writer.writerow(
|
|
{
|
|
"observed_product_id": "gobs_ice",
|
|
"canonical_product_id": "",
|
|
"retailer": "giant",
|
|
"raw_item_name": "SB BAGGED ICE 20LB",
|
|
"normalized_item_name": "BAGGED ICE",
|
|
"upc": "",
|
|
"line_total": "3.50",
|
|
}
|
|
)
|
|
|
|
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()
|