feat(address): ROA address cache refresh — 8-field format + manual refresh endpoint

Phase 5 address format upgrade (pre-existing working tree changes):
- import_service: extend vadrese_parteneri query to 8 fields (strada/numar/bloc/scara/apart/etaj/localitate/judet); strip trailing city name from address string passed to Oracle
- sync_service: extend _addr_match to compare bloc/scara/apart in addition to strada/numar
- 05_pack_import_parteneri.pck: updated PL/SQL package

New: address cache refresh mechanism:
- sqlite_service: add get_order_address_ids(), update_order_address_cache() (targeted 3-column update, no ANAF fields touched), get_orders_with_address_ids()
- sync.py: POST /api/orders/{order_number}/refresh-address endpoint (404/422/503/200); batch Oracle address refresh in refresh_invoices (single IN roundtrip, per-order mismatch recomputed)
- UI: refresh button (⟳) in ADRESE modal header (base.html); refreshOrderAddress() with loading state + toast (dashboard.js v43); window._detailOrderNumber global (shared.js v32)
- tests: TestRefreshOrderAddress — 4 tests (404, 422, 503, 200 with 8-field assert)

Oracle prod fix applied directly: ADRESE_PARTENERI id_adresa=4116 STRADA VASILE→VASILE GOLDIS

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Claude Agent
2026-04-07 12:35:18 +00:00
parent a8ad54a604
commit ecde7fe440
10 changed files with 377 additions and 10 deletions

View File

@@ -716,3 +716,128 @@ class TestAddrMatch:
def test_malformed_json_returns_true(self):
from app.services.sync_service import _addr_match
assert _addr_match("{bad json", '{"strada":"x"}') is True
def test_addr_match_structured_fields(self):
"""addrMatch compares GoMag free-text vs ROA structured fields."""
from app.services.sync_service import _addr_match
import json
gomag = json.dumps({"address": "Str Vasile Goldis Nr 19 Bl 30 Sc D Ap 78", "city": "Alba Iulia", "region": "Alba"})
roa = json.dumps({"strada": "STRADA VASILE GOLDIS", "numar": "19", "bloc": "30", "scara": "D", "apart": "78", "localitate": "ALBA IULIA", "judet": "ALBA"})
assert _addr_match(gomag, roa) is True
def test_addr_match_mismatch_structured(self):
"""addrMatch detects real mismatches with structured ROA fields."""
from app.services.sync_service import _addr_match
import json
gomag = json.dumps({"address": "Str Dacia 10", "city": "Bucuresti", "region": "Bucuresti"})
roa = json.dumps({"strada": "STRADA VASILE GOLDIS", "numar": "19", "bloc": "", "scara": "", "apart": "", "localitate": "ALBA IULIA", "judet": "ALBA"})
assert _addr_match(gomag, roa) is False
class TestFormatAddressForOracle:
"""Tests for format_address_for_oracle city stripping."""
def test_city_strip_from_address_end(self):
"""City name at end of address is stripped."""
from app.services.import_service import format_address_for_oracle
result = format_address_for_oracle("Strada Vasile Goldis nr 19 Alba Iulia", "Alba Iulia", "Alba")
assert result == "JUD:Alba;Alba Iulia;Strada Vasile Goldis nr 19"
def test_city_strip_case_insensitive(self):
"""City strip works regardless of case."""
from app.services.import_service import format_address_for_oracle
result = format_address_for_oracle("Str Dacia alba iulia", "Alba Iulia", "Alba")
assert result == "JUD:Alba;Alba Iulia;Str Dacia"
def test_city_no_strip_when_not_at_end(self):
"""Don't strip city if it's in the middle of the address."""
from app.services.import_service import format_address_for_oracle
result = format_address_for_oracle("Alba Iulia Str Dacia 5", "Alba Iulia", "Alba")
assert "Alba Iulia Str Dacia 5" in result
def test_city_no_strip_when_empty_remains(self):
"""Don't strip if it would leave empty address."""
from app.services.import_service import format_address_for_oracle
result = format_address_for_oracle("Alba Iulia", "Alba Iulia", "Alba")
assert "Alba Iulia" in result # address preserved
# ===========================================================================
# Group 11: TestRefreshOrderAddress
# ===========================================================================
import sqlite3
@pytest.fixture(scope="module")
def client():
"""TestClient for refresh-address endpoint tests (Oracle mocked via FORCE_THIN_MODE)."""
from fastapi.testclient import TestClient
from app.main import app
with TestClient(app, raise_server_exceptions=False) as c:
yield c
@pytest.fixture
def db():
"""Synchronous SQLite connection to the test DB."""
conn = sqlite3.connect(_sqlite_path)
conn.row_factory = sqlite3.Row
yield conn
# Clean up test rows inserted during the test
conn.execute("DELETE FROM orders WHERE order_number LIKE 'test-%'")
conn.commit()
conn.close()
class TestRefreshOrderAddress:
"""Tests for POST /api/orders/{order_number}/refresh-address endpoint."""
def test_order_not_found_returns_404(self, client):
"""Non-existent order_number returns 404."""
res = client.post("/api/orders/nonexistent-99999/refresh-address")
assert res.status_code == 404
def test_null_address_ids_returns_422(self, client, db):
"""Orders without Oracle address IDs return 422."""
db.execute("INSERT OR IGNORE INTO orders (order_number, status) VALUES ('test-no-addr', 'IMPORTED')")
db.commit()
res = client.post("/api/orders/test-no-addr/refresh-address")
assert res.status_code == 422
def test_oracle_unavailable_returns_503(self, client, db, monkeypatch):
"""Oracle connection failure returns 503."""
db.execute("INSERT OR IGNORE INTO orders (order_number, status, id_adresa_livrare) VALUES ('test-oracle-fail', 'IMPORTED', 4116)")
db.commit()
import asyncio as _asyncio
async def _mock_to_thread(fn, *args, **kwargs):
raise Exception("Oracle down")
monkeypatch.setattr(_asyncio, "to_thread", _mock_to_thread)
res = client.post("/api/orders/test-oracle-fail/refresh-address")
assert res.status_code == 503
def test_refresh_returns_8_fields(self, client, db, monkeypatch):
"""Successful refresh returns 8-field address dict."""
db.execute("INSERT OR IGNORE INTO orders (order_number, status, id_adresa_livrare) VALUES ('test-refresh-ok', 'IMPORTED', 4116)")
db.commit()
mock_result = (
{"strada": "VASILE GOLDIS", "numar": "19", "bloc": "30",
"scara": "D", "apart": "78", "etaj": None,
"localitate": "ALBA IULIA", "judet": "ALBA"},
None,
)
import asyncio as _asyncio
async def _mock_to_thread(fn, *args, **kwargs):
return mock_result
monkeypatch.setattr(_asyncio, "to_thread", _mock_to_thread)
res = client.post("/api/orders/test-refresh-ok/refresh-address")
assert res.status_code == 200
data = res.json()
assert data["adresa_livrare_roa"]["strada"] == "VASILE GOLDIS"
assert data["adresa_livrare_roa"]["bloc"] == "30"