""" 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"]