import copy from datetime import datetime, timezone import unittest from reporting import ReportError, Reporter, date_window, records NOW = datetime(2026, 10, 15, tzinfo=timezone.utc).timestamp() BASE = "https://api.lightspeedapp.com/API/V3/Account/42" def sale(sid="1", when="2026-10-14T12:00:00+00:00", store="1", subtotal="10", discount="1", tax="1.17", total="10.17"): return {"saleID": sid, "shopID": store, "completeTime": when, "completed": "true", "voided": "false", "calcSubtotal": subtotal, "calcDiscount": discount, "calcTax1": tax, "calcTax2": "0", "calcTotal": total, "customerID": "PRIVATE", "note": "PRIVATE", "SaleLines": {"SaleLine": {"saleLineID": sid, "itemID": "9", "unitQuantity": "1"}}} def page(kind, values, next_url=""): return {"@attributes": {"next": next_url}, kind: values} class Fixture: def __init__(self, pages): self.pages = list(pages) self.urls = [] def __call__(self, url, token): self.urls.append(url) if not self.pages: raise AssertionError("Unexpected additional API request") result = self.pages.pop(0) if isinstance(result, Exception): raise result return copy.deepcopy(result) class ReportingTests(unittest.TestCase): def reporter(self, pages): self.fetch = Fixture(pages) return Reporter("42", "a" * 48, fetch=self.fetch, clock=lambda: NOW) def test_dst_day_bounds(self): start, end = date_window("2026-03-08", "2026-03-08") self.assertEqual((end - start).total_seconds(), 23 * 3600) start, end = date_window("2026-11-01", "2026-11-01") self.assertEqual((end - start).total_seconds(), 25 * 3600) def test_bad_dates_rejected(self): for pair in [("2026-02-30", "2026-03-01"), ("2026-02-01", "2026-01-01"), ("bad", "bad")]: with self.assertRaises(ReportError): date_window(*pair) def test_singleton_and_empty_records(self): self.assertEqual(records({"Shop": {"shopID": "1"}}, "Shop"), [{"shopID": "1"}]) self.assertEqual(records({}, "Shop"), []) with self.assertRaises(ReportError): records({"Shop": "bad"}, "Shop") def test_pins_account(self): report = self.reporter([{"Account": {"accountID": "43"}}, {"Account": [{"accountID": "42"}]}]) self.assertFalse(report.account_allowed("test-token")) self.assertTrue(report.account_allowed("test-token")) def test_inventory_excludes_summary_store_and_private_fields(self): product = {"itemID": "9", "description": "Dog food", "customSku": "00123", "defaultCost": "999", "ItemShops": {"ItemShop": [{"shopID": "0", "qoh": "30"}, {"shopID": "1", "qoh": "10"}, {"shopID": "2", "qoh": "20"}]}} report = self.reporter([page("Item", product)]) result = report.inventory("test-token") item = result["items"][0] self.assertEqual([s["shopID"] for s in item["store_inventory"]], ["1", "2"]) self.assertEqual(item["customSku"], "00123") self.assertNotIn("defaultCost", item) self.assertTrue(result["complete"]) def test_missing_inventory_is_unknown(self): item = Reporter.clean_item({"itemID": "9"}) self.assertIsNone(item["store_inventory"]) self.assertFalse(item["inventory_available"]) def test_partial_inventory_is_explicit(self): report = self.reporter([page("Item", [], BASE + "/Item.json?after=abc&limit=100")]) result = report.inventory("test-token") self.assertFalse(result["complete"]) self.assertTrue(result["next_cursor"]) def test_next_urls_cannot_leak_credentials(self): bad = ["https://evil.example/Item.json", "http://api.lightspeedapp.com/API/V3/Account/42/Item.json", BASE.replace("42", "43") + "/Item.json", BASE + "/Sale.json", "https://api.lightspeedapp.com@evil.example/API/V3/Account/42/Item.json"] for url in bad: with self.subTest(url=url): report = self.reporter([page("Item", [], url)]) with self.assertRaises(ReportError): report.inventory("test-token") self.assertEqual(len(self.fetch.urls), 1) def test_query_bound_cursor(self): report = self.reporter([page("Item", [], BASE + "/Item.json?after=abc")]) cursor = report.inventory("test-token")["next_cursor"] with self.assertRaises(ReportError): report.inventory("test-token", cursor, archived=True) with self.assertRaises(ReportError): report.inventory("test-token", cursor + "x") report.clock = lambda: NOW + 3601 with self.assertRaises(ReportError): report.inventory("test-token", cursor) def test_missing_pagination_fails_closed(self): report = self.reporter([{"Item": [{"itemID": "9"}]}]) with self.assertRaises(ReportError): report.inventory("test-token") def test_error_json_is_not_zero_sales(self): report = self.reporter([{"error": "denied"}]) with self.assertRaises(ReportError): report.summary("test-token", "2026-10-14", "2026-10-14") def test_empty_confirmed_is_zero(self): report = self.reporter([{"@attributes": {"count": "0"}}]) result = report.summary("test-token", "2026-10-14", "2026-10-14") self.assertTrue(result["complete"]) self.assertEqual(result["sales_count"], 0) def test_refunds_and_decimal_totals(self): rows = [sale(), sale("2", subtotal="-5", discount="0", tax="-0.65", total="-5.65"), sale("3", store="2", subtotal="0.10", discount="0", tax="0", total="0.10")] report = self.reporter([page("Sale", rows)]) result = report.summary("test-token", "2026-10-14", "2026-10-14") self.assertEqual(result["sales_count"], 3) one, two = result["stores"] self.assertEqual(one["subtotal_less_discount"], "4") self.assertEqual(one["total_including_tax"], "4.52") self.assertEqual(two["subtotal"], "0.10") def test_cross_page_totals(self): report = self.reporter([page("Sale", [sale("1")], BASE + "/Sale.json?after=abc"), page("Sale", [sale("2")])]) result = report.summary("test-token", "2026-10-14", "2026-10-14") self.assertEqual(result["sales_count"], 2) self.assertEqual(result["pages_scanned"], 2) self.assertEqual(result["stores"][0]["total_including_tax"], "20.34") def test_partial_report_never_returns_total(self): report = self.reporter([page("Sale", [sale()], BASE + "/Sale.json?after=abc")]) with self.assertRaisesRegex(ReportError, "No total"): report.summary("test-token", "2026-10-14", "2026-10-14", max_pages=1) def test_missing_amount_never_becomes_zero(self): row = sale() del row["calcTotal"] report = self.reporter([page("Sale", row)]) with self.assertRaises(ReportError): report.summary("test-token", "2026-10-14", "2026-10-14") def test_void_or_incomplete_filtered_response_rejected(self): for field, value in [("completed", "false"), ("voided", "true")]: row = sale() row[field] = value report = self.reporter([page("Sale", row)]) with self.assertRaises(ReportError): report.summary("test-token", "2026-10-14", "2026-10-14") def test_duplicate_sales_across_pages_rejected(self): report = self.reporter([page("Sale", sale(), BASE + "/Sale.json?after=abc"), page("Sale", sale())]) with self.assertRaisesRegex(ReportError, "Duplicate"): report.summary("test-token", "2026-10-14", "2026-10-14") def test_time_order_is_required_for_early_completion(self): report = self.reporter([page("Sale", [sale("1", "2026-10-13T12:00:00Z"), sale("2")])]) with self.assertRaisesRegex(ReportError, "time order"): report.sales("test-token", "2026-10-14", "2026-10-14") def test_local_midnight_boundary_and_stop(self): rows = [sale("1", "2026-10-15T04:00:00Z"), sale("2", "2026-10-15T03:59:59Z"), sale("3", "2026-10-14T04:00:00Z"), sale("4", "2026-10-14T03:59:59Z")] report = self.reporter([page("Sale", rows, BASE + "/Sale.json?after=older")]) report.clock = lambda: NOW + 86400 result = report.sales("test-token", "2026-10-14", "2026-10-14") self.assertEqual([s["saleID"] for s in result["sales"]], ["2", "3"]) self.assertTrue(result["complete"]) self.assertIsNone(result["next_cursor"]) self.assertNotIn("customerID", result["sales"][0]) self.assertNotIn("note", result["sales"][0]) def test_foreign_store_and_invalid_id_rejected(self): report = self.reporter([page("Sale", sale(store="2"))]) with self.assertRaises(ReportError): report.sales("test-token", "2026-10-14", "2026-10-14", shop_id="1") with self.assertRaises(ReportError): report.product("test-token", "../Sale") def test_upstream_failure_not_swallowed(self): report = self.reporter([ReportError("rate limit")]) with self.assertRaisesRegex(ReportError, "rate limit"): report.summary("test-token", "2026-10-14", "2026-10-14") if __name__ == "__main__": unittest.main()