202 lines
8.0 KiB
Python
202 lines
8.0 KiB
Python
import csv
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
import enrich_costco
|
|
import scrape_costco
|
|
import validate_cross_retailer_flow
|
|
|
|
|
|
class CostcoPipelineTests(unittest.TestCase):
|
|
def test_flatten_costco_data_preserves_discount_rows(self):
|
|
summary_payload = {
|
|
"data": {
|
|
"receiptsWithCounts": {
|
|
"receipts": [
|
|
{
|
|
"transactionBarcode": "abc",
|
|
"tenderArray": [{"tenderDescription": "VISA"}],
|
|
"couponArray": [{"upcnumberCoupon": "2100003746641"}],
|
|
}
|
|
]
|
|
}
|
|
}
|
|
}
|
|
detail_payloads = [
|
|
{
|
|
"data": {
|
|
"receiptsWithCounts": {
|
|
"receipts": [
|
|
{
|
|
"transactionBarcode": "abc",
|
|
"transactionDate": "2026-03-12",
|
|
"receiptType": "In-Warehouse",
|
|
"total": 10.0,
|
|
"totalItemCount": 2,
|
|
"instantSavings": 5.0,
|
|
"warehouseName": "MT VERNON",
|
|
"warehouseNumber": 1115,
|
|
"warehouseAddress1": "7940 RICHMOND HWY",
|
|
"warehouseCity": "ALEXANDRIA",
|
|
"warehouseState": "VA",
|
|
"warehousePostalCode": "22306",
|
|
"itemArray": [
|
|
{
|
|
"itemNumber": "4873222",
|
|
"itemDescription01": "ALL F&C",
|
|
"itemDescription02": "200OZ 160LOADS P104",
|
|
"itemDepartmentNumber": 14,
|
|
"transDepartmentNumber": 14,
|
|
"unit": 1,
|
|
"itemIdentifier": "E",
|
|
"amount": 19.99,
|
|
"itemUnitPriceAmount": 19.99,
|
|
},
|
|
{
|
|
"itemNumber": "374664",
|
|
"itemDescription01": "/ 4873222",
|
|
"itemDescription02": None,
|
|
"itemDepartmentNumber": 14,
|
|
"transDepartmentNumber": 14,
|
|
"unit": -1,
|
|
"itemIdentifier": None,
|
|
"amount": -5,
|
|
"itemUnitPriceAmount": 0,
|
|
},
|
|
],
|
|
}
|
|
]
|
|
}
|
|
}
|
|
}
|
|
]
|
|
|
|
orders, items = scrape_costco.flatten_costco_data(
|
|
summary_payload, detail_payloads, Path("costco_output/raw")
|
|
)
|
|
|
|
self.assertEqual(1, len(orders))
|
|
self.assertEqual(2, len(items))
|
|
self.assertEqual("false", items[0]["is_discount_line"])
|
|
self.assertEqual("true", items[1]["is_discount_line"])
|
|
self.assertEqual("true", items[1]["is_coupon_line"])
|
|
|
|
def test_costco_enricher_parses_size_pack_and_discount(self):
|
|
row = enrich_costco.parse_costco_item(
|
|
order_id="abc",
|
|
order_date="2026-03-12",
|
|
raw_path=Path("costco_output/raw/abc.json"),
|
|
line_no=1,
|
|
item={
|
|
"itemNumber": "60357",
|
|
"itemDescription01": "MIXED PEPPER",
|
|
"itemDescription02": "6-PACK",
|
|
"itemDepartmentNumber": 65,
|
|
"transDepartmentNumber": 65,
|
|
"unit": 1,
|
|
"itemIdentifier": "E",
|
|
"amount": 7.49,
|
|
"itemUnitPriceAmount": 7.49,
|
|
},
|
|
)
|
|
self.assertEqual("60357", row["retailer_item_id"])
|
|
self.assertEqual("MIXED PEPPER", row["item_name_norm"])
|
|
self.assertEqual("6", row["pack_qty"])
|
|
self.assertEqual("count", row["measure_type"])
|
|
|
|
discount = enrich_costco.parse_costco_item(
|
|
order_id="abc",
|
|
order_date="2026-03-12",
|
|
raw_path=Path("costco_output/raw/abc.json"),
|
|
line_no=2,
|
|
item={
|
|
"itemNumber": "374664",
|
|
"itemDescription01": "/ 4873222",
|
|
"itemDescription02": None,
|
|
"itemDepartmentNumber": 14,
|
|
"transDepartmentNumber": 14,
|
|
"unit": -1,
|
|
"itemIdentifier": None,
|
|
"amount": -5,
|
|
"itemUnitPriceAmount": 0,
|
|
},
|
|
)
|
|
self.assertEqual("true", discount["is_discount_line"])
|
|
self.assertEqual("true", discount["is_coupon_line"])
|
|
|
|
def test_cross_retailer_validation_writes_proof_example(self):
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
giant_csv = Path(tmpdir) / "giant_items_enriched.csv"
|
|
costco_csv = Path(tmpdir) / "costco_items_enriched.csv"
|
|
outdir = Path(tmpdir) / "combined"
|
|
|
|
fieldnames = enrich_costco.OUTPUT_FIELDS
|
|
giant_row = {field: "" for field in fieldnames}
|
|
giant_row.update(
|
|
{
|
|
"retailer": "giant",
|
|
"order_id": "g1",
|
|
"line_no": "1",
|
|
"order_date": "2026-03-01",
|
|
"retailer_item_id": "100",
|
|
"item_name": "FRESH BANANA",
|
|
"item_name_norm": "BANANA",
|
|
"upc": "4011",
|
|
"measure_type": "weight",
|
|
"is_store_brand": "false",
|
|
"is_fee": "false",
|
|
"is_discount_line": "false",
|
|
"is_coupon_line": "false",
|
|
"line_total": "1.29",
|
|
}
|
|
)
|
|
costco_row = {field: "" for field in fieldnames}
|
|
costco_row.update(
|
|
{
|
|
"retailer": "costco",
|
|
"order_id": "c1",
|
|
"line_no": "1",
|
|
"order_date": "2026-03-12",
|
|
"retailer_item_id": "30669",
|
|
"item_name": "BANANAS 3 LB / 1.36 KG",
|
|
"item_name_norm": "BANANA",
|
|
"upc": "",
|
|
"size_value": "3",
|
|
"size_unit": "lb",
|
|
"measure_type": "weight",
|
|
"is_store_brand": "false",
|
|
"is_fee": "false",
|
|
"is_discount_line": "false",
|
|
"is_coupon_line": "false",
|
|
"line_total": "2.98",
|
|
}
|
|
)
|
|
|
|
with giant_csv.open("w", newline="", encoding="utf-8") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=fieldnames)
|
|
writer.writeheader()
|
|
writer.writerow(giant_row)
|
|
with costco_csv.open("w", newline="", encoding="utf-8") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=fieldnames)
|
|
writer.writeheader()
|
|
writer.writerow(costco_row)
|
|
|
|
validate_cross_retailer_flow.main.callback(
|
|
giant_items_enriched_csv=str(giant_csv),
|
|
costco_items_enriched_csv=str(costco_csv),
|
|
outdir=str(outdir),
|
|
)
|
|
|
|
proof_path = outdir / "proof_examples.csv"
|
|
self.assertTrue(proof_path.exists())
|
|
with proof_path.open(newline="", encoding="utf-8") as handle:
|
|
rows = list(csv.DictReader(handle))
|
|
self.assertEqual(1, len(rows))
|
|
self.assertEqual("banana", rows[0]["proof_name"])
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|