63 lines
1.8 KiB
Python
63 lines
1.8 KiB
Python
"""Tests for the geo cache repository."""
|
|
|
|
from pathlib import Path
|
|
|
|
import aiosqlite
|
|
import pytest
|
|
|
|
from app.repositories import geo_cache_repo
|
|
|
|
|
|
async def _create_geo_cache_table(db: aiosqlite.Connection) -> None:
|
|
await db.execute(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS geo_cache (
|
|
ip TEXT PRIMARY KEY,
|
|
country_code TEXT,
|
|
country_name TEXT,
|
|
asn TEXT,
|
|
org TEXT,
|
|
cached_at TEXT NOT NULL DEFAULT (strftime('%Y-%m-%dT%H:%M:%fZ', 'now'))
|
|
)
|
|
"""
|
|
)
|
|
await db.commit()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_unresolved_ips_returns_empty_when_none_exist(tmp_path: Path) -> None:
|
|
db_path = str(tmp_path / "geo_cache.db")
|
|
async with aiosqlite.connect(db_path) as db:
|
|
await _create_geo_cache_table(db)
|
|
await db.execute(
|
|
"INSERT INTO geo_cache (ip, country_code, country_name, asn, org) VALUES (?, ?, ?, ?, ?)",
|
|
("1.1.1.1", "DE", "Germany", "AS123", "Test"),
|
|
)
|
|
await db.commit()
|
|
|
|
async with aiosqlite.connect(db_path) as db:
|
|
ips = await geo_cache_repo.get_unresolved_ips(db)
|
|
|
|
assert ips == []
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_get_unresolved_ips_returns_pending_ips(tmp_path: Path) -> None:
|
|
db_path = str(tmp_path / "geo_cache.db")
|
|
async with aiosqlite.connect(db_path) as db:
|
|
await _create_geo_cache_table(db)
|
|
await db.executemany(
|
|
"INSERT INTO geo_cache (ip, country_code) VALUES (?, ?)",
|
|
[
|
|
("2.2.2.2", None),
|
|
("3.3.3.3", None),
|
|
("4.4.4.4", "US"),
|
|
],
|
|
)
|
|
await db.commit()
|
|
|
|
async with aiosqlite.connect(db_path) as db:
|
|
ips = await geo_cache_repo.get_unresolved_ips(db)
|
|
|
|
assert sorted(ips) == ["2.2.2.2", "3.3.3.3"]
|