diff --git a/tests/test_API.py b/tests/test_API.py index 9966f66b..ff0798e4 100644 --- a/tests/test_API.py +++ b/tests/test_API.py @@ -12,1013 +12,1013 @@ ## Status -# def test_status(): -# response = client.get("/v1/status/") -# assert response.status_code == 200 - - -# ## Login -# def test_login_url(): -# response = client.get("/v1/auth/login/") -# assert response.status_code == 200 - - -# def test_login_auth_me(): -# headers = {"access-token": access_token} -# response = client.get("/v1/auth/me/", headers=headers) -# assert response.status_code == 200 - - -# ## Countries - - -# def test_countries_endpoint(): -# response = client.get("/v1/countries/?q=nepal") -# assert response.status_code == 200 - - -# ## test osm_id - - -# def test_osm_id_endpoint(): -# response = client.get("/v1/osm_id/?osm_id=421498318") -# assert response.status_code == 200 - - -# ## Snapshot -# def test_snapshot(): -# response = client.post( -# "/v1/snapshot/", -# json={ -# "geometry": { -# "type": "Polygon", -# "coordinates": [ -# [ -# [83.96919250488281, 28.194446860487773], -# [83.99751663208006, 28.194446860487773], -# [83.99751663208006, 28.214869548073377], -# [83.96919250488281, 28.214869548073377], -# [83.96919250488281, 28.194446860487773], -# ] -# ], -# } -# }, -# ) -# assert response.status_code == 200 -# res = response.json() -# track_link = res["track_link"] -# max_attempts = 6 -# interval_seconds = 10 -# for attempt in range(1, max_attempts + 1): -# time.sleep(interval_seconds) # wait for worker to complete task - -# response = client.get(f"/v1{track_link}") -# assert response.status_code == 200 -# res = response.json() -# check_status = res["status"] - -# if check_status == "SUCCESS": -# break # exit the loop if the status is SUCCESS - -# if attempt == max_attempts: -# # If max_attempts reached and status is not SUCCESS, raise an AssertionError -# assert ( -# False -# ), f"Task did not complete successfully after {max_attempts} attempts" - - -# def test_snapshot_featurecollection(): -# response = client.post( -# "/v1/snapshot/", -# json={ -# "geometry": { -# "type": "FeatureCollection", -# "features": [ -# { -# "type": "Feature", -# "properties": {}, -# "geometry": { -# "coordinates": [ -# [ -# [83.97346137271688, 28.217525272345284], -# [83.97346137271688, 28.192595937414737], -# [84.01473909818759, 28.192595937414737], -# [84.01473909818759, 28.217525272345284], -# [83.97346137271688, 28.217525272345284], -# ] -# ], -# "type": "Polygon", -# }, -# } -# ], -# } -# }, -# ) -# assert response.status_code == 200 -# res = response.json() -# track_link = res["track_link"] -# max_attempts = 6 -# interval_seconds = 10 -# for attempt in range(1, max_attempts + 1): -# time.sleep(interval_seconds) # wait for worker to complete task - -# response = client.get(f"/v1{track_link}") -# assert response.status_code == 200 -# res = response.json() -# check_status = res["status"] - -# if check_status == "SUCCESS": -# break # exit the loop if the status is SUCCESS - -# if attempt == max_attempts: -# # If max_attempts reached and status is not SUCCESS, raise an AssertionError -# assert ( -# False -# ), f"Task did not complete successfully after {max_attempts} attempts" - - -# def test_snapshot_feature(): -# response = client.post( -# "/v1/snapshot/", -# json={ -# "geometry": { -# "type": "Feature", -# "properties": {}, -# "geometry": { -# "coordinates": [ -# [ -# [83.97346137271688, 28.217525272345284], -# [83.97346137271688, 28.192595937414737], -# [84.01473909818759, 28.192595937414737], -# [84.01473909818759, 28.217525272345284], -# [83.97346137271688, 28.217525272345284], -# ] -# ], -# "type": "Polygon", -# }, -# } -# }, -# ) -# assert response.status_code == 200 -# res = response.json() -# track_link = res["track_link"] -# max_attempts = 6 -# interval_seconds = 10 -# for attempt in range(1, max_attempts + 1): -# time.sleep(interval_seconds) # wait for worker to complete task - -# response = client.get(f"/v1{track_link}") -# assert response.status_code == 200 -# res = response.json() -# check_status = res["status"] - -# if check_status == "SUCCESS": -# break # exit the loop if the status is SUCCESS - -# if attempt == max_attempts: -# # If max_attempts reached and status is not SUCCESS, raise an AssertionError -# assert ( -# False -# ), f"Task did not complete successfully after {max_attempts} attempts" - - -# def test_snapshot_centroid(): -# response = client.post( -# "/v1/snapshot/", -# json={ -# "centroid": True, -# "geometry": { -# "type": "Polygon", -# "coordinates": [ -# [ -# [83.96919250488281, 28.194446860487773], -# [83.99751663208006, 28.194446860487773], -# [83.99751663208006, 28.214869548073377], -# [83.96919250488281, 28.214869548073377], -# [83.96919250488281, 28.194446860487773], -# ] -# ], -# }, -# }, -# ) -# assert response.status_code == 200 -# res = response.json() -# track_link = res["track_link"] -# max_attempts = 6 -# interval_seconds = 10 -# for attempt in range(1, max_attempts + 1): -# time.sleep(interval_seconds) # wait for worker to complete task - -# response = client.get(f"/v1{track_link}") -# assert response.status_code == 200 -# res = response.json() -# check_status = res["status"] - -# if check_status == "SUCCESS": -# break # exit the loop if the status is SUCCESS - -# if attempt == max_attempts: -# # If max_attempts reached and status is not SUCCESS, raise an AssertionError -# assert ( -# False -# ), f"Task did not complete successfully after {max_attempts} attempts" - - -# def test_snapshot_filters(): -# response = client.post( -# "/v1/snapshot/", -# json={ -# "fileName": "Example export with all features", -# "geometry": { -# "type": "Polygon", -# "coordinates": [ -# [ -# [83.585701, 28.046607], -# [83.585701, 28.382561], -# [84.391823, 28.382561], -# [84.391823, 28.046607], -# [83.585701, 28.046607], -# ] -# ], -# }, -# "outputType": "geojson", -# "geometryType": ["point", "line", "polygon"], -# "filters": { -# "tags": { -# "point": { -# "join_or": { -# "amenity": [ -# "bank", -# "ferry_terminal", -# "bus_station", -# "fuel", -# "kindergarten", -# "school", -# "college", -# "university", -# "place_of_worship", -# "marketplace", -# "clinic", -# "hospital", -# "police", -# "fire_station", -# ], -# "building": [ -# "bank", -# "aerodrome", -# "ferry_terminal", -# "train_station", -# "bus_station", -# "pumping_station", -# "power_substation", -# "kindergarten", -# "school", -# "college", -# "university", -# "mosque ", -# " church ", -# " temple", -# "supermarket", -# "marketplace", -# "clinic", -# "hospital", -# "police", -# "fire_station", -# "stadium ", -# " sports_centre", -# "governor_office ", -# " townhall ", -# " subdistrict_office ", -# " village_office ", -# " community_group_office", -# "government_office", -# ], -# "man_made": ["tower", "water_tower", "pumping_station"], -# "tower:type": ["communication"], -# "aeroway": ["aerodrome"], -# "railway": ["station"], -# "emergency": ["fire_hydrant"], -# "landuse": ["reservoir", "recreation_gound"], -# "waterway": ["floodgate"], -# "natural": ["spring"], -# "power": ["tower", "substation"], -# "shop": ["supermarket"], -# "leisure": [ -# "stadium ", -# " sports_centre ", -# " pitch ", -# " swimming_pool", -# "park", -# ], -# "office": ["government"], -# } -# }, -# "line": { -# "join_or": { -# "highway": [ -# "motorway ", -# " trunk ", -# " primary ", -# " secondary ", -# " tertiary ", -# " service ", -# " residential ", -# " pedestrian ", -# " path ", -# " living_street ", -# " track", -# ], -# "railway": ["rail"], -# "man_made": ["embankment"], -# "waterway": [], -# } -# }, -# "polygon": { -# "join_or": { -# "amenity": [ -# "bank", -# "ferry_terminal", -# "bus_station", -# "fuel", -# "kindergarten", -# "school", -# "college", -# "university", -# "place_of_worship", -# "marketplace", -# "clinic", -# "hospital", -# "police", -# "fire_station", -# ], -# "building": [ -# "bank", -# "aerodrome", -# "ferry_terminal", -# "train_station", -# "bus_station", -# "pumping_station", -# "power_substation", -# "power_plant", -# "kindergarten", -# "school", -# "college", -# "university", -# "mosque ", -# " church ", -# " temple", -# "supermarket", -# "marketplace", -# "clinic", -# "hospital", -# "police", -# "fire_station", -# "stadium ", -# " sports_centre", -# "governor_office ", -# " townhall ", -# " subdistrict_office ", -# " village_office ", -# " community_group_office", -# "government_office", -# ], -# "man_made": ["tower", "water_tower", "pumping_station"], -# "tower:type": ["communication"], -# "aeroway": ["aerodrome"], -# "railway": ["station"], -# "landuse": ["reservoir", "recreation_gound"], -# "waterway": [], -# "natural": ["spring"], -# "power": ["substation", "plant"], -# "shop": ["supermarket"], -# "leisure": [ -# "stadium ", -# " sports_centre ", -# " pitch ", -# " swimming_pool", -# "park", -# ], -# "office": ["government"], -# "type": ["boundary"], -# "boundary": ["administrative"], -# }, -# }, -# }, -# "attributes": { -# "point": [ -# "building", -# "ground_floor:height", -# "capacity:persons", -# "building:structure", -# "building:condition", -# "name", -# "admin_level", -# "building:material", -# "office", -# "building:roof", -# "backup_generator", -# "access:roof", -# "building:levels", -# "building:floor", -# "addr:full", -# "addr:city", -# "source", -# ], -# "line": ["width", "source", "waterway", "name"], -# "polygon": [ -# "landslide_prone", -# "name", -# "admin_level", -# "type", -# "is_in:town", -# "flood_prone", -# "is_in:province", -# "is_in:city", -# "is_in:municipality", -# "is_in:RW", -# "is_in:village", -# "source", -# "boundary", -# ], -# }, -# }, -# }, -# ) -# assert response.status_code == 200 -# res = response.json() -# track_link = res["track_link"] -# max_attempts = 6 -# interval_seconds = 10 -# for attempt in range(1, max_attempts + 1): -# time.sleep(interval_seconds) # wait for worker to complete task - -# response = client.get(f"/v1{track_link}") -# assert response.status_code == 200 -# res = response.json() -# check_status = res["status"] - -# if check_status == "SUCCESS": -# break # exit the loop if the status is SUCCESS - -# if attempt == max_attempts: -# # If max_attempts reached and status is not SUCCESS, raise an AssertionError -# assert ( -# False -# ), f"Task did not complete successfully after {max_attempts} attempts" - - -# def test_snapshot_and_filter(): -# response = client.post( -# "/v1/snapshot/", -# json={ -# "fileName": "Destroyed_Buildings_Turkey", -# "geometry": { -# "type": "Polygon", -# "coordinates": [ -# [ -# [36.70588085657477, 37.1979648807274], -# [36.70588085657477, 37.1651408422983], -# [36.759267544807194, 37.1651408422983], -# [36.759267544807194, 37.1979648807274], -# [36.70588085657477, 37.1979648807274], -# ] -# ], -# }, -# "outputType": "geojson", -# "geometryType": ["polygon"], -# "filters": { -# "tags": { -# "point": {}, -# "line": {}, -# "polygon": { -# "join_or": {}, -# "join_and": { -# "destroyed:building": ["yes"], -# "damage:date": ["2023-02-06"], -# }, -# }, -# }, -# "attributes": { -# "point": [], -# "line": [], -# "polygon": [ -# "building", -# "destroyed:building", -# "damage:date", -# "name", -# "source", -# ], -# }, -# }, -# }, -# ) -# assert response.status_code == 200 -# res = response.json() -# track_link = res["track_link"] -# max_attempts = 6 -# interval_seconds = 10 -# for attempt in range(1, max_attempts + 1): -# time.sleep(interval_seconds) # wait for worker to complete task - -# response = client.get(f"/v1{track_link}") -# assert response.status_code == 200 -# res = response.json() -# check_status = res["status"] - -# if check_status == "SUCCESS": -# break # exit the loop if the status is SUCCESS - -# if attempt == max_attempts: -# # If max_attempts reached and status is not SUCCESS, raise an AssertionError -# assert ( -# False -# ), f"Task did not complete successfully after {max_attempts} attempts" - - -# def test_snapshot_authentication_uuid(): -# headers = {"access-token": access_token} -# payload = { -# "geometry": { -# "type": "Polygon", -# "coordinates": [ -# [ -# [83.96919250488281, 28.194446860487773], -# [83.99751663208006, 28.194446860487773], -# [83.99751663208006, 28.214869548073377], -# [83.96919250488281, 28.214869548073377], -# [83.96919250488281, 28.194446860487773], -# ] -# ], -# }, -# "uuid": False, -# } - -# response = client.post("/v1/snapshot/", json=payload, headers=headers) - -# assert response.status_code == 200 -# res = response.json() -# track_link = res["track_link"] -# max_attempts = 6 -# interval_seconds = 10 -# for attempt in range(1, max_attempts + 1): -# time.sleep(interval_seconds) # wait for worker to complete task - -# response = client.get(f"/v1{track_link}") -# assert response.status_code == 200 -# res = response.json() -# check_status = res["status"] - -# if check_status == "SUCCESS": -# break # exit the loop if the status is SUCCESS - -# if attempt == max_attempts: -# # If max_attempts reached and status is not SUCCESS, raise an AssertionError -# assert ( -# False -# ), f"Task did not complete successfully after {max_attempts} attempts" - - -# def test_snapshot_bind_zip(): -# headers = {"access-token": access_token} -# payload = { -# "geometry": { -# "type": "Polygon", -# "coordinates": [ -# [ -# [83.96919250488281, 28.194446860487773], -# [83.99751663208006, 28.194446860487773], -# [83.99751663208006, 28.214869548073377], -# [83.96919250488281, 28.214869548073377], -# [83.96919250488281, 28.194446860487773], -# ] -# ], -# }, -# "bindZip": False, -# } - -# response = client.post("/v1/snapshot/", json=payload, headers=headers) - -# assert response.status_code == 200 -# res = response.json() -# track_link = res["track_link"] -# max_attempts = 6 -# interval_seconds = 10 -# for attempt in range(1, max_attempts + 1): -# time.sleep(interval_seconds) # wait for worker to complete task - -# response = client.get(f"/v1{track_link}") -# assert response.status_code == 200 -# res = response.json() -# check_status = res["status"] - -# if check_status == "SUCCESS": -# break # exit the loop if the status is SUCCESS - -# if attempt == max_attempts: -# # If max_attempts reached and status is not SUCCESS, raise an AssertionError -# assert ( -# False -# ), f"Task did not complete successfully after {max_attempts} attempts" - - -# ## Snapshot Plain - - -# def test_snapshot_plain(): -# response = client.post( -# "/v1/snapshot/plain/", -# json={ -# "geometry": { -# "type": "Polygon", -# "coordinates": [ -# [ -# [83.96919250488281, 28.194446860487773], -# [83.99751663208006, 28.194446860487773], -# [83.99751663208006, 28.214869548073377], -# [83.96919250488281, 28.214869548073377], -# [83.96919250488281, 28.194446860487773], -# ] -# ], -# } -# }, -# ) -# assert response.status_code == 200 - - -# ## Stats - - -# def test_stats_endpoint_custom_polygon(): -# headers = {"access-token": access_token} -# payload = { -# "geometry": { -# "type": "Polygon", -# "coordinates": [ -# [ -# [83.96919250488281, 28.194446860487773], -# [83.99751663208006, 28.194446860487773], -# [83.99751663208006, 28.214869548073377], -# [83.96919250488281, 28.214869548073377], -# [83.96919250488281, 28.194446860487773], -# ] -# ], -# } -# } - -# response = client.post("/v1/stats/polygon/", json=payload, headers=headers) - -# assert response.status_code == 200 -# res = response.json() -# assert ( -# res["meta"]["indicators"] -# == "https://github.com/hotosm/raw-data-api/tree/develop/docs/src/stats/indicators.md" -# ) - - -# def test_stats_endpoint_iso3(): -# headers = {"access-token": access_token} -# payload = {"iso3": "npl"} - -# response = client.post("/v1/stats/polygon/", json=payload, headers=headers) - -# assert response.status_code == 200 -# res = response.json() -# assert ( -# res["meta"]["indicators"] -# == "https://github.com/hotosm/raw-data-api/tree/develop/docs/src/stats/indicators.md" -# ) - - -## HDX - - -# def test_hdx_submit_normal_iso3(): -# headers = {"access-token": access_token} -# payload = { -# "iso3": "NPL", -# "hdx_upload": False, -# "categories": [ -# { -# "Roads": { -# "hdx": { -# "tags": ["roads", "transportation", "geodata"], -# "caveats": "OpenStreetMap data is crowd sourced and cannot be considered to be exhaustive", -# }, -# "types": ["lines"], -# "select": ["name", "highway"], -# "where": "tags['highway'] IS NOT NULL", -# "formats": ["geojson"], -# } -# } -# ], -# } - -# response = client.post("/v1/custom/snapshot/", json=payload, headers=headers) - -# assert response.status_code == 200 -# res = response.json() -# track_link = res["track_link"] -# max_attempts = 6 -# interval_seconds = 10 -# for attempt in range(1, max_attempts + 1): -# time.sleep(interval_seconds) # wait for worker to complete task - -# response = client.get(f"/v1{track_link}") -# assert response.status_code == 200 -# res = response.json() -# check_status = res["status"] - -# if check_status == "SUCCESS": -# break # exit the loop if the status is SUCCESS - -# if attempt == max_attempts: -# # If max_attempts reached and status is not SUCCESS, raise an AssertionError -# assert ( -# False -# ), f"Task did not complete successfully after {max_attempts} attempts" - - -# def test_hdx_submit_normal_iso3_multiple_format(): -# headers = {"access-token": access_token} -# payload = { -# "iso3": "NPL", -# "hdx_upload": False, -# "categories": [ -# { -# "Roads": { -# "hdx": { -# "tags": ["roads", "transportation", "geodata"], -# "caveats": "OpenStreetMap data is crowd sourced and cannot be considered to be exhaustive", -# }, -# "types": ["lines"], -# "select": ["name", "highway"], -# "where": "tags['highway'] IS NOT NULL", -# "formats": ["geojson", "gpkg", "kml", "shp"], -# } -# } -# ], -# } - -# response = client.post("/v1/custom/snapshot/", json=payload, headers=headers) - -# assert response.status_code == 200 -# res = response.json() -# track_link = res["track_link"] -# max_attempts = 6 -# interval_seconds = 10 -# for attempt in range(1, max_attempts + 1): -# time.sleep(interval_seconds) # wait for worker to complete task - -# response = client.get(f"/v1{track_link}") -# assert response.status_code == 200 -# res = response.json() -# check_status = res["status"] - -# if check_status == "SUCCESS": -# break # exit the loop if the status is SUCCESS - -# if attempt == max_attempts: -# # If max_attempts reached and status is not SUCCESS, raise an AssertionError -# assert ( -# False -# ), f"Task did not complete successfully after {max_attempts} attempts" - - -# def test_hdx_submit_normal_custom_polygon(): -# headers = {"access-token": access_token} -# payload = { -# "geometry": { -# "type": "Polygon", -# "coordinates": [ -# [ -# [83.96919250488281, 28.194446860487773], -# [83.99751663208006, 28.194446860487773], -# [83.99751663208006, 28.214869548073377], -# [83.96919250488281, 28.214869548073377], -# [83.96919250488281, 28.194446860487773], -# ] -# ], -# }, -# "hdx_upload": False, -# "dataset": { -# "subnational": True, -# "dataset_title": "Pokhara", -# "dataset_prefix": "hotosm_pkr", -# "dataset_locations": ["npl"], -# }, -# "categories": [ -# { -# "Roads": { -# "hdx": { -# "tags": ["roads", "transportation", "geodata"], -# "caveats": "OpenStreetMap data is crowd sourced and cannot be considered to be exhaustive", -# }, -# "types": ["lines"], -# "select": ["name", "highway"], -# "where": "tags['highway'] IS NOT NULL", -# "formats": ["geojson"], -# } -# } -# ], -# } - -# response = client.post("/v1/custom/snapshot/", json=payload, headers=headers) - -# assert response.status_code == 200 -# res = response.json() -# track_link = res["track_link"] -# max_attempts = 6 -# interval_seconds = 10 -# for attempt in range(1, max_attempts + 1): -# time.sleep(interval_seconds) # wait for worker to complete task - -# response = client.get(f"/v1{track_link}") -# assert response.status_code == 200 -# res = response.json() -# check_status = res["status"] - -# if check_status == "SUCCESS": -# break # exit the loop if the status is SUCCESS - -# if attempt == max_attempts: -# # If max_attempts reached and status is not SUCCESS, raise an AssertionError -# assert ( -# False -# ), f"Task did not complete successfully after {max_attempts} attempts" - - -# def test_custom_submit_normal_custom_polygon_TM_project(): -# headers = {"access-token": access_token} -# payload = { -# "geometry": { -# "type": "Polygon", -# "coordinates": [ -# [ -# [83.96919250488281, 28.194446860487773], -# [83.99751663208006, 28.194446860487773], -# [83.99751663208006, 28.214869548073377], -# [83.96919250488281, 28.214869548073377], -# [83.96919250488281, 28.194446860487773], -# ] -# ], -# }, -# "queue": "raw_ondemand", -# "dataset": { -# "dataset_prefix": "hotosm_project_1", -# "dataset_folder": "TM", -# "dataset_title": "Tasking Manger Project 1", -# }, -# "categories": [ -# { -# "Buildings": { -# "types": ["polygons"], -# "select": [ -# "name", -# "building", -# "building:levels", -# "building:materials", -# "addr:full", -# "addr:housenumber", -# "addr:street", -# "addr:city", -# "office", -# "source", -# ], -# "where": "tags['building'] IS NOT NULL", -# "formats": ["geojson", "shp", "kml"], -# }, -# "Roads": { -# "types": ["lines"], -# "select": [ -# "name", -# "highway", -# "surface", -# "smoothness", -# "width", -# "lanes", -# "oneway", -# "bridge", -# "layer", -# "source", -# ], -# "where": "tags['highway'] IS NOT NULL", -# "formats": ["geojson", "shp", "kml"], -# }, -# "Waterways": { -# "types": ["lines", "polygons"], -# "select": [ -# "name", -# "waterway", -# "covered", -# "width", -# "depth", -# "layer", -# "blockage", -# "tunnel", -# "natural", -# "water", -# "source", -# ], -# "where": "tags['waterway'] IS NOT NULL OR tags['water'] IS NOT NULL OR tags['natural'] IN ('water','wetland','bay')", -# "formats": ["geojson", "shp", "kml"], -# }, -# "Landuse": { -# "types": ["points", "polygons"], -# "select": ["name", "amenity", "landuse", "leisure"], -# "where": "tags['landuse'] IS NOT NULL", -# "formats": ["geojson", "shp", "kml"], -# }, -# } -# ], -# } - -# response = client.post("/v1/custom/snapshot/", json=payload, headers=headers) - -# assert response.status_code == 200 -# res = response.json() -# track_link = res["track_link"] -# max_attempts = 6 -# interval_seconds = 10 -# for attempt in range(1, max_attempts + 1): -# time.sleep(interval_seconds) # wait for worker to complete task - -# response = client.get(f"/v1{track_link}") -# assert response.status_code == 200 -# res = response.json() -# check_status = res["status"] - -# if check_status == "SUCCESS": -# break # exit the loop if the status is SUCCESS - -# if attempt == max_attempts: -# # If max_attempts reached and status is not SUCCESS, raise an AssertionError -# assert ( -# False -# ), f"Task did not complete successfully after {max_attempts} attempts" - - -# def test_hdx_submit_normal_custom_polygon_upload(): -# headers = {"access-token": access_token} -# payload = { -# "geometry": { -# "type": "Polygon", -# "coordinates": [ -# [ -# [83.96919250488281, 28.194446860487773], -# [83.99751663208006, 28.194446860487773], -# [83.99751663208006, 28.214869548073377], -# [83.96919250488281, 28.214869548073377], -# [83.96919250488281, 28.194446860487773], -# ] -# ], -# }, -# "hdx_upload": True, -# "dataset": { -# "subnational": True, -# "dataset_title": "Pokhara", -# "dataset_folder": "Test", -# "dataset_prefix": "hotosm_pkr", -# "dataset_locations": ["npl"], -# }, -# "categories": [ -# { -# "Roads": { -# "hdx": { -# "tags": ["roads", "transportation", "geodata"], -# "caveats": "OpenStreetMap data is crowd sourced and cannot be considered to be exhaustive", -# }, -# "types": ["lines"], -# "select": ["name", "highway"], -# "where": "tags['highway'] IS NOT NULL", -# "formats": ["geojson"], -# } -# } -# ], -# } - -# response = client.post("/v1/custom/snapshot/", json=payload, headers=headers) - -# assert response.status_code == 200 -# res = response.json() -# track_link = res["track_link"] -# max_attempts = 6 -# interval_seconds = 10 -# for attempt in range(1, max_attempts + 1): -# time.sleep(interval_seconds) # wait for worker to complete task - -# response = client.get(f"/v1{track_link}") -# assert response.status_code == 200 -# res = response.json() -# check_status = res["status"] - -# if check_status == "SUCCESS": -# break # exit the loop if the status is SUCCESS - -# if attempt == max_attempts: -# # If max_attempts reached and status is not SUCCESS, raise an AssertionError -# assert ( -# False -# ), f"Task did not complete successfully after {max_attempts} attempts" +def test_status(): + response = client.get("/v1/status/") + assert response.status_code == 200 + + +## Login +def test_login_url(): + response = client.get("/v1/auth/login/") + assert response.status_code == 200 + + +def test_login_auth_me(): + headers = {"access-token": access_token} + response = client.get("/v1/auth/me/", headers=headers) + assert response.status_code == 200 + + +## Countries + + +def test_countries_endpoint(): + response = client.get("/v1/countries/?q=nepal") + assert response.status_code == 200 + + +## test osm_id + + +def test_osm_id_endpoint(): + response = client.get("/v1/osm_id/?osm_id=421498318") + assert response.status_code == 200 + + +## Snapshot +def test_snapshot(): + response = client.post( + "/v1/snapshot/", + json={ + "geometry": { + "type": "Polygon", + "coordinates": [ + [ + [83.96919250488281, 28.194446860487773], + [83.99751663208006, 28.194446860487773], + [83.99751663208006, 28.214869548073377], + [83.96919250488281, 28.214869548073377], + [83.96919250488281, 28.194446860487773], + ] + ], + } + }, + ) + assert response.status_code == 200 + res = response.json() + track_link = res["track_link"] + max_attempts = 6 + interval_seconds = 10 + for attempt in range(1, max_attempts + 1): + time.sleep(interval_seconds) # wait for worker to complete task + + response = client.get(f"/v1{track_link}") + assert response.status_code == 200 + res = response.json() + check_status = res["status"] + + if check_status == "SUCCESS": + break # exit the loop if the status is SUCCESS + + if attempt == max_attempts: + # If max_attempts reached and status is not SUCCESS, raise an AssertionError + assert ( + False + ), f"Task did not complete successfully after {max_attempts} attempts" + + +def test_snapshot_featurecollection(): + response = client.post( + "/v1/snapshot/", + json={ + "geometry": { + "type": "FeatureCollection", + "features": [ + { + "type": "Feature", + "properties": {}, + "geometry": { + "coordinates": [ + [ + [83.97346137271688, 28.217525272345284], + [83.97346137271688, 28.192595937414737], + [84.01473909818759, 28.192595937414737], + [84.01473909818759, 28.217525272345284], + [83.97346137271688, 28.217525272345284], + ] + ], + "type": "Polygon", + }, + } + ], + } + }, + ) + assert response.status_code == 200 + res = response.json() + track_link = res["track_link"] + max_attempts = 6 + interval_seconds = 10 + for attempt in range(1, max_attempts + 1): + time.sleep(interval_seconds) # wait for worker to complete task + + response = client.get(f"/v1{track_link}") + assert response.status_code == 200 + res = response.json() + check_status = res["status"] + + if check_status == "SUCCESS": + break # exit the loop if the status is SUCCESS + + if attempt == max_attempts: + # If max_attempts reached and status is not SUCCESS, raise an AssertionError + assert ( + False + ), f"Task did not complete successfully after {max_attempts} attempts" + + +def test_snapshot_feature(): + response = client.post( + "/v1/snapshot/", + json={ + "geometry": { + "type": "Feature", + "properties": {}, + "geometry": { + "coordinates": [ + [ + [83.97346137271688, 28.217525272345284], + [83.97346137271688, 28.192595937414737], + [84.01473909818759, 28.192595937414737], + [84.01473909818759, 28.217525272345284], + [83.97346137271688, 28.217525272345284], + ] + ], + "type": "Polygon", + }, + } + }, + ) + assert response.status_code == 200 + res = response.json() + track_link = res["track_link"] + max_attempts = 6 + interval_seconds = 10 + for attempt in range(1, max_attempts + 1): + time.sleep(interval_seconds) # wait for worker to complete task + + response = client.get(f"/v1{track_link}") + assert response.status_code == 200 + res = response.json() + check_status = res["status"] + + if check_status == "SUCCESS": + break # exit the loop if the status is SUCCESS + + if attempt == max_attempts: + # If max_attempts reached and status is not SUCCESS, raise an AssertionError + assert ( + False + ), f"Task did not complete successfully after {max_attempts} attempts" + + +def test_snapshot_centroid(): + response = client.post( + "/v1/snapshot/", + json={ + "centroid": True, + "geometry": { + "type": "Polygon", + "coordinates": [ + [ + [83.96919250488281, 28.194446860487773], + [83.99751663208006, 28.194446860487773], + [83.99751663208006, 28.214869548073377], + [83.96919250488281, 28.214869548073377], + [83.96919250488281, 28.194446860487773], + ] + ], + }, + }, + ) + assert response.status_code == 200 + res = response.json() + track_link = res["track_link"] + max_attempts = 6 + interval_seconds = 10 + for attempt in range(1, max_attempts + 1): + time.sleep(interval_seconds) # wait for worker to complete task + + response = client.get(f"/v1{track_link}") + assert response.status_code == 200 + res = response.json() + check_status = res["status"] + + if check_status == "SUCCESS": + break # exit the loop if the status is SUCCESS + + if attempt == max_attempts: + # If max_attempts reached and status is not SUCCESS, raise an AssertionError + assert ( + False + ), f"Task did not complete successfully after {max_attempts} attempts" + + +def test_snapshot_filters(): + response = client.post( + "/v1/snapshot/", + json={ + "fileName": "Example export with all features", + "geometry": { + "type": "Polygon", + "coordinates": [ + [ + [83.585701, 28.046607], + [83.585701, 28.382561], + [84.391823, 28.382561], + [84.391823, 28.046607], + [83.585701, 28.046607], + ] + ], + }, + "outputType": "geojson", + "geometryType": ["point", "line", "polygon"], + "filters": { + "tags": { + "point": { + "join_or": { + "amenity": [ + "bank", + "ferry_terminal", + "bus_station", + "fuel", + "kindergarten", + "school", + "college", + "university", + "place_of_worship", + "marketplace", + "clinic", + "hospital", + "police", + "fire_station", + ], + "building": [ + "bank", + "aerodrome", + "ferry_terminal", + "train_station", + "bus_station", + "pumping_station", + "power_substation", + "kindergarten", + "school", + "college", + "university", + "mosque ", + " church ", + " temple", + "supermarket", + "marketplace", + "clinic", + "hospital", + "police", + "fire_station", + "stadium ", + " sports_centre", + "governor_office ", + " townhall ", + " subdistrict_office ", + " village_office ", + " community_group_office", + "government_office", + ], + "man_made": ["tower", "water_tower", "pumping_station"], + "tower:type": ["communication"], + "aeroway": ["aerodrome"], + "railway": ["station"], + "emergency": ["fire_hydrant"], + "landuse": ["reservoir", "recreation_gound"], + "waterway": ["floodgate"], + "natural": ["spring"], + "power": ["tower", "substation"], + "shop": ["supermarket"], + "leisure": [ + "stadium ", + " sports_centre ", + " pitch ", + " swimming_pool", + "park", + ], + "office": ["government"], + } + }, + "line": { + "join_or": { + "highway": [ + "motorway ", + " trunk ", + " primary ", + " secondary ", + " tertiary ", + " service ", + " residential ", + " pedestrian ", + " path ", + " living_street ", + " track", + ], + "railway": ["rail"], + "man_made": ["embankment"], + "waterway": [], + } + }, + "polygon": { + "join_or": { + "amenity": [ + "bank", + "ferry_terminal", + "bus_station", + "fuel", + "kindergarten", + "school", + "college", + "university", + "place_of_worship", + "marketplace", + "clinic", + "hospital", + "police", + "fire_station", + ], + "building": [ + "bank", + "aerodrome", + "ferry_terminal", + "train_station", + "bus_station", + "pumping_station", + "power_substation", + "power_plant", + "kindergarten", + "school", + "college", + "university", + "mosque ", + " church ", + " temple", + "supermarket", + "marketplace", + "clinic", + "hospital", + "police", + "fire_station", + "stadium ", + " sports_centre", + "governor_office ", + " townhall ", + " subdistrict_office ", + " village_office ", + " community_group_office", + "government_office", + ], + "man_made": ["tower", "water_tower", "pumping_station"], + "tower:type": ["communication"], + "aeroway": ["aerodrome"], + "railway": ["station"], + "landuse": ["reservoir", "recreation_gound"], + "waterway": [], + "natural": ["spring"], + "power": ["substation", "plant"], + "shop": ["supermarket"], + "leisure": [ + "stadium ", + " sports_centre ", + " pitch ", + " swimming_pool", + "park", + ], + "office": ["government"], + "type": ["boundary"], + "boundary": ["administrative"], + }, + }, + }, + "attributes": { + "point": [ + "building", + "ground_floor:height", + "capacity:persons", + "building:structure", + "building:condition", + "name", + "admin_level", + "building:material", + "office", + "building:roof", + "backup_generator", + "access:roof", + "building:levels", + "building:floor", + "addr:full", + "addr:city", + "source", + ], + "line": ["width", "source", "waterway", "name"], + "polygon": [ + "landslide_prone", + "name", + "admin_level", + "type", + "is_in:town", + "flood_prone", + "is_in:province", + "is_in:city", + "is_in:municipality", + "is_in:RW", + "is_in:village", + "source", + "boundary", + ], + }, + }, + }, + ) + assert response.status_code == 200 + res = response.json() + track_link = res["track_link"] + max_attempts = 6 + interval_seconds = 10 + for attempt in range(1, max_attempts + 1): + time.sleep(interval_seconds) # wait for worker to complete task + + response = client.get(f"/v1{track_link}") + assert response.status_code == 200 + res = response.json() + check_status = res["status"] + + if check_status == "SUCCESS": + break # exit the loop if the status is SUCCESS + + if attempt == max_attempts: + # If max_attempts reached and status is not SUCCESS, raise an AssertionError + assert ( + False + ), f"Task did not complete successfully after {max_attempts} attempts" + + +def test_snapshot_and_filter(): + response = client.post( + "/v1/snapshot/", + json={ + "fileName": "Destroyed_Buildings_Turkey", + "geometry": { + "type": "Polygon", + "coordinates": [ + [ + [36.70588085657477, 37.1979648807274], + [36.70588085657477, 37.1651408422983], + [36.759267544807194, 37.1651408422983], + [36.759267544807194, 37.1979648807274], + [36.70588085657477, 37.1979648807274], + ] + ], + }, + "outputType": "geojson", + "geometryType": ["polygon"], + "filters": { + "tags": { + "point": {}, + "line": {}, + "polygon": { + "join_or": {}, + "join_and": { + "destroyed:building": ["yes"], + "damage:date": ["2023-02-06"], + }, + }, + }, + "attributes": { + "point": [], + "line": [], + "polygon": [ + "building", + "destroyed:building", + "damage:date", + "name", + "source", + ], + }, + }, + }, + ) + assert response.status_code == 200 + res = response.json() + track_link = res["track_link"] + max_attempts = 6 + interval_seconds = 10 + for attempt in range(1, max_attempts + 1): + time.sleep(interval_seconds) # wait for worker to complete task + + response = client.get(f"/v1{track_link}") + assert response.status_code == 200 + res = response.json() + check_status = res["status"] + + if check_status == "SUCCESS": + break # exit the loop if the status is SUCCESS + + if attempt == max_attempts: + # If max_attempts reached and status is not SUCCESS, raise an AssertionError + assert ( + False + ), f"Task did not complete successfully after {max_attempts} attempts" + + +def test_snapshot_authentication_uuid(): + headers = {"access-token": access_token} + payload = { + "geometry": { + "type": "Polygon", + "coordinates": [ + [ + [83.96919250488281, 28.194446860487773], + [83.99751663208006, 28.194446860487773], + [83.99751663208006, 28.214869548073377], + [83.96919250488281, 28.214869548073377], + [83.96919250488281, 28.194446860487773], + ] + ], + }, + "uuid": False, + } + + response = client.post("/v1/snapshot/", json=payload, headers=headers) + + assert response.status_code == 200 + res = response.json() + track_link = res["track_link"] + max_attempts = 6 + interval_seconds = 10 + for attempt in range(1, max_attempts + 1): + time.sleep(interval_seconds) # wait for worker to complete task + + response = client.get(f"/v1{track_link}") + assert response.status_code == 200 + res = response.json() + check_status = res["status"] + + if check_status == "SUCCESS": + break # exit the loop if the status is SUCCESS + + if attempt == max_attempts: + # If max_attempts reached and status is not SUCCESS, raise an AssertionError + assert ( + False + ), f"Task did not complete successfully after {max_attempts} attempts" + + +def test_snapshot_bind_zip(): + headers = {"access-token": access_token} + payload = { + "geometry": { + "type": "Polygon", + "coordinates": [ + [ + [83.96919250488281, 28.194446860487773], + [83.99751663208006, 28.194446860487773], + [83.99751663208006, 28.214869548073377], + [83.96919250488281, 28.214869548073377], + [83.96919250488281, 28.194446860487773], + ] + ], + }, + "bindZip": False, + } + + response = client.post("/v1/snapshot/", json=payload, headers=headers) + + assert response.status_code == 200 + res = response.json() + track_link = res["track_link"] + max_attempts = 6 + interval_seconds = 10 + for attempt in range(1, max_attempts + 1): + time.sleep(interval_seconds) # wait for worker to complete task + + response = client.get(f"/v1{track_link}") + assert response.status_code == 200 + res = response.json() + check_status = res["status"] + + if check_status == "SUCCESS": + break # exit the loop if the status is SUCCESS + + if attempt == max_attempts: + # If max_attempts reached and status is not SUCCESS, raise an AssertionError + assert ( + False + ), f"Task did not complete successfully after {max_attempts} attempts" + + +## Snapshot Plain + + +def test_snapshot_plain(): + response = client.post( + "/v1/snapshot/plain/", + json={ + "geometry": { + "type": "Polygon", + "coordinates": [ + [ + [83.96919250488281, 28.194446860487773], + [83.99751663208006, 28.194446860487773], + [83.99751663208006, 28.214869548073377], + [83.96919250488281, 28.214869548073377], + [83.96919250488281, 28.194446860487773], + ] + ], + } + }, + ) + assert response.status_code == 200 + + +## Stats + + +def test_stats_endpoint_custom_polygon(): + headers = {"access-token": access_token} + payload = { + "geometry": { + "type": "Polygon", + "coordinates": [ + [ + [83.96919250488281, 28.194446860487773], + [83.99751663208006, 28.194446860487773], + [83.99751663208006, 28.214869548073377], + [83.96919250488281, 28.214869548073377], + [83.96919250488281, 28.194446860487773], + ] + ], + } + } + + response = client.post("/v1/stats/polygon/", json=payload, headers=headers) + + assert response.status_code == 200 + res = response.json() + assert ( + res["meta"]["indicators"] + == "https://github.com/hotosm/raw-data-api/tree/develop/docs/src/stats/indicators.md" + ) + + +def test_stats_endpoint_iso3(): + headers = {"access-token": access_token} + payload = {"iso3": "npl"} + + response = client.post("/v1/stats/polygon/", json=payload, headers=headers) + + assert response.status_code == 200 + res = response.json() + assert ( + res["meta"]["indicators"] + == "https://github.com/hotosm/raw-data-api/tree/develop/docs/src/stats/indicators.md" + ) + + +# HDX + + +def test_hdx_submit_normal_iso3(): + headers = {"access-token": access_token} + payload = { + "iso3": "NPL", + "hdx_upload": False, + "categories": [ + { + "Roads": { + "hdx": { + "tags": ["roads", "transportation", "geodata"], + "caveats": "OpenStreetMap data is crowd sourced and cannot be considered to be exhaustive", + }, + "types": ["lines"], + "select": ["name", "highway"], + "where": "tags['highway'] IS NOT NULL", + "formats": ["geojson"], + } + } + ], + } + + response = client.post("/v1/custom/snapshot/", json=payload, headers=headers) + + assert response.status_code == 200 + res = response.json() + track_link = res["track_link"] + max_attempts = 6 + interval_seconds = 10 + for attempt in range(1, max_attempts + 1): + time.sleep(interval_seconds) # wait for worker to complete task + + response = client.get(f"/v1{track_link}") + assert response.status_code == 200 + res = response.json() + check_status = res["status"] + + if check_status == "SUCCESS": + break # exit the loop if the status is SUCCESS + + if attempt == max_attempts: + # If max_attempts reached and status is not SUCCESS, raise an AssertionError + assert ( + False + ), f"Task did not complete successfully after {max_attempts} attempts" + + +def test_hdx_submit_normal_iso3_multiple_format(): + headers = {"access-token": access_token} + payload = { + "iso3": "NPL", + "hdx_upload": False, + "categories": [ + { + "Roads": { + "hdx": { + "tags": ["roads", "transportation", "geodata"], + "caveats": "OpenStreetMap data is crowd sourced and cannot be considered to be exhaustive", + }, + "types": ["lines"], + "select": ["name", "highway"], + "where": "tags['highway'] IS NOT NULL", + "formats": ["geojson", "gpkg", "kml", "shp"], + } + } + ], + } + + response = client.post("/v1/custom/snapshot/", json=payload, headers=headers) + + assert response.status_code == 200 + res = response.json() + track_link = res["track_link"] + max_attempts = 6 + interval_seconds = 10 + for attempt in range(1, max_attempts + 1): + time.sleep(interval_seconds) # wait for worker to complete task + + response = client.get(f"/v1{track_link}") + assert response.status_code == 200 + res = response.json() + check_status = res["status"] + + if check_status == "SUCCESS": + break # exit the loop if the status is SUCCESS + + if attempt == max_attempts: + # If max_attempts reached and status is not SUCCESS, raise an AssertionError + assert ( + False + ), f"Task did not complete successfully after {max_attempts} attempts" + + +def test_hdx_submit_normal_custom_polygon(): + headers = {"access-token": access_token} + payload = { + "geometry": { + "type": "Polygon", + "coordinates": [ + [ + [83.96919250488281, 28.194446860487773], + [83.99751663208006, 28.194446860487773], + [83.99751663208006, 28.214869548073377], + [83.96919250488281, 28.214869548073377], + [83.96919250488281, 28.194446860487773], + ] + ], + }, + "hdx_upload": False, + "dataset": { + "subnational": True, + "dataset_title": "Pokhara", + "dataset_prefix": "hotosm_pkr", + "dataset_locations": ["npl"], + }, + "categories": [ + { + "Roads": { + "hdx": { + "tags": ["roads", "transportation", "geodata"], + "caveats": "OpenStreetMap data is crowd sourced and cannot be considered to be exhaustive", + }, + "types": ["lines"], + "select": ["name", "highway"], + "where": "tags['highway'] IS NOT NULL", + "formats": ["geojson"], + } + } + ], + } + + response = client.post("/v1/custom/snapshot/", json=payload, headers=headers) + + assert response.status_code == 200 + res = response.json() + track_link = res["track_link"] + max_attempts = 6 + interval_seconds = 10 + for attempt in range(1, max_attempts + 1): + time.sleep(interval_seconds) # wait for worker to complete task + + response = client.get(f"/v1{track_link}") + assert response.status_code == 200 + res = response.json() + check_status = res["status"] + + if check_status == "SUCCESS": + break # exit the loop if the status is SUCCESS + + if attempt == max_attempts: + # If max_attempts reached and status is not SUCCESS, raise an AssertionError + assert ( + False + ), f"Task did not complete successfully after {max_attempts} attempts" + + +def test_custom_submit_normal_custom_polygon_TM_project(): + headers = {"access-token": access_token} + payload = { + "geometry": { + "type": "Polygon", + "coordinates": [ + [ + [83.96919250488281, 28.194446860487773], + [83.99751663208006, 28.194446860487773], + [83.99751663208006, 28.214869548073377], + [83.96919250488281, 28.214869548073377], + [83.96919250488281, 28.194446860487773], + ] + ], + }, + "queue": "raw_ondemand", + "dataset": { + "dataset_prefix": "hotosm_project_1", + "dataset_folder": "TM", + "dataset_title": "Tasking Manger Project 1", + }, + "categories": [ + { + "Buildings": { + "types": ["polygons"], + "select": [ + "name", + "building", + "building:levels", + "building:materials", + "addr:full", + "addr:housenumber", + "addr:street", + "addr:city", + "office", + "source", + ], + "where": "tags['building'] IS NOT NULL", + "formats": ["geojson", "shp", "kml"], + }, + "Roads": { + "types": ["lines"], + "select": [ + "name", + "highway", + "surface", + "smoothness", + "width", + "lanes", + "oneway", + "bridge", + "layer", + "source", + ], + "where": "tags['highway'] IS NOT NULL", + "formats": ["geojson", "shp", "kml"], + }, + "Waterways": { + "types": ["lines", "polygons"], + "select": [ + "name", + "waterway", + "covered", + "width", + "depth", + "layer", + "blockage", + "tunnel", + "natural", + "water", + "source", + ], + "where": "tags['waterway'] IS NOT NULL OR tags['water'] IS NOT NULL OR tags['natural'] IN ('water','wetland','bay')", + "formats": ["geojson", "shp", "kml"], + }, + "Landuse": { + "types": ["points", "polygons"], + "select": ["name", "amenity", "landuse", "leisure"], + "where": "tags['landuse'] IS NOT NULL", + "formats": ["geojson", "shp", "kml"], + }, + } + ], + } + + response = client.post("/v1/custom/snapshot/", json=payload, headers=headers) + + assert response.status_code == 200 + res = response.json() + track_link = res["track_link"] + max_attempts = 6 + interval_seconds = 10 + for attempt in range(1, max_attempts + 1): + time.sleep(interval_seconds) # wait for worker to complete task + + response = client.get(f"/v1{track_link}") + assert response.status_code == 200 + res = response.json() + check_status = res["status"] + + if check_status == "SUCCESS": + break # exit the loop if the status is SUCCESS + + if attempt == max_attempts: + # If max_attempts reached and status is not SUCCESS, raise an AssertionError + assert ( + False + ), f"Task did not complete successfully after {max_attempts} attempts" + + +def test_hdx_submit_normal_custom_polygon_upload(): + headers = {"access-token": access_token} + payload = { + "geometry": { + "type": "Polygon", + "coordinates": [ + [ + [83.96919250488281, 28.194446860487773], + [83.99751663208006, 28.194446860487773], + [83.99751663208006, 28.214869548073377], + [83.96919250488281, 28.214869548073377], + [83.96919250488281, 28.194446860487773], + ] + ], + }, + "hdx_upload": True, + "dataset": { + "subnational": True, + "dataset_title": "Pokhara", + "dataset_folder": "Test", + "dataset_prefix": "hotosm_pkr", + "dataset_locations": ["npl"], + }, + "categories": [ + { + "Roads": { + "hdx": { + "tags": ["roads", "transportation", "geodata"], + "caveats": "OpenStreetMap data is crowd sourced and cannot be considered to be exhaustive", + }, + "types": ["lines"], + "select": ["name", "highway"], + "where": "tags['highway'] IS NOT NULL", + "formats": ["geojson"], + } + } + ], + } + + response = client.post("/v1/custom/snapshot/", json=payload, headers=headers) + + assert response.status_code == 200 + res = response.json() + track_link = res["track_link"] + max_attempts = 6 + interval_seconds = 10 + for attempt in range(1, max_attempts + 1): + time.sleep(interval_seconds) # wait for worker to complete task + + response = client.get(f"/v1{track_link}") + assert response.status_code == 200 + res = response.json() + check_status = res["status"] + + if check_status == "SUCCESS": + break # exit the loop if the status is SUCCESS + + if attempt == max_attempts: + # If max_attempts reached and status is not SUCCESS, raise an AssertionError + assert ( + False + ), f"Task did not complete successfully after {max_attempts} attempts" def test_full_hdx_set_iso(): diff --git a/tests/test_app.py b/tests/test_app.py index 9cdbfeda..3ce2c796 100644 --- a/tests/test_app.py +++ b/tests/test_app.py @@ -1,274 +1,274 @@ -# # Copyright (C) 2021 Humanitarian OpenStreetmap Team +# Copyright (C) 2021 Humanitarian OpenStreetmap Team -# # This program is free software: you can redistribute it and/or modify -# # it under the terms of the GNU Affero General Public License as -# # published by the Free Software Foundation, either version 3 of the -# # License, or (at your option) any later version. +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU Affero General Public License as +# published by the Free Software Foundation, either version 3 of the +# License, or (at your option) any later version. -# # This program is distributed in the hope that it will be useful, -# # but WITHOUT ANY WARRANTY; without even the implied warranty of -# # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the -# # GNU Affero General Public License for more details. +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU Affero General Public License for more details. -# # You should have received a copy of the GNU Affero General Public License -# # along with this program. If not, see . +# You should have received a copy of the GNU Affero General Public License +# along with this program. If not, see . -# # Humanitarian OpenStreetmap Team -# # 1100 13th Street NW Suite 800 Washington, D.C. 20005 -# # +# Humanitarian OpenStreetmap Team +# 1100 13th Street NW Suite 800 Washington, D.C. 20005 +# -# from json import dumps +from json import dumps -# from src.query_builder.builder import raw_currentdata_extraction_query -# from src.validation.models import RawDataCurrentParams +from src.query_builder.builder import raw_currentdata_extraction_query +from src.validation.models import RawDataCurrentParams -# def test_rawdata_current_snapshot_geometry_query(): -# test_param = { -# "geometry": { -# "type": "Polygon", -# "coordinates": [ -# [ -# [84.92431640625, 27.766190642387496], -# [85.31982421875, 27.766190642387496], -# [85.31982421875, 28.02592458049937], -# [84.92431640625, 28.02592458049937], -# [84.92431640625, 27.766190642387496], -# ] -# ], -# }, -# "outputType": "geojson", -# "useStWithin": False, -# "filters": { -# "tags": {"point": {"join_or": {"amenity": ["shop", "toilet"]}}}, -# "attributes": {"point": ["name"]}, -# }, -# } -# validated_params = RawDataCurrentParams(**test_param) -# expected_query = """select ST_AsGeoJSON(t0.*) from (select -# osm_id , tableoid::regclass AS osm_type , tags ->> 'name' as name , geom -# from -# nodes -# where -# ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}'))))) and (tags ->> 'amenity' IN ( 'shop' , 'toilet' ))) t0 UNION ALL select ST_AsGeoJSON(t1.*) from (select -# osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom -# from -# ways_line -# where -# ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t1 UNION ALL select ST_AsGeoJSON(t2.*) from (select -# osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom -# from -# ways_poly -# where -# ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t2 UNION ALL select ST_AsGeoJSON(t3.*) from (select -# osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom -# from -# relations -# where -# ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t3""" +def test_rawdata_current_snapshot_geometry_query(): + test_param = { + "geometry": { + "type": "Polygon", + "coordinates": [ + [ + [84.92431640625, 27.766190642387496], + [85.31982421875, 27.766190642387496], + [85.31982421875, 28.02592458049937], + [84.92431640625, 28.02592458049937], + [84.92431640625, 27.766190642387496], + ] + ], + }, + "outputType": "geojson", + "useStWithin": False, + "filters": { + "tags": {"point": {"join_or": {"amenity": ["shop", "toilet"]}}}, + "attributes": {"point": ["name"]}, + }, + } + validated_params = RawDataCurrentParams(**test_param) + expected_query = """select ST_AsGeoJSON(t0.*) from (select + osm_id , tableoid::regclass AS osm_type , tags ->> 'name' as name , geom + from + nodes + where + ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}'))))) and (tags ->> 'amenity' IN ( 'shop' , 'toilet' ))) t0 UNION ALL select ST_AsGeoJSON(t1.*) from (select + osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom + from + ways_line + where + ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t1 UNION ALL select ST_AsGeoJSON(t2.*) from (select + osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom + from + ways_poly + where + ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t2 UNION ALL select ST_AsGeoJSON(t3.*) from (select + osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom + from + relations + where + ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t3""" -# query_result = raw_currentdata_extraction_query( -# validated_params, -# ) -# assert query_result.encode("utf-8") == expected_query.encode("utf-8") + query_result = raw_currentdata_extraction_query( + validated_params, + ) + assert query_result.encode("utf-8") == expected_query.encode("utf-8") -# def test_rawdata_current_snapshot_normal_query(): -# test_param = { -# "geometry": { -# "type": "Polygon", -# "coordinates": [ -# [ -# [84.92431640625, 27.766190642387496], -# [85.31982421875, 27.766190642387496], -# [85.31982421875, 28.02592458049937], -# [84.92431640625, 28.02592458049937], -# [84.92431640625, 27.766190642387496], -# ] -# ], -# }, -# "useStWithin": False, -# "outputType": "geojson", -# } -# validated_params = RawDataCurrentParams(**test_param) -# expected_query = """select ST_AsGeoJSON(t0.*) from (select -# osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom -# from -# nodes -# where -# ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t0 UNION ALL select ST_AsGeoJSON(t1.*) from (select -# osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom -# from -# ways_line -# where -# ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t1 UNION ALL select ST_AsGeoJSON(t2.*) from (select -# osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom -# from -# ways_poly -# where -# ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t2 UNION ALL select ST_AsGeoJSON(t3.*) from (select -# osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom -# from -# relations -# where -# ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t3""" -# query_result = raw_currentdata_extraction_query( -# validated_params, -# ) -# assert query_result.encode("utf-8") == expected_query.encode("utf-8") +def test_rawdata_current_snapshot_normal_query(): + test_param = { + "geometry": { + "type": "Polygon", + "coordinates": [ + [ + [84.92431640625, 27.766190642387496], + [85.31982421875, 27.766190642387496], + [85.31982421875, 28.02592458049937], + [84.92431640625, 28.02592458049937], + [84.92431640625, 27.766190642387496], + ] + ], + }, + "useStWithin": False, + "outputType": "geojson", + } + validated_params = RawDataCurrentParams(**test_param) + expected_query = """select ST_AsGeoJSON(t0.*) from (select + osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom + from + nodes + where + ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t0 UNION ALL select ST_AsGeoJSON(t1.*) from (select + osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom + from + ways_line + where + ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t1 UNION ALL select ST_AsGeoJSON(t2.*) from (select + osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom + from + ways_poly + where + ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t2 UNION ALL select ST_AsGeoJSON(t3.*) from (select + osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom + from + relations + where + ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t3""" + query_result = raw_currentdata_extraction_query( + validated_params, + ) + assert query_result.encode("utf-8") == expected_query.encode("utf-8") -# def test_rawdata_current_snapshot_normal_query_ST_within(): -# test_param = { -# "geometry": { -# "type": "Polygon", -# "coordinates": [ -# [ -# [84.92431640625, 27.766190642387496], -# [85.31982421875, 27.766190642387496], -# [85.31982421875, 28.02592458049937], -# [84.92431640625, 28.02592458049937], -# [84.92431640625, 27.766190642387496], -# ] -# ], -# }, -# "outputType": "geojson", -# } -# validated_params = RawDataCurrentParams(**test_param) -# expected_query = """select ST_AsGeoJSON(t0.*) from (select -# osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom -# from -# nodes -# where -# ST_within(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t0 UNION ALL select ST_AsGeoJSON(t1.*) from (select -# osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom -# from -# ways_line -# where -# ST_within(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t1 UNION ALL select ST_AsGeoJSON(t2.*) from (select -# osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom -# from -# ways_poly -# where -# ST_within(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t2 UNION ALL select ST_AsGeoJSON(t3.*) from (select -# osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom -# from -# relations -# where -# ST_within(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t3""" -# query_result = raw_currentdata_extraction_query( -# validated_params, -# ) -# assert query_result.encode("utf-8") == expected_query.encode("utf-8") +def test_rawdata_current_snapshot_normal_query_ST_within(): + test_param = { + "geometry": { + "type": "Polygon", + "coordinates": [ + [ + [84.92431640625, 27.766190642387496], + [85.31982421875, 27.766190642387496], + [85.31982421875, 28.02592458049937], + [84.92431640625, 28.02592458049937], + [84.92431640625, 27.766190642387496], + ] + ], + }, + "outputType": "geojson", + } + validated_params = RawDataCurrentParams(**test_param) + expected_query = """select ST_AsGeoJSON(t0.*) from (select + osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom + from + nodes + where + ST_within(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t0 UNION ALL select ST_AsGeoJSON(t1.*) from (select + osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom + from + ways_line + where + ST_within(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t1 UNION ALL select ST_AsGeoJSON(t2.*) from (select + osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom + from + ways_poly + where + ST_within(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t2 UNION ALL select ST_AsGeoJSON(t3.*) from (select + osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom + from + relations + where + ST_within(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[84.92431640625, 27.766190642387496], [85.31982421875, 27.766190642387496], [85.31982421875, 28.02592458049937], [84.92431640625, 28.02592458049937], [84.92431640625, 27.766190642387496]]]}')))))) t3""" + query_result = raw_currentdata_extraction_query( + validated_params, + ) + assert query_result.encode("utf-8") == expected_query.encode("utf-8") -# def test_attribute_filter_rawdata(): -# test_param = { -# "geometry": { -# "type": "Polygon", -# "coordinates": [ -# [ -# [83.502574, 27.569073], -# [83.502574, 28.332758], -# [85.556417, 28.332758], -# [85.556417, 27.569073], -# [83.502574, 27.569073], -# ] -# ], -# }, -# "outputType": "geojson", -# "useStWithin": False, -# "geometryType": ["polygon", "line"], -# "filters": { -# "attributes": {"line": ["name"]}, -# "tags": {"all_geometry": {"join_or": {"building": ["yes"]}}}, -# }, -# } -# validated_params = RawDataCurrentParams(**test_param) -# expected_query = """select ST_AsGeoJSON(t0.*) from (select -# osm_id , tableoid::regclass AS osm_type , tags ->> 'name' as name , geom -# from -# ways_line -# where -# ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[83.502574, 27.569073], [83.502574, 28.332758], [85.556417, 28.332758], [85.556417, 27.569073], [83.502574, 27.569073]]]}'))))) and (tags ->> 'building' = 'yes')) t0 UNION ALL select ST_AsGeoJSON(t1.*) from (select -# osm_id , tableoid::regclass AS osm_type , tags ->> 'name' as name , geom -# from -# relations -# where -# ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[83.502574, 27.569073], [83.502574, 28.332758], [85.556417, 28.332758], [85.556417, 27.569073], [83.502574, 27.569073]]]}'))))) and (tags ->> 'building' = 'yes') and (geometrytype(geom)='MULTILINESTRING')) t1 UNION ALL select ST_AsGeoJSON(t2.*) from (select -# osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom -# from -# ways_poly -# where -# (grid = 1187 OR grid = 1188) and (ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[83.502574, 27.569073], [83.502574, 28.332758], [85.556417, 28.332758], [85.556417, 27.569073], [83.502574, 27.569073]]]}')))))) and (tags ->> 'building' = 'yes')) t2 UNION ALL select ST_AsGeoJSON(t3.*) from (select -# osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom -# from -# relations -# where -# ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[83.502574, 27.569073], [83.502574, 28.332758], [85.556417, 28.332758], [85.556417, 27.569073], [83.502574, 27.569073]]]}'))))) and (tags ->> 'building' = 'yes') and (geometrytype(geom)='POLYGON' or geometrytype(geom)='MULTIPOLYGON')) t3""" -# query_result = raw_currentdata_extraction_query( -# validated_params, -# g_id=[[1187], [1188]], -# ) -# assert query_result.encode("utf-8") == expected_query.encode("utf-8") +def test_attribute_filter_rawdata(): + test_param = { + "geometry": { + "type": "Polygon", + "coordinates": [ + [ + [83.502574, 27.569073], + [83.502574, 28.332758], + [85.556417, 28.332758], + [85.556417, 27.569073], + [83.502574, 27.569073], + ] + ], + }, + "outputType": "geojson", + "useStWithin": False, + "geometryType": ["polygon", "line"], + "filters": { + "attributes": {"line": ["name"]}, + "tags": {"all_geometry": {"join_or": {"building": ["yes"]}}}, + }, + } + validated_params = RawDataCurrentParams(**test_param) + expected_query = """select ST_AsGeoJSON(t0.*) from (select + osm_id , tableoid::regclass AS osm_type , tags ->> 'name' as name , geom + from + ways_line + where + ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[83.502574, 27.569073], [83.502574, 28.332758], [85.556417, 28.332758], [85.556417, 27.569073], [83.502574, 27.569073]]]}'))))) and (tags ->> 'building' = 'yes')) t0 UNION ALL select ST_AsGeoJSON(t1.*) from (select + osm_id , tableoid::regclass AS osm_type , tags ->> 'name' as name , geom + from + relations + where + ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[83.502574, 27.569073], [83.502574, 28.332758], [85.556417, 28.332758], [85.556417, 27.569073], [83.502574, 27.569073]]]}'))))) and (tags ->> 'building' = 'yes') and (geometrytype(geom)='MULTILINESTRING')) t1 UNION ALL select ST_AsGeoJSON(t2.*) from (select + osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom + from + ways_poly + where + (grid = 1187 OR grid = 1188) and (ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[83.502574, 27.569073], [83.502574, 28.332758], [85.556417, 28.332758], [85.556417, 27.569073], [83.502574, 27.569073]]]}')))))) and (tags ->> 'building' = 'yes')) t2 UNION ALL select ST_AsGeoJSON(t3.*) from (select + osm_id, tableoid::regclass AS osm_type, version,tags,changeset,timestamp,geom + from + relations + where + ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[83.502574, 27.569073], [83.502574, 28.332758], [85.556417, 28.332758], [85.556417, 27.569073], [83.502574, 27.569073]]]}'))))) and (tags ->> 'building' = 'yes') and (geometrytype(geom)='POLYGON' or geometrytype(geom)='MULTIPOLYGON')) t3""" + query_result = raw_currentdata_extraction_query( + validated_params, + g_id=[[1187], [1188]], + ) + assert query_result.encode("utf-8") == expected_query.encode("utf-8") -# def test_and_filters(): -# test_param = { -# "fileName": "Destroyed_Buildings_Turkey", -# "geometry": { -# "type": "Polygon", -# "coordinates": [ -# [ -# [36.70588085657477, 37.1979648807274], -# [36.70588085657477, 37.1651408422983], -# [36.759267544807194, 37.1651408422983], -# [36.759267544807194, 37.1979648807274], -# [36.70588085657477, 37.1979648807274], -# ] -# ], -# }, -# "outputType": "geojson", -# "useStWithin": False, -# "geometryType": ["polygon"], -# "filters": { -# "tags": { -# "point": {}, -# "line": {}, -# "polygon": { -# "join_or": {}, -# "join_and": { -# "destroyed:building": ["yes"], -# "damage:date": ["2023-02-06"], -# }, -# }, -# }, -# "attributes": { -# "point": [], -# "line": [], -# "polygon": [ -# "building", -# "destroyed:building", -# "damage:date", -# "name", -# "source", -# ], -# }, -# }, -# } -# validated_params = RawDataCurrentParams(**test_param) -# expected_query = """select ST_AsGeoJSON(t0.*) from (select -# osm_id , tableoid::regclass AS osm_type , tags ->> 'building' as building , tags ->> 'destroyed:building' as destroyed_building , tags ->> 'damage:date' as damage_date , tags ->> 'name' as name , tags ->> 'source' as source , geom -# from -# ways_poly -# where -# ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[36.70588085657477, 37.1979648807274], [36.70588085657477, 37.1651408422983], [36.759267544807194, 37.1651408422983], [36.759267544807194, 37.1979648807274], [36.70588085657477, 37.1979648807274]]]}'))))) and (tags ->> 'destroyed:building' = 'yes' AND tags ->> 'damage:date' = '2023-02-06')) t0 UNION ALL select ST_AsGeoJSON(t1.*) from (select -# osm_id , tableoid::regclass AS osm_type , tags ->> 'building' as building , tags ->> 'destroyed:building' as destroyed_building , tags ->> 'damage:date' as damage_date , tags ->> 'name' as name , tags ->> 'source' as source , geom -# from -# relations -# where -# ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[36.70588085657477, 37.1979648807274], [36.70588085657477, 37.1651408422983], [36.759267544807194, 37.1651408422983], [36.759267544807194, 37.1979648807274], [36.70588085657477, 37.1979648807274]]]}'))))) and (tags ->> 'destroyed:building' = 'yes' AND tags ->> 'damage:date' = '2023-02-06') and (geometrytype(geom)='POLYGON' or geometrytype(geom)='MULTIPOLYGON')) t1""" -# query_result = raw_currentdata_extraction_query( -# validated_params, -# ) -# assert query_result.encode("utf-8") == expected_query.encode("utf-8") +def test_and_filters(): + test_param = { + "fileName": "Destroyed_Buildings_Turkey", + "geometry": { + "type": "Polygon", + "coordinates": [ + [ + [36.70588085657477, 37.1979648807274], + [36.70588085657477, 37.1651408422983], + [36.759267544807194, 37.1651408422983], + [36.759267544807194, 37.1979648807274], + [36.70588085657477, 37.1979648807274], + ] + ], + }, + "outputType": "geojson", + "useStWithin": False, + "geometryType": ["polygon"], + "filters": { + "tags": { + "point": {}, + "line": {}, + "polygon": { + "join_or": {}, + "join_and": { + "destroyed:building": ["yes"], + "damage:date": ["2023-02-06"], + }, + }, + }, + "attributes": { + "point": [], + "line": [], + "polygon": [ + "building", + "destroyed:building", + "damage:date", + "name", + "source", + ], + }, + }, + } + validated_params = RawDataCurrentParams(**test_param) + expected_query = """select ST_AsGeoJSON(t0.*) from (select + osm_id , tableoid::regclass AS osm_type , tags ->> 'building' as building , tags ->> 'destroyed:building' as destroyed_building , tags ->> 'damage:date' as damage_date , tags ->> 'name' as name , tags ->> 'source' as source , geom + from + ways_poly + where + ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[36.70588085657477, 37.1979648807274], [36.70588085657477, 37.1651408422983], [36.759267544807194, 37.1651408422983], [36.759267544807194, 37.1979648807274], [36.70588085657477, 37.1979648807274]]]}'))))) and (tags ->> 'destroyed:building' = 'yes' AND tags ->> 'damage:date' = '2023-02-06')) t0 UNION ALL select ST_AsGeoJSON(t1.*) from (select + osm_id , tableoid::regclass AS osm_type , tags ->> 'building' as building , tags ->> 'destroyed:building' as destroyed_building , tags ->> 'damage:date' as damage_date , tags ->> 'name' as name , tags ->> 'source' as source , geom + from + relations + where + ST_intersects(geom,(select ST_Union(ST_makeValid(ST_GEOMFROMGEOJSON('{"type": "Polygon", "coordinates": [[[36.70588085657477, 37.1979648807274], [36.70588085657477, 37.1651408422983], [36.759267544807194, 37.1651408422983], [36.759267544807194, 37.1979648807274], [36.70588085657477, 37.1979648807274]]]}'))))) and (tags ->> 'destroyed:building' = 'yes' AND tags ->> 'damage:date' = '2023-02-06') and (geometrytype(geom)='POLYGON' or geometrytype(geom)='MULTIPOLYGON')) t1""" + query_result = raw_currentdata_extraction_query( + validated_params, + ) + assert query_result.encode("utf-8") == expected_query.encode("utf-8")