113 lines
4.0 KiB
Python
113 lines
4.0 KiB
Python
"""Pure-logic tests for the report library and NL analytics.
|
|
|
|
No DB and no live API calls: everything here is filter resolution, input
|
|
validation, and serializer math — the parts that decide what a report or an
|
|
answer is allowed to contain before any query runs.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
from fastapi import HTTPException
|
|
|
|
from analytics.ask import MAX_QUESTION_CHARS, _clamp, _parse_date, ask_analytics
|
|
from analytics.serializers import serialize_source_count
|
|
from reports.runner import FILTER_KEYS, REPORT_TYPES, resolve_filters
|
|
from reports.views import _clean_filters, _validate_report_type
|
|
|
|
|
|
# ---------------------------------------------------------------- runner filters
|
|
|
|
def test_resolve_filters_empty_means_unbounded():
|
|
f = resolve_filters(None)
|
|
assert f["from_date"] is None and f["to_date"] is None
|
|
assert f["months"] == 7 and f["top"] == 10
|
|
|
|
|
|
def test_resolve_filters_window_days_is_rolling():
|
|
f = resolve_filters({"window_days": 90})
|
|
assert f["from_date"] is not None and f["to_date"] is not None
|
|
span = f["to_date"] - f["from_date"]
|
|
assert span == timedelta(days=90)
|
|
# resolved against "now", not a stored date
|
|
assert abs((datetime.now(timezone.utc) - f["to_date"]).total_seconds()) < 60
|
|
|
|
|
|
def test_resolve_filters_explicit_dates_beat_window_days():
|
|
f = resolve_filters({
|
|
"from_date": "2026-01-01",
|
|
"to_date": "2026-02-01T00:00:00Z",
|
|
"window_days": 90,
|
|
})
|
|
assert f["from_date"] == datetime(2026, 1, 1, tzinfo=timezone.utc)
|
|
assert f["to_date"] == datetime(2026, 2, 1, tzinfo=timezone.utc)
|
|
|
|
|
|
def test_resolve_filters_clamps_and_survives_garbage():
|
|
f = resolve_filters({"months": 999, "top": -3, "window_days": "junk", "department": " "})
|
|
assert f["months"] == 24
|
|
assert f["top"] == 1
|
|
assert f["from_date"] is None # unparseable window resolves to unbounded
|
|
assert f["department"] is None
|
|
|
|
|
|
# ---------------------------------------------------------------- views validation
|
|
|
|
def test_clean_filters_whitelists_keys():
|
|
cleaned = _clean_filters({"window_days": 30, "evil": "1; DROP TABLE", "department": ""})
|
|
assert cleaned == {"window_days": 30}
|
|
assert set(cleaned) <= set(FILTER_KEYS)
|
|
|
|
|
|
def test_clean_filters_rejects_non_object():
|
|
with pytest.raises(HTTPException) as exc:
|
|
_clean_filters("window_days=30")
|
|
assert exc.value.status_code == 422
|
|
|
|
|
|
def test_report_type_must_be_whitelisted():
|
|
for rt in REPORT_TYPES:
|
|
assert _validate_report_type(rt) == rt
|
|
with pytest.raises(HTTPException) as exc:
|
|
_validate_report_type("select_star")
|
|
assert exc.value.status_code == 422
|
|
|
|
|
|
# ---------------------------------------------------------------- ask analytics
|
|
|
|
def test_ask_rejects_blank_and_oversized_questions():
|
|
with pytest.raises(ValueError):
|
|
asyncio.run(ask_analytics(None, " "))
|
|
with pytest.raises(ValueError):
|
|
asyncio.run(ask_analytics(None, "x" * (MAX_QUESTION_CHARS + 1)))
|
|
|
|
|
|
def test_ask_date_parsing_is_forgiving():
|
|
assert _parse_date("2026-03-01") == datetime(2026, 3, 1, tzinfo=timezone.utc)
|
|
assert _parse_date("2026-03-01T05:00:00Z") is not None
|
|
assert _parse_date("last tuesday") is None
|
|
assert _parse_date(None) is None
|
|
|
|
|
|
def test_ask_clamp():
|
|
assert _clamp("12", 1, 24, 7) == 12
|
|
assert _clamp(999, 1, 24, 7) == 24
|
|
assert _clamp("junk", 1, 24, 7) == 7
|
|
|
|
|
|
# ---------------------------------------------------------------- source serializer
|
|
|
|
def test_source_row_cost_per_application():
|
|
row = serialize_source_count("LinkedIn", 40, source_id=3, spend=1000.0)
|
|
assert row["cost_per_application"] == 25.0
|
|
assert row["spend"] == 1000.0 and row["count"] == 40
|
|
|
|
|
|
def test_source_row_no_fabricated_ratio():
|
|
# no spend -> no ratio; no applications -> no ratio (never a division blowup)
|
|
assert serialize_source_count("X", 40)["cost_per_application"] is None
|
|
assert serialize_source_count("X", 0, spend=500.0)["cost_per_application"] is None
|