Exclude cancelled reservations before ARR field review

tests/test_arr_direct_data.py

@@ -291,6 +291,33 @@ class DirectDataPostgresTests(unittest.TestCase):
         self.assertEqual(self.monthly_worker().process_next().status, "published")
         self.assertEqual(self.coordinator.get_price_review(outcome.job_id, 50, 0)["case_status"], "completed")
 
+    def test_cancelled_direct_source_keeps_audit_but_no_room_review_or_monthly_row(self):
+        row = self.transport.rows[0]
+        row["reservationStatus"] = "Cancelled"
+        row["roomStay"].pop("roomId", None)
+        row["roomStay"].pop("currentRoomInfo", None)
+        for rate in row["roomStay"]["roomRates"]:
+            rate.pop("roomId", None)
+        self.transport.calendar_rooms = []
+        self.assertEqual(self.execute().status, "succeeded")
+        self.assertEqual(self.sql("SELECT result_schema_version,source_rows,retained_rows,excluded_cancelled_rows FROM finance.daily_versions"),
+                         [("5.0", 6, 5, 1)])
+        self.assertEqual(self.sql("SELECT outcome FROM finance.daily_records WHERE source_sequence=1"), [("excluded_cancelled",)])
+        self.assertEqual(self.count("finance.v_active_daily_facts"), 5)
+        self.assertEqual(self.monthly_worker().process_next().status, "published")
+        key, size = self.sql("SELECT a.object_key,a.byte_size FROM reporting.monthly_runs r JOIN ingestion.artifacts a ON a.id=r.workbook_artifact_id")[0]
+        output = self.files / "monthly.xlsx"
+        self.store.materialize(key, output, size)
+        workbook = load_workbook(output, data_only=False)
+        try:
+            self.assertEqual(sum(c.data_type == "f" for sheet in workbook for row in sheet for c in row), 5)
+        finally:
+            workbook.close()
+        calls = len(self.transport.calls)
+        self.assertEqual(self.execute().status, "succeeded")
+        self.assertEqual(len(self.transport.calls), calls)
+        self.assertEqual(self.count("finance.daily_versions"), 1)
+
     def test_commit_ack_loss_retry_has_one_result_and_no_new_fetch(self):
         state = {"armed": True}
         self.repository._connect = lambda dsn: pg_helpers.LoseCommitConnection(self.database.connect(dsn), state)