Added and test floorspace upserting

.gitlab-ci.yml

 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

rabotnikobm/rules/get_building.py

@@ -25,6 +25,7 @@ logger = logging.getLogger(__name__)
 
 
 async def upsert(storage: StorageBase, table: str, data: tuple):
+    """Insert `data` into `storage` and update if `data` is already present."""
     osm_id, geometry = data
     await storage.execute(
         f"INSERT INTO {table} VALUES (%s, %s) ON CONFLICT (osm_id) DO UPDATE SET geometry = %s",

rabotnikobm/rules/get_floorspace.py

@@ -7,6 +7,15 @@ from rabotnik import Rule  # pylint: disable=import-error
 logger = logging.getLogger(__name__)
 
 
+async def upsert(storage: RabotnikStorage, osm_id: int, floorspace: float):
+    """Insert data into `storage` and update if data is already present."""
+    await storage.execute(
+        "INSERT INTO obm_buildings (osm_id, floorspace) VALUES (%s, %s) "
+        "ON CONFLICT (osm_id) DO UPDATE SET floorspace = %s",
+        (osm_id, floorspace, floorspace),
+    )
+
+
 class GetFloorspace(Rule):
     def __init__(self, storage: RabotnikStorage, storage_target: RabotnikStorage):
         """Calculate floorspace of buildings.
@@ -22,7 +31,7 @@ class GetFloorspace(Rule):
         building_id = payload["building_id"]
         logger.debug("evaluating %s", building_id)
 
-        query_data = f"""
+        query_floorspace = f"""
             SELECT ST_Area(ST_Transform(geometry, 4326)) *
                 CAST(br.tags->'building:levels' AS float)
             FROM osm_building_relations AS br
@@ -37,14 +46,5 @@ class GetFloorspace(Rule):
             AND osm_id = {building_id};
         """
 
-        write_data = """
-        INSERT INTO obm_buildings (osm_id, floorspace) VALUES ({}, {})
-        """
-
-        async for data in self.storage.iter_results(query_data):
-            assert len(data) == 1
-            data = data[0]
-            async for response in self.storage_target.iter_results(
-                write_data.format(*(building_id, data))
-            ):
-                logger.debug(response)
+        floorspace = await self.storage.expect_one(query_floorspace)
+        await upsert(self.storage_target, building_id, floorspace)

tests/conftest.py

@@ -8,10 +8,17 @@ logger = logging.getLogger(__name__)
 
 def pytest_addoption(parser):
     parser.addoption(
-        "--storage-configuration",
-        help="Source obm-database instance that can delivers data for testing",
+        "--storage-consumer",
+        help="obm-database configuration that can delivers data for testing",
         action="store",
-        default=os.getenv("RABOTNIK_OBM_TEST_STORAGE_CONFIGURATION"),
+        default=os.getenv("RABOTNIK_OBM_STORAGE_CONSUMER"),
+    )
+
+    parser.addoption(
+        "--storage-contributor",
+        help="obm-database configuration that can delivers and digest data for testing",
+        action="store",
+        default=os.getenv("RABOTNIK_OBM_STORAGE_CONTRIBUTOR"),
     )
 
 
@@ -29,8 +36,22 @@ async def mock_storage():
 
 @pytest.fixture
 @pytest.mark.asyncio
-async def connected_storage(pytestconfig):
-    storage_configuration = pytestconfig.getoption("storage_configuration")
+async def storage_contributor(pytestconfig):
+    storage_configuration = pytestconfig.getoption("storage_contributor")
+
+    storage = rabotnik.Rabotnik.get_storage("postgresql")
+    await storage.connect(config_file=storage_configuration)
+
+    yield storage
+
+    storage.disconnect()
+    await storage.pool.wait_closed()
+
+
+@pytest.fixture
+@pytest.mark.asyncio
+async def storage_consumer(pytestconfig):
+    storage_configuration = pytestconfig.getoption("storage_consumer")
 
     storage = rabotnik.Rabotnik.get_storage("postgresql")
     await storage.connect(config_file=storage_configuration)
@@ -42,7 +63,7 @@ async def connected_storage(pytestconfig):
 
 
 def pytest_collection_modifyitems(config, items):
-    storage_configuration = config.getoption("storage_configuration")
+    storage_configuration = config.getoption("storage_contributor")
     if storage_configuration:
         return
 

tests/storage-testing-docker.yml

----
-user: postgres
-dbname: postgres
-host: docker-obm-database
-port: 5433
-password: postgres

tests/test_get_building.py

@@ -4,20 +4,20 @@ from rabotnikobm.rules.get_building import GetBuilding, upsert
 
 @pytest.mark.requires_storage
 @pytest.mark.asyncio
-async def test_upsert(connected_storage):
+async def test_upsert(storage_contributor):
 
     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(
+    await upsert(storage_contributor, table=table, data=test_data)
+    data_out = await storage_contributor.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(
+    await upsert(storage_contributor, table=table, data=test_data_update)
+    data_out_updated = await storage_contributor.get_results(
         f"SELECT * FROM {table} WHERE osm_id = {osm_id};"
     )
     assert len(data_out) == 1
@@ -26,11 +26,14 @@ async def test_upsert(connected_storage):
 
 @pytest.mark.requires_storage
 @pytest.mark.asyncio
-async def test_get_building(connected_storage):
+async def test_get_building(storage_consumer, storage_contributor):
 
-    rule = GetBuilding(storage_from=connected_storage, storage_to=connected_storage)
+    rule = GetBuilding(storage_from=storage_consumer, storage_to=storage_contributor)
+
+    building_id = 848970775
+    await rule.evaluate(payload={"building_id": building_id})
+    data_out = await storage_contributor.get_results(
+        f"SELECT * FROM obm_buildings WHERE osm_id = {building_id};"
+    )
 
-    payload = {"building_id": 848970775}
-    await rule.evaluate(payload=payload)
-    data_out = connected_storage.get_results("SELECT * FROM obm_buildings;")
     assert len(data_out) == 1

tests/test_get_building_land_use.py

@@ -25,8 +25,8 @@ from rabotnikobm.rules.get_building_land_use import (
 
 @pytest.mark.requires_storage
 @pytest.mark.asyncio
-async def test_get_points_in_building(connected_storage):
-    rule = GetBuildingLandUse(storage=connected_storage)
+async def test_get_points_in_building(storage_consumer):
+    rule = GetBuildingLandUse(storage=storage_consumer)
 
     payload = {"building_id": 193799}
     result = [result async for result in rule.evaluate(payload=payload)]

tests/test_get_floorspace.py

@@ -25,19 +25,13 @@ from rabotnikobm.rules.get_floorspace import (
 
 @pytest.mark.requires_storage
 @pytest.mark.asyncio
-async def test_get_floorspace(connected_storage, mock_storage):
-    c = mock_storage.connection.cursor()
-    c.execute(
-        """CREATE TABLE obm_buildings (
-        osm_id INTEGER,
-        floorspace FLOAT
-    )
-    """
-    )
+async def test_get_floorspace(storage_consumer, storage_contributor):
+    rule = GetFloorspace(storage=storage_consumer, storage_target=storage_contributor)
 
-    rule = GetFloorspace(storage=connected_storage, storage_target=mock_storage)
-
-    payload = {"building_id": 848970775}
+    osm_id = 848970775
+    payload = {"building_id": osm_id}
     await rule.evaluate(payload=payload)
-    data_out = await mock_storage.get_results("SELECT * FROM obm_buildings;")
-    assert data_out == [(848970775, 2.1424237391628443e-09)]
+    data_out = await storage_contributor.get_results(
+        f"SELECT osm_id, floorspace FROM obm_buildings WHERE osm_id = {osm_id};"
+    )
+    assert data_out == [(osm_id, 2.1424237391628443e-09)]

tests/test_get_points_in_building.py

@@ -25,8 +25,8 @@ from rabotnikobm.rules.get_points_in_building import (
 
 @pytest.mark.requires_storage
 @pytest.mark.asyncio
-async def test_get_points_in_building(connected_storage):
-    rule = GetPointsInBuilding(storage=connected_storage)
+async def test_get_points_in_building(storage_consumer):
+    rule = GetPointsInBuilding(storage=storage_consumer)
 
     payload = {"building_id": -12412829}
     result = [nbuildings async for nbuildings in rule.evaluate(payload=payload)]