chaim
4bc054f315
The PDF host (free-justice.openapi.gov.il) requires the same x-client-id header as the search API, so a normal browser click on the URL fails. This tool carries the auth header automatically and saves the PDF to a configurable local directory. Safety properties: - SSRF guard: only download from free-justice.openapi.gov.il and pub-justice.openapi.gov.il. - Path-traversal guard: filename is reduced to its basename; arbitrary paths are stripped. - Content-type guard: rejects 200-OK responses whose body is not a real PDF (gateway sometimes returns JSON-error 200s). - Atomic write via .tmp + rename so partial downloads never replace the cached copy. - Caches by destination path; re-downloads are no-ops unless overwrite=True. DECISIVE_APPRAISER_DOWNLOAD_DIR env var configures the output dir (default: ./downloads/decisive_appraisals). 6 new unit tests cover the happy path, caching, and all four guards. End-to-end live test confirmed a 1.4MB real PDF lands on disk with valid `%PDF-1.7` header. Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
310 lines
13 KiB
Python
310 lines
13 KiB
Python
"""
|
|
Unit tests for the DecisiveAppraiser (gov.il / Ministry of Justice) client.
|
|
|
|
These tests mock the curl_cffi session entirely so they're fast and offline.
|
|
A separate integration test (marked `@pytest.mark.integration`) exercises
|
|
the real upstream — run with `pytest -m integration` only when needed.
|
|
"""
|
|
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
|
|
from nadlan_mcp.govil import DecisiveAppraiserClient
|
|
from nadlan_mcp.govil.client import _format_date
|
|
from nadlan_mcp.govil.models import (
|
|
AppraisalDecision,
|
|
AppraisalDocument,
|
|
DecisiveAppraiserSearchResponse,
|
|
)
|
|
|
|
# A trimmed-but-realistic upstream payload, matching the live shape captured
|
|
# during development.
|
|
SAMPLE_RESPONSE = {
|
|
"Results": [
|
|
{
|
|
"Data": {
|
|
"AppraisalHeader": "הכרעת שמאי מכריע מיום 01-04-2026 בעניין היטל השבחה",
|
|
"AppraisalType": "היטל השבחה",
|
|
"AppraisalVersion": "שומה מקורית",
|
|
"DecisiveAppraiser": "דדון דוד",
|
|
"AppraiserType": "שמאי מכריע",
|
|
"Block": "6212",
|
|
"Plot": "894",
|
|
"Committee": "תל אביב-יפו",
|
|
"DecisionDate": "2026-04-01T00:00:00+03:00",
|
|
"PublicityDate": "2026-04-11T00:00:00+03:00",
|
|
"Document": [
|
|
{
|
|
"FileName": "https://free-justice.openapi.gov.il/.../abc=",
|
|
"DisplayName": "הכרעת שמאי מכריע",
|
|
"Extension": "pdf",
|
|
}
|
|
],
|
|
"DocSummary": {},
|
|
}
|
|
}
|
|
],
|
|
"Status": None,
|
|
"message": None,
|
|
"TotalResults": 333,
|
|
}
|
|
|
|
|
|
class TestModels:
|
|
def test_decision_parses_pascalcase_payload(self):
|
|
decision = AppraisalDecision.model_validate(SAMPLE_RESPONSE["Results"][0]["Data"])
|
|
assert decision.block == "6212"
|
|
assert decision.plot == "894"
|
|
assert decision.decisive_appraiser == "דדון דוד"
|
|
assert decision.committee == "תל אביב-יפו"
|
|
assert decision.decision_date is not None
|
|
assert len(decision.documents) == 1
|
|
assert isinstance(decision.documents[0], AppraisalDocument)
|
|
assert decision.documents[0].file_url.endswith("abc=")
|
|
assert decision.documents[0].extension == "pdf"
|
|
|
|
def test_decision_handles_missing_fields(self):
|
|
decision = AppraisalDecision.model_validate({"Block": "1234"})
|
|
assert decision.block == "1234"
|
|
assert decision.plot is None
|
|
assert decision.documents == []
|
|
assert decision.doc_summary == {}
|
|
|
|
|
|
class TestFormatDate:
|
|
def test_none_passthrough(self):
|
|
assert _format_date(None) is None
|
|
|
|
def test_empty_string(self):
|
|
assert _format_date("") is None
|
|
assert _format_date(" ") is None
|
|
|
|
def test_dd_mm_yyyy_string_passthrough(self):
|
|
assert _format_date("01-04-2026") == "01-04-2026"
|
|
|
|
def test_datetime_formatting(self):
|
|
from datetime import datetime as dt
|
|
|
|
assert _format_date(dt(2026, 4, 1)) == "01-04-2026"
|
|
|
|
def test_date_formatting(self):
|
|
from datetime import date as date_type
|
|
|
|
assert _format_date(date_type(2026, 4, 1)) == "01-04-2026"
|
|
|
|
|
|
class TestDecisiveAppraiserClient:
|
|
@patch("nadlan_mcp.govil.client.cf_requests.Session")
|
|
def test_search_by_block_builds_correct_body(self, mock_session_class):
|
|
"""Block filter is sent as a flat top-level field, not under `filters`."""
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = SAMPLE_RESPONSE
|
|
mock_session = Mock()
|
|
mock_session.post.return_value = mock_response
|
|
mock_session_class.return_value = mock_session
|
|
|
|
client = DecisiveAppraiserClient()
|
|
result = client.search_decisions(block="6212")
|
|
|
|
# Body shape we discovered: flat keys, not nested filters.
|
|
call_kwargs = mock_session.post.call_args.kwargs
|
|
assert call_kwargs["json"] == {"skip": 0, "Block": "6212"}
|
|
assert isinstance(result, DecisiveAppraiserSearchResponse)
|
|
assert result.total_results == 333
|
|
assert len(result.results) == 1
|
|
assert result.results[0].block == "6212"
|
|
|
|
@patch("nadlan_mcp.govil.client.cf_requests.Session")
|
|
def test_search_combines_all_filters(self, mock_session_class):
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = SAMPLE_RESPONSE
|
|
mock_session = Mock()
|
|
mock_session.post.return_value = mock_response
|
|
mock_session_class.return_value = mock_session
|
|
|
|
client = DecisiveAppraiserClient()
|
|
client.search_decisions(
|
|
block="6212",
|
|
plot="894",
|
|
decisive_appraiser="דדון דוד",
|
|
committee="תל אביב-יפו",
|
|
decision_date_from="01-01-2025",
|
|
decision_date_to="31-12-2026",
|
|
search_text="היטל השבחה",
|
|
skip=20,
|
|
)
|
|
body = mock_session.post.call_args.kwargs["json"]
|
|
assert body["skip"] == 20
|
|
assert body["Block"] == "6212"
|
|
assert body["Plot"] == "894"
|
|
assert body["DecisiveAppraiser"] == "דדון דוד"
|
|
assert body["Committee"] == "תל אביב-יפו"
|
|
assert body["DecisionDate_from"] == "01-01-2025"
|
|
assert body["DecisionDate_to"] == "31-12-2026"
|
|
assert body["SearchText"] == "היטל השבחה"
|
|
|
|
@patch("nadlan_mcp.govil.client.cf_requests.Session")
|
|
def test_500_response_raises_immediately(self, mock_session_class):
|
|
"""500 from this gateway is fatal (config issue), not retryable."""
|
|
mock_response = Mock()
|
|
mock_response.status_code = 500
|
|
mock_response.text = '{"code":500,"message":"Internal Server Error"}'
|
|
mock_session = Mock()
|
|
mock_session.post.return_value = mock_response
|
|
mock_session_class.return_value = mock_session
|
|
|
|
client = DecisiveAppraiserClient()
|
|
with pytest.raises(ValueError, match="500"):
|
|
client.search_decisions(block="6212")
|
|
# Should not have retried — single call.
|
|
assert mock_session.post.call_count == 1
|
|
|
|
@patch("nadlan_mcp.govil.client.cf_requests.Session")
|
|
def test_empty_filters_only_sends_skip(self, mock_session_class):
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.json.return_value = {"Results": [], "TotalResults": 0}
|
|
mock_session = Mock()
|
|
mock_session.post.return_value = mock_response
|
|
mock_session_class.return_value = mock_session
|
|
|
|
client = DecisiveAppraiserClient()
|
|
client.search_decisions()
|
|
assert mock_session.post.call_args.kwargs["json"] == {"skip": 0}
|
|
|
|
@patch("nadlan_mcp.govil.client.cf_requests.Session")
|
|
def test_paged_collects_across_pages(self, mock_session_class):
|
|
"""Pagination should keep calling skip+=10 until max_results is reached."""
|
|
|
|
def make_page(skip: int):
|
|
# Three pages of 10, then empty.
|
|
results = (
|
|
[{"Data": {"Block": "6212", "Plot": str(skip + i)}} for i in range(10)]
|
|
if skip < 30
|
|
else []
|
|
)
|
|
return {"Results": results, "TotalResults": 30}
|
|
|
|
# Track call sequence
|
|
call_log = []
|
|
|
|
def post_side_effect(url, **kwargs):
|
|
call_log.append(kwargs["json"]["skip"])
|
|
mock_resp = Mock()
|
|
mock_resp.status_code = 200
|
|
mock_resp.json.return_value = make_page(kwargs["json"]["skip"])
|
|
return mock_resp
|
|
|
|
mock_session = Mock()
|
|
mock_session.post.side_effect = post_side_effect
|
|
mock_session_class.return_value = mock_session
|
|
|
|
client = DecisiveAppraiserClient()
|
|
result = client.search_decisions_paged(max_results=25, block="6212")
|
|
|
|
assert call_log == [0, 10, 20] # 3 pages
|
|
assert len(result.results) == 25 # truncated to max_results
|
|
assert result.total_results == 30
|
|
|
|
def test_paged_rejects_explicit_skip(self):
|
|
client = DecisiveAppraiserClient()
|
|
with pytest.raises(ValueError, match="manages skip"):
|
|
client.search_decisions_paged(max_results=10, skip=20, block="6212")
|
|
|
|
def test_init_sets_required_headers(self):
|
|
client = DecisiveAppraiserClient()
|
|
headers = client._session.headers
|
|
# The static client_id is the gateway's auth gate — without it every
|
|
# request returns 500.
|
|
assert "x-client-id" in headers
|
|
assert headers["x-client-id"] == "149a5bad-edde-49a6-9fb9-188bd17d4788"
|
|
assert headers["Origin"] == "https://www.gov.il"
|
|
|
|
|
|
class TestDownloadPdf:
|
|
@patch("nadlan_mcp.govil.client.cf_requests.Session")
|
|
def test_download_writes_pdf_to_disk(self, mock_session_class, tmp_path):
|
|
pdf_bytes = b"%PDF-1.7\n" + b"x" * 1000
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.content = pdf_bytes
|
|
mock_response.headers = {"content-type": "application/pdf"}
|
|
mock_session = Mock()
|
|
mock_session.get.return_value = mock_response
|
|
mock_session_class.return_value = mock_session
|
|
|
|
client = DecisiveAppraiserClient()
|
|
url = "https://free-justice.openapi.gov.il/free/moj/portal/rest/searchpredefinedapi/v1/SearchPredefinedApi/Documents/DecisiveAppraiser/abc123="
|
|
result = client.download_pdf(url, output_dir=str(tmp_path), filename="decision.pdf")
|
|
|
|
assert result["from_cache"] is False
|
|
assert result["size_bytes"] == len(pdf_bytes)
|
|
assert result["content_type"] == "application/pdf"
|
|
saved = tmp_path / "decision.pdf"
|
|
assert saved.exists()
|
|
assert saved.read_bytes() == pdf_bytes
|
|
|
|
@patch("nadlan_mcp.govil.client.cf_requests.Session")
|
|
def test_download_caches_existing_file(self, mock_session_class, tmp_path):
|
|
# Pre-create the destination
|
|
existing = tmp_path / "decision.pdf"
|
|
existing.write_bytes(b"%PDF-1.7\nold")
|
|
|
|
mock_session = Mock()
|
|
mock_session_class.return_value = mock_session
|
|
|
|
client = DecisiveAppraiserClient()
|
|
url = "https://free-justice.openapi.gov.il/free/moj/portal/rest/searchpredefinedapi/v1/SearchPredefinedApi/Documents/DecisiveAppraiser/abc123="
|
|
result = client.download_pdf(url, output_dir=str(tmp_path), filename="decision.pdf")
|
|
|
|
assert result["from_cache"] is True
|
|
# Network call should NOT have happened.
|
|
mock_session.get.assert_not_called()
|
|
|
|
def test_download_rejects_external_host(self, tmp_path):
|
|
client = DecisiveAppraiserClient()
|
|
with pytest.raises(ValueError, match="untrusted host"):
|
|
client.download_pdf("https://evil.com/foo.pdf", output_dir=str(tmp_path))
|
|
|
|
def test_download_rejects_non_https(self, tmp_path):
|
|
client = DecisiveAppraiserClient()
|
|
with pytest.raises(ValueError, match="non-http"):
|
|
client.download_pdf("file:///etc/passwd", output_dir=str(tmp_path))
|
|
|
|
@patch("nadlan_mcp.govil.client.cf_requests.Session")
|
|
def test_download_rejects_non_pdf_response(self, mock_session_class, tmp_path):
|
|
# Gateway sometimes returns 200 with a JSON error body — we must catch it.
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.content = b'{"error":"forbidden"}'
|
|
mock_response.headers = {"content-type": "application/json"}
|
|
mock_session = Mock()
|
|
mock_session.get.return_value = mock_response
|
|
mock_session_class.return_value = mock_session
|
|
|
|
client = DecisiveAppraiserClient()
|
|
url = "https://free-justice.openapi.gov.il/free/moj/portal/rest/searchpredefinedapi/v1/SearchPredefinedApi/Documents/DecisiveAppraiser/abc="
|
|
with pytest.raises(ValueError, match="not a PDF"):
|
|
client.download_pdf(url, output_dir=str(tmp_path))
|
|
|
|
@patch("nadlan_mcp.govil.client.cf_requests.Session")
|
|
def test_download_strips_path_components_from_filename(self, mock_session_class, tmp_path):
|
|
"""Path-traversal guard: filename must not escape the output dir."""
|
|
mock_response = Mock()
|
|
mock_response.status_code = 200
|
|
mock_response.content = b"%PDF-1.7\n"
|
|
mock_response.headers = {"content-type": "application/pdf"}
|
|
mock_session = Mock()
|
|
mock_session.get.return_value = mock_response
|
|
mock_session_class.return_value = mock_session
|
|
|
|
client = DecisiveAppraiserClient()
|
|
url = "https://free-justice.openapi.gov.il/free/moj/portal/rest/searchpredefinedapi/v1/SearchPredefinedApi/Documents/DecisiveAppraiser/abc="
|
|
result = client.download_pdf(url, output_dir=str(tmp_path), filename="../../../escape.pdf")
|
|
# File should land inside tmp_path, not outside.
|
|
assert (tmp_path / "escape.pdf").exists()
|
|
assert str(tmp_path) in result["path"]
|