🧪 Add tests for check_fresh in reputation.py (#9)

* 🧪 Add tests for check_fresh and improve coverage in reputation.py

* Extract `check_fresh` to a module-level function for testability.
* Add comprehensive tests for `check_fresh` covering missing, fresh, and stale cache files.
* Add a missing test for `reverse_ip` with IPv6 to hit 100% coverage on `reputation.py`.

Co-authored-by: Horizontal42 <75383519+Horizontal42@users.noreply.github.com>

* drop redundant pytest-cov dep addition

---------

Co-authored-by: google-labs-jules[bot] <161369871+google-labs-jules[bot]@users.noreply.github.com>
This commit is contained in:
Horizontal42 2026-08-19 17:09:13 +03:00 • committed by GitHub
parent 322dcf59b9
commit fb28b0ac37
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
2 changed files with 34 additions and 7 deletions

View file

@ -69,6 +69,13 @@ class NetsetIndex:
return found
def check_fresh(path: Path, refresh_hours: int) -> bool:
return (
path.exists()
and (time.time() - os.path.getmtime(path)) < refresh_hours * 3600
)
async def refresh_netsets(
client: httpx.AsyncClient,
providers: Providers,
@ -86,13 +93,7 @@ async def refresh_netsets(
name = url.rsplit("/", 1)[-1].removesuffix(".netset")
path = directory / f"{name}.netset"
def check_fresh() -> bool:
return (
path.exists()
and (time.time() - os.path.getmtime(path)) < providers.firehol_refresh_hours * 3600
)
fresh = await asyncio.to_thread(check_fresh)
fresh = await asyncio.to_thread(check_fresh, path, providers.firehol_refresh_hours)
if not fresh:
try:
response = await client.get(url)

View file

@ -16,6 +16,7 @@ from netsleuth.reputation import (
ReputationContext,
build_reputation,
captcha_risk,
check_fresh,
decode_dnsbl,
fetch_abuseipdb,
fetch_internetdb,
@ -23,10 +24,30 @@ from netsleuth.reputation import (
parse_netset,
query_dnsbl,
refresh_netsets,
reverse_ip,
summarize_dnsbl,
)
def test_check_fresh_missing_file(tmp_path):
path = tmp_path / "nonexistent.netset"
assert check_fresh(path, refresh_hours=24) is False
def test_check_fresh_recent_file(tmp_path):
path = tmp_path / "recent.netset"
path.touch()
assert check_fresh(path, refresh_hours=24) is True
def test_check_fresh_stale_file(tmp_path):
path = tmp_path / "stale.netset"
path.touch()
stale_time = time.time() - (24 * 3600 + 100)
os.utime(path, (stale_time, stale_time))
assert check_fresh(path, refresh_hours=24) is False
def test_parse_netset_strips_comments_and_blank_lines(fixtures_dir):
cidrs = parse_netset((fixtures_dir / "api" / "firehol_sample.netset").read_text(encoding="utf-8"))
assert cidrs == [
@ -39,6 +60,11 @@ def test_parse_netset_strips_comments_and_blank_lines(fixtures_dir):
]
def test_reverse_ip_raises_value_error_for_ipv6():
with pytest.raises(ValueError, match="classic DNSBL zones accept IPv4 only"):
reverse_ip("2001:db8::1")
def test_parse_netset_of_an_empty_file_is_empty():
assert parse_netset("# only a comment\n\n") == []