95 lines
3.7 KiB
Python
95 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
import tempfile
|
|
import unittest
|
|
from pathlib import Path
|
|
|
|
from shanghai_housing.db import connect, import_template_file
|
|
from shanghai_housing.import_templates import validate_csv_file, validate_import_rows
|
|
from shanghai_housing.raw_artifacts import fingerprint_file
|
|
|
|
|
|
class ImportTemplateTests(unittest.TestCase):
|
|
def test_validates_area_monthly_metrics_template_file(self) -> None:
|
|
result = validate_csv_file(
|
|
"area_monthly_metrics",
|
|
Path("docs/import_templates/area_monthly_metrics.csv"),
|
|
)
|
|
|
|
self.assertTrue(result.valid)
|
|
self.assertEqual(result.row_count, 1)
|
|
self.assertEqual(result.errors, [])
|
|
|
|
def test_reports_missing_required_value_and_invalid_range(self) -> None:
|
|
rows = [
|
|
{
|
|
"area_id": "",
|
|
"month": "2026-13",
|
|
"transaction_count": "-1",
|
|
"transaction_price_psm": "abc",
|
|
"listing_count": "10",
|
|
"listing_price_psm": "100",
|
|
"median_days_on_market": "30",
|
|
"rent_price_psm": "3",
|
|
"new_supply_units": "0",
|
|
"land_residential_gfa_sqm": "0",
|
|
"mortgage_rate_pct": "3.35",
|
|
"policy_signal": "3",
|
|
"source": "manual",
|
|
}
|
|
]
|
|
|
|
result = validate_import_rows("area_monthly_metrics", rows)
|
|
|
|
self.assertFalse(result.valid)
|
|
messages = {(error.column, error.message) for error in result.errors}
|
|
self.assertIn(("area_id", "value is required"), messages)
|
|
self.assertIn(("month", "must be YYYY-MM"), messages)
|
|
self.assertIn(("transaction_count", "must be >= 0"), messages)
|
|
self.assertIn(("transaction_price_psm", "must be a number"), messages)
|
|
self.assertIn(("policy_signal", "must be <= 2"), messages)
|
|
|
|
def test_reports_unexpected_headers_from_csv(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
path = Path(tmpdir) / "bad.csv"
|
|
with path.open("w", encoding="utf-8", newline="") as handle:
|
|
writer = csv.DictWriter(handle, fieldnames=["area_id", "unknown"])
|
|
writer.writeheader()
|
|
writer.writerow({"area_id": "qiantan", "unknown": "value"})
|
|
|
|
result = validate_csv_file("area_monthly_metrics", path)
|
|
|
|
self.assertFalse(result.valid)
|
|
messages = {(error.row, error.column, error.message) for error in result.errors}
|
|
self.assertIn((0, "unknown", "unexpected header"), messages)
|
|
self.assertIn((0, "month", "missing required header"), messages)
|
|
|
|
def test_imports_valid_template_file_to_sqlite(self) -> None:
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
db_path = Path(tmpdir) / "research.sqlite"
|
|
count = import_template_file(
|
|
db_path,
|
|
"policy_events",
|
|
Path("docs/import_templates/policy_events.csv"),
|
|
)
|
|
with connect(db_path) as conn:
|
|
rows = conn.execute("SELECT * FROM policy_events").fetchall()
|
|
|
|
self.assertEqual(count, 1)
|
|
self.assertEqual(len(rows), 1)
|
|
self.assertEqual(rows[0]["level"], "shanghai")
|
|
|
|
def test_fingerprints_import_file_without_local_path(self) -> None:
|
|
fingerprint = fingerprint_file(Path("docs/import_templates/policy_events.csv"))
|
|
|
|
self.assertEqual(fingerprint.original_filename, "policy_events.csv")
|
|
self.assertEqual(len(fingerprint.sha256), 64)
|
|
self.assertGreater(fingerprint.size_bytes, 0)
|
|
self.assertTrue(fingerprint.raw_uri.startswith("raw://sha256/"))
|
|
self.assertNotIn("/Users/", fingerprint.raw_uri)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|