Cheng Zhou
69 lines
2.2 KiB
Python
69 lines
2.2 KiB
Python
from datetime import datetime, timedelta, timezone
|
|
|
|
import pytest
|
|
|
|
from app.models.source_location_snapshot import SourceLocationSnapshot
|
|
from app.services.source_location_aggregator import _identity_hash, get_source_location_timeline
|
|
|
|
|
|
def _snapshot(bucket_time: datetime, *, allowed: int, denied: int) -> SourceLocationSnapshot:
|
|
return SourceLocationSnapshot(
|
|
bucket_time=bucket_time,
|
|
ip_hash=_identity_hash("ip", "203.0.113.10"),
|
|
user_hash=_identity_hash("user", "user-1"),
|
|
country="United States",
|
|
country_code="US",
|
|
province="California",
|
|
region_code="",
|
|
city="San Jose",
|
|
isp="Example ISP",
|
|
location="United States / California / San Jose",
|
|
longitude=-121.8863,
|
|
latitude=37.3382,
|
|
accuracy_level="city",
|
|
allowed_count=allowed,
|
|
denied_count=denied,
|
|
security_event_count=denied,
|
|
high_risk_count=0,
|
|
auth_failure_count=denied,
|
|
first_seen_at=bucket_time,
|
|
last_seen_at=bucket_time + timedelta(minutes=30),
|
|
)
|
|
|
|
|
|
def test_source_location_identity_hash_is_stable_and_does_not_expose_ip():
|
|
first = _identity_hash("ip", "203.0.113.10")
|
|
second = _identity_hash("ip", "203.0.113.10")
|
|
|
|
assert first == second
|
|
assert len(first) == 64
|
|
assert "203.0.113.10" not in first
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_source_location_timeline_supports_hour_and_day_rollups(db_session):
|
|
start = datetime(2026, 7, 10, 8, tzinfo=timezone.utc)
|
|
db_session.add(_snapshot(start, allowed=4, denied=1))
|
|
db_session.add(_snapshot(start + timedelta(hours=1), allowed=3, denied=2))
|
|
await db_session.commit()
|
|
|
|
hourly = await get_source_location_timeline(
|
|
db_session,
|
|
start_at=start,
|
|
end_at=start + timedelta(hours=2),
|
|
granularity="hour",
|
|
)
|
|
daily = await get_source_location_timeline(
|
|
db_session,
|
|
start_at=start,
|
|
end_at=start + timedelta(hours=2),
|
|
granularity="day",
|
|
)
|
|
|
|
assert [item["total_count"] for item in hourly] == [5, 5]
|
|
assert len(daily) == 1
|
|
assert daily[0]["total_count"] == 10
|
|
assert daily[0]["denied_count"] == 3
|
|
assert daily[0]["unique_ip_count"] == 1
|
|
assert daily[0]["unique_user_count"] == 1
|