Spaces:
Configuration error
Configuration error
File size: 7,699 Bytes
7af851c | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 | """
Property-based tests for Analytics Reports base engine.
Uses Hypothesis to validate invariants across arbitrary inputs.
Properties tested:
2. Projection Correctness
3. Pagination Invariant
4. Period Resolution Correctness
5. Aging Bucket Invariant
6. Export Row Count Consistency
7. Export Format Correctness (CSV and XLSX)
"""
from hypothesis import given, settings, HealthCheck
from hypothesis import strategies as st
from app.reports.base.engine import apply_projection, resolve_period, stream_export
settings.register_profile("ci", max_examples=100, suppress_health_check=[HealthCheck.too_slow])
settings.load_profile("ci")
# ---------------------------------------------------------------------------
# Property 2: Projection Correctness
# ---------------------------------------------------------------------------
# Feature: analytics-reports, Property 2: Projection Correctness
@given(
rows=st.lists(st.dictionaries(st.text(min_size=1, max_size=10), st.integers()), min_size=1, max_size=20),
projection=st.lists(st.text(min_size=1, max_size=10), min_size=1, max_size=5),
)
@settings(max_examples=100)
def test_projection_correctness(rows, projection):
result = apply_projection(rows, projection)
for row in result:
assert "_id" not in row
for key in row:
assert key in projection
# Feature: analytics-reports, Property 2: Projection Correctness (None case)
@given(rows=st.lists(st.dictionaries(st.text(min_size=1, max_size=10), st.integers()), min_size=1, max_size=20))
@settings(max_examples=100)
def test_projection_absent_returns_all(rows):
# When no projection, all keys returned (minus _id)
result = apply_projection(rows, None)
for orig, proj in zip(rows, result):
for k, v in orig.items():
if k != "_id":
assert k in proj
assert proj[k] == v
# ---------------------------------------------------------------------------
# Property 3: Pagination Invariant
# ---------------------------------------------------------------------------
# Feature: analytics-reports, Property 3: Pagination Invariant
@given(
limit=st.integers(min_value=1, max_value=1000),
rows=st.lists(st.dictionaries(st.text(min_size=1, max_size=5), st.integers()), max_size=2000),
)
@settings(max_examples=100)
def test_pagination_invariant(limit, rows):
skip = 0
paginated = rows[skip:skip + limit]
assert len(paginated) <= limit
envelope = {"success": True, "total": len(rows), "skip": skip, "limit": limit, "data": paginated}
for key in ("success", "total", "skip", "limit", "data"):
assert key in envelope
assert envelope["total"] >= 0
assert envelope["skip"] == skip
# ---------------------------------------------------------------------------
# Property 4: Period Resolution Correctness
# ---------------------------------------------------------------------------
# Feature: analytics-reports, Property 4: Period Resolution Correctness
@given(period=st.sampled_from(["today", "last_7_days", "mtd", "ytd"]))
@settings(max_examples=100)
def test_period_resolution_correctness(period):
from datetime import date, timedelta
start, end = resolve_period(period, None, None)
today = date.today()
assert start <= end
assert end <= today
if period == "today":
assert start == today and end == today
elif period == "last_7_days":
assert start == today - timedelta(days=7)
elif period == "mtd":
assert start == today.replace(day=1)
elif period == "ytd":
assert start == today.replace(month=1, day=1)
# ---------------------------------------------------------------------------
# Property 5: Aging Bucket Invariant
# ---------------------------------------------------------------------------
# Feature: analytics-reports, Property 5: Aging Bucket Invariant
@given(
b0=st.floats(min_value=0, max_value=1e6, allow_nan=False, allow_infinity=False),
b1=st.floats(min_value=0, max_value=1e6, allow_nan=False, allow_infinity=False),
b2=st.floats(min_value=0, max_value=1e6, allow_nan=False, allow_infinity=False),
b3=st.floats(min_value=0, max_value=1e6, allow_nan=False, allow_infinity=False),
)
@settings(max_examples=100)
def test_aging_bucket_invariant(b0, b1, b2, b3):
total = b0 + b1 + b2 + b3
assert abs((b0 + b1 + b2 + b3) - total) < 1e-9
# ---------------------------------------------------------------------------
# Async helper — consume a StreamingResponse body synchronously
# ---------------------------------------------------------------------------
def _collect_response_body(response) -> bytes:
"""Drain a StreamingResponse body_iterator into bytes, handling both
sync iterables (``iter([...])``) and async generators."""
import asyncio, inspect
iterator = response.body_iterator
if inspect.isasyncgen(iterator) or hasattr(iterator, "__aiter__"):
async def _drain():
chunks = []
async for chunk in iterator:
chunks.append(chunk if isinstance(chunk, bytes) else chunk.encode())
return b"".join(chunks)
return asyncio.run(_drain())
# sync iterable
chunks = []
for chunk in iterator:
chunks.append(chunk if isinstance(chunk, bytes) else chunk.encode())
return b"".join(chunks)
# ---------------------------------------------------------------------------
# Property 6: Export Row Count Consistency
# ---------------------------------------------------------------------------
# Feature: analytics-reports, Property 6: Export Row Count Consistency
@given(rows=st.lists(
st.fixed_dictionaries({"name": st.text(min_size=1, max_size=10), "value": st.integers()}),
min_size=0, max_size=100,
))
@settings(max_examples=100)
def test_export_row_count_consistency(rows):
import asyncio, csv, io
response = asyncio.run(stream_export(rows, "test-slug", "csv"))
body = _collect_response_body(response)
reader = list(csv.DictReader(io.StringIO(body.decode("utf-8"))))
assert len(reader) == len(rows)
# ---------------------------------------------------------------------------
# Property 7: Export Format Correctness (CSV)
# ---------------------------------------------------------------------------
# Feature: analytics-reports, Property 7: Export Format Correctness (CSV)
@given(rows=st.lists(
st.fixed_dictionaries({"col_a": st.integers(), "col_b": st.text(min_size=1, max_size=5)}),
min_size=1, max_size=50,
))
@settings(max_examples=100)
def test_export_csv_has_header(rows):
import asyncio, io
response = asyncio.run(stream_export(rows, "test-slug", "csv"))
body = _collect_response_body(response)
lines = body.decode("utf-8").splitlines()
header = lines[0].split(",")
assert "col_a" in header
assert "col_b" in header
# ---------------------------------------------------------------------------
# Property 7: Export Format Correctness (XLSX)
# ---------------------------------------------------------------------------
# Feature: analytics-reports, Property 7: Export Format Correctness (XLSX)
@given(rows=st.lists(
st.fixed_dictionaries({
"col_a": st.integers(),
"col_b": st.text(min_size=1, max_size=5, alphabet=st.characters(whitelist_categories=("L", "N", "P", "Zs"))),
}),
min_size=1, max_size=50,
))
@settings(max_examples=100)
def test_export_xlsx_valid_workbook(rows):
import asyncio, io, openpyxl
response = asyncio.run(stream_export(rows, "my-report", "xlsx"))
body = _collect_response_body(response)
wb = openpyxl.load_workbook(io.BytesIO(body))
assert "my-report" in wb.sheetnames
|