46 lines
1.3 KiB
Python
46 lines
1.3 KiB
Python
import asyncio
|
|
from datetime import date
|
|
|
|
from app.infra.db.postgresql.gis_attributes import GisAttributeRepository
|
|
from tests.conftest import FakeConnection, FakeCursor
|
|
|
|
|
|
def test_list_element_groups_returns_empty_when_source_gis_is_not_provisioned():
|
|
cursor = FakeCursor(fetchone_results=[{"relation": None}])
|
|
|
|
result = asyncio.run(
|
|
GisAttributeRepository.list_element_groups(
|
|
FakeConnection(cursor), "pipes", "P-1"
|
|
)
|
|
)
|
|
|
|
assert result == []
|
|
assert len(cursor.executed) == 1
|
|
|
|
|
|
def test_list_element_groups_reads_linked_attributes():
|
|
row = {
|
|
"dataset_key": "fx_pipe_inventory_20200811",
|
|
"dataset_title": "奉贤 DN100 及以上管网台账",
|
|
"source_date": date(2020, 8, 11),
|
|
"layer_key": "fx_20200811_dn100_pipes",
|
|
"layer_name": "DN100 及以上管线台账",
|
|
"source_fid": 42,
|
|
"match_method": "geometry",
|
|
"match_distance_m": 0.12,
|
|
"attributes": {"PIPE_D": 300, "MATERIAL": "DI"},
|
|
}
|
|
cursor = FakeCursor(
|
|
fetchone_results=[{"relation": "source_gis.element_links"}],
|
|
fetchall_results=[[row]],
|
|
)
|
|
|
|
result = asyncio.run(
|
|
GisAttributeRepository.list_element_groups(
|
|
FakeConnection(cursor), "pipes", "P-42"
|
|
)
|
|
)
|
|
|
|
assert result == [row]
|
|
assert cursor.executed[1][1] == ("pipes", "P-42")
|