fix(campaign): persist full crawl snapshots
All checks were successful
CD Pipeline / deploy (push) Successful in 2m22s
All checks were successful
CD Pipeline / deploy (push) Successful in 2m22s
This commit is contained in:
40
tests/test_promo_persistence_contract.py
Normal file
40
tests/test_promo_persistence_contract.py
Normal file
@@ -0,0 +1,40 @@
|
||||
from pathlib import Path
|
||||
|
||||
|
||||
ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
def test_promo_crawlers_append_full_snapshots_for_unchanged_items():
|
||||
scheduler_source = (ROOT / "scheduler.py").read_text(encoding="utf-8")
|
||||
|
||||
assert 'status_change=status_change if is_changed else "ACTIVE"' in scheduler_source
|
||||
assert "snapshot_count = 0" in scheduler_source
|
||||
assert "snapshot_count += 1" in scheduler_source
|
||||
assert '"snapshot_records": snapshot_count' in scheduler_source
|
||||
assert "無變動快照已寫入" in scheduler_source
|
||||
assert "Snapshot Written" in scheduler_source
|
||||
assert "Skip Write" not in scheduler_source
|
||||
assert "無變動跳過" not in scheduler_source
|
||||
|
||||
|
||||
def test_promo_product_model_contains_crawled_fields_needed_for_history():
|
||||
model_source = (ROOT / "database/edm_models.py").read_text(encoding="utf-8")
|
||||
|
||||
for field in [
|
||||
"batch_id",
|
||||
"crawled_at",
|
||||
"time_slot",
|
||||
"activity_time_text",
|
||||
"session_time_text",
|
||||
"i_code",
|
||||
"name",
|
||||
"price",
|
||||
"discount_text",
|
||||
"url",
|
||||
"image_url",
|
||||
"previous_price",
|
||||
"remain_qty",
|
||||
"status_change",
|
||||
"page_type",
|
||||
]:
|
||||
assert field in model_source
|
||||
Reference in New Issue
Block a user