have an identifier of the form '<NRW><osm id>[:<class>]'. All
expected rows are expected to be present with at least one database row.
"""
- with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
- for row in context.table:
- nid = NominatimID(row['object'])
- nid.row_by_place_id(cur, 'search_name',
- ['ST_X(centroid) as cx', 'ST_Y(centroid) as cy'])
- assert cur.rowcount > 0, "No rows found for " + row['object']
+ tokenizer = tokenizer_factory.get_tokenizer_for_db(context.nominatim.get_test_config())
+
+ with tokenizer.name_analyzer() as analyzer:
+ with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
+ for row in context.table:
+ nid = NominatimID(row['object'])
+ nid.row_by_place_id(cur, 'search_name',
+ ['ST_X(centroid) as cx', 'ST_Y(centroid) as cy'])
+ assert cur.rowcount > 0, "No rows found for " + row['object']
+
+ for res in cur:
+ db_row = DBRow(nid, res, context)
+ for name, value in zip(row.headings, row.cells):
+ if name in ('name_vector', 'nameaddress_vector'):
+ items = [x.strip() for x in value.split(',')]
+ tokens = analyzer.get_word_token_info(items)
- for res in cur:
- db_row = DBRow(nid, res, context)
- for name, value in zip(row.headings, row.cells):
- if name in ('name_vector', 'nameaddress_vector'):
- items = [x.strip() for x in value.split(',')]
- with context.db.cursor() as subcur:
- subcur.execute(""" SELECT word_id, word_token
- FROM word, (SELECT unnest(%s::TEXT[]) as term) t
- WHERE word_token = make_standard_name(t.term)
- and class is null and country_code is null
- and operator is null
- UNION
- SELECT word_id, word_token
- FROM word, (SELECT unnest(%s::TEXT[]) as term) t
- WHERE word_token = ' ' || make_standard_name(t.term)
- and class is null and country_code is null
- and operator is null
- """,
- (list(filter(lambda x: not x.startswith('#'), items)),
- list(filter(lambda x: x.startswith('#'), items))))
if not exclude:
- assert subcur.rowcount >= len(items), \
- "No word entry found for {}. Entries found: {!s}".format(value, subcur.rowcount)
- for wid in subcur:
- present = wid[0] in res[name]
+ assert len(tokens) >= len(items), \
+ "No word entry found for {}. Entries found: {!s}".format(value, len(tokens))
+ for word, token, wid in tokens:
if exclude:
- assert not present, "Found term for {}/{}: {}".format(row['object'], name, wid[1])
+ assert wid not in res[name], \
+ "Found term for {}/{}: {}".format(nid, name, wid)
else:
- assert present, "Missing term for {}/{}: {}".format(row['object'], name, wid[1])
- elif name != 'object':
- assert db_row.contains(name, value), db_row.assert_msg(name, value)
+ assert wid in res[name], \
+ "Missing term for {}/{}: {}".format(nid, name, wid)
+ elif name != 'object':
+ assert db_row.contains(name, value), db_row.assert_msg(name, value)
@then("search_name has no entry for (?P<oid>.*)")
def check_search_name_has_entry(context, oid):
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
cur.execute("SELECT *, ST_AsText(geometry) as geomtxt FROM location_postcode")
assert cur.rowcount == len(list(context.table)), \
- "Postcode table has {} rows, expected {}.".foramt(cur.rowcount, len(list(context.table)))
+ "Postcode table has {} rows, expected {}.".format(cur.rowcount, len(list(context.table)))
results = {}
for row in cur:
db_row.assert_row(row, ('country', 'postcode'))
-@then("word contains(?P<exclude> not)?")
-def check_word_table(context, exclude):
- """ Check the contents of the word table. Each row represents a table row
- and all data must match. Data not present in the expected table, may
- be arbitry. The rows are identified via all given columns.
+@then("there are(?P<exclude> no)? word tokens for postcodes (?P<postcodes>.*)")
+def check_word_table_for_postcodes(context, exclude, postcodes):
+ """ Check that the tokenizer produces postcode tokens for the given
+ postcodes. The postcodes are a comma-separated list of postcodes.
+ Whitespace matters.
"""
+ nctx = context.nominatim
+ tokenizer = tokenizer_factory.get_tokenizer_for_db(nctx.get_test_config())
+ with tokenizer.name_analyzer() as ana:
+ plist = [ana.normalize_postcode(p) for p in postcodes.split(',')]
+
+ plist.sort()
+
with context.db.cursor(cursor_factory=psycopg2.extras.DictCursor) as cur:
- for row in context.table:
- wheres = ' AND '.join(["{} = %s".format(h) for h in row.headings])
- cur.execute("SELECT * from word WHERE " + wheres, list(row.cells))
- if exclude:
- assert cur.rowcount == 0, "Row still in word table: %s" % '/'.join(values)
- else:
- assert cur.rowcount > 0, "Row not in word table: %s" % '/'.join(values)
+ if nctx.tokenizer == 'icu':
+ cur.execute("SELECT word FROM word WHERE type = 'P' and word = any(%s)",
+ (plist,))
+ else:
+ cur.execute("""SELECT word FROM word WHERE word = any(%s)
+ and class = 'place' and type = 'postcode'""",
+ (plist,))
+
+ found = [row[0] for row in cur]
+ assert len(found) == len(set(found)), f"Duplicate rows for postcodes: {found}"
+
+ if exclude:
+ assert len(found) == 0, f"Unexpected postcodes: {found}"
+ else:
+ assert set(found) == set(plist), \
+ f"Missing postcodes {set(plist) - set(found)}. Found: {found}"
@then("place_addressline contains")
def check_place_addressline(context):
projects / nominatim.git / blobdiff