100 lines
3.8 KiB
Python
100 lines
3.8 KiB
Python
import hashlib
|
|
import json
|
|
import tempfile
|
|
import unittest
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
from urllib.parse import parse_qs
|
|
|
|
from app.bot_tourism import BotTourismSource, TourismSourceError, parse_bot_tourism_report
|
|
|
|
|
|
FIXTURES = Path(__file__).parent / "fixtures"
|
|
REPORT_HTML = (FIXTURES / "bot_tourism_report.html").read_text(encoding="utf-8")
|
|
FORM_HTML = (FIXTURES / "bot_tourism_form.html").read_text(encoding="utf-8")
|
|
|
|
|
|
class FakeResponse:
|
|
def __init__(self, body):
|
|
self.body = body.encode("utf-8")
|
|
self.status = 200
|
|
|
|
def read(self):
|
|
return self.body
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
|
|
class FakeOpener:
|
|
def __init__(self):
|
|
self.calls = []
|
|
|
|
def open(self, request, timeout):
|
|
body = FORM_HTML if request.data is None else REPORT_HTML
|
|
self.calls.append({"url": request.full_url, "body": request.data, "timeout": timeout})
|
|
return FakeResponse(body)
|
|
|
|
|
|
class BotTourismSourceTests(unittest.TestCase):
|
|
def test_default_source_opener_is_constructible(self):
|
|
source = BotTourismSource()
|
|
self.assertIsNotNone(source.opener)
|
|
|
|
def test_parser_builds_point_in_time_snapshot_from_bot_table(self):
|
|
raw_hash = hashlib.sha256(REPORT_HTML.encode("utf-8")).hexdigest()
|
|
snapshot = parse_bot_tourism_report(
|
|
REPORT_HTML,
|
|
retrieved_at="2026-08-23T02:00:00+00:00",
|
|
raw_payload_hash=raw_hash,
|
|
)
|
|
|
|
self.assertEqual(snapshot["as_of"], "2026-06-30")
|
|
self.assertEqual(snapshot["data_quality"], "provisional")
|
|
self.assertEqual(snapshot["source"]["source_id"], "bot.ec_ei_028_s2")
|
|
self.assertEqual(snapshot["source"]["published_at"], "2026-07-31T14:30:00+07:00")
|
|
self.assertEqual(snapshot["source"]["raw_payload_hash"], raw_hash)
|
|
observation = snapshot["observations"][0]
|
|
self.assertEqual(observation["metric_key"], "foreign_arrivals_yoy")
|
|
self.assertEqual(observation["period"], "2026-06")
|
|
self.assertLess(observation["value"], 0)
|
|
self.assertTrue(observation["scale"] > 0)
|
|
self.assertEqual(observation["history_points"], 12)
|
|
|
|
def test_parser_rejects_report_without_tourism_row(self):
|
|
with self.assertRaisesRegex(TourismSourceError, "tourism table"):
|
|
parse_bot_tourism_report("<html><table id='dgExcel'><tr><th>wrong</th></tr></table></html>")
|
|
|
|
def test_source_fetches_history_and_persists_raw_and_normalized_snapshot(self):
|
|
opener = FakeOpener()
|
|
with tempfile.TemporaryDirectory() as temp_dir:
|
|
root = Path(temp_dir)
|
|
source = BotTourismSource(
|
|
opener=opener,
|
|
clock=lambda: datetime(2026, 8, 23, 2, 0, tzinfo=timezone.utc),
|
|
raw_dir=root / "raw",
|
|
snapshot_dir=root / "snapshots",
|
|
)
|
|
snapshot = source.fetch()
|
|
|
|
self.assertEqual(len(opener.calls), 2)
|
|
posted = parse_qs(opener.calls[1]["body"].decode("utf-8"))
|
|
self.assertEqual(posted["drpFromMonth"], ["xxxx01xx"])
|
|
self.assertEqual(posted["drpFromYear"], ["2015xxxx"])
|
|
self.assertEqual(posted["drpToMonth"], ["xxxx06xx"])
|
|
self.assertEqual(posted["drpToYear"], ["2026xxxx"])
|
|
self.assertEqual(posted["btnSubmit"], ["Submit"])
|
|
|
|
source_meta = snapshot["source"]
|
|
raw_path = root / "raw" / source_meta["raw_snapshot_file"]
|
|
snapshot_path = root / "snapshots" / source_meta["snapshot_file"]
|
|
self.assertEqual(raw_path.read_text(encoding="utf-8"), REPORT_HTML)
|
|
self.assertEqual(json.loads(snapshot_path.read_text(encoding="utf-8")), snapshot)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|