Commit 9c585a42 authored by Marius Kriegerowski's avatar Marius Kriegerowski
Browse files

Added upserting to get_building

parent 3681950b
Pipeline #27345 passed with stage
in 2 minutes and 26 seconds
image: python:3-buster
services:
- git.gfz-potsdam.de:5000/dynamicexposure/server-components/containers/docker-obm-database:master
before_script:
- python3 -V
- pip3 install virtualenv
......
......@@ -19,10 +19,19 @@
import logging
from rabotnik import Rule
from rabotnik.storages.base import StorageBase
logger = logging.getLogger(__name__)
async def upsert(storage: StorageBase, table: str, data: tuple):
osm_id, geometry = data
await storage.execute(
f"INSERT INTO {table} VALUES (%s, %s) ON CONFLICT (osm_id) DO UPDATE SET geometry = %s",
(osm_id, geometry, geometry),
)
class GetBuilding(Rule):
"""A rule to copy a building entry from a source database to a destination database."""
......@@ -41,5 +50,6 @@ class GetBuilding(Rule):
for row in await self.storage_from.get_results(
f"SELECT osm_id, geometry FROM {table_from} WHERE osm_id={building_id}"
):
# Push the building data to storage_to
await self.storage_to.execute(f"INSERT INTO {table_to} VALUES (%s, %s)", row)
await upsert(self.storage_to, table_to, row)
......@@ -9,6 +9,7 @@ logger = logging.getLogger(__name__)
def pytest_addoption(parser):
parser.addoption(
"--storage-configuration",
help="Source obm-database instance that can delivers data for testing",
action="store",
default=os.getenv("RABOTNIK_OBM_TEST_STORAGE_CONFIGURATION"),
)
......
---
user: postgres
dbname: postgres
host: docker-obm-database
port: 5433
password: postgres
import pytest
from rabotnikobm.rules.get_building import GetBuilding
from rabotnikobm.rules.get_building import GetBuilding, upsert
@pytest.mark.xfail(reason="need to test this rule against postgresql database")
@pytest.mark.requires_storage
@pytest.mark.asyncio
async def test_get_building(connected_storage, mock_storage):
async def test_upsert(connected_storage):
await mock_storage.execute(
"""CREATE TABLE obm_buildings (
osm_id INTEGER,
geometry TEXT
floorspace DOUBLE PRECISION
table = "obm_buildings"
osm_id = 0
test_data = (osm_id, "POLYGON((0 0, 1 0, 1 1, 0 0))")
test_data_update = (osm_id, "POLYGON((0 0, 1 0, 1 1, 0 1, 0 0))")
await upsert(connected_storage, table=table, data=test_data)
data_out = await connected_storage.get_results(
f"SELECT * FROM {table} WHERE osm_id = {osm_id};"
)
"""
await upsert(connected_storage, table=table, data=test_data_update)
data_out_updated = await connected_storage.get_results(
f"SELECT * FROM {table} WHERE osm_id = {osm_id};"
)
assert len(data_out) == 1
assert data_out_updated != data_out
@pytest.mark.requires_storage
@pytest.mark.asyncio
async def test_get_building(connected_storage):
rule = GetBuilding(storage_from=connected_storage, storage_to=mock_storage)
rule = GetBuilding(storage_from=connected_storage, storage_to=connected_storage)
payload = {"building_id": 848970775}
await rule.evaluate(payload=payload)
data_out = mock_storage.get_results("SELECT * FROM obm_buildings;")
data_out = connected_storage.get_results("SELECT * FROM obm_buildings;")
assert len(data_out) == 1
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment