72 lines
2.2 KiB
Python
72 lines
2.2 KiB
Python
import tempfile
|
|
from datetime import UTC, datetime, timedelta
|
|
from pathlib import Path
|
|
|
|
from finn_eiendom.cache import (
|
|
get_eiendom_unit,
|
|
get_finn_ad,
|
|
get_search_page,
|
|
get_similar_units,
|
|
init_db,
|
|
save_eiendom_unit,
|
|
save_finn_ad,
|
|
save_search_page,
|
|
save_similar_units,
|
|
)
|
|
from finn_eiendom.models import EiendomUnit, FinnAd, SimilarUnit
|
|
|
|
|
|
def test_cache_roundtrip():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
db_path = Path(tmpdir) / "cache.sqlite"
|
|
conn = init_db(str(db_path))
|
|
|
|
ad = FinnAd(finnkode="1234", url="https://example.com", title="Test")
|
|
save_finn_ad(conn, ad)
|
|
loaded_ad = get_finn_ad(conn, "1234")
|
|
assert loaded_ad is not None
|
|
assert loaded_ad.finnkode == "1234"
|
|
assert loaded_ad.url == "https://example.com"
|
|
|
|
unit = EiendomUnit(unit_code="abc", address="Oslo")
|
|
save_eiendom_unit(conn, unit)
|
|
loaded_unit = get_eiendom_unit(conn, "abc")
|
|
assert loaded_unit is not None
|
|
assert loaded_unit.address == "Oslo"
|
|
|
|
comps = [
|
|
SimilarUnit(unit_code="x1"),
|
|
SimilarUnit(unit_code="x2"),
|
|
]
|
|
save_similar_units(conn, "abc", "RECENTLY_SOLD", comps)
|
|
loaded_comps = get_similar_units(conn, "abc", "RECENTLY_SOLD")
|
|
assert len(loaded_comps) == 2
|
|
assert loaded_comps[0].unit_code == "x1"
|
|
|
|
|
|
def test_search_page_cache_roundtrip():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
conn = init_db(str(Path(tmpdir) / "cache.sqlite"))
|
|
|
|
html = "<html><body>search page</body></html>"
|
|
url = "https://www.finn.no/realestate/homes/search.html"
|
|
|
|
save_search_page(conn, url, html, ttl_minutes=5)
|
|
loaded_html = get_search_page(conn, url)
|
|
assert loaded_html == html
|
|
|
|
|
|
def test_finn_ad_cache_ttl_expiration():
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
conn = init_db(str(Path(tmpdir) / "cache.sqlite"))
|
|
|
|
ad = FinnAd(
|
|
finnkode="1234",
|
|
url="https://example.com",
|
|
title="Test",
|
|
detail_fetched_at=datetime.now(UTC) - timedelta(hours=2),
|
|
)
|
|
save_finn_ad(conn, ad)
|
|
expired_ad = get_finn_ad(conn, "1234", ttl_hours=1)
|
|
assert expired_ad is None
|