diff --git a/nadlan_mcp/config.py b/nadlan_mcp/config.py index f28e8ea..3d45391 100644 --- a/nadlan_mcp/config.py +++ b/nadlan_mcp/config.py @@ -69,8 +69,9 @@ class GovmapConfig: # Percentage-based backup filtering (catches extreme outliers in heterogeneous data) analysis_use_percentage_backup: bool = field( - default_factory=lambda: os.getenv("ANALYSIS_USE_PERCENTAGE_BACKUP", "true").lower() - == "true" + default_factory=lambda: ( + os.getenv("ANALYSIS_USE_PERCENTAGE_BACKUP", "true").lower() == "true" + ) ) analysis_percentage_threshold: float = field( default_factory=lambda: float(os.getenv("ANALYSIS_PERCENTAGE_THRESHOLD", "0.4")) @@ -91,8 +92,9 @@ class GovmapConfig: # Statistical Robustness (for investment analysis) analysis_use_robust_volatility: bool = field( - default_factory=lambda: os.getenv("ANALYSIS_USE_ROBUST_VOLATILITY", "true").lower() - == "true" + default_factory=lambda: ( + os.getenv("ANALYSIS_USE_ROBUST_VOLATILITY", "true").lower() == "true" + ) ) analysis_use_robust_trends: bool = field( default_factory=lambda: os.getenv("ANALYSIS_USE_ROBUST_TRENDS", "true").lower() == "true" @@ -100,8 +102,9 @@ class GovmapConfig: # Reporting analysis_include_unfiltered_stats: bool = field( - default_factory=lambda: os.getenv("ANALYSIS_INCLUDE_UNFILTERED_STATS", "true").lower() - == "true" + default_factory=lambda: ( + os.getenv("ANALYSIS_INCLUDE_UNFILTERED_STATS", "true").lower() == "true" + ) ) # Distance Filtering for Deal Relevance @@ -119,48 +122,59 @@ class GovmapConfig: # MCP Tool Availability (enable/disable specific tools) tool_autocomplete_address_enabled: bool = field( - default_factory=lambda: os.getenv("TOOL_AUTOCOMPLETE_ADDRESS_ENABLED", "true").lower() - == "true" + default_factory=lambda: ( + os.getenv("TOOL_AUTOCOMPLETE_ADDRESS_ENABLED", "true").lower() == "true" + ) ) tool_get_deals_by_radius_enabled: bool = field( - default_factory=lambda: os.getenv("TOOL_GET_DEALS_BY_RADIUS_ENABLED", "false").lower() - == "true" + default_factory=lambda: ( + os.getenv("TOOL_GET_DEALS_BY_RADIUS_ENABLED", "false").lower() == "true" + ) ) tool_get_street_deals_enabled: bool = field( - default_factory=lambda: os.getenv("TOOL_GET_STREET_DEALS_ENABLED", "false").lower() - == "true" + default_factory=lambda: ( + os.getenv("TOOL_GET_STREET_DEALS_ENABLED", "false").lower() == "true" + ) ) tool_get_neighborhood_deals_enabled: bool = field( - default_factory=lambda: os.getenv("TOOL_GET_NEIGHBORHOOD_DEALS_ENABLED", "true").lower() - == "true" + default_factory=lambda: ( + os.getenv("TOOL_GET_NEIGHBORHOOD_DEALS_ENABLED", "true").lower() == "true" + ) ) tool_find_recent_deals_for_address_enabled: bool = field( - default_factory=lambda: os.getenv( - "TOOL_FIND_RECENT_DEALS_FOR_ADDRESS_ENABLED", "true" - ).lower() - == "true" + default_factory=lambda: ( + os.getenv("TOOL_FIND_RECENT_DEALS_FOR_ADDRESS_ENABLED", "true").lower() == "true" + ) ) tool_analyze_market_trends_enabled: bool = field( - default_factory=lambda: os.getenv("TOOL_ANALYZE_MARKET_TRENDS_ENABLED", "true").lower() - == "true" + default_factory=lambda: ( + os.getenv("TOOL_ANALYZE_MARKET_TRENDS_ENABLED", "true").lower() == "true" + ) ) tool_compare_addresses_enabled: bool = field( - default_factory=lambda: os.getenv("TOOL_COMPARE_ADDRESSES_ENABLED", "true").lower() - == "true" + default_factory=lambda: ( + os.getenv("TOOL_COMPARE_ADDRESSES_ENABLED", "true").lower() == "true" + ) ) tool_get_valuation_comparables_enabled: bool = field( - default_factory=lambda: os.getenv("TOOL_GET_VALUATION_COMPARABLES_ENABLED", "true").lower() - == "true" + default_factory=lambda: ( + os.getenv("TOOL_GET_VALUATION_COMPARABLES_ENABLED", "true").lower() == "true" + ) ) tool_get_deal_statistics_enabled: bool = field( - default_factory=lambda: os.getenv("TOOL_GET_DEAL_STATISTICS_ENABLED", "true").lower() - == "true" + default_factory=lambda: ( + os.getenv("TOOL_GET_DEAL_STATISTICS_ENABLED", "true").lower() == "true" + ) ) tool_get_market_activity_metrics_enabled: bool = field( - default_factory=lambda: os.getenv( - "TOOL_GET_MARKET_ACTIVITY_METRICS_ENABLED", "false" - ).lower() - == "true" + default_factory=lambda: ( + os.getenv("TOOL_GET_MARKET_ACTIVITY_METRICS_ENABLED", "false").lower() == "true" + ) + ) + tool_search_decisive_appraisals_enabled: bool = field( + default_factory=lambda: ( + os.getenv("TOOL_SEARCH_DECISIVE_APPRAISALS_ENABLED", "true").lower() == "true" + ) ) def __post_init__(self): diff --git a/nadlan_mcp/fastmcp_server.py b/nadlan_mcp/fastmcp_server.py index 88a12fe..53d3345 100644 --- a/nadlan_mcp/fastmcp_server.py +++ b/nadlan_mcp/fastmcp_server.py @@ -14,6 +14,7 @@ from mcp.server.fastmcp import FastMCP from starlette.responses import JSONResponse from nadlan_mcp.config import get_config +from nadlan_mcp.govil import DecisiveAppraiserClient from nadlan_mcp.govmap import GovmapClient from nadlan_mcp.govmap.models import Deal from nadlan_mcp.govmap.outlier_detection import filter_deals_for_analysis @@ -28,6 +29,9 @@ mcp = FastMCP("nadlan-mcp") # Initialize the Govmap client client = GovmapClient() +# Initialize the Decisive Appraiser (gov.il / Ministry of Justice) client +decisive_appraiser_client = DecisiveAppraiserClient() + def conditional_tool(config_flag: str): """ @@ -1310,6 +1314,112 @@ def get_market_activity_metrics(address: str, years_back: int = 2, radius_meters return f"Error analyzing market activity: {str(e)}" +@conditional_tool("tool_search_decisive_appraisals_enabled") +def search_decisive_appraisals( + block: Optional[str] = None, + plot: Optional[str] = None, + decisive_appraiser: Optional[str] = None, + committee: Optional[str] = None, + decision_date_from: Optional[str] = None, + decision_date_to: Optional[str] = None, + publicity_date_from: Optional[str] = None, + publicity_date_to: Optional[str] = None, + search_text: Optional[str] = None, + appraisal_header: Optional[str] = None, + max_results: int = 30, +) -> str: + """Search published "decisive appraiser" (שמאי מכריע) decisions from the Israeli + Ministry of Justice public registry. The registry holds 30,000+ written + decisions on land betterment levy, expropriation, real-estate disputes etc. + + All filters are optional. Use block + plot for property-level lookups + (גוש + חלקה), or `decisive_appraiser` to find every decision by a given + appraiser. Page size is fixed by the upstream at 10; this tool pages + automatically up to `max_results`. + + Args: + block: Land block number, גוש (e.g. "6212"). Exact match. + plot: Land plot number, חלקה (e.g. "894"). Use with block. + decisive_appraiser: Appraiser name in Hebrew (e.g. "דדון דוד"). + committee: Local planning committee (e.g. "תל אביב-יפו"). + decision_date_from / decision_date_to: Date range for the decision date. + Format: dd-MM-yyyy (e.g. "01-01-2024"). Either bound is optional. + publicity_date_from / publicity_date_to: Date range for publication. + search_text: Free-text search inside the decision document. + appraisal_header: Search within the decision title. + max_results: Maximum number of decisions to return (default 30). + + Returns: + JSON string with `total_results` (database-wide), `returned`, the + applied `filters`, and a `decisions` array. Each decision contains + the appraiser name, block/plot, committee, dates, the decision title + and a `pdf_url` to download the original PDF. + """ + log_mcp_call( + "search_decisive_appraisals", + block=block, + plot=plot, + decisive_appraiser=decisive_appraiser, + committee=committee, + decision_date_from=decision_date_from, + decision_date_to=decision_date_to, + publicity_date_from=publicity_date_from, + publicity_date_to=publicity_date_to, + search_text=search_text, + appraisal_header=appraisal_header, + max_results=max_results, + ) + try: + response = decisive_appraiser_client.search_decisions_paged( + max_results=max_results, + block=block, + plot=plot, + decisive_appraiser=decisive_appraiser, + committee=committee, + decision_date_from=decision_date_from, + decision_date_to=decision_date_to, + publicity_date_from=publicity_date_from, + publicity_date_to=publicity_date_to, + search_text=search_text, + appraisal_header=appraisal_header, + ) + + decisions = [] + for idx, decision in enumerate(response.results, start=1): + data = decision.model_dump(mode="json", exclude_none=True) + documents = data.pop("documents", []) or [] + primary_pdf = documents[0]["file_url"] if documents else None + data["id"] = idx + data["pdf_url"] = primary_pdf + data["all_documents"] = documents + decisions.append(data) + + return json.dumps( + { + "total_results": response.total_results, + "returned": len(decisions), + "filters": { + "block": block, + "plot": plot, + "decisive_appraiser": decisive_appraiser, + "committee": committee, + "decision_date_from": decision_date_from, + "decision_date_to": decision_date_to, + "publicity_date_from": publicity_date_from, + "publicity_date_to": publicity_date_to, + "search_text": search_text, + "appraisal_header": appraisal_header, + }, + "decisions": decisions, + }, + ensure_ascii=False, + indent=None, + ) + except Exception as e: + logger.error(f"Error in search_decisive_appraisals: {e}", exc_info=True) + return f"Error searching decisive appraisals: {str(e)}" + + # Health check endpoint for HTTP deployments @mcp.custom_route("/health", methods=["GET"]) async def health_check(request): diff --git a/nadlan_mcp/govil/__init__.py b/nadlan_mcp/govil/__init__.py new file mode 100644 index 0000000..d3132d7 --- /dev/null +++ b/nadlan_mcp/govil/__init__.py @@ -0,0 +1,21 @@ +""" +Israeli government open API (gov.il) clients. + +This package provides clients for gov.il public APIs distinct from Govmap — +notably the Ministry of Justice "Decisive Appraiser" search service that +exposes published decisions of certified property appraisers. +""" + +from .client import DecisiveAppraiserClient +from .models import ( + AppraisalDecision, + AppraisalDocument, + DecisiveAppraiserSearchResponse, +) + +__all__ = [ + "DecisiveAppraiserClient", + "AppraisalDecision", + "AppraisalDocument", + "DecisiveAppraiserSearchResponse", +] diff --git a/nadlan_mcp/govil/client.py b/nadlan_mcp/govil/client.py new file mode 100644 index 0000000..3bee89e --- /dev/null +++ b/nadlan_mcp/govil/client.py @@ -0,0 +1,263 @@ +""" +Client for the Ministry of Justice "Decisive Appraiser" search API. + +The upstream endpoint sits behind an F5 WAF that fingerprints TLS clients +and rejects standard `requests`. We use `curl_cffi` with Chrome +impersonation to traverse it. A static `x-client-id` header issued to the +gov.il SPA is required by the gateway; without it every request returns a +generic 500 "General Error". +""" + +from datetime import datetime +import logging +import time +from typing import Any, Dict, List, Optional + +from curl_cffi import requests as cf_requests + +from nadlan_mcp.config import GovmapConfig, get_config + +from .models import ( + AppraisalDecision, + DecisiveAppraiserSearchResponse, +) + +logger = logging.getLogger(__name__) + +# Endpoint is public and stable; embedding it avoids forcing every operator +# to set yet another env var. Override-able via config if the path moves. +DEFAULT_API_URL = ( + "https://pub-justice.openapi.gov.il/pub/moj/portal/rest/searchpredefinedapi/v1" + "/SearchPredefinedApi/DecisiveAppraiser/SearchDecisions" +) +DEFAULT_REFERER = "https://www.gov.il/he/departments/dynamiccollectors/decisive_appraisal_decisions" +# Public client id assigned to the gov.il SPA and visible in any browser +# DevTools session. Not a secret, but required by the gateway. +DEFAULT_CLIENT_ID = "149a5bad-edde-49a6-9fb9-188bd17d4788" + +# Filters the upstream API understands. Anything else is silently dropped. +_VALID_FILTER_KEYS = { + "SearchText", + "AppraisalHeader", + "DecisiveAppraiser", + "Block", + "Plot", + "Committee", + "AppraisalType", + "AppraiserType", + "AppraisalVersion", + "PublicityDate_from", + "PublicityDate_to", + "DecisionDate_from", + "DecisionDate_to", +} + + +def _format_date(value: Optional[Any]) -> Optional[str]: + """Coerce a date / datetime / string into the dd-MM-yyyy format the API expects.""" + if value is None: + return None + if isinstance(value, datetime): + return value.strftime("%d-%m-%Y") + if hasattr(value, "strftime"): + return value.strftime("%d-%m-%Y") + s = str(value).strip() + if not s: + return None + return s + + +class DecisiveAppraiserClient: + """ + Search and download published decisive-appraiser decisions. + + Uses curl_cffi with Chrome 120 impersonation to satisfy the F5 WAF in + front of pub-justice.openapi.gov.il. Honours the same retry / rate-limit + knobs as `GovmapClient` so operators don't manage two configs. + """ + + def __init__( + self, + config: Optional[GovmapConfig] = None, + api_url: str = DEFAULT_API_URL, + client_id: str = DEFAULT_CLIENT_ID, + impersonate: str = "chrome120", + ): + self.config = config or get_config() + self.api_url = api_url + self.client_id = client_id + self._impersonate = impersonate + self._session = cf_requests.Session(impersonate=impersonate) + self._session.headers.update( + { + "Accept": "application/json, text/plain, */*", + "Accept-Language": "he-IL,he;q=0.9,en-US;q=0.8,en;q=0.7", + "Content-Type": "application/json;charset=UTF-8", + "Origin": "https://www.gov.il", + "Referer": DEFAULT_REFERER, + "Sec-Fetch-Dest": "empty", + "Sec-Fetch-Mode": "cors", + "Sec-Fetch-Site": "cross-site", + "x-client-id": self.client_id, + } + ) + self._last_request_time = 0.0 + + def _rate_limit(self) -> None: + min_interval = 1.0 / self.config.requests_per_second + elapsed = time.time() - self._last_request_time + if elapsed < min_interval: + time.sleep(min_interval - elapsed) + self._last_request_time = time.time() + + def _post(self, body: Dict[str, Any]) -> Dict[str, Any]: + """POST `body` to the search endpoint with retries on transient failures.""" + last_exc: Optional[Exception] = None + for attempt in range(self.config.max_retries + 1): + self._rate_limit() + try: + response = self._session.post( + self.api_url, + json=body, + timeout=(self.config.connect_timeout, self.config.read_timeout), + ) + # 500 from this gateway typically means "you got past TLS but + # the gateway rejected your shape/headers"; treat as fatal so + # retries don't mask a misconfiguration. + if response.status_code == 500: + raise ValueError( + f"DecisiveAppraiser API returned 500 — request shape or " + f"x-client-id is likely wrong. Body keys sent: {list(body.keys())}" + ) + if response.status_code >= 500: + raise cf_requests.RequestsError( + f"Upstream {response.status_code}: {response.text[:200]}" + ) + if response.status_code != 200: + raise ValueError( + f"DecisiveAppraiser API returned {response.status_code}: " + f"{response.text[:200]}" + ) + return response.json() + except (cf_requests.RequestsError, OSError) as e: + last_exc = e + if attempt < self.config.max_retries: + wait = min( + self.config.retry_max_wait, + self.config.retry_min_wait * (2**attempt), + ) + logger.warning( + f"DecisiveAppraiser request failed (attempt {attempt + 1}/" + f"{self.config.max_retries + 1}): {e}. Retrying in {wait}s" + ) + time.sleep(wait) + else: + raise + # Unreachable, but satisfies the type checker. + raise last_exc if last_exc else RuntimeError("retry loop exited unexpectedly") + + def search_decisions( + self, + block: Optional[str] = None, + plot: Optional[str] = None, + decisive_appraiser: Optional[str] = None, + committee: Optional[str] = None, + decision_date_from: Optional[Any] = None, + decision_date_to: Optional[Any] = None, + publicity_date_from: Optional[Any] = None, + publicity_date_to: Optional[Any] = None, + search_text: Optional[str] = None, + appraisal_header: Optional[str] = None, + skip: int = 0, + ) -> DecisiveAppraiserSearchResponse: + """ + Search published decisive-appraiser decisions. + + All filters are optional — the API will return the latest 10 results + when called with just `skip=0`. Page size is fixed at 10 server-side; + use `skip` to paginate. + + Args: + block: Land block (גוש) — exact string match. + plot: Land plot (חלקה) — exact string match within `block`. + decisive_appraiser: Appraiser name (substring/exact, Hebrew). + committee: Local committee name (e.g. "תל אביב-יפו"). + decision_date_from / decision_date_to: Filter by decision date. + Accepts datetime, date, or "dd-MM-yyyy" string. + publicity_date_from / publicity_date_to: Filter by publication date. + search_text: Free-text search across the document body. + appraisal_header: Search within the appraisal header text. + skip: Pagination offset (multiples of 10). + """ + body: Dict[str, Any] = {"skip": int(skip)} + if block: + body["Block"] = str(block).strip() + if plot: + body["Plot"] = str(plot).strip() + if decisive_appraiser: + body["DecisiveAppraiser"] = decisive_appraiser.strip() + if committee: + body["Committee"] = committee.strip() + if search_text: + body["SearchText"] = search_text.strip() + if appraisal_header: + body["AppraisalHeader"] = appraisal_header.strip() + df = _format_date(decision_date_from) + if df: + body["DecisionDate_from"] = df + dt = _format_date(decision_date_to) + if dt: + body["DecisionDate_to"] = dt + pf = _format_date(publicity_date_from) + if pf: + body["PublicityDate_from"] = pf + pt = _format_date(publicity_date_to) + if pt: + body["PublicityDate_to"] = pt + + raw = self._post(body) + # Unwrap {"Results": [{"Data": {...}}, ...]} into [decision, ...] + decisions: List[AppraisalDecision] = [] + for item in raw.get("Results", []) or []: + data = item.get("Data") if isinstance(item, dict) else None + if isinstance(data, dict): + decisions.append(AppraisalDecision.model_validate(data)) + return DecisiveAppraiserSearchResponse( + results=decisions, + total_results=int(raw.get("TotalResults") or 0), + status=raw.get("Status"), + message=raw.get("message"), + ) + + def search_decisions_paged( + self, + max_results: int = 50, + **filters: Any, + ) -> DecisiveAppraiserSearchResponse: + """ + Page through `search_decisions` until `max_results` decisions are + collected (or the upstream runs out). + + `filters` are forwarded to `search_decisions`; do not pass `skip`. + """ + if "skip" in filters: + raise ValueError("search_decisions_paged manages skip itself") + + page_size = 10 # Server-side fixed. + collected: List[AppraisalDecision] = [] + total = 0 + skip = 0 + while len(collected) < max_results: + page = self.search_decisions(skip=skip, **filters) + total = page.total_results + if not page.results: + break + collected.extend(page.results) + skip += page_size + if skip >= total: + break + + return DecisiveAppraiserSearchResponse( + results=collected[:max_results], + total_results=total, + ) diff --git a/nadlan_mcp/govil/models.py b/nadlan_mcp/govil/models.py new file mode 100644 index 0000000..b193332 --- /dev/null +++ b/nadlan_mcp/govil/models.py @@ -0,0 +1,63 @@ +""" +Pydantic models for the Ministry of Justice "Decisive Appraiser" API. + +The upstream API at pub-justice.openapi.gov.il returns decisions in a +nested wrapper: + + {"Results": [{"Data": {...decision fields...}}, ...], + "TotalResults": 30398, "Status": ..., "message": ...} + +These models flatten the wrapper for ergonomic use, while preserving the +original Hebrew field labels via aliases. +""" + +from datetime import datetime +from typing import Any, Dict, List, Optional + +from pydantic import BaseModel, ConfigDict, Field + + +class AppraisalDocument(BaseModel): + """A single PDF attached to an appraisal decision.""" + + model_config = ConfigDict(populate_by_name=True, extra="ignore") + + file_url: str = Field(alias="FileName") + display_name: Optional[str] = Field(default=None, alias="DisplayName") + extension: Optional[str] = Field(default=None, alias="Extension") + + +class AppraisalDecision(BaseModel): + """ + One published decision by a certified decisive appraiser (שמאי מכריע). + + Field names are snake_case in Python; aliases match the upstream + PascalCase JSON keys so the model can be constructed from a raw API + payload via `AppraisalDecision.model_validate(data)`. + """ + + model_config = ConfigDict(populate_by_name=True, extra="ignore") + + appraisal_header: Optional[str] = Field(default=None, alias="AppraisalHeader") + appraisal_type: Optional[str] = Field(default=None, alias="AppraisalType") + appraisal_version: Optional[str] = Field(default=None, alias="AppraisalVersion") + decisive_appraiser: Optional[str] = Field(default=None, alias="DecisiveAppraiser") + appraiser_type: Optional[str] = Field(default=None, alias="AppraiserType") + block: Optional[str] = Field(default=None, alias="Block") + plot: Optional[str] = Field(default=None, alias="Plot") + committee: Optional[str] = Field(default=None, alias="Committee") + decision_date: Optional[datetime] = Field(default=None, alias="DecisionDate") + publicity_date: Optional[datetime] = Field(default=None, alias="PublicityDate") + documents: List[AppraisalDocument] = Field(default_factory=list, alias="Document") + doc_summary: Dict[str, Any] = Field(default_factory=dict, alias="DocSummary") + + +class DecisiveAppraiserSearchResponse(BaseModel): + """Top-level wrapper of the SearchDecisions endpoint.""" + + model_config = ConfigDict(populate_by_name=True, extra="ignore") + + total_results: int = Field(default=0, alias="TotalResults") + results: List[AppraisalDecision] = Field(default_factory=list) + status: Optional[Any] = Field(default=None, alias="Status") + message: Optional[Any] = Field(default=None) diff --git a/pyproject.toml b/pyproject.toml index e0e17f7..7d252ba 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -27,6 +27,7 @@ dependencies = [ "python-dotenv>=1.0.0", "pydantic>=2.0.0", "fastmcp>=2.13.0,<3.0.0", + "curl_cffi>=0.5.0", ] [project.optional-dependencies] diff --git a/tests/test_decisive_appraiser_client.py b/tests/test_decisive_appraiser_client.py new file mode 100644 index 0000000..076ba44 --- /dev/null +++ b/tests/test_decisive_appraiser_client.py @@ -0,0 +1,224 @@ +""" +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"