robot
4e7ba7adea
CI Pipeline / backend-test (push) Successful in 12s
CI Pipeline / frontend-test (push) Successful in 27s
CI Pipeline / backend-lint (push) Successful in 9s
CI Pipeline / backend-test (pull_request) Successful in 12s
CI Pipeline / frontend-test (pull_request) Successful in 26s
CI Pipeline / backend-lint (pull_request) Successful in 8s
CI Pipeline / quality-gate (push) Successful in 2s
CI Pipeline / quality-gate (pull_request) Successful in 2s
230 lines
8.0 KiB
Python
230 lines
8.0 KiB
Python
"""Tests for log_parser — IP resolution, log line parsing, and geo lookups."""
|
|
|
|
import os
|
|
import tempfile
|
|
import pytest
|
|
from unittest.mock import patch, MagicMock
|
|
|
|
from app.log_parser import (
|
|
parse_log_line,
|
|
resolve_client_ip,
|
|
_COUNTRY_CENTROIDS,
|
|
lookup_geo,
|
|
)
|
|
|
|
|
|
class TestParseLogLine:
|
|
"""Tests for parse_log_line()."""
|
|
|
|
def test_valid_json_line(self):
|
|
line = '{"remote_addr": "192.168.1.1", "method": "GET"}'
|
|
result = parse_log_line(line)
|
|
assert result is not None
|
|
assert result["remote_addr"] == "192.168.1.1"
|
|
assert result["method"] == "GET"
|
|
|
|
def test_empty_line_returns_none(self):
|
|
assert parse_log_line("") is None
|
|
|
|
def test_whitespace_only_returns_none(self):
|
|
assert parse_log_line(" ") is None
|
|
|
|
def test_invalid_json_returns_none(self):
|
|
assert parse_log_line("not json at all") is None
|
|
|
|
def test_partial_json_returns_none(self):
|
|
assert parse_log_line('{"remote_addr": ') is None
|
|
|
|
def test_strips_whitespace(self):
|
|
line = ' {"key": "value"} \n'
|
|
result = parse_log_line(line)
|
|
assert result is not None
|
|
assert result["key"] == "value"
|
|
|
|
|
|
class TestResolveClientIp:
|
|
"""Tests for resolve_client_ip()."""
|
|
|
|
def test_remote_addr_without_xff(self):
|
|
entry = {"remote_addr": "203.0.113.5"}
|
|
assert resolve_client_ip(entry) == "203.0.113.5"
|
|
|
|
def test_xff_single_ip(self):
|
|
entry = {
|
|
"remote_addr": "10.0.0.1",
|
|
"x_forwarded_for": "203.0.113.50",
|
|
}
|
|
assert resolve_client_ip(entry) == "203.0.113.50"
|
|
|
|
def test_xff_multiple_ips_returns_leftmost(self):
|
|
entry = {
|
|
"remote_addr": "10.0.0.1",
|
|
"x_forwarded_for": "203.0.113.100, 10.1.2.3, 172.16.0.5",
|
|
}
|
|
assert resolve_client_ip(entry) == "203.0.113.100"
|
|
|
|
def test_xff_dash_falls_back_to_remote_addr(self):
|
|
entry = {"remote_addr": "192.168.0.1", "x_forwarded_for": "-"}
|
|
assert resolve_client_ip(entry) == "192.168.0.1"
|
|
|
|
def test_missing_fields_returns_empty(self):
|
|
entry = {}
|
|
assert resolve_client_ip(entry) == ""
|
|
|
|
def test_xff_with_whitespace(self):
|
|
entry = {
|
|
"remote_addr": "10.0.0.1",
|
|
"x_forwarded_for": " 203.0.113.75 ",
|
|
}
|
|
assert resolve_client_ip(entry) == "203.0.113.75"
|
|
|
|
|
|
class TestCountryCentroids:
|
|
"""Tests for _COUNTRY_CENTROIDS data integrity."""
|
|
|
|
def test_centroids_are_float_tuples(self):
|
|
for code, coords in _COUNTRY_CENTROIDS.items():
|
|
assert isinstance(coords, tuple)
|
|
assert len(coords) == 2
|
|
assert isinstance(coords[0], float)
|
|
assert isinstance(coords[1], float)
|
|
|
|
def test_known_countries_present(self):
|
|
expected = {"US", "GB", "DE", "FR", "JP", "AU", "BR", "CA", "IN"}
|
|
assert expected.issubset(_COUNTRY_CENTROIDS.keys())
|
|
|
|
def test_centroid_values_are_reasonable(self):
|
|
"""Latitudes should be in [-90, 90], longitudes in [-180, 180]."""
|
|
for code, (lat, lon) in _COUNTRY_CENTROIDS.items():
|
|
assert -90 <= lat <= 90, f"{code} latitude {lat} out of range"
|
|
assert -180 <= lon <= 180, f"{code} longitude {lon} out of range"
|
|
|
|
|
|
class TestParseLogFile:
|
|
"""Tests for parse_log_file() — reads from real temp files."""
|
|
|
|
def test_parses_multiple_lines(self, tmp_path):
|
|
log_content = (
|
|
'{"remote_addr": "1.2.3.4"}\n'
|
|
'{"remote_addr": "5.6.7.8"}\n'
|
|
'{"remote_addr": "1.2.3.4"}\n'
|
|
)
|
|
log_file = tmp_path / "test.log"
|
|
log_file.write_text(log_content)
|
|
|
|
from app.log_parser import parse_log_file
|
|
result = parse_log_file(str(log_file))
|
|
# Should have 2 unique IPs (1.2.3.4 appears twice)
|
|
assert "1.2.3.4" in result["unique_ips"]
|
|
assert "5.6.7.8" in result["unique_ips"]
|
|
assert len(result["unique_ips"]) == 2
|
|
|
|
def test_skips_invalid_lines(self, tmp_path):
|
|
log_content = (
|
|
'{"remote_addr": "1.2.3.4"}\n'
|
|
'not valid json\n'
|
|
'\n'
|
|
'{"remote_addr": "5.6.7.8"}\n'
|
|
)
|
|
log_file = tmp_path / "test.log"
|
|
log_file.write_text(log_content)
|
|
|
|
from app.log_parser import parse_log_file
|
|
result = parse_log_file(str(log_file))
|
|
assert len(result["unique_ips"]) == 2
|
|
|
|
def test_xff_takes_precedence(self, tmp_path):
|
|
log_content = (
|
|
'{"remote_addr": "10.0.0.1", "x_forwarded_for": "203.0.113.1"}\n'
|
|
)
|
|
log_file = tmp_path / "test.log"
|
|
log_file.write_text(log_content)
|
|
|
|
from app.log_parser import parse_log_file
|
|
result = parse_log_file(str(log_file))
|
|
assert "203.0.113.1" in result["unique_ips"]
|
|
assert "10.0.0.1" not in result["unique_ips"]
|
|
|
|
|
|
class TestLookupGeo:
|
|
"""Tests for lookup_geo() — verify IPs map to countries via our APIs."""
|
|
|
|
def _reset_geo_reader(self):
|
|
"""Clear the cached reader so mocks take effect."""
|
|
import app.log_parser as lp
|
|
|
|
lp._geo_reader = None
|
|
|
|
@patch("app.log_parser.settings")
|
|
@patch("app.log_parser.os.path.exists", return_value=True)
|
|
def test_ip_maps_to_country(self, mock_exists, mock_settings):
|
|
"""Verify a known IP resolves to a country code via the GeoLite2 reader."""
|
|
self._reset_geo_reader()
|
|
mock_settings.GEOLITE2_DB_PATH = "/tmp/fake-geo.mmdb"
|
|
|
|
# Create a fake reader that returns US data
|
|
fake_reader = MagicMock()
|
|
fake_reader.get.return_value = {
|
|
"country": {"iso_code": "US", "names": {"en": "United States"}},
|
|
"city": {"names": {"en": "New York"}},
|
|
"location": {"latitude": 40.7128, "longitude": -74.006},
|
|
}
|
|
|
|
with patch("maxminddb.open_database", return_value=fake_reader):
|
|
result = lookup_geo("8.8.8.8")
|
|
|
|
assert result["country_code"] == "US"
|
|
assert result["country"] == "United States"
|
|
assert result["city"] == "New York"
|
|
assert result["latitude"] == 40.7128
|
|
assert result["longitude"] == -74.006
|
|
|
|
@patch("app.log_parser.settings")
|
|
@patch("app.log_parser.os.path.exists", return_value=True)
|
|
def test_ip_without_coordinates_uses_centroid_fallback(self, mock_exists, mock_settings):
|
|
"""When the free GeoLite2 DB lacks coordinates, centroid fallback kicks in."""
|
|
self._reset_geo_reader()
|
|
mock_settings.GEOLITE2_DB_PATH = "/tmp/fake-geo.mmdb"
|
|
|
|
# Simulate free GeoLite2 response — has country but no coordinates
|
|
fake_reader = MagicMock()
|
|
fake_reader.get.return_value = {
|
|
"country": {"iso_code": "DE", "names": {"en": "Germany"}},
|
|
"city": {"names": {"en": ""}},
|
|
"location": {}, # No lat/lon
|
|
}
|
|
|
|
with patch("maxminddb.open_database", return_value=fake_reader):
|
|
result = lookup_geo("1.1.1.1")
|
|
|
|
assert result["country_code"] == "DE"
|
|
assert result["country"] == "Germany"
|
|
# Falls back to _COUNTRY_CENTROIDS["DE"]
|
|
assert result["latitude"] == _COUNTRY_CENTROIDS["DE"][0]
|
|
assert result["longitude"] == _COUNTRY_CENTROIDS["DE"][1]
|
|
|
|
@patch("app.log_parser.settings")
|
|
def test_unknown_ip_returns_empty_dict(self, mock_settings):
|
|
"""An IP not in the database returns an empty dict."""
|
|
self._reset_geo_reader()
|
|
mock_settings.GEOLITE2_DB_PATH = "/tmp/fake-geo.mmdb"
|
|
|
|
fake_reader = MagicMock()
|
|
fake_reader.get.return_value = None
|
|
|
|
with patch("maxminddb.open_database", return_value=fake_reader):
|
|
result = lookup_geo("255.255.255.255")
|
|
|
|
assert result == {}
|
|
|
|
@patch("app.log_parser.settings")
|
|
def test_reader_error_returns_empty_dict(self, mock_settings):
|
|
"""When the reader raises an exception, lookup_geo() returns safely."""
|
|
self._reset_geo_reader()
|
|
mock_settings.GEOLITE2_DB_PATH = "/tmp/fake-geo.mmdb"
|
|
|
|
with patch("maxminddb.open_database", side_effect=Exception("reader failed")):
|
|
result = lookup_geo("1.2.3.4")
|
|
|
|
assert result == {}
|