from pathlib import Path from datetime import date from services import import_service from services.import_service import _build_in_clause def test_build_in_clause_binds_each_value(): clause, params = _build_in_clause("d", ["2026-05-01", "x' OR 1=1 --"]) assert clause == ":d_0, :d_1" assert params == {"d_0": "2026-05-01", "d_1": "x' OR 1=1 --"} def test_import_service_does_not_interpolate_date_values_into_in_clauses(): source = Path("services/import_service.py").read_text(encoding="utf-8") assert "join([f\"'{d}'\"" not in source assert "join([f\"'{d}'\" for d in" not in source def test_daily_snapshot_delete_casts_text_date_column_on_postgres(monkeypatch): monkeypatch.setattr(import_service, "_db_dialect_name", lambda: "postgresql") assert import_service._date_filter_expr("snapshot_date") == "snapshot_date::date" assert import_service._normalise_date_values_for_sql(["2026-05-01"]) == [date(2026, 5, 1)] def test_daily_snapshot_delete_query_uses_snapshot_date_cast_helper(): source = Path("services/import_service.py").read_text(encoding="utf-8") assert 'snapshot_date_expr = _date_filter_expr("snapshot_date")' in source assert "DELETE FROM {table_name} WHERE {snapshot_date_expr} IN" in source assert "DELETE FROM daily_sales_snapshot WHERE snapshot_date IN" not in source def test_daily_snapshot_delete_uses_iso_dates_on_sqlite(monkeypatch): monkeypatch.setattr(import_service, "_db_dialect_name", lambda: "sqlite") assert import_service._date_filter_expr("snapshot_date") == "date(snapshot_date)" assert import_service._normalise_date_values_for_sql([date(2026, 5, 1)]) == ["2026-05-01"] def test_monthly_summary_import_does_not_replace_entire_table(): source = Path("routes/import_routes.py").read_text(encoding="utf-8") start = source.index("DELETE FROM monthly_summary_analysis WHERE year = :y AND month = :m") end = source.index("月份總表資料匯入成功", start) monthly_import_block = source[start:end] assert "if_exists='append'" in monthly_import_block assert "if_exists='replace'" not in monthly_import_block