Merge pull request #3997 from lonvia/fix-postcode-index

Reenable index on centroid column for location_postcodes

.github/actions/setup-postgresql-windows/action.yml

@@ -0,0 +1,95 @@
+name: 'Setup Postgresql and Postgis on Windows'
+
+description: 'Installs PostgreSQL and PostGIS for Windows and configures it for CI tests'
+
+inputs:
+    postgresql-version:
+        description: 'Version of PostgreSQL to install'
+        required: true
+
+runs:
+    using: "composite"  
+
+    steps:
+        - name: Set up PostgreSQL variables
+          shell: pwsh
+          run: |
+            $version = "${{ inputs.postgresql-version }}"
+            $root = "C:\Program Files\PostgreSQL\$version"
+            $bin = "$root\bin"
+
+            echo "PGROOT=$root" | Out-File -FilePath $env:GITHUB_ENV -Encoding utf8 -Append
+            echo "PGBIN=$bin"   | Out-File -FilePath $env:GITHUB_ENV -Encoding utf8 -Append
+
+            echo "$bin" | Out-File -FilePath $env:GITHUB_PATH -Encoding utf8 -Append
+
+        - name: Decide Postgis version (Windows)
+          id: postgis-ver
+          shell: pwsh
+          run: |
+            echo "PowerShell version: ${PSVersionTable.PSVersion}"
+            $PG_VERSION = Split-Path $env:PGROOT -Leaf
+            $postgis_page = "https://download.osgeo.org/postgis/windows/pg$PG_VERSION"
+            echo "Detecting PostGIS version from $postgis_page for PostgreSQL $PG_VERSION"
+            $pgis_bundle = (Invoke-WebRequest -Uri $postgis_page -ErrorAction Stop).Links.Where({$_.href -match "^postgis.*zip$"}).href
+            if (!$pgis_bundle) {
+              Write-Error "Could not find latest PostGIS version in $postgis_page that would match  ^postgis.*zip$  pattern"
+              exit 1
+            }
+            $pgis_bundle = [IO.Path]::ChangeExtension($pgis_bundle, [NullString]::Value)
+            $pgis_bundle_url = "$postgis_page/$pgis_bundle.zip"
+            Add-Content $env:GITHUB_OUTPUT "postgis_file=$pgis_bundle"
+            Add-Content $env:GITHUB_OUTPUT "postgis_bundle_url=$pgis_bundle_url"
+
+        - uses: actions/cache@v4
+          with:
+            path: |
+              C:/postgis.zip
+            key: postgis-cache-${{ steps.postgis-ver.outputs.postgis_file }}
+
+        - name: Download postgis
+          shell: pwsh
+          run: |
+            if (!(Test-Path "C:\postgis.zip")){(new-object net.webclient).DownloadFile($env:PGIS_BUNDLE_URL, "c:\postgis.zip")}
+            if (Test-path "c:\postgis_archive"){Remove-Item "c:\postgis_archive" -Recurse -Force}
+            7z x c:\postgis.zip -oc:\postgis_archive
+          env: 
+            PGIS_BUNDLE_URL: ${{ steps.postgis-ver.outputs.postgis_bundle_url }}
+
+        - name: Install postgis
+          shell: bash
+          run: |
+            echo "Root: $PGROOT, Bin: $PGBIN"
+            cp -r c:/postgis_archive/postgis-bundle-*/* "$PGROOT" 
+
+        - name: Start PostgreSQL on Windows
+          run: |
+            $pgService = Get-Service -Name postgresql*
+            Set-Service -InputObject $pgService -Status running -StartupType automatic
+            Start-Process -FilePath "$env:PGBIN\pg_isready" -Wait -PassThru
+          shell: pwsh
+
+        - name: Adapt postgresql configuration
+          shell: pwsh
+          env:
+            PGPASSWORD: root
+          run: |
+            & "$env:PGBIN\psql" -U postgres -d postgres -c "ALTER SYSTEM SET fsync = 'off';"
+            & "$env:PGBIN\psql" -U postgres -d postgres -c "ALTER SYSTEM SET synchronous_commit = 'off';"
+            & "$env:PGBIN\psql" -U postgres -d postgres -c "ALTER SYSTEM SET full_page_writes = 'off';"
+            & "$env:PGBIN\psql" -U postgres -d postgres -c "ALTER SYSTEM SET shared_buffers = '1GB';"
+            & "$env:PGBIN\psql" -U postgres -d postgres -c "ALTER SYSTEM SET port = 5432;"
+
+            Restart-Service -Name postgresql*
+            Start-Process -FilePath "$env:PGBIN\pg_isready" -Wait -PassThru
+
+        - name: Setup database users
+          shell: pwsh
+          env:
+            PGPASSWORD: root
+          run: |
+            & "$env:PGBIN\createuser" -U postgres -S www-data
+            & "$env:PGBIN\createuser" -U postgres -s runner
+        
+
+

.github/actions/setup-postgresql/action.yml

@@ -1,5 +1,7 @@
 name: 'Setup Postgresql and Postgis'
 
+description: 'Installs PostgreSQL and PostGIS and configures it for CI tests'
+
 inputs:
     postgresql-version:
         description: 'Version of PostgreSQL to install'

.github/workflows/ci-tests.yml

@@ -140,6 +140,65 @@ jobs:
                   ../venv/bin/python -m pytest test/bdd --nominatim-purge
               working-directory: Nominatim
 
+    tests-windows:
+        needs: create-archive
+        runs-on: windows-latest
+
+        steps:
+            - uses: actions/download-artifact@v4
+              with:
+                  name: full-source
+
+            - name: Unpack Nominatim
+              run: tar xf nominatim-src.tar.bz2
+
+            - uses: ./Nominatim/.github/actions/setup-postgresql-windows
+              with:
+                  postgresql-version: 17
+
+            - name: Set up Python
+              uses: actions/setup-python@v5
+              with:
+                  python-version: '3.14'
+
+            - name: Install Spatialite
+              run: |
+                  Invoke-WebRequest -Uri "https://www.gaia-gis.it/gaia-sins/windows-bin-amd64/mod_spatialite-5.1.0-win-amd64.7z" -OutFile "spatialite.7z"
+                  7z x spatialite.7z -o"C:\spatialite"
+                  echo "C:\spatialite\mod_spatialite-5.1.0-win-amd64" | Out-File -FilePath $env:GITHUB_PATH -Encoding utf8 -Append
+
+            - name: Install osm2pgsql
+              run: |
+                  Invoke-WebRequest -Uri "https://osm2pgsql.org/download/windows/osm2pgsql-latest-x64.zip" -OutFile "osm2pgsql.zip"
+                  Expand-Archive -Path "osm2pgsql.zip" -DestinationPath "C:\osm2pgsql"
+                  $BinDir = Get-ChildItem -Path "C:\osm2pgsql" -Recurse -Filter "osm2pgsql.exe" | Select-Object -ExpandProperty DirectoryName | Select-Object -First 1
+                  if (-not $BinDir) {
+                      Write-Error "Could not find osm2pgsql.exe"
+                      exit 1
+                  }
+                  echo "$BinDir" | Out-File -FilePath $env:GITHUB_PATH -Encoding utf8 -Append
+                  $FullExePath = Join-Path $BinDir "osm2pgsql.exe"
+                  echo "NOMINATIM_OSM2PGSQL_BINARY=$FullExePath" | Out-File -FilePath $env:GITHUB_ENV -Encoding utf8 -Append
+
+            - name: Set UTF-8 encoding
+              run: |
+                  echo "PYTHONUTF8=1" >> $env:GITHUB_ENV
+                  [System.Console]::OutputEncoding = [System.Text.Encoding]::UTF8
+
+            - name: Install PyICU from wheel
+              run: |
+                  python -m pip install https://github.com/cgohlke/pyicu-build/releases/download/v2.16.0/pyicu-2.16-cp314-cp314-win_amd64.whl
+                  
+            - name: Install test prerequisites
+              run: |
+                  python -m pip install -U pip
+                  python -m pip install pytest pytest-asyncio "psycopg[binary]!=3.3.0" python-dotenv pyyaml jinja2 psutil sqlalchemy pytest-bdd falcon starlette uvicorn asgi_lifespan aiosqlite osmium mwparserfromhell
+
+            - name: Python unit tests
+              run: |
+                  python -m pytest test/python -k "not (import_osm or run_osm2pgsql)"
+              working-directory: Nominatim
+
     install:
         runs-on: ubuntu-latest
         needs: create-archive

lib-sql/functions/placex_triggers.sql

@@ -89,7 +89,7 @@ BEGIN
 
       -- Add the linked-place (e.g. city) name as a searchable placename in the default language (if any)
       default_language := get_country_language_code(location.country_code);
-      IF default_language is not NULL AND NOT location.name ? ('name:' || default_language) THEN
+      IF default_language is not NULL AND location.name ? 'name' AND NOT location.name ? ('name:' || default_language) THEN
         location.name := location.name || hstore('name:' || default_language, location.name->'name');
       END IF;
 

lib-sql/functions/utils.sql

@@ -153,8 +153,7 @@ BEGIN
     IF ST_GeometryType(geom) in ('ST_Polygon','ST_MultiPolygon') THEN
       SELECT min(postcode), count(*) FROM
         (SELECT postcode FROM location_postcodes
-           WHERE geom && location_postcodes.geometry -- want to use the index
+           WHERE ST_Contains(geom, location_postcodes.centroid)
-                 AND ST_Contains(geom, location_postcodes.centroid)
                  AND country_code = country
            LIMIT 2) sub
         INTO outcode, cnt;

lib-sql/tables/postcodes.sql

@@ -23,6 +23,8 @@ CREATE UNIQUE INDEX idx_location_postcodes_id ON location_postcodes
   USING BTREE (place_id) {{db.tablespace.search_index}};
 CREATE INDEX idx_location_postcodes_geometry ON location_postcodes
   USING GIST (geometry) {{db.tablespace.search_index}};
+CREATE INDEX idx_location_postcodes_centroid ON location_postcodes
+  USING GIST (centroid) {{db.tablespace.search_index}};
 CREATE INDEX IF NOT EXISTS idx_location_postcodes_postcode ON location_postcodes
   USING BTREE (postcode, country_code) {{db.tablespace.search_index}};
 CREATE INDEX IF NOT EXISTS idx_location_postcodes_osmid ON location_postcodes

lib-sql/tables/search_name.sql

@@ -7,7 +7,7 @@
 
 DROP TABLE IF EXISTS search_name;
 
-{% if not db.reverse_only %}
+{% if not create_reverse_only %}
 
 CREATE TABLE search_name (
   place_id BIGINT NOT NULL,

src/nominatim_api/server/falcon/server.py

@@ -184,6 +184,10 @@ class APIMiddleware:
         formatter = load_format_dispatcher('v1', self.api.config.project_dir)
         for name, func in await api_impl.get_routes(self.api):
             endpoint = EndpointWrapper(name, func, self.api, formatter)
+            # If func is a LazySearchEndpoint, give it a reference to wrapper
+            # so it can replace wrapper.func dynamically
+            if hasattr(func, 'set_wrapper'):
+                func.set_wrapper(endpoint)
             self.app.add_route(f"/{name}", endpoint)
             if legacy_urls:
                 self.app.add_route(f"/{name}.php", endpoint)

src/nominatim_api/v1/server_glue.py

@@ -12,6 +12,7 @@ from typing import Optional, Any, Type, Dict, cast, Sequence, Tuple
 from functools import reduce
 import dataclasses
 from urllib.parse import urlencode
+import asyncio
 
 import sqlalchemy as sa
 
@@ -124,6 +125,12 @@ def parse_geometry_details(adaptor: ASGIAdaptor, fmt: str) -> Dict[str, Any]:
             }
 
 
+def has_search_name(conn: sa.engine.Connection) -> bool:
+    """ Check if the search_name table exists in the database.
+    """
+    return sa.inspect(conn).has_table('search_name')
+
+
 async def status_endpoint(api: NominatimAPIAsync, params: ASGIAdaptor) -> Any:
     """ Server glue for /status endpoint. See API docs for details.
     """
@@ -441,6 +448,61 @@ async def polygons_endpoint(api: NominatimAPIAsync, params: ASGIAdaptor) -> Any:
     return build_response(params, params.formatting().format_result(results, fmt, {}))
 
 
+async def search_unavailable_endpoint(api: NominatimAPIAsync, params: ASGIAdaptor) -> Any:
+    """ Server glue for /search endpoint in reverse-only mode.
+        Returns 404 when search functionality is not available.
+    """
+    params.raise_error('Search not available (reverse-only mode)', 404)
+
+
+class LazySearchEndpoint:
+    """
+    Lazy-loading search endpoint that replaces itself after first successful check.
+
+    - Falcon: EndpointWrapper stores this instance in wrapper.func
+      On first request, replace wrapper.func directly with real endpoint
+
+    - Starlette: _wrap_endpoint wraps this instance in a callback
+      store a delegate function and call it on subsequent requests
+    """
+    def __init__(self, api: NominatimAPIAsync, real_endpoint: EndpointFunc):
+        self.api = api
+        self.real_endpoint = real_endpoint
+        self._lock = asyncio.Lock()
+        self._wrapper: Any = None  # Store reference to Falcon's EndpointWrapper
+        self._delegate: Optional[EndpointFunc] = None
+
+    def set_wrapper(self, wrapper: Any) -> None:
+        self._wrapper = wrapper
+
+    async def __call__(self, api: NominatimAPIAsync, params: ASGIAdaptor) -> Any:
+        if self._delegate is None:
+            async with self._lock:
+                # Double-check after acquiring lock (thread safety)
+                if self._delegate is None:
+                    try:
+                        async with api.begin() as conn:
+                            has_table = await conn.connection.run_sync(
+                                has_search_name)
+
+                        if has_table:
+                            # For Starlette
+                            self._delegate = self.real_endpoint
+                            # For Falcon
+                            if self._wrapper is not None:
+                                self._wrapper.func = self.real_endpoint
+                        else:
+                            self._delegate = search_unavailable_endpoint
+                            if self._wrapper is not None:
+                                self._wrapper.func = search_unavailable_endpoint
+
+                    except (PGCORE_ERROR, sa.exc.OperationalError, OSError):
+                        # No _delegate set, so retry on next request
+                        params.raise_error('Search temporarily unavailable', 503)
+
+        return await self._delegate(api, params)
+
+
 async def get_routes(api: NominatimAPIAsync) -> Sequence[Tuple[str, EndpointFunc]]:
     routes = [
         ('status', status_endpoint),
@@ -451,15 +513,13 @@ async def get_routes(api: NominatimAPIAsync) -> Sequence[Tuple[str, EndpointFunc
         ('polygons', polygons_endpoint),
     ]
 
-    def has_search_name(conn: sa.engine.Connection) -> bool:
-        insp = sa.inspect(conn)
-        return insp.has_table('search_name')
-
     try:
         async with api.begin() as conn:
             if await conn.connection.run_sync(has_search_name):
                 routes.append(('search', search_endpoint))
-    except (PGCORE_ERROR, sa.exc.OperationalError):
+            else:
-        pass  # ignored
+                routes.append(('search', search_unavailable_endpoint))
+    except (PGCORE_ERROR, sa.exc.OperationalError, OSError):
+        routes.append(('search', LazySearchEndpoint(api, search_endpoint)))
 
     return routes

src/nominatim_db/clicmd/args.py

@@ -120,6 +120,7 @@ class NominatimArgs:
     data_object: Sequence[Tuple[str, int]]
     data_area: Sequence[Tuple[str, int]]
     ro_access: bool
+    postcode_force_reimport: bool
 
     # Arguments to 'replication'
     init: bool

src/nominatim_db/clicmd/refresh.py

@@ -84,6 +84,10 @@ class UpdateRefresh:
                            help='Do not enable code for propagating updates')
         group.add_argument('--enable-debug-statements', action='store_true',
                            help='Enable debug warning statements in functions')
+        group = parser.add_argument_group('Arguments for postcode refresh')
+        group.add_argument('--force-reimport', action='store_true',
+                           dest='postcode_force_reimport',
+                           help='Recompute the postcodes from scratch instead of updating')
 
     def run(self, args: NominatimArgs) -> int:
         from ..tools import refresh, postcodes
@@ -96,7 +100,8 @@ class UpdateRefresh:
                 LOG.warning("Update postcodes centroid")
                 tokenizer = self._get_tokenizer(args.config)
                 postcodes.update_postcodes(args.config.get_libpq_dsn(),
-                                           args.project_dir, tokenizer)
+                                           args.project_dir, tokenizer,
+                                           force_reimport=args.postcode_force_reimport)
                 indexer = Indexer(args.config.get_libpq_dsn(), tokenizer,
                                   args.threads or 1)
                 asyncio.run(indexer.index_postcodes())

src/nominatim_db/db/query_pool.py

@@ -38,6 +38,7 @@ class QueryPool:
         """ Schedule a query for execution.
         """
         if self.is_cancelled:
+            self.clear_queue()
             await self.finish()
             return
 
@@ -47,6 +48,7 @@ class QueryPool:
         await asyncio.sleep(0)
 
         if self.is_cancelled:
+            self.clear_queue()
             await self.finish()
 
     async def finish(self) -> None:

src/nominatim_db/tokenizer/sanitizers/clean_housenumbers.py

@@ -2,7 +2,7 @@
 #
 # This file is part of Nominatim. (https://nominatim.org)
 #
-# Copyright (C) 2024 by the Nominatim developer community.
+# Copyright (C) 2026 by the Nominatim developer community.
 # For a full list of authors see the git log.
 """
 Sanitizer that preprocesses address tags for house numbers. The sanitizer
@@ -10,6 +10,7 @@ allows to
 
 * define which tags are to be considered house numbers (see 'filter-kind')
 * split house number lists into individual numbers (see 'delimiters')
+* expand interpolated house numbers
 
 Arguments:
     delimiters: Define the set of characters to be used for
@@ -23,13 +24,19 @@ Arguments:
                      instead of a house number. Either takes a single string
                      or a list of strings, where each string is a regular
                      expression that must match the full house number value.
+    expand-interpolations: When true, expand house number ranges to separate numbers
+                           when an 'interpolation' is present. (default: true)
+
 """
-from typing import Callable, Iterator, List
+from typing import Callable, Iterator, Iterable, Union
+import re
 
 from ...data.place_name import PlaceName
 from .base import ProcessInfo
 from .config import SanitizerConfig
 
+RANGE_REGEX = re.compile(r'\d+-\d+')
+
 
 class _HousenumberSanitizer:
 
@@ -38,21 +45,40 @@ class _HousenumberSanitizer:
         self.split_regexp = config.get_delimiter()
 
         self.filter_name = config.get_filter('convert-to-name', 'FAIL_ALL')
+        self.expand_interpolations = config.get_bool('expand-interpolations', True)
 
     def __call__(self, obj: ProcessInfo) -> None:
         if not obj.address:
             return
 
-        new_address: List[PlaceName] = []
+        itype: Union[int, str, None] = None
+        if self.expand_interpolations:
+            itype = next((i.name for i in obj.address if i.kind == 'interpolation'), None)
+            if itype is not None:
+                if itype == 'all':
+                    itype = 1
+                elif len(itype) == 1 and itype.isdigit():
+                    itype = int(itype)
+                elif itype not in ('odd', 'even'):
+                    itype = None
+
+        new_address: list[PlaceName] = []
         for item in obj.address:
             if self.filter_kind(item.kind):
+                if itype is not None and RANGE_REGEX.fullmatch(item.name):
+                    hnrs = self._expand_range(itype, item.name)
+                    if hnrs:
+                        new_address.extend(item.clone(kind='housenumber', name=str(hnr))
+                                           for hnr in hnrs)
+                        continue
+
                 if self.filter_name(item.name):
                     obj.names.append(item.clone(kind='housenumber'))
                 else:
                     new_address.extend(item.clone(kind='housenumber', name=n)
                                        for n in self.sanitize(item.name))
-            else:
+            elif item.kind != 'interpolation':
-                # Don't touch other address items.
+                # Ignore interpolation, otherwise don't touch other address items.
                 new_address.append(item)
 
         obj.address = new_address
@@ -70,6 +96,22 @@ class _HousenumberSanitizer:
     def _regularize(self, hnr: str) -> Iterator[str]:
         yield hnr
 
+    def _expand_range(self, itype: Union[str, int], hnr: str) -> Iterable[int]:
+        first, last = (int(i) for i in hnr.split('-'))
+
+        if isinstance(itype, int):
+            step = itype
+        else:
+            step = 2
+            if (itype == 'even' and first % 2 == 1)\
+               or (itype == 'odd' and first % 2 == 0):
+                first += 1
+
+        if (last + 1 - first) / step < 10:
+            return range(first, last + 1, step)
+
+        return []
+
 
 def create(config: SanitizerConfig) -> Callable[[ProcessInfo], None]:
     """ Create a housenumber processing function.

src/nominatim_db/tools/database_import.py

@@ -152,12 +152,11 @@ def create_tables(conn: Connection, config: Configuration, reverse_only: bool =
         When `reverse_only` is True, then the main table for searching will
         be skipped and only reverse search is possible.
     """
-    sql = SQLPreprocessor(conn, config)
+    SQLPreprocessor(conn, config).run_sql_file(conn, 'tables.sql',
-    sql.env.globals['db']['reverse_only'] = reverse_only
+                                               create_reverse_only=reverse_only)
 
-    sql.run_sql_file(conn, 'tables.sql')
+    # reinitiate the preprocessor to get all the newly created tables
-
+    SQLPreprocessor(conn, config).run_sql_file(conn, 'grants.sql')
-    sql.run_sql_file(conn, 'grants.sql')
 
 
 def create_table_triggers(conn: Connection, config: Configuration) -> None:

src/nominatim_db/tools/postcodes.py

@@ -78,7 +78,7 @@ class _PostcodeCollector:
                 self.collected[normalized] += (x, y)
 
     def commit(self, conn: Connection, analyzer: AbstractAnalyzer,
-               project_dir: Optional[Path]) -> None:
+               project_dir: Optional[Path], is_initial: bool) -> None:
         """ Update postcodes for the country from the postcodes selected so far.
 
             When 'project_dir' is set, then any postcode files found in this
@@ -87,11 +87,14 @@ class _PostcodeCollector:
         if project_dir is not None:
             self._update_from_external(analyzer, project_dir)
 
-        with conn.cursor() as cur:
+        if is_initial:
-            cur.execute("""SELECT postcode FROM location_postcodes
+            to_delete = []
-                           WHERE country_code = %s AND osm_id is null""",
+        else:
-                        (self.country, ))
+            with conn.cursor() as cur:
-            to_delete = [row[0] for row in cur if row[0] not in self.collected]
+                cur.execute("""SELECT postcode FROM location_postcodes
+                               WHERE country_code = %s AND osm_id is null""",
+                            (self.country, ))
+                to_delete = [row[0] for row in cur if row[0] not in self.collected]
 
         to_add = [dict(zip(('pc', 'x', 'y'), (k, *v.centroid())))
                   for k, v in self.collected.items()]
@@ -102,22 +105,32 @@ class _PostcodeCollector:
 
         with conn.cursor() as cur:
             if to_add:
-                cur.executemany(pysql.SQL(
+                columns = ['country_code',
-                    """INSERT INTO location_postcodes
+                           'rank_search',
-                         (country_code, rank_search, postcode, centroid, geometry)
+                           'postcode',
-                       VALUES ({}, {}, %(pc)s,
+                           'centroid',
-                               ST_SetSRID(ST_MakePoint(%(x)s, %(y)s), 4326),
+                           'geometry']
-                               expand_by_meters(ST_SetSRID(ST_MakePoint(%(x)s, %(y)s), 4326), {}))
+                values = [pysql.Literal(self.country),
-                    """).format(pysql.Literal(self.country),
+                          pysql.Literal(_extent_to_rank(self.extent)),
-                                pysql.Literal(_extent_to_rank(self.extent)),
+                          pysql.Placeholder('pc'),
-                                pysql.Literal(self.extent)),
+                          pysql.SQL('ST_SetSRID(ST_MakePoint(%(x)s, %(y)s), 4326)'),
-                    to_add)
+                          pysql.SQL("""expand_by_meters(
+                                           ST_SetSRID(ST_MakePoint(%(x)s, %(y)s), 4326), {})""")
+                               .format(pysql.Literal(self.extent))]
+                if is_initial:
+                    columns.extend(('place_id', 'indexed_status'))
+                    values.extend((pysql.SQL("nextval('seq_place')"), pysql.Literal(1)))
+
+                cur.executemany(pysql.SQL("INSERT INTO location_postcodes ({}) VALUES ({})")
+                                     .format(pysql.SQL(',')
+                                                  .join(pysql.Identifier(c) for c in columns),
+                                             pysql.SQL(',').join(values)),
+                                to_add)
             if to_delete:
                 cur.execute("""DELETE FROM location_postcodes
                                WHERE country_code = %s and postcode = any(%s)
                                      AND osm_id is null
                             """, (self.country, to_delete))
-            cur.execute("ANALYSE location_postcodes")
 
     def _update_from_external(self, analyzer: AbstractAnalyzer, project_dir: Path) -> None:
         """ Look for an external postcode file for the active country in
@@ -164,7 +177,8 @@ class _PostcodeCollector:
         return None
 
 
-def update_postcodes(dsn: str, project_dir: Optional[Path], tokenizer: AbstractTokenizer) -> None:
+def update_postcodes(dsn: str, project_dir: Optional[Path],
+                     tokenizer: AbstractTokenizer, force_reimport: bool = False) -> None:
     """ Update the table of postcodes from the input tables
         placex and place_postcode.
     """
@@ -176,45 +190,76 @@ def update_postcodes(dsn: str, project_dir: Optional[Path], tokenizer: AbstractT
                               SET country_code = get_country_code(centroid)
                               WHERE country_code is null
                          """)
+            if force_reimport:
+                conn.execute("TRUNCATE location_postcodes")
+                is_initial = True
+            else:
+                is_initial = _is_postcode_table_empty(conn)
+            if is_initial:
+                conn.execute("""ALTER TABLE location_postcodes
+                                DISABLE TRIGGER location_postcodes_before_insert""")
             # Now update first postcode areas
-            _update_postcode_areas(conn, analyzer, matcher)
+            _update_postcode_areas(conn, analyzer, matcher, is_initial)
             # Then fill with estimated postcode centroids from other info
-            _update_guessed_postcode(conn, analyzer, matcher, project_dir)
+            _update_guessed_postcode(conn, analyzer, matcher, project_dir, is_initial)
+            if is_initial:
+                conn.execute("""ALTER TABLE location_postcodes
+                                ENABLE TRIGGER location_postcodes_before_insert""")
             conn.commit()
 
         analyzer.update_postcodes_from_db()
 
 
+def _is_postcode_table_empty(conn: Connection) -> bool:
+    """ Check if there are any entries in the location_postcodes table yet.
+    """
+    with conn.cursor() as cur:
+        cur.execute("SELECT place_id FROM location_postcodes LIMIT 1")
+        return cur.fetchone() is None
+
+
 def _insert_postcode_areas(conn: Connection, country_code: str,
-                           extent: int, pcs: list[dict[str, str]]) -> None:
+                           extent: int, pcs: list[dict[str, str]],
+                           is_initial: bool) -> None:
     if pcs:
         with conn.cursor() as cur:
+            columns = ['osm_id', 'country_code',
+                       'rank_search', 'postcode',
+                       'centroid', 'geometry']
+            values = [pysql.Identifier('osm_id'), pysql.Identifier('country_code'),
+                      pysql.Literal(_extent_to_rank(extent)), pysql.Placeholder('out'),
+                      pysql.Identifier('centroid'), pysql.Identifier('geometry')]
+            if is_initial:
+                columns.extend(('place_id', 'indexed_status'))
+                values.extend((pysql.SQL("nextval('seq_place')"), pysql.Literal(1)))
+
             cur.executemany(
                 pysql.SQL(
-                    """ INSERT INTO location_postcodes
+                    """ INSERT INTO location_postcodes ({})
-                            (osm_id, country_code, rank_search, postcode, centroid, geometry)
+                            SELECT {} FROM place_postcode
-                            SELECT osm_id, country_code, {}, %(out)s, centroid, geometry
-                            FROM place_postcode
                             WHERE osm_type = 'R'
                                   and country_code = {} and postcode = %(in)s
                                   and geometry is not null
-                    """).format(pysql.Literal(_extent_to_rank(extent)),
+                    """).format(pysql.SQL(',')
+                                     .join(pysql.Identifier(c) for c in columns),
+                                pysql.SQL(',').join(values),
                                 pysql.Literal(country_code)),
                 pcs)
 
 
 def _update_postcode_areas(conn: Connection, analyzer: AbstractAnalyzer,
-                           matcher: PostcodeFormatter) -> None:
+                           matcher: PostcodeFormatter, is_initial: bool) -> None:
     """ Update the postcode areas made from postcode boundaries.
     """
     # first delete all areas that have gone
-    conn.execute(""" DELETE FROM location_postcodes pc
+    if not is_initial:
-                     WHERE pc.osm_id is not null
+        conn.execute(""" DELETE FROM location_postcodes pc
-                       AND NOT EXISTS(
+                         WHERE pc.osm_id is not null
-                              SELECT * FROM place_postcode pp
+                           AND NOT EXISTS(
-                              WHERE pp.osm_type = 'R' and pp.osm_id = pc.osm_id
+                                  SELECT * FROM place_postcode pp
-                                    and geometry is not null)
+                                  WHERE pp.osm_type = 'R' and pp.osm_id = pc.osm_id
-                """)
+                                        and geometry is not null)
+                    """)
     # now insert all in country batches, triggers will ensure proper updates
     with conn.cursor() as cur:
         cur.execute(""" SELECT country_code, postcode FROM place_postcode
@@ -230,7 +275,8 @@ def _update_postcode_areas(conn: Connection, analyzer: AbstractAnalyzer,
                 fmt = matcher.get_matcher(country_code)
             elif country_code != cc:
                 _insert_postcode_areas(conn, country_code,
-                                       matcher.get_postcode_extent(country_code), pcs)
+                                       matcher.get_postcode_extent(country_code), pcs,
+                                       is_initial)
                 country_code = cc
                 fmt = matcher.get_matcher(country_code)
                 pcs = []
@@ -241,21 +287,26 @@ def _update_postcode_areas(conn: Connection, analyzer: AbstractAnalyzer,
 
         if country_code is not None and pcs:
             _insert_postcode_areas(conn, country_code,
-                                   matcher.get_postcode_extent(country_code), pcs)
+                                   matcher.get_postcode_extent(country_code), pcs,
+                                   is_initial)
 
 
 def _update_guessed_postcode(conn: Connection, analyzer: AbstractAnalyzer,
-                             matcher: PostcodeFormatter, project_dir: Optional[Path]) -> None:
+                             matcher: PostcodeFormatter, project_dir: Optional[Path],
+                             is_initial: bool) -> None:
     """ Computes artificial postcode centroids from the placex table,
         potentially enhances it with external data and then updates the
         postcodes in the table 'location_postcodes'.
     """
     # First get the list of countries that currently have postcodes.
     # (Doing this before starting to insert, so it is fast on import.)
-    with conn.cursor() as cur:
+    if is_initial:
-        cur.execute("""SELECT DISTINCT country_code FROM location_postcodes
+        todo_countries: set[str] = set()
-                        WHERE osm_id is null""")
+    else:
-        todo_countries = {row[0] for row in cur}
+        with conn.cursor() as cur:
+            cur.execute("""SELECT DISTINCT country_code FROM location_postcodes
+                            WHERE osm_id is null""")
+            todo_countries = {row[0] for row in cur}
 
     # Next, get the list of postcodes that are already covered by areas.
     area_pcs = defaultdict(set)
@@ -275,6 +326,7 @@ def _update_guessed_postcode(conn: Connection, analyzer: AbstractAnalyzer,
                         FROM place_postcode WHERE geometry is not null)
                     """)
         cur.execute("CREATE INDEX ON _global_postcode_area USING gist(geometry)")
+
     # Recompute the list of valid postcodes from placex.
     with conn.cursor(name="placex_postcodes") as cur:
         cur.execute("""
@@ -296,7 +348,7 @@ def _update_guessed_postcode(conn: Connection, analyzer: AbstractAnalyzer,
         for country, postcode, x, y in cur:
             if collector is None or country != collector.country:
                 if collector is not None:
-                    collector.commit(conn, analyzer, project_dir)
+                    collector.commit(conn, analyzer, project_dir, is_initial)
                 collector = _PostcodeCollector(country, matcher.get_matcher(country),
                                                matcher.get_postcode_extent(country),
                                                exclude=area_pcs[country])
@@ -304,14 +356,14 @@ def _update_guessed_postcode(conn: Connection, analyzer: AbstractAnalyzer,
             collector.add(postcode, x, y)
 
         if collector is not None:
-            collector.commit(conn, analyzer, project_dir)
+            collector.commit(conn, analyzer, project_dir, is_initial)
 
     # Now handle any countries that are only in the postcode table.
     for country in todo_countries:
         fmt = matcher.get_matcher(country)
         ext = matcher.get_postcode_extent(country)
         _PostcodeCollector(country, fmt, ext,
-                           exclude=area_pcs[country]).commit(conn, analyzer, project_dir)
+                           exclude=area_pcs[country]).commit(conn, analyzer, project_dir, False)
 
     conn.execute("DROP TABLE IF EXISTS _global_postcode_area")
 

test/bdd/features/db/import/linking.feature

@@ -299,18 +299,23 @@ Feature: Linking of places
 
     Scenario: Linked places expand default language names
         Given the grid with origin CO
-            | 1 |   | 2 |
+            | 1 |   | 2 |  | 5 |    | 6 |
-            |   | 9 |   |
+            |   | 9 |   |  |   | 10 |   |
-            | 4 |   | 3 |
+            | 4 |   | 3 |  | 8 |    | 7 |
-        Given the places
+        And the places
             | osm | class    | type           | name+name                | geometry    |
             | N9  | place    | city           | Popayán                  | 9           |
-        Given the places
+        And the places
+            | osm | class    | type           | name+name:en             | geometry    |
+            | N10 | place    | city           | Open                     | 10          |
+        And the places
             | osm | class    | type           | name+name                | geometry    | admin |
             | R1  | boundary | administrative | Perímetro Urbano Popayán | (1,2,3,4,1) | 8     |
+            | R2  | boundary | administrative | Abre                     | (5,6,7,8,5) | 8     |
         And the relations
-            | id | members  |
+            | id | members   |
-            | 1  | N9:label |
+            | 1  | N9:label  |
+            | 2  | N10:label |
         When importing
         Then placex contains
             | object      | linked_place_id |

test/bdd/features/db/query/housenumbers.feature

@@ -318,3 +318,64 @@ Feature: Searching of house numbers
         Then the result set contains
          | object |
          | W20 |
+
+    Scenario: A housenumber with interpolation is found
+        Given the places
+         | osm | class    | type | housenr | addr+interpolation | geometry |
+         | N1  | building | yes  | 1-5     | odd                | 9        |
+        And the places
+         | osm | class   | type | name      | geometry |
+         | W10 | highway | path | Rue Paris | 1,2,3    |
+        When importing
+        When geocoding "Rue Paris 1"
+        Then the result set contains
+         | object | address+house_number |
+         | N1     | 1-5 |
+        When geocoding "Rue Paris 3"
+        Then the result set contains
+         | object | address+house_number |
+         | N1     | 1-5 |
+        When geocoding "Rue Paris 5"
+        Then the result set contains
+         | object | address+house_number |
+         | N1     | 1-5 |
+        When geocoding "Rue Paris 2"
+        Then the result set contains
+         | object |
+         | W10  |
+
+    Scenario: A housenumber with bad interpolation is ignored
+        Given the places
+         | osm | class    | type | housenr | addr+interpolation | geometry |
+         | N1  | building | yes  | 1-5     | bad                | 9        |
+        And the places
+         | osm | class   | type | name      | geometry |
+         | W10 | highway | path | Rue Paris | 1,2,3    |
+        When importing
+        When geocoding "Rue Paris 1-5"
+        Then the result set contains
+         | object | address+house_number |
+         | N1     | 1-5 |
+        When geocoding "Rue Paris 3"
+        Then the result set contains
+         | object |
+         | W10    |
+
+
+    Scenario: A bad housenumber with a good interpolation is just a housenumber
+        Given the places
+         | osm | class    | type | housenr | addr+interpolation | geometry |
+         | N1  | building | yes  | 1-100   | all                | 9        |
+        And the places
+         | osm | class   | type | name      | geometry |
+         | W10 | highway | path | Rue Paris | 1,2,3    |
+        When importing
+        When geocoding "Rue Paris 1-100"
+        Then the result set contains
+         | object | address+house_number |
+         | N1     | 1-100 |
+        When geocoding "Rue Paris 3"
+        Then the result set contains
+         | object |
+         | W10    |
+

test/python/config/test_config.py

@@ -200,14 +200,15 @@ def test_get_path_empty(make_config):
     assert not config.get_path('TOKENIZER_CONFIG')
 
 
-def test_get_path_absolute(make_config, monkeypatch):
+def test_get_path_absolute(make_config, monkeypatch, tmp_path):
     config = make_config()
 
-    monkeypatch.setenv('NOMINATIM_FOOBAR', '/dont/care')
+    p = (tmp_path / "does_not_exist").resolve()
+    monkeypatch.setenv('NOMINATIM_FOOBAR', str(p))
     result = config.get_path('FOOBAR')
 
     assert isinstance(result, Path)
-    assert str(result) == '/dont/care'
+    assert str(result) == str(p)
 
 
 def test_get_path_relative(make_config, monkeypatch, tmp_path):

test/python/conftest.py

@@ -2,12 +2,16 @@
 #
 # This file is part of Nominatim. (https://nominatim.org)
 #
-# Copyright (C) 2025 by the Nominatim developer community.
+# Copyright (C) 2026 by the Nominatim developer community.
 # For a full list of authors see the git log.
 import itertools
 import sys
+import asyncio
 from pathlib import Path
 
+if sys.platform == 'win32':
+    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())
+
 import psycopg
 from psycopg import sql as pysql
 import pytest
@@ -17,12 +21,11 @@ SRC_DIR = (Path(__file__) / '..' / '..' / '..').resolve()
 sys.path.insert(0, str(SRC_DIR / 'src'))
 
 from nominatim_db.config import Configuration
-from nominatim_db.db import connection
+from nominatim_db.db import connection, properties
 from nominatim_db.db.sql_preprocessor import SQLPreprocessor
 import nominatim_db.tokenizer.factory
 
 import dummy_tokenizer
-import mocks
 from cursor import CursorForTesting
 
 
@@ -132,28 +135,50 @@ def project_env(tmp_path):
 
 
 @pytest.fixture
-def property_table(table_factory, temp_db_conn):
+def country_table(table_factory):
-    table_factory('nominatim_properties', 'property TEXT, value TEXT')
+    table_factory('country_name', 'partition INT, country_code varchar(2), name hstore')
-
-    return mocks.MockPropertyTable(temp_db_conn)
 
 
 @pytest.fixture
-def status_table(table_factory):
+def country_row(country_table, temp_db_cursor):
+    def _add(partition=None, country=None, names=None):
+        temp_db_cursor.insert_row('country_name', partition=partition,
+                                  country_code=country, name=names)
+
+    return _add
+
+
+@pytest.fixture
+def load_sql(temp_db_conn, country_table):
+    conf = Configuration(None)
+
+    def _run(*filename, **kwargs):
+        for fn in filename:
+            SQLPreprocessor(temp_db_conn, conf).run_sql_file(temp_db_conn, fn, **kwargs)
+
+    return _run
+
+
+@pytest.fixture
+def property_table(load_sql, temp_db_conn):
+    load_sql('tables/nominatim_properties.sql')
+
+    class _PropTable:
+        def set(self, name, value):
+            properties.set_property(temp_db_conn, name, value)
+
+        def get(self, name):
+            return properties.get_property(temp_db_conn, name)
+
+    return _PropTable()
+
+
+@pytest.fixture
+def status_table(load_sql):
     """ Create an empty version of the status table and
         the status logging table.
     """
-    table_factory('import_status',
+    load_sql('tables/status.sql')
-                  """lastimportdate timestamp with time zone NOT NULL,
-                     sequence_id integer,
-                     indexed boolean""")
-    table_factory('import_osmosis_log',
-                  """batchend timestamp,
-                     batchseq integer,
-                     batchsize bigint,
-                     starttime timestamp,
-                     endtime timestamp,
-                     event text""")
 
 
 @pytest.fixture
@@ -178,12 +203,14 @@ def place_row(place_table, temp_db_cursor):
         prerequisite to the fixture.
     """
     idseq = itertools.count(1001)
+
     def _insert(osm_type='N', osm_id=None, cls='amenity', typ='cafe', names=None,
-                admin_level=None, address=None, extratags=None, geom=None):
+                admin_level=None, address=None, extratags=None, geom='POINT(0 0)'):
-        temp_db_cursor.execute("INSERT INTO place VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s)",
+        args = {'osm_type': osm_type, 'osm_id': osm_id or next(idseq),
-                               (osm_id or next(idseq), osm_type, cls, typ, names,
+                'class': cls, 'type': typ, 'name': names, 'admin_level': admin_level,
-                                admin_level, address, extratags,
+                'address': address, 'extratags': extratags,
-                                geom or 'SRID=4326;POINT(0 0)'))
+                'geometry': _with_srid(geom)}
+        temp_db_cursor.insert_row('place', **args)
 
     return _insert
 
@@ -203,50 +230,104 @@ def place_postcode_table(temp_db_with_extensions, table_factory):
 
 @pytest.fixture
 def place_postcode_row(place_postcode_table, temp_db_cursor):
-    """ A factory for rows in the place table. The table is created as a
+    """ A factory for rows in the place_postcode table. The table is created as a
         prerequisite to the fixture.
     """
     idseq = itertools.count(5001)
+
     def _insert(osm_type='N', osm_id=None, postcode=None, country=None,
-                centroid=None, geom=None):
+                centroid='POINT(12.0 4.0)', geom=None):
-        temp_db_cursor.execute("INSERT INTO place_postcode VALUES (%s, %s, %s, %s, %s, %s)",
+        temp_db_cursor.insert_row('place_postcode',
-                               (osm_type, osm_id or next(idseq),
+                                  osm_type=osm_type, osm_id=osm_id or next(idseq),
-                                postcode, country,
+                                  postcode=postcode, country_code=country,
-                                _with_srid(centroid, 'POINT(12.0 4.0)'),
+                                  centroid=_with_srid(centroid),
-                                _with_srid(geom)))
+                                  geometry=_with_srid(geom))
 
     return _insert
 
 
 @pytest.fixture
-def placex_table(temp_db_with_extensions, temp_db_conn):
+def placex_table(temp_db_with_extensions, temp_db_conn, load_sql, place_table):
-    """ Create an empty version of the place table.
+    """ Create an empty version of the placex table.
     """
-    return mocks.MockPlacexTable(temp_db_conn)
+    load_sql('tables/placex.sql')
+    temp_db_conn.execute("CREATE SEQUENCE IF NOT EXISTS seq_place START 1")
 
 
 @pytest.fixture
-def osmline_table(temp_db_with_extensions, table_factory):
+def placex_row(placex_table, temp_db_cursor):
-    table_factory('location_property_osmline',
+    """ A factory for rows in the placex table. The table is created as a
-                  """place_id BIGINT,
+        prerequisite to the fixture.
-                     osm_id BIGINT,
+    """
-                     parent_place_id BIGINT,
+    idseq = itertools.count(1001)
-                     geometry_sector INTEGER,
+
-                     indexed_date TIMESTAMP,
+    def _add(osm_type='N', osm_id=None, cls='amenity', typ='cafe', names=None,
-                     startnumber INTEGER,
+             admin_level=None, address=None, extratags=None, geom='POINT(10 4)',
-                     endnumber INTEGER,
+             country=None, housenumber=None, rank_search=30, rank_address=30,
-                     partition SMALLINT,
+             centroid='POINT(10 4)', indexed_status=0, indexed_date=None):
-                     indexed_status SMALLINT,
+        args = {'place_id': pysql.SQL("nextval('seq_place')"),
-                     linegeo GEOMETRY,
+                'osm_type': osm_type, 'osm_id': osm_id or next(idseq),
-                     interpolationtype TEXT,
+                'class': cls, 'type': typ, 'name': names, 'admin_level': admin_level,
-                     address HSTORE,
+                'address': address, 'housenumber': housenumber,
-                     postcode TEXT,
+                'rank_search': rank_search, 'rank_address': rank_address,
-                     country_code VARCHAR(2)""")
+                'extratags': extratags,
+                'centroid': _with_srid(centroid), 'geometry': _with_srid(geom),
+                'country_code': country,
+                'indexed_status': indexed_status, 'indexed_date': indexed_date,
+                'partition': pysql.Literal(0), 'geometry_sector': pysql.Literal(1)}
+        return temp_db_cursor.insert_row('placex', **args)
+
+    return _add
 
 
 @pytest.fixture
-def sql_preprocessor_cfg(tmp_path, table_factory, temp_db_with_extensions):
+def osmline_table(temp_db_with_extensions, load_sql):
-    table_factory('country_name', 'partition INT', ((0, ), (1, ), (2, )))
+    load_sql('tables/interpolation.sql')
+
+
+@pytest.fixture
+def osmline_row(osmline_table, temp_db_cursor):
+    idseq = itertools.count(20001)
+
+    def _add(osm_id=None, geom='LINESTRING(12.0 11.0, 12.003 11.0)'):
+        return temp_db_cursor.insert_row(
+            'location_property_osmline',
+            place_id=pysql.SQL("nextval('seq_place')"),
+            osm_id=osm_id or next(idseq),
+            geometry_sector=pysql.Literal(20),
+            partition=pysql.Literal(0),
+            indexed_status=1,
+            linegeo=_with_srid(geom))
+
+    return _add
+
+
+@pytest.fixture
+def postcode_table(temp_db_with_extensions, load_sql):
+    load_sql('tables/postcodes.sql')
+
+
+@pytest.fixture
+def postcode_row(postcode_table, temp_db_cursor):
+    def _add(country, postcode, x=34.5, y=-9.33):
+        geom = _with_srid(f"POINT({x} {y})")
+        return temp_db_cursor.insert_row(
+            'location_postcodes',
+            place_id=pysql.SQL("nextval('seq_place')"),
+            indexed_status=pysql.Literal(1),
+            country_code=country, postcode=postcode,
+            centroid=geom,
+            rank_search=pysql.Literal(16),
+            geometry=('ST_Expand(%s::geometry, 0.005)', geom))
+
+    return _add
+
+
+@pytest.fixture
+def sql_preprocessor_cfg(tmp_path, table_factory, temp_db_with_extensions, country_row):
+    for part in range(3):
+        country_row(partition=part)
+
     cfg = Configuration(None)
     cfg.set_libdirs(sql=tmp_path)
     return cfg

test/python/cursor.py

@@ -2,12 +2,13 @@
 #
 # This file is part of Nominatim. (https://nominatim.org)
 #
-# Copyright (C) 2025 by the Nominatim developer community.
+# Copyright (C) 2026 by the Nominatim developer community.
 # For a full list of authors see the git log.
 """
 Specialised psycopg cursor with shortcut functions useful for testing.
 """
 import psycopg
+from psycopg import sql as pysql
 
 
 class CursorForTesting(psycopg.Cursor):
@@ -52,7 +53,49 @@ class CursorForTesting(psycopg.Cursor):
     def table_rows(self, table, where=None):
         """ Return the number of rows in the given table.
         """
-        if where is None:
+        sql = pysql.SQL('SELECT count(*) FROM') + pysql.Identifier(table)
-            return self.scalar('SELECT count(*) FROM ' + table)
+        if where is not None:
+            sql += pysql.SQL('WHERE') + pysql.SQL(where)
 
-        return self.scalar('SELECT count(*) FROM {} WHERE {}'.format(table, where))
+        return self.scalar(sql)
+
+    def insert_row(self, table, **data):
+        """ Insert a row into the given table.
+
+            'data' is a dictionary of column names and associated values.
+            When the value is a pysql.Literal or pysql.SQL, then the expression
+            will be inserted as is instead of loading the value. When the
+            value is a tuple, then the first element will be added as an
+            SQL expression for the value and the second element is treated
+            as the actual value to insert. The SQL expression must contain
+            a %s placeholder in that case.
+
+            If data contains a 'place_id' column, then the value of the
+            place_id column after insert is returned. Otherwise the function
+            returns nothing.
+        """
+        columns = []
+        placeholders = []
+        values = []
+        for k, v in data.items():
+            columns.append(pysql.Identifier(k))
+            if isinstance(v, tuple):
+                placeholders.append(pysql.SQL(v[0]))
+                values.append(v[1])
+            elif isinstance(v, (pysql.Literal, pysql.SQL)):
+                placeholders.append(v)
+            else:
+                placeholders.append(pysql.Placeholder())
+                values.append(v)
+
+        sql = pysql.SQL("INSERT INTO {table} ({columns}) VALUES({values})")\
+                   .format(table=pysql.Identifier(table),
+                           columns=pysql.SQL(',').join(columns),
+                           values=pysql.SQL(',').join(placeholders))
+
+        if 'place_id' in data:
+            sql += pysql.SQL('RETURNING place_id')
+
+        self.execute(sql, values)
+
+        return self.fetchone()[0] if 'place_id' in data else None

test/python/data/test_country_info.py

@@ -53,11 +53,10 @@ def test_setup_country_tables(src_dir, temp_db_with_extensions, dsn, temp_db_cur
 
 @pytest.mark.parametrize("languages", (None, ['fr', 'en']))
 def test_create_country_names(temp_db_with_extensions, temp_db_conn, temp_db_cursor,
-                              table_factory, tokenizer_mock, languages, loaded_country):
+                              country_row, tokenizer_mock, languages, loaded_country):
-
+    temp_db_cursor.execute('TRUNCATE country_name')
-    table_factory('country_name', 'country_code varchar(2), name hstore',
+    country_row(country='us', names={"name": "us1", "name:af": "us2"})
-                  content=(('us', '"name"=>"us1","name:af"=>"us2"'),
+    country_row(country='fr', names={"name": "Fra", "name:en": "Fren"})
-                           ('fr', '"name"=>"Fra", "name:en"=>"Fren"')))
 
     assert temp_db_cursor.scalar("SELECT count(*) FROM country_name") == 2
 

test/python/indexer/test_indexing.py

@@ -1,13 +1,12 @@
-# SPDX-License-Identifier: GPL-3.0-or-later
 #
 # This file is part of Nominatim. (https://nominatim.org)
 #
-# Copyright (C) 2025 by the Nominatim developer community.
+# Copyright (C) 2026 by the Nominatim developer community.
 # For a full list of authors see the git log.
 """
 Tests for running the indexing.
 """
-import itertools
+
 import pytest
 import pytest_asyncio  # noqa
 
@@ -15,129 +14,57 @@ from nominatim_db.indexer import indexer
 from nominatim_db.tokenizer import factory
 
 
-class IndexerTestDB:
+class TestIndexing:
+    @pytest.fixture(autouse=True)
+    def setup(self, temp_db_conn, project_env, tokenizer_mock,
+              placex_table, postcode_table, osmline_table):
+        self.conn = temp_db_conn
+        temp_db_conn.execute("""
+            CREATE OR REPLACE FUNCTION date_update() RETURNS TRIGGER AS $$
+            BEGIN
+              IF NEW.indexed_status = 0 and OLD.indexed_status != 0 THEN
+                NEW.indexed_date = now();
+              END IF;
+              RETURN NEW;
+            END; $$ LANGUAGE plpgsql;
 
-    def __init__(self, conn):
+            DROP TYPE IF EXISTS prepare_update_info CASCADE;
-        self.placex_id = itertools.count(100000)
+            CREATE TYPE prepare_update_info AS (
-        self.osmline_id = itertools.count(500000)
+                         name HSTORE,
-        self.postcode_id = itertools.count(700000)
+                         address HSTORE,
+                         rank_address SMALLINT,
+                         country_code TEXT,
+                         class TEXT,
+                         type TEXT,
+                         linked_place_id BIGINT
+                       );
+            CREATE OR REPLACE FUNCTION placex_indexing_prepare(p placex,
+                                                 OUT result prepare_update_info) AS $$
+            BEGIN
+              result.address := p.address;
+              result.name := p.name;
+              result.class := p.class;
+              result.type := p.type;
+              result.country_code := p.country_code;
+              result.rank_address := p.rank_address;
+            END; $$ LANGUAGE plpgsql STABLE;
 
-        self.conn = conn
+            CREATE OR REPLACE FUNCTION get_interpolation_address(in_address HSTORE, wayid BIGINT)
-        self.conn.autocimmit = True
+            RETURNS HSTORE AS $$ SELECT in_address $$ LANGUAGE sql STABLE;
-        with self.conn.cursor() as cur:
+        """)
-            cur.execute("""CREATE TABLE placex (place_id BIGINT,
-                                                name HSTORE,
-                                                class TEXT,
-                                                type TEXT,
-                                                linked_place_id BIGINT,
-                                                rank_address SMALLINT,
-                                                rank_search SMALLINT,
-                                                indexed_status SMALLINT,
-                                                indexed_date TIMESTAMP,
-                                                partition SMALLINT,
-                                                admin_level SMALLINT,
-                                                country_code TEXT,
-                                                address HSTORE,
-                                                token_info JSONB,
-                                                geometry_sector INTEGER)""")
-            cur.execute("""CREATE TABLE location_property_osmline (
-                               place_id BIGINT,
-                               osm_id BIGINT,
-                               address HSTORE,
-                               token_info JSONB,
-                               indexed_status SMALLINT,
-                               indexed_date TIMESTAMP,
-                               geometry_sector INTEGER)""")
-            cur.execute("""CREATE TABLE location_postcodes (
-                               place_id BIGINT,
-                               indexed_status SMALLINT,
-                               indexed_date TIMESTAMP,
-                               country_code varchar(2),
-                               postcode TEXT)""")
-            cur.execute("""CREATE OR REPLACE FUNCTION date_update() RETURNS TRIGGER
-                           AS $$
-                           BEGIN
-                             IF NEW.indexed_status = 0 and OLD.indexed_status != 0 THEN
-                               NEW.indexed_date = now();
-                             END IF;
-                             RETURN NEW;
-                           END; $$ LANGUAGE plpgsql;""")
-            cur.execute("DROP TYPE IF EXISTS prepare_update_info CASCADE")
-            cur.execute("""CREATE TYPE prepare_update_info AS (
-                             name HSTORE,
-                             address HSTORE,
-                             rank_address SMALLINT,
-                             country_code TEXT,
-                             class TEXT,
-                             type TEXT,
-                             linked_place_id BIGINT
-                           )""")
-            cur.execute("""CREATE OR REPLACE FUNCTION placex_indexing_prepare(p placex,
-                                                     OUT result prepare_update_info)
-                           AS $$
-                           BEGIN
-                             result.address := p.address;
-                             result.name := p.name;
-                             result.class := p.class;
-                             result.type := p.type;
-                             result.country_code := p.country_code;
-                             result.rank_address := p.rank_address;
-                           END;
-                           $$ LANGUAGE plpgsql STABLE;
-                        """)
-            cur.execute("""CREATE OR REPLACE FUNCTION
-                             get_interpolation_address(in_address HSTORE, wayid BIGINT)
-                           RETURNS HSTORE AS $$
-                           BEGIN
-                             RETURN in_address;
-                           END;
-                           $$ LANGUAGE plpgsql STABLE;
-                        """)
 
-            for table in ('placex', 'location_property_osmline', 'location_postcodes'):
+        for table in ('placex', 'location_property_osmline', 'location_postcodes'):
-                cur.execute("""CREATE TRIGGER {0}_update BEFORE UPDATE ON {0}
+            temp_db_conn.execute("""CREATE TRIGGER {0}_update BEFORE UPDATE ON {0}
-                               FOR EACH ROW EXECUTE PROCEDURE date_update()
+                                    FOR EACH ROW EXECUTE PROCEDURE date_update()
-                            """.format(table))
+                                 """.format(table))
+
+        self.tokenizer = factory.create_tokenizer(project_env)
 
     def scalar(self, query):
         with self.conn.cursor() as cur:
             cur.execute(query)
             return cur.fetchone()[0]
 
-    def add_place(self, cls='place', typ='locality',
-                  rank_search=30, rank_address=30, sector=20):
-        next_id = next(self.placex_id)
-        with self.conn.cursor() as cur:
-            cur.execute("""INSERT INTO placex
-                              (place_id, class, type, rank_search, rank_address,
-                               indexed_status, geometry_sector)
-                              VALUES (%s, %s, %s, %s, %s, 1, %s)""",
-                        (next_id, cls, typ, rank_search, rank_address, sector))
-        return next_id
-
-    def add_admin(self, **kwargs):
-        kwargs['cls'] = 'boundary'
-        kwargs['typ'] = 'administrative'
-        return self.add_place(**kwargs)
-
-    def add_osmline(self, sector=20):
-        next_id = next(self.osmline_id)
-        with self.conn.cursor() as cur:
-            cur.execute("""INSERT INTO location_property_osmline
-                              (place_id, osm_id, indexed_status, geometry_sector)
-                              VALUES (%s, %s, 1, %s)""",
-                        (next_id, next_id, sector))
-        return next_id
-
-    def add_postcode(self, country, postcode):
-        next_id = next(self.postcode_id)
-        with self.conn.cursor() as cur:
-            cur.execute("""INSERT INTO location_postcodes
-                            (place_id, indexed_status, country_code, postcode)
-                            VALUES (%s, 1, %s, %s)""",
-                        (next_id, country, postcode))
-        return next_id
-
     def placex_unindexed(self):
         return self.scalar('SELECT count(*) from placex where indexed_status > 0')
 
@@ -145,148 +72,133 @@ class IndexerTestDB:
         return self.scalar("""SELECT count(*) from location_property_osmline
                               WHERE indexed_status > 0""")
 
+    @pytest.mark.parametrize("threads", [1, 15])
+    @pytest.mark.asyncio
+    async def test_index_all_by_rank(self, dsn, threads, placex_row, osmline_row):
+        for rank in range(31):
+            placex_row(rank_address=rank, rank_search=rank, indexed_status=1)
+        osmline_row()
 
-@pytest.fixture
+        assert self.placex_unindexed() == 31
-def test_db(temp_db_conn):
+        assert self.osmline_unindexed() == 1
-    yield IndexerTestDB(temp_db_conn)
 
+        idx = indexer.Indexer(dsn, self.tokenizer, threads)
+        await idx.index_by_rank(0, 30)
 
-@pytest.fixture
+        assert self.placex_unindexed() == 0
-def test_tokenizer(tokenizer_mock, project_env):
+        assert self.osmline_unindexed() == 0
-    return factory.create_tokenizer(project_env)
 
+        assert self.scalar("""SELECT count(*) from placex
+                                 WHERE indexed_status = 0 and indexed_date is null""") == 0
+        # ranks come in order of rank address
+        assert self.scalar("""
+            SELECT count(*) FROM placex p WHERE rank_address > 0
+              AND indexed_date >= (SELECT min(indexed_date) FROM placex o
+                                   WHERE p.rank_address < o.rank_address)""") == 0
+        # placex address ranked objects come before interpolations
+        assert self.scalar(
+            """SELECT count(*) FROM placex WHERE rank_address > 0
+                 AND indexed_date >
+                       (SELECT min(indexed_date) FROM location_property_osmline)""") == 0
+        # rank 0 comes after all other placex objects
+        assert self.scalar(
+            """SELECT count(*) FROM placex WHERE rank_address > 0
+                 AND indexed_date >
+                       (SELECT min(indexed_date) FROM placex WHERE rank_address = 0)""") == 0
 
-@pytest.mark.parametrize("threads", [1, 15])
+    @pytest.mark.parametrize("threads", [1, 15])
-@pytest.mark.asyncio
+    @pytest.mark.asyncio
-async def test_index_all_by_rank(test_db, threads, test_tokenizer):
+    async def test_index_partial_without_30(self, dsn, threads, placex_row, osmline_row):
-    for rank in range(31):
+        for rank in range(31):
-        test_db.add_place(rank_address=rank, rank_search=rank)
+            placex_row(rank_address=rank, rank_search=rank, indexed_status=1)
-    test_db.add_osmline()
+        osmline_row()
 
-    assert test_db.placex_unindexed() == 31
+        assert self.placex_unindexed() == 31
-    assert test_db.osmline_unindexed() == 1
+        assert self.osmline_unindexed() == 1
 
-    idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
+        idx = indexer.Indexer(dsn, self.tokenizer, threads)
-    await idx.index_by_rank(0, 30)
+        await idx.index_by_rank(4, 15)
 
-    assert test_db.placex_unindexed() == 0
+        assert self.placex_unindexed() == 19
-    assert test_db.osmline_unindexed() == 0
+        assert self.osmline_unindexed() == 1
 
-    assert test_db.scalar("""SELECT count(*) from placex
+        assert self.scalar("""
-                             WHERE indexed_status = 0 and indexed_date is null""") == 0
+                        SELECT count(*) FROM placex
-    # ranks come in order of rank address
+                          WHERE indexed_status = 0 AND not rank_address between 4 and 15""") == 0
-    assert test_db.scalar("""
-        SELECT count(*) FROM placex p WHERE rank_address > 0
-          AND indexed_date >= (SELECT min(indexed_date) FROM placex o
-                               WHERE p.rank_address < o.rank_address)""") == 0
-    # placex address ranked objects come before interpolations
-    assert test_db.scalar(
-        """SELECT count(*) FROM placex WHERE rank_address > 0
-             AND indexed_date >
-                   (SELECT min(indexed_date) FROM location_property_osmline)""") == 0
-    # rank 0 comes after all other placex objects
-    assert test_db.scalar(
-        """SELECT count(*) FROM placex WHERE rank_address > 0
-             AND indexed_date >
-                   (SELECT min(indexed_date) FROM placex WHERE rank_address = 0)""") == 0
 
+    @pytest.mark.parametrize("threads", [1, 15])
+    @pytest.mark.asyncio
+    async def test_index_partial_with_30(self, dsn, threads, placex_row, osmline_row):
+        for rank in range(31):
+            placex_row(rank_address=rank, rank_search=rank, indexed_status=1)
+        osmline_row()
 
-@pytest.mark.parametrize("threads", [1, 15])
+        assert self.placex_unindexed() == 31
-@pytest.mark.asyncio
+        assert self.osmline_unindexed() == 1
-async def test_index_partial_without_30(test_db, threads, test_tokenizer):
-    for rank in range(31):
-        test_db.add_place(rank_address=rank, rank_search=rank)
-    test_db.add_osmline()
 
-    assert test_db.placex_unindexed() == 31
+        idx = indexer.Indexer(dsn, self.tokenizer, threads)
-    assert test_db.osmline_unindexed() == 1
+        await idx.index_by_rank(28, 30)
 
-    idx = indexer.Indexer('dbname=test_nominatim_python_unittest',
+        assert self.placex_unindexed() == 28
-                          test_tokenizer, threads)
+        assert self.osmline_unindexed() == 0
-    await idx.index_by_rank(4, 15)
 
-    assert test_db.placex_unindexed() == 19
+        assert self.scalar("""
-    assert test_db.osmline_unindexed() == 1
+                        SELECT count(*) FROM placex
+                          WHERE indexed_status = 0 AND rank_address between 0 and 27""") == 0
 
-    assert test_db.scalar("""
+    @pytest.mark.parametrize("threads", [1, 15])
-                    SELECT count(*) FROM placex
+    @pytest.mark.asyncio
-                      WHERE indexed_status = 0 AND not rank_address between 4 and 15""") == 0
+    async def test_index_boundaries(self, dsn, threads, placex_row, osmline_row):
+        for rank in range(4, 10):
+            placex_row(cls='boundary', typ='administrative',
+                       rank_address=rank, rank_search=rank, indexed_status=1)
+        for rank in range(31):
+            placex_row(rank_address=rank, rank_search=rank, indexed_status=1)
+        osmline_row()
 
+        assert self.placex_unindexed() == 37
+        assert self.osmline_unindexed() == 1
 
-@pytest.mark.parametrize("threads", [1, 15])
+        idx = indexer.Indexer(dsn, self.tokenizer, threads)
-@pytest.mark.asyncio
+        await idx.index_boundaries()
-async def test_index_partial_with_30(test_db, threads, test_tokenizer):
-    for rank in range(31):
-        test_db.add_place(rank_address=rank, rank_search=rank)
-    test_db.add_osmline()
 
-    assert test_db.placex_unindexed() == 31
+        assert self.placex_unindexed() == 31
-    assert test_db.osmline_unindexed() == 1
+        assert self.osmline_unindexed() == 1
 
-    idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
+        assert self.scalar("""
-    await idx.index_by_rank(28, 30)
+                        SELECT count(*) FROM placex
+                          WHERE indexed_status = 0 AND class != 'boundary'""") == 0
 
-    assert test_db.placex_unindexed() == 28
+    @pytest.mark.parametrize("threads", [1, 15])
-    assert test_db.osmline_unindexed() == 0
+    @pytest.mark.asyncio
+    async def test_index_postcodes(self, dsn, threads, postcode_row):
+        for postcode in range(1000):
+            postcode_row(country='de', postcode=postcode)
+        for postcode in range(32000, 33000):
+            postcode_row(country='us', postcode=postcode)
 
-    assert test_db.scalar("""
+        idx = indexer.Indexer(dsn, self.tokenizer, threads)
-                    SELECT count(*) FROM placex
+        await idx.index_postcodes()
-                      WHERE indexed_status = 0 AND rank_address between 0 and 27""") == 0
 
+        assert self.scalar("""SELECT count(*) FROM location_postcodes
+                                      WHERE indexed_status != 0""") == 0
 
-@pytest.mark.parametrize("threads", [1, 15])
+    @pytest.mark.parametrize("analyse", [True, False])
-@pytest.mark.asyncio
+    @pytest.mark.asyncio
-async def test_index_boundaries(test_db, threads, test_tokenizer):
+    async def test_index_full(self, dsn, analyse, placex_row, osmline_row, postcode_row):
-    for rank in range(4, 10):
+        for rank in range(4, 10):
-        test_db.add_admin(rank_address=rank, rank_search=rank)
+            placex_row(cls='boundary', typ='administrative',
-    for rank in range(31):
+                       rank_address=rank, rank_search=rank, indexed_status=1)
-        test_db.add_place(rank_address=rank, rank_search=rank)
+        for rank in range(31):
-    test_db.add_osmline()
+            placex_row(rank_address=rank, rank_search=rank, indexed_status=1)
+        osmline_row()
+        for postcode in range(1000):
+            postcode_row(country='de', postcode=postcode)
 
-    assert test_db.placex_unindexed() == 37
+        idx = indexer.Indexer(dsn, self.tokenizer, 4)
-    assert test_db.osmline_unindexed() == 1
+        await idx.index_full(analyse=analyse)
 
-    idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
+        assert self.placex_unindexed() == 0
-    await idx.index_boundaries()
+        assert self.osmline_unindexed() == 0
-
+        assert self.scalar("""SELECT count(*) FROM location_postcodes
-    assert test_db.placex_unindexed() == 31
+                                 WHERE indexed_status != 0""") == 0
-    assert test_db.osmline_unindexed() == 1
-
-    assert test_db.scalar("""
-                    SELECT count(*) FROM placex
-                      WHERE indexed_status = 0 AND class != 'boundary'""") == 0
-
-
-@pytest.mark.parametrize("threads", [1, 15])
-@pytest.mark.asyncio
-async def test_index_postcodes(test_db, threads, test_tokenizer):
-    for postcode in range(1000):
-        test_db.add_postcode('de', postcode)
-    for postcode in range(32000, 33000):
-        test_db.add_postcode('us', postcode)
-
-    idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, threads)
-    await idx.index_postcodes()
-
-    assert test_db.scalar("""SELECT count(*) FROM location_postcodes
-                                  WHERE indexed_status != 0""") == 0
-
-
-@pytest.mark.parametrize("analyse", [True, False])
-@pytest.mark.asyncio
-async def test_index_full(test_db, analyse, test_tokenizer):
-    for rank in range(4, 10):
-        test_db.add_admin(rank_address=rank, rank_search=rank)
-    for rank in range(31):
-        test_db.add_place(rank_address=rank, rank_search=rank)
-    test_db.add_osmline()
-    for postcode in range(1000):
-        test_db.add_postcode('de', postcode)
-
-    idx = indexer.Indexer('dbname=test_nominatim_python_unittest', test_tokenizer, 4)
-    await idx.index_full(analyse=analyse)
-
-    assert test_db.placex_unindexed() == 0
-    assert test_db.osmline_unindexed() == 0
-    assert test_db.scalar("""SELECT count(*) FROM location_postcodes
-                             WHERE indexed_status != 0""") == 0

test/python/mocks.py

@@ -1,85 +0,0 @@
-# SPDX-License-Identifier: GPL-3.0-or-later
-#
-# This file is part of Nominatim. (https://nominatim.org)
-#
-# Copyright (C) 2025 by the Nominatim developer community.
-# For a full list of authors see the git log.
-"""
-Custom mocks for testing.
-"""
-import itertools
-
-from nominatim_db.db import properties
-
-
-class MockPlacexTable:
-    """ A placex table for testing.
-    """
-    def __init__(self, conn):
-        self.idseq = itertools.count(10000)
-        self.conn = conn
-        with conn.cursor() as cur:
-            cur.execute("""CREATE TABLE placex (
-                               place_id BIGINT,
-                               parent_place_id BIGINT,
-                               linked_place_id BIGINT,
-                               importance FLOAT,
-                               indexed_date TIMESTAMP,
-                               geometry_sector INTEGER,
-                               rank_address SMALLINT,
-                               rank_search SMALLINT,
-                               partition SMALLINT,
-                               indexed_status SMALLINT,
-                               osm_id int8,
-                               osm_type char(1),
-                               class text,
-                               type text,
-                               name hstore,
-                               admin_level smallint,
-                               address hstore,
-                               extratags hstore,
-                               token_info jsonb,
-                               geometry Geometry(Geometry,4326),
-                               wikipedia TEXT,
-                               country_code varchar(2),
-                               housenumber TEXT,
-                               postcode TEXT,
-                               centroid GEOMETRY(Geometry, 4326))""")
-            cur.execute("CREATE SEQUENCE IF NOT EXISTS seq_place")
-        conn.commit()
-
-    def add(self, osm_type='N', osm_id=None, cls='amenity', typ='cafe', names=None,
-            admin_level=None, address=None, extratags=None, geom='POINT(10 4)',
-            country=None, housenumber=None, rank_search=30, centroid=None):
-        with self.conn.cursor() as cur:
-            cur.execute("""INSERT INTO placex (place_id, osm_type, osm_id, class,
-                                               type, name, admin_level, address,
-                                               housenumber, rank_search,
-                                               extratags, centroid, geometry, country_code)
-                            VALUES(nextval('seq_place'), %s, %s, %s, %s, %s, %s,
-                                   %s, %s, %s, %s, %s, %s, %s)
-                            RETURNING place_id""",
-                        (osm_type, osm_id or next(self.idseq), cls, typ, names,
-                         admin_level, address, housenumber, rank_search,
-                         extratags, centroid, 'SRID=4326;' + geom,
-                         country))
-            place_id = cur.fetchone()[0]
-        self.conn.commit()
-        return place_id
-
-
-class MockPropertyTable:
-    """ A property table for testing.
-    """
-    def __init__(self, conn):
-        self.conn = conn
-
-    def set(self, name, value):
-        """ Set a property in the table to the given value.
-        """
-        properties.set_property(self.conn, name, value)
-
-    def get(self, name):
-        """ Set a property in the table to the given value.
-        """
-        return properties.get_property(self.conn, name)

test/python/tokenizer/sanitizers/test_clean_housenumbers.py

@@ -2,7 +2,7 @@
 #
 # This file is part of Nominatim. (https://nominatim.org)
 #
-# Copyright (C) 2025 by the Nominatim developer community.
+# Copyright (C) 2026 by the Nominatim developer community.
 # For a full list of authors see the git log.
 """
 Tests for the sanitizer that normalizes housenumbers.
@@ -67,3 +67,25 @@ def test_convert_to_name_unconverted(def_config, number):
 
     assert 'housenumber' not in set(p.kind for p in names)
     assert ('housenumber', number) in set((p.kind, p.name) for p in address)
+
+
+@pytest.mark.parametrize('hnr,itype,out', [
+                            ('1-5', 'all', (1, 2, 3, 4, 5)),
+                            ('1-5', 'odd', (1, 3, 5)),
+                            ('1-5', 'even', (2, 4)),
+                            ('6-9', '1', (6, 7, 8, 9)),
+                            ('6-9', '2', (6, 8)),
+                            ('6-9', '3', (6, 9)),
+                            ('6-9', '5', (6,)),
+                            ('6-9', 'odd', (7, 9)),
+                            ('6-9', 'even', (6, 8)),
+                            ('6-22', 'even', (6, 8, 10, 12, 14, 16, 18, 20, 22))
+                            ])
+def test_convert_interpolations(sanitize, hnr, itype, out):
+    assert set(sanitize(housenumber=hnr, interpolation=itype)) \
+            == {('housenumber', str(i)) for i in out}
+
+
+@pytest.mark.parametrize('hnr', ('23', '23-', '3z-f', '1-10', '5-1', '1-4-5'))
+def test_ignore_interpolation_with_bad_housenumber(sanitize, hnr):
+    assert sanitize(housenumber=hnr, interpolation='all') == [('housenumber', hnr)]

test/python/tokenizer/test_icu.py

@@ -2,7 +2,7 @@
 #
 # This file is part of Nominatim. (https://nominatim.org)
 #
-# Copyright (C) 2025 by the Nominatim developer community.
+# Copyright (C) 2026 by the Nominatim developer community.
 # For a full list of authors see the git log.
 """
 Tests for ICU tokenizer.
@@ -15,7 +15,6 @@ import pytest
 from nominatim_db.tokenizer import icu_tokenizer
 import nominatim_db.tokenizer.icu_rule_loader
 from nominatim_db.db import properties
-from nominatim_db.db.sql_preprocessor import SQLPreprocessor
 from nominatim_db.data.place_info import PlaceInfo
 
 from mock_icu_word_table import MockIcuWordTable
@@ -90,13 +89,9 @@ def analyzer(tokenizer_factory, test_config, monkeypatch,
 
 
 @pytest.fixture
-def sql_functions(temp_db_conn, def_config, src_dir):
+def sql_functions(load_sql):
-    orig_sql = def_config.lib_dir.sql
+    load_sql('functions/utils.sql')
-    def_config.lib_dir.sql = src_dir / 'lib-sql'
+    load_sql('tokenizer/icu_tokenizer.sql')
-    sqlproc = SQLPreprocessor(temp_db_conn, def_config)
-    sqlproc.run_sql_file(temp_db_conn, 'functions/utils.sql')
-    sqlproc.run_sql_file(temp_db_conn, 'tokenizer/icu_tokenizer.sql')
-    def_config.lib_dir.sql = orig_sql
 
 
 @pytest.fixture
@@ -653,22 +648,21 @@ class TestUpdateWordTokens:
         self.tok.update_word_tokens()
         assert word_table.count_housenumbers() == 1
 
-    def test_keep_housenumbers_from_placex_table(self, add_housenumber, word_table,
+    def test_keep_housenumbers_from_placex_table(self, add_housenumber, word_table, placex_row):
-                                                 placex_table):
         add_housenumber(9999, '5432a')
         add_housenumber(9990, '34z')
-        placex_table.add(housenumber='34z')
+        placex_row(housenumber='34z')
-        placex_table.add(housenumber='25432a')
+        placex_row(housenumber='25432a')
 
         assert word_table.count_housenumbers() == 2
         self.tok.update_word_tokens()
         assert word_table.count_housenumbers() == 1
 
     def test_keep_housenumbers_from_placex_table_hnr_list(self, add_housenumber,
-                                                          word_table, placex_table):
+                                                          word_table, placex_row):
         add_housenumber(9991, '9 b')
         add_housenumber(9990, '34z')
-        placex_table.add(housenumber='9 a;9 b;9 c')
+        placex_row(housenumber='9 a;9 b;9 c')
 
         assert word_table.count_housenumbers() == 2
         self.tok.update_word_tokens()

test/python/tools/test_admin.py

@@ -2,17 +2,17 @@
 #
 # This file is part of Nominatim. (https://nominatim.org)
 #
-# Copyright (C) 2025 by the Nominatim developer community.
+# Copyright (C) 2026 by the Nominatim developer community.
 # For a full list of authors see the git log.
 """
 Tests for maintenance and analysis functions.
 """
 import pytest
+import datetime as dt
 
 from nominatim_db.errors import UsageError
 from nominatim_db.tools import admin
 from nominatim_db.tokenizer import factory
-from nominatim_db.db.sql_preprocessor import SQLPreprocessor
 
 
 @pytest.fixture(autouse=True)
@@ -61,15 +61,14 @@ def test_analyse_indexing_unknown_osmid(project_env):
         admin.analyse_indexing(project_env, osm_id='W12345674')
 
 
-def test_analyse_indexing_with_place_id(project_env, temp_db_cursor):
+def test_analyse_indexing_with_place_id(project_env, placex_row):
-    temp_db_cursor.execute("INSERT INTO placex (place_id) VALUES(12345)")
+    place_id = placex_row()
 
-    admin.analyse_indexing(project_env, place_id=12345)
+    admin.analyse_indexing(project_env, place_id=place_id)
 
 
-def test_analyse_indexing_with_osm_id(project_env, temp_db_cursor):
+def test_analyse_indexing_with_osm_id(project_env, placex_row):
-    temp_db_cursor.execute("""INSERT INTO placex (place_id, osm_type, osm_id)
+    placex_row(osm_type='N', osm_id=10000)
-                              VALUES(9988, 'N', 10000)""")
 
     admin.analyse_indexing(project_env, osm_id='N10000')
 
@@ -77,8 +76,8 @@ def test_analyse_indexing_with_osm_id(project_env, temp_db_cursor):
 class TestAdminCleanDeleted:
 
     @pytest.fixture(autouse=True)
-    def setup_polygon_delete(self, project_env, table_factory, place_table,
+    def setup_polygon_delete(self, project_env, table_factory, place_table, placex_row,
-                             osmline_table, temp_db_cursor, temp_db_conn, def_config, src_dir):
+                             osmline_table, temp_db_cursor, load_sql):
         """ Set up place_force_delete function and related tables
         """
         self.project_env = project_env
@@ -91,12 +90,15 @@ class TestAdminCleanDeleted:
                       ((100, 'N', 'boundary', 'administrative'),
                        (145, 'N', 'boundary', 'administrative'),
                        (175, 'R', 'landcover', 'grass')))
-        temp_db_cursor.execute("""
+
-            INSERT INTO placex (place_id, osm_id, osm_type, class, type,
+        now = dt.datetime.now()
-                                indexed_date, indexed_status)
+        placex_row(osm_type='N', osm_id=100, cls='boundary', typ='administrative',
-            VALUES(1, 100, 'N', 'boundary', 'administrative', current_date - INTERVAL '1 month', 1),
+                   indexed_status=1, indexed_date=now - dt.timedelta(days=30))
-                  (2, 145, 'N', 'boundary', 'administrative', current_date - INTERVAL '3 month', 1),
+        placex_row(osm_type='N', osm_id=145, cls='boundary', typ='administrative',
-                  (3, 175, 'R', 'landcover', 'grass', current_date - INTERVAL '3 months', 1)""")
+                   indexed_status=1, indexed_date=now - dt.timedelta(days=90))
+        placex_row(osm_type='R', osm_id=175, cls='landcover', typ='grass',
+                   indexed_status=1, indexed_date=now - dt.timedelta(days=90))
+
         # set up tables and triggers for utils function
         table_factory('place_to_be_deleted',
                       """osm_id BIGINT,
@@ -104,7 +106,6 @@ class TestAdminCleanDeleted:
                       class TEXT NOT NULL,
                       type TEXT NOT NULL,
                       deferred BOOLEAN""")
-        table_factory('country_name', 'partition INT')
         table_factory('import_polygon_error', """osm_id BIGINT,
                       osm_type CHAR(1),
                       class TEXT NOT NULL,
@@ -115,11 +116,7 @@ class TestAdminCleanDeleted:
                                $$ LANGUAGE plpgsql;""")
         temp_db_cursor.execute("""CREATE TRIGGER place_before_delete BEFORE DELETE ON place
                                FOR EACH ROW EXECUTE PROCEDURE place_delete();""")
-        orig_sql = def_config.lib_dir.sql
+        load_sql('functions/utils.sql')
-        def_config.lib_dir.sql = src_dir / 'lib-sql'
-        sqlproc = SQLPreprocessor(temp_db_conn, def_config)
-        sqlproc.run_sql_file(temp_db_conn, 'functions/utils.sql')
-        def_config.lib_dir.sql = orig_sql
 
     def test_admin_clean_deleted_no_records(self):
         admin.clean_deleted_relations(self.project_env, age='1 year')

test/python/tools/test_check_database.py

@@ -2,7 +2,7 @@
 #
 # This file is part of Nominatim. (https://nominatim.org)
 #
-# Copyright (C) 2025 by the Nominatim developer community.
+# Copyright (C) 2026 by the Nominatim developer community.
 # For a full list of authors see the git log.
 """
 Tests for database integrity checks.
@@ -46,8 +46,7 @@ def test_check_database_version_bad(property_table, temp_db_conn, def_config):
     assert chkdb.check_database_version(temp_db_conn, def_config) == chkdb.CheckState.FATAL
 
 
-def test_check_placex_table_good(table_factory, temp_db_conn, def_config):
+def test_check_placex_table_good(placex_table, temp_db_conn, def_config):
-    table_factory('placex')
     assert chkdb.check_placex_table(temp_db_conn, def_config) == chkdb.CheckState.OK
 
 
@@ -55,13 +54,13 @@ def test_check_placex_table_bad(temp_db_conn, def_config):
     assert chkdb.check_placex_table(temp_db_conn, def_config) == chkdb.CheckState.FATAL
 
 
-def test_check_placex_table_size_good(table_factory, temp_db_conn, def_config):
+def test_check_placex_table_size_good(placex_row, temp_db_conn, def_config):
-    table_factory('placex', content=((1, ), (2, )))
+    for _ in range(2):
+        placex_row()
     assert chkdb.check_placex_size(temp_db_conn, def_config) == chkdb.CheckState.OK
 
 
-def test_check_placex_table_size_bad(table_factory, temp_db_conn, def_config):
+def test_check_placex_table_size_bad(placex_table, temp_db_conn, def_config):
-    table_factory('placex')
     assert chkdb.check_placex_size(temp_db_conn, def_config) == chkdb.CheckState.FATAL
 
 
@@ -84,15 +83,22 @@ def test_check_tokenizer(temp_db_conn, def_config, monkeypatch,
     assert chkdb.check_tokenizer(temp_db_conn, def_config) == state
 
 
-def test_check_indexing_good(table_factory, temp_db_conn, def_config):
+def test_check_indexing_good(placex_row, temp_db_conn, def_config):
-    table_factory('placex', 'place_id int, indexed_status smallint',
+    for _ in range(2):
-                  content=((1, 0), (2, 0)))
+        placex_row(indexed_status=0)
     assert chkdb.check_indexing(temp_db_conn, def_config) == chkdb.CheckState.OK
 
 
-def test_check_indexing_bad(table_factory, temp_db_conn, def_config):
+def test_check_indexing_bad(placex_row, temp_db_conn, def_config):
-    table_factory('placex', 'place_id int, indexed_status smallint',
+    for status in (0, 2):
-                  content=((1, 0), (2, 2)))
+        placex_row(indexed_status=status)
+    assert chkdb.check_indexing(temp_db_conn, def_config) == chkdb.CheckState.FAIL
+
+
+def test_check_indexing_bad_frozen(placex_row, temp_db_conn, def_config):
+    for status in (0, 2):
+        placex_row(indexed_status=status)
+    temp_db_conn.execute('DROP TABLE place')
     assert chkdb.check_indexing(temp_db_conn, def_config) == chkdb.CheckState.WARN
 
 

test/python/tools/test_database_import.py

@@ -78,8 +78,8 @@ def test_setup_skeleton_already_exists(temp_db):
         database_import.setup_database_skeleton(f'dbname={temp_db}')
 
 
-def test_import_osm_data_simple(table_factory, osm2pgsql_options, capfd):
+def test_import_osm_data_simple(place_row, osm2pgsql_options, capfd):
-    table_factory('place', content=((1, ), ))
+    place_row()
 
     database_import.import_osm_data(Path('file.pbf'), osm2pgsql_options)
     captured = capfd.readouterr()
@@ -92,8 +92,8 @@ def test_import_osm_data_simple(table_factory, osm2pgsql_options, capfd):
     assert 'file.pbf' in captured.out
 
 
-def test_import_osm_data_multifile(table_factory, tmp_path, osm2pgsql_options, capfd):
+def test_import_osm_data_multifile(place_row, tmp_path, osm2pgsql_options, capfd):
-    table_factory('place', content=((1, ), ))
+    place_row()
     osm2pgsql_options['osm2pgsql_cache'] = 0
 
     files = [tmp_path / 'file1.osm', tmp_path / 'file2.osm']
@@ -107,22 +107,19 @@ def test_import_osm_data_multifile(table_factory, tmp_path, osm2pgsql_options, c
     assert 'file2.osm' in captured.out
 
 
-def test_import_osm_data_simple_no_data(table_factory, osm2pgsql_options):
+def test_import_osm_data_simple_no_data(place_row, osm2pgsql_options):
-    table_factory('place')
-
     with pytest.raises(UsageError, match='No data imported'):
         database_import.import_osm_data(Path('file.pbf'), osm2pgsql_options)
 
 
-def test_import_osm_data_simple_ignore_no_data(table_factory, osm2pgsql_options):
+def test_import_osm_data_simple_ignore_no_data(place_table, osm2pgsql_options):
-    table_factory('place')
-
     database_import.import_osm_data(Path('file.pbf'), osm2pgsql_options,
                                     ignore_errors=True)
 
 
-def test_import_osm_data_drop(table_factory, temp_db_cursor, tmp_path, osm2pgsql_options):
+def test_import_osm_data_drop(place_row, table_factory, temp_db_cursor,
-    table_factory('place', content=((1, ), ))
+                              tmp_path, osm2pgsql_options):
+    place_row()
     table_factory('planet_osm_nodes')
 
     flatfile = tmp_path / 'flatfile'
@@ -136,8 +133,8 @@ def test_import_osm_data_drop(table_factory, temp_db_cursor, tmp_path, osm2pgsql
     assert not temp_db_cursor.table_exists('planet_osm_nodes')
 
 
-def test_import_osm_data_default_cache(table_factory, osm2pgsql_options, capfd):
+def test_import_osm_data_default_cache(place_row, osm2pgsql_options, capfd):
-    table_factory('place', content=((1, ), ))
+    place_row()
 
     osm2pgsql_options['osm2pgsql_cache'] = 0
 
@@ -170,14 +167,41 @@ def test_truncate_database_tables(temp_db_conn, temp_db_cursor, table_factory, w
 @pytest.mark.asyncio
 async def test_load_data(dsn, place_row, placex_table, osmline_table,
                          temp_db_cursor, threads):
-    for func in ('precompute_words', 'getorcreate_housenumber_id', 'make_standard_name'):
-        temp_db_cursor.execute(pysql.SQL("""CREATE FUNCTION {} (src TEXT)
-                                            RETURNS TEXT AS $$ SELECT 'a'::TEXT $$ LANGUAGE SQL
-                                         """).format(pysql.Identifier(func)))
     for oid in range(100, 130):
         place_row(osm_id=oid)
     place_row(osm_type='W', osm_id=342, cls='place', typ='houses',
-              geom='SRID=4326;LINESTRING(0 0, 10 10)')
+              geom='LINESTRING(0 0, 10 10)')
+
+    temp_db_cursor.execute("""
+        CREATE OR REPLACE FUNCTION placex_insert() RETURNS TRIGGER AS $$
+        BEGIN
+          NEW.place_id := nextval('seq_place');
+          NEW.indexed_status := 1;
+          NEW.centroid := ST_Centroid(NEW.geometry);
+          NEW.partition := 0;
+          NEW.geometry_sector := 2424;
+          NEW.rank_address := 30;
+          NEW.rank_search := 30;
+        RETURN NEW;
+        END; $$ LANGUAGE plpgsql STABLE PARALLEL SAFE;
+
+        CREATE OR REPLACE FUNCTION osmline_insert() RETURNS TRIGGER AS $$
+        BEGIN
+          NEW.place_id := nextval('seq_place');
+          IF NEW.indexed_status IS NULL THEN
+            NEW.indexed_status := 1;
+            NEW.partition := 0;
+            NEW.geometry_sector := 2424;
+          END IF;
+        RETURN NEW;
+        END; $$ LANGUAGE plpgsql STABLE PARALLEL SAFE;
+
+        CREATE TRIGGER placex_before_insert BEFORE INSERT ON placex
+        FOR EACH ROW EXECUTE PROCEDURE placex_insert();
+
+        CREATE TRIGGER osmline_before_insert BEFORE INSERT ON location_property_osmline
+        FOR EACH ROW EXECUTE PROCEDURE osmline_insert();
+    """)
 
     await database_import.load_data(dsn, threads)
 
@@ -188,52 +212,53 @@ async def test_load_data(dsn, place_row, placex_table, osmline_table,
 class TestSetupSQL:
 
     @pytest.fixture(autouse=True)
-    def init_env(self, temp_db, tmp_path, def_config, sql_preprocessor_cfg):
+    def osm2ppsql_skel(self, def_config, temp_db_with_extensions, place_row,
-        def_config.lib_dir.sql = tmp_path / 'sql'
+                       country_table, table_factory, temp_db_conn):
-        def_config.lib_dir.sql.mkdir()
-
         self.config = def_config
+        place_row()
+        table_factory('osm2pgsql_properties', 'property TEXT, value TEXT',
+                      (('db_format', 2),))
 
-    def write_sql(self, fname, content):
+        table_factory('planet_osm_rels', 'id BIGINT, members JSONB, tags JSONB')
-        (self.config.lib_dir.sql / fname).write_text(content, encoding='utf-8')
+        temp_db_conn.execute("""
+            CREATE OR REPLACE FUNCTION planet_osm_member_ids(jsonb, character)
+              RETURNS bigint[] AS $$
+                SELECT array_agg((el->>'ref')::int8)
+                FROM jsonb_array_elements($1) AS el WHERE el->>'type' = $2
+              $$ LANGUAGE sql IMMUTABLE;
+            """)
 
     @pytest.mark.parametrize("reverse", [True, False])
-    def test_create_tables(self, temp_db_conn, temp_db_cursor, reverse):
+    def test_create_tables(self, table_factory, temp_db_conn, temp_db_cursor, reverse):
-        self.write_sql('tables.sql',
+        table_factory('country_osm_grid')
-                       """CREATE FUNCTION test() RETURNS bool
-                          AS $$ SELECT {{db.reverse_only}} $$ LANGUAGE SQL""")
-
-        self.write_sql('grants.sql', "-- Mock grants file for testing\n")
 
         database_import.create_tables(temp_db_conn, self.config, reverse)
 
-        temp_db_cursor.scalar('SELECT test()') == reverse
+        assert temp_db_cursor.table_exists('placex')
+        assert not reverse == temp_db_cursor.table_exists('search_name')
 
-    def test_create_table_triggers(self, temp_db_conn, temp_db_cursor):
+    def test_create_table_triggers(self, temp_db_conn, placex_table, osmline_table,
-        self.write_sql('table-triggers.sql',
+                                   postcode_table, load_sql):
-                       """CREATE FUNCTION test() RETURNS TEXT
+        load_sql('functions.sql')
-                          AS $$ SELECT 'a'::text $$ LANGUAGE SQL""")
 
         database_import.create_table_triggers(temp_db_conn, self.config)
 
-        temp_db_cursor.scalar('SELECT test()') == 'a'
+    def test_create_partition_tables(self, country_row, temp_db_conn, temp_db_cursor, load_sql):
-
+        for i in range(3):
-    def test_create_partition_tables(self, temp_db_conn, temp_db_cursor):
+            country_row(partition=i)
-        self.write_sql('partition-tables.src.sql',
+        load_sql('tables/location_area.sql')
-                       """CREATE FUNCTION test() RETURNS TEXT
-                          AS $$ SELECT 'b'::text $$ LANGUAGE SQL""")
 
         database_import.create_partition_tables(temp_db_conn, self.config)
 
-        temp_db_cursor.scalar('SELECT test()') == 'b'
+        for i in range(3):
+            assert temp_db_cursor.table_exists(f"location_area_large_{i}")
+            assert temp_db_cursor.table_exists(f"search_name_{i}")
 
     @pytest.mark.parametrize("drop", [True, False])
     @pytest.mark.asyncio
-    async def test_create_search_indices(self, temp_db_conn, temp_db_cursor, drop):
+    async def test_create_search_indices(self, temp_db_conn, temp_db_cursor, drop, load_sql):
-        self.write_sql('indices.sql',
+        load_sql('tables.sql', 'functions/ranking.sql')
-                       """CREATE FUNCTION test() RETURNS bool
-                          AS $$ SELECT {{drop}} $$ LANGUAGE SQL""")
-
         await database_import.create_search_indices(temp_db_conn, self.config, drop)
 
-        temp_db_cursor.scalar('SELECT test()') == drop
+        assert temp_db_cursor.index_exists('placex', 'idx_placex_geometry')
+        assert not drop == temp_db_cursor.index_exists('placex', 'idx_placex_geometry_buildings')

test/python/tools/test_import_special_phrases.py

@@ -2,7 +2,7 @@
 #
 # This file is part of Nominatim. (https://nominatim.org)
 #
-# Copyright (C) 2025 by the Nominatim developer community.
+# Copyright (C) 2026 by the Nominatim developer community.
 # For a full list of authors see the git log.
 """
     Tests for import special phrases methods
@@ -125,9 +125,8 @@ def test_grant_access_to_web_user(temp_db_conn, temp_db_cursor, table_factory,
                               phrase_class, phrase_type)
 
 
-def test_create_place_classtype_table_and_indexes(
+def test_create_place_classtype_table_and_indexes(temp_db_cursor, def_config, placex_row,
-        temp_db_cursor, def_config, placex_table,
+                                                  sp_importer, temp_db_conn, monkeypatch):
-        sp_importer, temp_db_conn, monkeypatch):
     """
         Test that _create_place_classtype_table_and_indexes()
         create the right place_classtype tables and place_id indexes
@@ -136,7 +135,7 @@ def test_create_place_classtype_table_and_indexes(
     """
     pairs = set([('class1', 'type1'), ('class2', 'type2')])
     for pair in pairs:
-        placex_table.add(cls=pair[0], typ=pair[1])   # adding to db
+        placex_row(cls=pair[0], typ=pair[1])   # adding to db
     sp_importer._create_classtype_table_and_indexes(pairs)
     temp_db_conn.commit()
 
@@ -178,7 +177,7 @@ def test_remove_non_existent_tables_from_db(sp_importer, default_phrases,
 
 @pytest.mark.parametrize("should_replace", [(True), (False)])
 def test_import_phrases(monkeypatch, temp_db_cursor, def_config, sp_importer,
-                        placex_table, table_factory, tokenizer_mock,
+                        placex_row, table_factory, tokenizer_mock,
                         xml_wiki_content, should_replace):
     """
         Check that the main import_phrases() method is well executed.
@@ -199,8 +198,8 @@ def test_import_phrases(monkeypatch, temp_db_cursor, def_config, sp_importer,
     type_test = 'zip_line'
 
     tokenizer = tokenizer_mock()
-    placex_table.add(cls=class_test, typ=type_test)  # in db for special phrase filtering
+    placex_row(cls=class_test, typ=type_test)  # in db for special phrase filtering
-    placex_table.add(cls='amenity', typ='animal_shelter')  # in db for special phrase filtering
+    placex_row(cls='amenity', typ='animal_shelter')  # in db for special phrase filtering
     sp_importer.import_phrases(tokenizer, should_replace)
 
     assert len(tokenizer.analyser_cache['special_phrases']) == 19
@@ -257,7 +256,7 @@ def check_placeid_and_centroid_indexes(temp_db_cursor, phrase_class, phrase_type
 
 @pytest.mark.parametrize("should_replace", [(True), (False)])
 def test_import_phrases_special_phrase_filtering(monkeypatch, temp_db_cursor, def_config,
-                                                 sp_importer, placex_table, tokenizer_mock,
+                                                 sp_importer, placex_row, tokenizer_mock,
                                                  xml_wiki_content, should_replace):
 
     monkeypatch.setattr('nominatim_db.tools.special_phrases.sp_wiki_loader._get_wiki_content',
@@ -266,7 +265,7 @@ def test_import_phrases_special_phrase_filtering(monkeypatch, temp_db_cursor, de
     class_test = 'aerialway'
     type_test = 'zip_line'
 
-    placex_table.add(cls=class_test, typ=type_test)  # add to the database to make valid
+    placex_row(cls=class_test, typ=type_test)  # add to the database to make valid
     tokenizer = tokenizer_mock()
     sp_importer.import_phrases(tokenizer, should_replace)
 
@@ -276,11 +275,11 @@ def test_import_phrases_special_phrase_filtering(monkeypatch, temp_db_cursor, de
     assert check_grant_access(temp_db_cursor, def_config.DATABASE_WEBUSER, class_test, type_test)
 
 
-def test_get_classtype_pairs_directly(placex_table, temp_db_conn, sp_importer):
+def test_get_classtype_pairs_directly(placex_row, temp_db_conn, sp_importer):
     for _ in range(101):
-        placex_table.add(cls='highway', typ='residential')
+        placex_row(cls='highway', typ='residential')
     for _ in range(99):
-        placex_table.add(cls='amenity', typ='toilet')
+        placex_row(cls='amenity', typ='toilet')
 
     temp_db_conn.commit()
 

test/python/tools/test_postcodes.py

@@ -2,7 +2,7 @@
 #
 # This file is part of Nominatim. (https://nominatim.org)
 #
-# Copyright (C) 2025 by the Nominatim developer community.
+# Copyright (C) 2026 by the Nominatim developer community.
 # For a full list of authors see the git log.
 """
 Tests for functions to maintain the artificial postcode table.
@@ -11,88 +11,27 @@ import subprocess
 
 import pytest
 
+from psycopg.rows import tuple_row
+
 from nominatim_db.tools import postcodes
 from nominatim_db.data import country_info
-from nominatim_db.db.sql_preprocessor import SQLPreprocessor
 
 import dummy_tokenizer
 
 
-class MockPostcodeTable:
-    """ A location_postcodes table for testing.
-    """
-    def __init__(self, conn, config):
-        self.conn = conn
-        SQLPreprocessor(conn, config).run_sql_file(conn, 'functions/postcode_triggers.sql')
-        with conn.cursor() as cur:
-            cur.execute("""CREATE TABLE location_postcodes (
-                               place_id BIGINT,
-                               osm_id BIGINT,
-                               parent_place_id BIGINT,
-                               rank_search SMALLINT,
-                               indexed_status SMALLINT,
-                               indexed_date TIMESTAMP,
-                               country_code varchar(2),
-                               postcode TEXT,
-                               geometry GEOMETRY(Geometry, 4326),
-                               centroid GEOMETRY(Point, 4326))""")
-            cur.execute("""CREATE OR REPLACE FUNCTION token_normalized_postcode(postcode TEXT)
-                           RETURNS TEXT AS $$ BEGIN RETURN postcode; END; $$ LANGUAGE plpgsql;
-
-                           CREATE OR REPLACE FUNCTION get_country_code(place geometry)
-                           RETURNS TEXT AS $$ BEGIN
-                           RETURN null;
-                           END; $$ LANGUAGE plpgsql;
-                        """)
-            cur.execute("""CREATE OR REPLACE FUNCTION expand_by_meters(geom GEOMETRY, meters FLOAT)
-                           RETURNS GEOMETRY AS $$
-                           SELECT ST_Envelope(ST_Buffer(geom::geography, meters, 1)::geometry)
-                           $$ LANGUAGE sql;""")
-
-        conn.commit()
-
-    def add(self, country, postcode, x, y):
-        with self.conn.cursor() as cur:
-            cur.execute(
-                """INSERT INTO location_postcodes
-                       (place_id, indexed_status, country_code, postcode, centroid, geometry)
-                     VALUES (nextval('seq_place'), 1, %(cc)s, %(pc)s,
-                             ST_SetSRID(ST_MakePoint(%(x)s, %(y)s), 4326),
-                             ST_Expand(ST_SetSRID(ST_MakePoint(%(x)s, %(y)s), 4326), 0.005))""",
-                {'cc': country, 'pc': postcode, 'x': x, 'y': y})
-
-        self.conn.commit()
-
-    @property
-    def row_set(self):
-        with self.conn.cursor() as cur:
-            cur.execute("""SELECT osm_id, country_code, postcode,
-                                  ST_X(centroid), ST_Y(centroid)
-                           FROM location_postcodes""")
-            return set((tuple(row) for row in cur))
-
-
 @pytest.fixture
-def postcode_table(def_config, temp_db_conn, placex_table, table_factory):
+def insert_implicit_postcode(placex_row, place_postcode_row):
-    country_info.setup_country_config(def_config)
-    table_factory('country_name', 'partition INT', ((0, ), (1, ), (2, )))
-    return MockPostcodeTable(temp_db_conn, def_config)
-
-
-@pytest.fixture
-def insert_implicit_postcode(placex_table, place_postcode_row):
     """ Insert data into the placex and place table
         which can then be used to compute one postcode.
     """
     def _insert_implicit_postcode(osm_id, country, geometry, postcode, in_placex=False):
         if in_placex:
-            placex_table.add(osm_id=osm_id, country=country, geom=geometry,
+            placex_row(osm_id=osm_id, country=country, geom=geometry,
-                             centroid=f'SRID=4326;{geometry}',
+                       centroid=geometry,
-                             address={'postcode': postcode})
+                       address={'postcode': postcode})
         else:
             place_postcode_row(osm_id=osm_id, centroid=geometry,
                                country=country, postcode=postcode)
-
     return _insert_implicit_postcode
 
 
@@ -105,7 +44,6 @@ def insert_postcode_area(place_postcode_row):
         place_postcode_row(osm_type='R', osm_id=osm_id, postcode=postcode, country=country,
                            centroid=f"POINT({x} {y})",
                            geom=f"POLYGON(({x1} {y1}, {x1} {y2}, {x2} {y2}, {x2} {y1}, {x1} {y1}))")
-
     return _do
 
 
@@ -125,186 +63,198 @@ def postcode_update(dsn, temp_db_conn):
                             BEFORE INSERT ON location_postcodes
                             FOR EACH ROW EXECUTE PROCEDURE postcodes_insert()""")
         temp_db_conn.commit()
-
         postcodes.update_postcodes(dsn, data_path, tokenizer)
-
     return _do
 
 
-def test_postcodes_empty(postcode_update, postcode_table, place_postcode_table):
+class TestPostcodes:
-    postcode_update()
+    @pytest.fixture(autouse=True)
+    def setup(self, def_config, postcode_table, placex_table, place_postcode_table,
+              load_sql, temp_db_conn):
+        self.conn = temp_db_conn
+        country_info.setup_country_config(def_config)
+        load_sql('functions/postcode_triggers.sql')
 
-    assert not postcode_table.row_set
+        temp_db_conn.execute("""
+            CREATE OR REPLACE FUNCTION token_normalized_postcode(postcode TEXT)
+            RETURNS TEXT AS $$
+              SELECT postcode
+            $$ LANGUAGE sql;
 
+            CREATE OR REPLACE FUNCTION get_country_code(place geometry)
+            RETURNS TEXT AS $$
+              SELECT NULL
+            $$ LANGUAGE sql;
 
-@pytest.mark.parametrize('in_placex', [True, False])
+            CREATE OR REPLACE FUNCTION expand_by_meters(geom GEOMETRY, meters FLOAT)
-def test_postcodes_add_new_point(postcode_update, postcode_table,
+            RETURNS GEOMETRY AS $$
-                                 insert_implicit_postcode, in_placex):
+              SELECT ST_Envelope(ST_Buffer(geom::geography, meters, 1)::geometry)
-    insert_implicit_postcode(1, 'xx', 'POINT(10 12)', '9486', in_placex)
+            $$ LANGUAGE sql;
-    postcode_table.add('yy', '9486', 99, 34)
+        """)
 
-    postcode_update()
+    @property
+    def row_set(self):
+        with self.conn.cursor(row_factory=tuple_row) as cur:
+            cur.execute("""SELECT osm_id, country_code, postcode,
+                                  ST_X(centroid), ST_Y(centroid)
+                           FROM location_postcodes""")
+            return {r for r in cur}
 
-    assert postcode_table.row_set == {(None, 'xx', '9486', 10, 12), }
+    def test_postcodes_empty(self, postcode_update):
+        postcode_update()
 
+        assert not self.row_set
 
-def test_postcodes_add_new_area(postcode_update, insert_postcode_area, postcode_table):
+    @pytest.mark.parametrize('in_placex', [True, False])
-    insert_postcode_area(345, 'de', '10445', 23.5, 46.2)
+    def test_postcodes_add_new_point(self, postcode_update, postcode_row,
+                                     insert_implicit_postcode, in_placex):
+        insert_implicit_postcode(1, 'xx', 'POINT(10 12)', '9486', in_placex)
+        postcode_row('yy', '9486', 99, 34)
 
-    postcode_update()
+        postcode_update()
 
-    assert postcode_table.row_set == {(345, 'de', '10445', 23.5, 46.2)}
+        assert self.row_set == {(None, 'xx', '9486', 10, 12), }
 
+    def test_postcodes_add_new_area(self, postcode_update, insert_postcode_area):
+        insert_postcode_area(345, 'de', '10445', 23.5, 46.2)
 
-@pytest.mark.parametrize('in_placex', [True, False])
+        postcode_update()
-def test_postcodes_add_area_and_point(postcode_update, insert_postcode_area,
-                                      insert_implicit_postcode, postcode_table, in_placex):
-    insert_implicit_postcode(1, 'xx', 'POINT(10 12)', '10445', in_placex)
-    insert_postcode_area(345, 'xx', '10445', 23.5, 46.2)
 
-    postcode_update()
+        assert self.row_set == {(345, 'de', '10445', 23.5, 46.2)}
 
-    assert postcode_table.row_set == {(345, 'xx', '10445', 23.5, 46.2)}
+    @pytest.mark.parametrize('in_placex', [True, False])
+    def test_postcodes_add_area_and_point(self, postcode_update, insert_postcode_area,
+                                          insert_implicit_postcode, in_placex):
+        insert_implicit_postcode(1, 'xx', 'POINT(10 12)', '10445', in_placex)
+        insert_postcode_area(345, 'xx', '10445', 23.5, 46.2)
 
+        postcode_update()
 
-@pytest.mark.parametrize('in_placex', [True, False])
+        assert self.row_set == {(345, 'xx', '10445', 23.5, 46.2)}
-def test_postcodes_add_point_within_area(postcode_update, insert_postcode_area,
-                                         insert_implicit_postcode, postcode_table, in_placex):
-    insert_implicit_postcode(1, 'xx', 'POINT(23.5 46.2)', '10446', in_placex)
-    insert_postcode_area(345, 'xx', '10445', 23.5, 46.2)
 
-    postcode_update()
+    @pytest.mark.parametrize('in_placex', [True, False])
+    def test_postcodes_add_point_within_area(self, postcode_update, insert_postcode_area,
+                                             insert_implicit_postcode, in_placex):
+        insert_implicit_postcode(1, 'xx', 'POINT(23.5 46.2)', '10446', in_placex)
+        insert_postcode_area(345, 'xx', '10445', 23.5, 46.2)
 
-    assert postcode_table.row_set == {(345, 'xx', '10445', 23.5, 46.2)}
+        postcode_update()
 
+        assert self.row_set == {(345, 'xx', '10445', 23.5, 46.2)}
 
-@pytest.mark.parametrize('coords', [(99, 34), (10, 34), (99, 12),
+    @pytest.mark.parametrize('coords', [(99, 34), (10, 34), (99, 12),
-                                    (9, 34), (9, 11), (23, 11)])
+                                        (9, 34), (9, 11), (23, 11)])
-def test_postcodes_replace_coordinates(postcode_update, postcode_table, tmp_path,
+    def test_postcodes_replace_coordinates(self, postcode_update, postcode_row, tmp_path,
-                                       insert_implicit_postcode, coords):
+                                           insert_implicit_postcode, coords):
-    insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
+        insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
-    postcode_table.add('xx', 'AB 4511', *coords)
+        postcode_row('xx', 'AB 4511', *coords)
 
-    postcode_update(tmp_path)
+        postcode_update(tmp_path)
 
-    assert postcode_table.row_set == {(None, 'xx', 'AB 4511', 10, 12)}
+        assert self.row_set == {(None, 'xx', 'AB 4511', 10, 12)}
 
+    def test_postcodes_replace_coordinates_close(self, postcode_update, postcode_row,
+                                                 insert_implicit_postcode):
+        insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
+        postcode_row('xx', 'AB 4511', 10, 11.99999999)
 
-def test_postcodes_replace_coordinates_close(postcode_update, postcode_table,
+        postcode_update()
-                                             insert_implicit_postcode):
-    insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
-    postcode_table.add('xx', 'AB 4511', 10, 11.99999999)
 
-    postcode_update()
+        assert self.row_set == {(None, 'xx', 'AB 4511', 10, 11.99999999)}
 
-    assert postcode_table.row_set == {(None, 'xx', 'AB 4511', 10, 11.99999999)}
+    def test_postcodes_remove_point(self, postcode_update, postcode_row,
+                                    insert_implicit_postcode):
+        insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
+        postcode_row('xx', 'badname', 10, 12)
 
+        postcode_update()
 
-def test_postcodes_remove_point(postcode_update, postcode_table,
+        assert self.row_set == {(None, 'xx', 'AB 4511', 10, 12)}
-                                insert_implicit_postcode):
-    insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
-    postcode_table.add('xx', 'badname', 10, 12)
 
-    postcode_update()
+    def test_postcodes_ignore_empty_country(self, postcode_update, insert_implicit_postcode):
+        insert_implicit_postcode(1, None, 'POINT(10 12)', 'AB 4511')
+        postcode_update()
+        assert not self.row_set
 
-    assert postcode_table.row_set == {(None, 'xx', 'AB 4511', 10, 12)}
+    def test_postcodes_remove_all(self, postcode_update, postcode_row, place_postcode_table):
+        postcode_row('ch', '5613', 10, 12)
+        postcode_update()
 
+        assert not self.row_set
 
-def test_postcodes_ignore_empty_country(postcode_update, postcode_table,
+    def test_postcodes_multi_country(self, postcode_update,
-                                        insert_implicit_postcode):
-    insert_implicit_postcode(1, None, 'POINT(10 12)', 'AB 4511')
-    postcode_update()
-    assert not postcode_table.row_set
-
-
-def test_postcodes_remove_all(postcode_update, postcode_table, place_postcode_table):
-    postcode_table.add('ch', '5613', 10, 12)
-    postcode_update()
-
-    assert not postcode_table.row_set
-
-
-def test_postcodes_multi_country(postcode_update, postcode_table,
-                                 insert_implicit_postcode):
-    insert_implicit_postcode(1, 'de', 'POINT(10 12)', '54451')
-    insert_implicit_postcode(2, 'cc', 'POINT(100 56)', 'DD23 T')
-    insert_implicit_postcode(3, 'de', 'POINT(10.3 11.0)', '54452')
-    insert_implicit_postcode(4, 'cc', 'POINT(10.3 11.0)', '54452')
-
-    postcode_update()
-
-    assert postcode_table.row_set == {(None, 'de', '54451', 10, 12),
-                                      (None, 'de', '54452', 10.3, 11.0),
-                                      (None, 'cc', '54452', 10.3, 11.0),
-                                      (None, 'cc', 'DD23 T', 100, 56)}
-
-
-@pytest.mark.parametrize("gzipped", [True, False])
-def test_postcodes_extern(postcode_update, postcode_table, tmp_path,
-                          insert_implicit_postcode, gzipped):
-    insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
-
-    extfile = tmp_path / 'xx_postcodes.csv'
-    extfile.write_text("postcode,lat,lon\nAB 4511,-4,-1\nCD 4511,-5, -10", encoding='utf-8')
-
-    if gzipped:
-        subprocess.run(['gzip', str(extfile)])
-        assert not extfile.is_file()
-
-    postcode_update(tmp_path)
-
-    assert postcode_table.row_set == {(None, 'xx', 'AB 4511', 10, 12),
-                                      (None, 'xx', 'CD 4511', -10, -5)}
-
-
-def test_postcodes_extern_bad_column(postcode_update, postcode_table, tmp_path,
                                      insert_implicit_postcode):
-    insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
+        insert_implicit_postcode(1, 'de', 'POINT(10 12)', '54451')
+        insert_implicit_postcode(2, 'cc', 'POINT(100 56)', 'DD23 T')
+        insert_implicit_postcode(3, 'de', 'POINT(10.3 11.0)', '54452')
+        insert_implicit_postcode(4, 'cc', 'POINT(10.3 11.0)', '54452')
 
-    extfile = tmp_path / 'xx_postcodes.csv'
+        postcode_update()
-    extfile.write_text("postode,lat,lon\nAB 4511,-4,-1\nCD 4511,-5, -10", encoding='utf-8')
 
-    postcode_update(tmp_path)
+        assert self.row_set == {(None, 'de', '54451', 10, 12),
+                                (None, 'de', '54452', 10.3, 11.0),
+                                (None, 'cc', '54452', 10.3, 11.0),
+                                (None, 'cc', 'DD23 T', 100, 56)}
 
-    assert postcode_table.row_set == {(None, 'xx', 'AB 4511', 10, 12)}
+    @pytest.mark.parametrize("gzipped", [True, False])
+    def test_postcodes_extern(self, postcode_update, tmp_path,
+                              insert_implicit_postcode, gzipped):
+        insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
 
+        extfile = tmp_path / 'xx_postcodes.csv'
+        extfile.write_text("postcode,lat,lon\nAB 4511,-4,-1\nCD 4511,-5, -10", encoding='utf-8')
 
-def test_postcodes_extern_bad_number(postcode_update, insert_implicit_postcode,
+        if gzipped:
-                                     postcode_table, tmp_path):
+            subprocess.run(['gzip', str(extfile)])
-    insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
+            assert not extfile.is_file()
 
-    extfile = tmp_path / 'xx_postcodes.csv'
+        postcode_update(tmp_path)
-    extfile.write_text(
-        "postcode,lat,lon\nXX 4511,-4,NaN\nCD 4511,-5, -10\n34,200,0", encoding='utf-8')
 
-    postcode_update(tmp_path)
+        assert self.row_set == {(None, 'xx', 'AB 4511', 10, 12),
+                                (None, 'xx', 'CD 4511', -10, -5)}
 
-    assert postcode_table.row_set == {(None, 'xx', 'AB 4511', 10, 12),
+    def test_postcodes_extern_bad_column(self, postcode_update, tmp_path,
-                                      (None, 'xx', 'CD 4511', -10, -5)}
+                                         insert_implicit_postcode):
+        insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
+
+        extfile = tmp_path / 'xx_postcodes.csv'
+        extfile.write_text("postode,lat,lon\nAB 4511,-4,-1\nCD 4511,-5, -10", encoding='utf-8')
+
+        postcode_update(tmp_path)
+
+        assert self.row_set == {(None, 'xx', 'AB 4511', 10, 12)}
+
+    def test_postcodes_extern_bad_number(self, postcode_update, insert_implicit_postcode,
+                                         tmp_path):
+        insert_implicit_postcode(1, 'xx', 'POINT(10 12)', 'AB 4511')
+
+        extfile = tmp_path / 'xx_postcodes.csv'
+        extfile.write_text(
+            "postcode,lat,lon\nXX 4511,-4,NaN\nCD 4511,-5, -10\n34,200,0", encoding='utf-8')
+
+        postcode_update(tmp_path)
+
+        assert self.row_set == {(None, 'xx', 'AB 4511', 10, 12),
+                                (None, 'xx', 'CD 4511', -10, -5)}
+
+    def test_no_placex_entry(self, postcode_update, temp_db_cursor, place_postcode_row):
+        # Rewrite the get_country_code function to verify its execution.
+        temp_db_cursor.execute("""
+            CREATE OR REPLACE FUNCTION get_country_code(place geometry) RETURNS TEXT AS $$
+              SELECT 'yy' $$ LANGUAGE sql""")
+        place_postcode_row(centroid='POINT(10 12)', postcode='AB 4511')
+        postcode_update()
+
+        assert self.row_set == {(None, 'yy', 'AB 4511', 10, 12)}
+
+    def test_discard_badly_formatted_postcodes(self, postcode_update, place_postcode_row):
+        place_postcode_row(centroid='POINT(10 12)', country='fr', postcode='AB 4511')
+        postcode_update()
+
+        assert not self.row_set
 
 
 def test_can_compute(dsn, table_factory):
     assert not postcodes.can_compute(dsn)
     table_factory('place_postcode')
     assert postcodes.can_compute(dsn)
-
-
-def test_no_placex_entry(postcode_update, temp_db_cursor, place_postcode_row, postcode_table):
-    # Rewrite the get_country_code function to verify its execution.
-    temp_db_cursor.execute("""
-        CREATE OR REPLACE FUNCTION get_country_code(place geometry)
-        RETURNS TEXT AS $$ BEGIN
-        RETURN 'yy';
-        END; $$ LANGUAGE plpgsql;
-    """)
-    place_postcode_row(centroid='POINT(10 12)', postcode='AB 4511')
-    postcode_update()
-
-    assert postcode_table.row_set == {(None, 'yy', 'AB 4511', 10, 12)}
-
-
-def test_discard_badly_formatted_postcodes(postcode_update, place_postcode_row, postcode_table):
-    place_postcode_row(centroid='POINT(10 12)', country='fr', postcode='AB 4511')
-    postcode_update()
-
-    assert not postcode_table.row_set

test/python/tools/test_refresh.py

@@ -42,8 +42,8 @@ def test_refresh_import_wikipedia(dsn, src_dir, table_factory, temp_db_cursor, r
 
 
 @pytest.mark.parametrize('osm_type', ('N', 'W', 'R'))
-def test_invalidate_osm_object_simple(placex_table, osm_type, temp_db_conn, temp_db_cursor):
+def test_invalidate_osm_object_simple(placex_row, osm_type, temp_db_conn, temp_db_cursor):
-    placex_table.add(osm_type=osm_type, osm_id=57283)
+    placex_row(osm_type=osm_type, osm_id=57283)
 
     refresh.invalidate_osm_object(osm_type, 57283, temp_db_conn, recursive=False)
     temp_db_conn.commit()
@@ -53,8 +53,8 @@ def test_invalidate_osm_object_simple(placex_table, osm_type, temp_db_conn, temp
                                       (osm_type, 57283))
 
 
-def test_invalidate_osm_object_nonexisting_simple(placex_table, temp_db_conn, temp_db_cursor):
+def test_invalidate_osm_object_nonexisting_simple(placex_row, temp_db_conn, temp_db_cursor):
-    placex_table.add(osm_type='W', osm_id=57283)
+    placex_row(osm_type='W', osm_id=57283)
 
     refresh.invalidate_osm_object('N', 57283, temp_db_conn, recursive=False)
     temp_db_conn.commit()
@@ -64,8 +64,8 @@ def test_invalidate_osm_object_nonexisting_simple(placex_table, temp_db_conn, te
 
 
 @pytest.mark.parametrize('osm_type', ('N', 'W', 'R'))
-def test_invalidate_osm_object_recursive(placex_table, osm_type, temp_db_conn, temp_db_cursor):
+def test_invalidate_osm_object_recursive(placex_row, osm_type, temp_db_conn, temp_db_cursor):
-    placex_table.add(osm_type=osm_type, osm_id=57283)
+    placex_row(osm_type=osm_type, osm_id=57283)
 
     temp_db_cursor.execute("""CREATE OR REPLACE FUNCTION place_force_update(placeid BIGINT)
                               RETURNS BOOLEAN AS $$

test/python/tools/test_refresh_wiki_data.py

@@ -2,7 +2,7 @@
 #
 # This file is part of Nominatim. (https://nominatim.org)
 #
-# Copyright (C) 2025 by the Nominatim developer community.
+# Copyright (C) 2026 by the Nominatim developer community.
 # For a full list of authors see the git log.
 """
 Tests for correctly assigning wikipedia pages to places.
@@ -38,7 +38,7 @@ def wiki_csv(tmp_path, sql_preprocessor):
                                    {'wikipedia': 'en:Test'},
                                    {'wikidata': 'Q123'}])
 def test_wikipedia(dsn, temp_db_conn, temp_db_cursor, table_factory,
-                   def_config, wiki_csv, placex_table, extra):
+                   def_config, wiki_csv, placex_row, extra):
     import_wikipedia_articles(dsn, wiki_csv([('en', 'Test', 0.3, 'Q123')]))
     create_functions(temp_db_conn, def_config)
 
@@ -46,7 +46,7 @@ def test_wikipedia(dsn, temp_db_conn, temp_db_cursor, table_factory,
         'SELECT language, title, importance, wikidata FROM wikimedia_importance')
     assert content == set([('en', 'Test', 0.3, 'Q123')])
 
-    place_id = placex_table.add(osm_id=12, extratags=extra)
+    place_id = placex_row(osm_id=12, extratags=extra)
     table_factory('search_name',
                   'place_id BIGINT, importance FLOAT',
                   [(place_id, 0.2)])
@@ -61,11 +61,11 @@ def test_wikipedia(dsn, temp_db_conn, temp_db_cursor, table_factory,
 
 
 def test_wikipedia_no_match(dsn, temp_db_conn, temp_db_cursor, def_config, wiki_csv,
-                            placex_table, table_factory):
+                            placex_row, table_factory):
     import_wikipedia_articles(dsn, wiki_csv([('de', 'Test', 0.3, 'Q123')]))
     create_functions(temp_db_conn, def_config)
 
-    place_id = placex_table.add(osm_id=12, extratags={'wikipedia': 'en:Test'}, rank_search=10)
+    place_id = placex_row(osm_id=12, extratags={'wikipedia': 'en:Test'}, rank_search=10)
     table_factory('search_name',
                   'place_id BIGINT, importance FLOAT',
                   [(place_id, 0.2)])

test/python/tools/test_sp_importer.py

@@ -1,69 +1,70 @@
+# SPDX-License-Identifier: GPL-3.0-or-later
+#
+# This file is part of Nominatim. (https://nominatim.org)
+#
+# Copyright (C) 2026 by the Nominatim developer community.
+# For a full list of authors see the git log.
+
 from nominatim_db.tools.special_phrases.sp_importer import SPImporter
 
 
 # Testing Database Class Pair Retrival using Conftest.py and placex
-def test_get_classtype_pair_data(placex_table, def_config, temp_db_conn):
+def test_get_classtype_pair_data(placex_row, def_config, temp_db_conn):
     for _ in range(100):
-        placex_table.add(cls='highway', typ='motorway')  # edge case 100
+        placex_row(cls='highway', typ='motorway')  # edge case 100
 
     for _ in range(99):
-        placex_table.add(cls='amenity', typ='prison')  # edge case 99
+        placex_row(cls='amenity', typ='prison')  # edge case 99
 
     for _ in range(150):
-        placex_table.add(cls='tourism', typ='hotel')
+        placex_row(cls='tourism', typ='hotel')
 
     importer = SPImporter(config=def_config, conn=temp_db_conn, sp_loader=None)
 
     result = importer.get_classtype_pairs(min=100)
 
-    expected = {
+    assert result == {
         ("highway", "motorway"),
         ("tourism", "hotel")
     }
 
-    assert result == expected, f"Expected {expected}, got {result}"
 
-
+def test_get_classtype_pair_data_more(placex_row, def_config, temp_db_conn):
-def test_get_classtype_pair_data_more(placex_table, def_config, temp_db_conn):
     for _ in range(99):
-        placex_table.add(cls='emergency', typ='firehydrant')  # edge case 99, not included
+        placex_row(cls='emergency', typ='firehydrant')  # edge case 99, not included
 
     for _ in range(199):
-        placex_table.add(cls='amenity', typ='prison')
+        placex_row(cls='amenity', typ='prison')
 
     for _ in range(3478):
-        placex_table.add(cls='tourism', typ='hotel')
+        placex_row(cls='tourism', typ='hotel')
 
     importer = SPImporter(config=def_config, conn=temp_db_conn, sp_loader=None)
 
     result = importer.get_classtype_pairs(min=100)
 
-    expected = {
+    assert result == {
         ("amenity", "prison"),
         ("tourism", "hotel")
     }
 
-    assert result == expected, f"Expected {expected}, got {result}"
 
-
+def test_get_classtype_pair_data_default(placex_row, def_config, temp_db_conn):
-def test_get_classtype_pair_data_default(placex_table, def_config, temp_db_conn):
     for _ in range(1):
-        placex_table.add(cls='emergency', typ='firehydrant')
+        placex_row(cls='emergency', typ='firehydrant')
 
     for _ in range(199):
-        placex_table.add(cls='amenity', typ='prison')
+        placex_row(cls='amenity', typ='prison')
 
     for _ in range(3478):
-        placex_table.add(cls='tourism', typ='hotel')
+        placex_row(cls='tourism', typ='hotel')
 
     importer = SPImporter(config=def_config, conn=temp_db_conn, sp_loader=None)
 
     result = importer.get_classtype_pairs()
 
-    expected = {
+    assert result == {
         ("amenity", "prison"),
         ("tourism", "hotel"),
         ("emergency", "firehydrant")
     }
-
-    assert result == expected, f"Expected {expected}, got {result}"

test/python/tools/test_tiger_data.py

@@ -2,7 +2,7 @@
 #
 # This file is part of Nominatim. (https://nominatim.org)
 #
-# Copyright (C) 2025 by the Nominatim developer community.
+# Copyright (C) 2026 by the Nominatim developer community.
 # For a full list of authors see the git log.
 """
 Test for tiger data function
@@ -13,59 +13,10 @@ from textwrap import dedent
 import pytest
 import pytest_asyncio  # noqa: F401
 
-from nominatim_db.db.connection import execute_scalar
+from nominatim_db.tools import tiger_data
-from nominatim_db.tools import tiger_data, freeze
 from nominatim_db.errors import UsageError
 
 
-class MockTigerTable:
-
-    def __init__(self, conn):
-        self.conn = conn
-        with conn.cursor() as cur:
-            cur.execute("""CREATE TABLE tiger (linegeo GEOMETRY,
-                                               start INTEGER,
-                                               stop INTEGER,
-                                               interpol TEXT,
-                                               token_info JSONB,
-                                               postcode TEXT)""")
-
-            # We need this table to determine if the database is frozen or not
-            cur.execute("CREATE TABLE place (number INTEGER)")
-            # We need this table to determine if the database is in reverse-only mode
-            cur.execute("CREATE TABLE search_name (place_id BIGINT)")
-
-    def count(self):
-        return execute_scalar(self.conn, "SELECT count(*) FROM tiger")
-
-    def row(self):
-        with self.conn.cursor() as cur:
-            cur.execute("SELECT * FROM tiger LIMIT 1")
-            return cur.fetchone()
-
-
-@pytest.fixture
-def tiger_table(def_config, temp_db_conn, sql_preprocessor,
-                temp_db_with_extensions, tmp_path):
-    def_config.lib_dir.sql = tmp_path / 'sql'
-    def_config.lib_dir.sql.mkdir()
-
-    (def_config.lib_dir.sql / 'tiger_import_start.sql').write_text(
-        """CREATE OR REPLACE FUNCTION tiger_line_import(linegeo GEOMETRY, start INTEGER,
-                                                        stop INTEGER, interpol TEXT,
-                                                        token_info JSONB, postcode TEXT)
-           RETURNS INTEGER AS $$
-            INSERT INTO tiger VALUES(linegeo, start, stop, interpol, token_info, postcode)
-            RETURNING 1
-           $$ LANGUAGE SQL;""", encoding='utf-8')
-    (def_config.lib_dir.sql / 'tiger_import_finish.sql').write_text(
-        """DROP FUNCTION tiger_line_import (linegeo GEOMETRY, in_startnumber INTEGER,
-                                 in_endnumber INTEGER, interpolationtype TEXT,
-                                 token_info JSONB, in_postcode TEXT);""", encoding='utf-8')
-
-    return MockTigerTable(temp_db_conn)
-
-
 @pytest.fixture
 def csv_factory(tmp_path):
     def _mk_file(fname, hnr_from=1, hnr_to=9, interpol='odd', street='Main St',
@@ -80,107 +31,110 @@ def csv_factory(tmp_path):
     return _mk_file
 
 
-@pytest.mark.parametrize("threads", (1, 5))
+class TestTiger:
-@pytest.mark.asyncio
-async def test_add_tiger_data(def_config, src_dir, tiger_table, tokenizer_mock, threads):
-    await tiger_data.add_tiger_data(str(src_dir / 'test' / 'testdb' / 'tiger'),
-                                    def_config, threads, tokenizer_mock())
 
-    assert tiger_table.count() == 6213
+    @pytest.fixture(autouse=True)
+    def setup(self, temp_db_conn, placex_row, load_sql):
+        load_sql('tables/search_name.sql', create_reverse_only=False)
+        load_sql('tables/tiger.sql')
 
+        # fake parent roads
+        for x in range(-870, -863):
+            for y in range(323, 328):
+                placex_row(rank_search=26, rank_address=26,
+                           geom=f"LINESTRING({x/10 - 0.1} {y/10}, {x/10 + 0.1} {y/10})")
 
-@pytest.mark.parametrize("threads", (1, 5))
+        temp_db_conn.execute("""
-@pytest.mark.asyncio
+            CREATE OR REPLACE FUNCTION get_partition(cc VARCHAR(10)) RETURNS INTEGER AS $$
-async def test_add_tiger_data_database_frozen(def_config, src_dir, temp_db_conn, tiger_table,
+              SELECT 0;
-                                              tokenizer_mock, threads):
+            $$ LANGUAGE sql;
-    freeze.drop_update_tables(temp_db_conn)
+            CREATE OR REPLACE FUNCTION token_matches_street(i JSONB, s INT[]) RETURNS BOOLEAN AS $$
+             SELECT false
+            $$ LANGUAGE SQL IMMUTABLE STRICT PARALLEL SAFE;
+        """)
 
-    await tiger_data.add_tiger_data(str(src_dir / 'test' / 'testdb' / 'tiger'),
+    @pytest.mark.parametrize("threads", (1, 5))
-                                    def_config, threads, tokenizer_mock())
+    @pytest.mark.asyncio
-
+    async def test_add_tiger_data_database_frozen(self, def_config, src_dir, temp_db_cursor,
-    assert tiger_table.count() == 6213
+                                                  tokenizer_mock, threads):
-
-
-@pytest.mark.asyncio
-async def test_add_tiger_data_reverse_only(def_config, src_dir, temp_db_conn, tiger_table,
-                                           tokenizer_mock):
-    with temp_db_conn.cursor() as cur:
-        cur.execute("DROP TABLE search_name")
-    temp_db_conn.commit()
-
-    with pytest.raises(UsageError,
-                       match="Cannot perform tiger import: required tables are missing. "
-                       "See https://github.com/osm-search/Nominatim/issues/2463 for details."):
         await tiger_data.add_tiger_data(str(src_dir / 'test' / 'testdb' / 'tiger'),
-                                        def_config, 1, tokenizer_mock())
+                                        def_config, threads, tokenizer_mock())
 
-    assert tiger_table.count() == 0
+        assert temp_db_cursor.table_rows('location_property_tiger') == 6209
 
+    @pytest.mark.asyncio
+    async def test_add_tiger_data_reverse_only(self, def_config, src_dir, temp_db_cursor,
+                                               tokenizer_mock):
+        temp_db_cursor.execute("DROP TABLE search_name")
 
-@pytest.mark.asyncio
+        with pytest.raises(UsageError,
-async def test_add_tiger_data_no_files(def_config, tiger_table, tokenizer_mock,
+                           match="Cannot perform tiger import: required tables are missing. "
-                                       tmp_path):
+                           "See https://github.com/osm-search/Nominatim/issues/2463 for details."):
-    await tiger_data.add_tiger_data(str(tmp_path), def_config, 1, tokenizer_mock())
+            await tiger_data.add_tiger_data(str(src_dir / 'test' / 'testdb' / 'tiger'),
+                                            def_config, 1, tokenizer_mock())
 
-    assert tiger_table.count() == 0
+        assert temp_db_cursor.table_rows('location_property_tiger') == 0
 
+    @pytest.mark.asyncio
+    async def test_add_tiger_data_no_files(self, def_config, temp_db_cursor, tokenizer_mock,
+                                           tmp_path):
+        await tiger_data.add_tiger_data(str(tmp_path), def_config, 1, tokenizer_mock())
 
-@pytest.mark.asyncio
+        assert temp_db_cursor.table_rows('location_property_tiger') == 0
-async def test_add_tiger_data_bad_file(def_config, tiger_table, tokenizer_mock,
-                                       tmp_path):
-    sqlfile = tmp_path / '1010.csv'
-    sqlfile.write_text("""Random text""", encoding='utf-8')
 
-    await tiger_data.add_tiger_data(str(tmp_path), def_config, 1, tokenizer_mock())
+    @pytest.mark.asyncio
+    async def test_add_tiger_data_bad_file(self, def_config, temp_db_cursor, tokenizer_mock,
+                                           tmp_path):
+        sqlfile = tmp_path / '1010.csv'
+        sqlfile.write_text('Random text', encoding='utf-8')
 
-    assert tiger_table.count() == 0
+        await tiger_data.add_tiger_data(str(tmp_path), def_config, 1, tokenizer_mock())
 
+        assert temp_db_cursor.table_rows('location_property_tiger') == 0
 
-@pytest.mark.asyncio
+    @pytest.mark.asyncio
-async def test_add_tiger_data_hnr_nan(def_config, tiger_table, tokenizer_mock,
+    async def test_add_tiger_data_hnr_nan(self, def_config, temp_db_cursor, tokenizer_mock,
-                                      csv_factory, tmp_path):
+                                          csv_factory, tmp_path):
-    csv_factory('file1', hnr_from=99)
+        csv_factory('file1', hnr_to=99)
-    csv_factory('file2', hnr_from='L12')
+        csv_factory('file2', hnr_from='L12')
-    csv_factory('file3', hnr_to='12.4')
+        csv_factory('file3', hnr_to='12.4')
 
-    await tiger_data.add_tiger_data(str(tmp_path), def_config, 1, tokenizer_mock())
+        await tiger_data.add_tiger_data(str(tmp_path), def_config, 1, tokenizer_mock())
 
-    assert tiger_table.count() == 1
+        rows = temp_db_cursor.row_set("""
-    assert tiger_table.row().start == 99
+            SELECT startnumber, endnumber FROM location_property_tiger""")
 
+        assert rows == {(1, 99)}
 
-@pytest.mark.parametrize("threads", (1, 5))
+    @pytest.mark.parametrize("threads", (1, 5))
-@pytest.mark.asyncio
+    @pytest.mark.asyncio
-async def test_add_tiger_data_tarfile(def_config, tiger_table, tokenizer_mock,
+    async def test_add_tiger_data_tarfile(self, def_config, temp_db_cursor, tokenizer_mock,
-                                      tmp_path, src_dir, threads):
+                                          tmp_path, src_dir, threads):
-    tar = tarfile.open(str(tmp_path / 'sample.tar.gz'), "w:gz")
+        tar = tarfile.open(str(tmp_path / 'sample.tar.gz'), "w:gz")
-    tar.add(str(src_dir / 'test' / 'testdb' / 'tiger' / '01001.csv'))
+        tar.add(str(src_dir / 'test' / 'testdb' / 'tiger' / '01001.csv'))
-    tar.close()
+        tar.close()
 
-    await tiger_data.add_tiger_data(str(tmp_path / 'sample.tar.gz'), def_config, threads,
+        await tiger_data.add_tiger_data(str(tmp_path / 'sample.tar.gz'), def_config, threads,
-                                    tokenizer_mock())
+                                        tokenizer_mock())
 
-    assert tiger_table.count() == 6213
+        assert temp_db_cursor.table_rows('location_property_tiger') == 6209
 
+    @pytest.mark.asyncio
+    async def test_add_tiger_data_bad_tarfile(self, def_config, tokenizer_mock, tmp_path):
+        tarfile = tmp_path / 'sample.tar.gz'
+        tarfile.write_text("""Random text""", encoding='utf-8')
 
-@pytest.mark.asyncio
+        with pytest.raises(UsageError):
-async def test_add_tiger_data_bad_tarfile(def_config, tiger_table, tokenizer_mock,
+            await tiger_data.add_tiger_data(str(tarfile), def_config, 1, tokenizer_mock())
-                                          tmp_path):
-    tarfile = tmp_path / 'sample.tar.gz'
-    tarfile.write_text("""Random text""", encoding='utf-8')
 
-    with pytest.raises(UsageError):
+    @pytest.mark.asyncio
-        await tiger_data.add_tiger_data(str(tarfile), def_config, 1, tokenizer_mock())
+    async def test_add_tiger_data_empty_tarfile(self, def_config, temp_db_cursor, tokenizer_mock,
+                                                tmp_path):
+        tar = tarfile.open(str(tmp_path / 'sample.tar.gz'), "w:gz")
+        tar.add(__file__)
+        tar.close()
 
+        await tiger_data.add_tiger_data(str(tmp_path / 'sample.tar.gz'), def_config, 1,
+                                        tokenizer_mock())
 
-@pytest.mark.asyncio
+        assert temp_db_cursor.table_rows('location_property_tiger') == 0
-async def test_add_tiger_data_empty_tarfile(def_config, tiger_table, tokenizer_mock,
-                                            tmp_path):
-    tar = tarfile.open(str(tmp_path / 'sample.tar.gz'), "w:gz")
-    tar.add(__file__)
-    tar.close()
-
-    await tiger_data.add_tiger_data(str(tmp_path / 'sample.tar.gz'), def_config, 1,
-                                    tokenizer_mock())
-
-    assert tiger_table.count() == 0