214 lines
8.2 KiB
Python
214 lines
8.2 KiB
Python
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",
|
|
}
|
|
]
|
|
|
|
rows = build_purchases.build_purchase_rows(
|
|
[giant_row],
|
|
[costco_row],
|
|
giant_orders,
|
|
costco_orders,
|
|
)
|
|
|
|
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),
|
|
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))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|