admin/Odoo-Modules
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(fusion_login_audit): async geo enrichment cron

Browse Source 
5-min cron processes up to 100 pending rows per pass: private IPs
short-circuit to state=private_ip; same-IP cache (30 days) avoids
duplicate ip-api.com calls; reverse DNS via socket with 1.5s timeout;
HTTP lookup respects ip-api''s X-Rl rate-limit header. Tests cover
private-IP shortcut, cache hit (no HTTP), and internal-state skip --
no network calls needed.

Per-row isolation uses cr.savepoint() instead of cr.commit() because
Odoo 19 TestCursor raises AssertionError on commit/rollback. Recorded
the gotcha as CLAUDE.md rule #14.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
This commit is contained in:
gsinghpal
2026-05-26 21:50:24 -04:00
parent 1b8038d8e8
commit 86b8e59c95
4 changed files with 225 additions and 1 deletions
Download Patch File Download Diff File Expand all files Collapse all files

61
fusion_login_audit/tests/test_login_audit.py
View File

@@ -440,3 +440,64 @@ class TestFusionLoginAuditModel(TransactionCase):
self.assertTrue(Audit.browse(ancient_id).exists(),
"retention_days=0 must keep everything")
def test_geo_private_ip_shortcut(self):
"""Private IPs short-circuit to state='private_ip' without HTTP."""
Audit = self.env['fusion.login.audit'].sudo()
rec = Audit.create({
'attempted_login': 'lan@example.com',
'result': 'success',
'ip_address': '192.168.1.40',
'geo_lookup_state': 'pending',
})
Audit._fc_geo_enrich_pending(limit=10)
rec.invalidate_recordset()
self.assertEqual(rec.geo_lookup_state, 'private_ip')
self.assertEqual(rec.country_code, '--')
def test_geo_cache_hit_avoids_http(self):
"""A second row with the same recent IP copies from cache."""
from unittest.mock import patch
Audit = self.env['fusion.login.audit'].sudo()
# Seed a "done" row from the same IP.
Audit.create({
'attempted_login': 'seed@example.com',
'result': 'success',
'ip_address': '203.0.113.99',
'geo_lookup_state': 'done',
'country_code': 'CA',
'country_name': 'Canada',
'city': 'Toronto',
'geo_state': 'Ontario',
})
target = Audit.create({
'attempted_login': 'hit@example.com',
'result': 'success',
'ip_address': '203.0.113.99',
'geo_lookup_state': 'pending',
})
with patch(
'odoo.addons.fusion_login_audit.models.fusion_login_audit.requests.get'
) as mock_get:
Audit._fc_geo_enrich_pending(limit=10)
mock_get.assert_not_called()
target.invalidate_recordset()
self.assertEqual(target.geo_lookup_state, 'done')
self.assertEqual(target.country_code, 'CA')
self.assertEqual(target.city, 'Toronto')
def test_geo_internal_skipped(self):
"""Rows with geo_lookup_state='internal' are not picked up."""
Audit = self.env['fusion.login.audit'].sudo()
rec = Audit.create({
'attempted_login': 'cron@example.com',
'result': 'success',
'ip_address': 'internal',
'geo_lookup_state': 'internal',
})
# Should be a no-op for 'internal' state (cron only picks 'pending').
Audit._fc_geo_enrich_pending(limit=10)
rec.invalidate_recordset()
self.assertEqual(rec.geo_lookup_state, 'internal')