Commit 08e5cb32 authored by Marius Kriegerowski's avatar Marius Kriegerowski
Browse files

Added get_building test and adopted sql argument passing

parent b49f3f4c
Pipeline #24461 passed with stage
in 2 minutes and 41 seconds
......@@ -35,12 +35,14 @@ class GetBuilding(Rule):
logger.debug("Processing building: %s", building_id)
table_from = "osm_building_polygons"
table_to = "obm_buildings"
# Pull the building data from storage_from
for row in await self.storage_from.get_results(
f"SELECT osm_id, geometry FROM osm_building_polygons WHERE osm_id={building_id}"
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(
"INSERT INTO obm_buildings (osm_id, geometry) VALUES (%s,%s)", row
f"INSERT INTO {table_to} (osm_id, geometry) VALUES (?, ?)", row
)
......@@ -37,6 +37,7 @@ async def connected_storage(pytestconfig):
yield storage
storage.disconnect()
await storage.pool.wait_closed()
def pytest_collection_modifyitems(config, items):
......
import pytest
from rabotnikobm.rules.get_building import GetBuilding
@pytest.mark.requires_storage
@pytest.mark.asyncio
async def test_get_building(connected_storage, mock_storage):
await mock_storage.execute(
"""CREATE TABLE obm_buildings (
osm_id INTEGER,
geometry TEXT
floorspace DOUBLE PRECISION
)
"""
)
rule = GetBuilding(storage_from=connected_storage, storage_to=mock_storage)
payload = {"building_id": 848970775}
await rule.evaluate(payload=payload)
data_out = mock_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