mirror of
https://github.com/geoserver/geoserver-cloud.git
synced 2025-12-08 20:16:08 +00:00
Use new coverage API
This commit is contained in:
Cécile Vuilleumier
Gabriel Roldan
parent
0cb1cd72b6
commit
5b14faaadb
@ -10,7 +10,9 @@ packages = [{ include = "tests" }]
|
||||
python = "^3.10"
|
||||
pytest = "^8.3.3"
|
||||
psycopg2-binary = "^2.9.9"
|
||||
geoservercloud = "^0.5.0"
|
||||
geoservercloud = "0.6.0"
|
||||
# For local development of the Python client, uncomment the following line and comment out the line above
|
||||
# geoservercloud = { path = "../../python-geoservercloud", develop = true }
|
||||
sqlalchemy = "^2.0.35"
|
||||
|
||||
|
||||
|
||||
@ -7,61 +7,42 @@ def test_create_cog_coverage(geoserver_factory):
|
||||
store_name = "land_shallow_topo_21600_NW_cog"
|
||||
coverage_name = "land_shallow_topo_NW"
|
||||
geoserver = geoserver_factory(workspace)
|
||||
rest_client = geoserver.rest_service.rest_client
|
||||
|
||||
# Create COG coverage store
|
||||
store_xml = f"""<coverageStore>
|
||||
<name>{store_name}</name>
|
||||
<type>GeoTIFF</type>
|
||||
<enabled>true</enabled>
|
||||
<workspace><name>{workspace}</name></workspace>
|
||||
<url>cog://https://test-data-cog-public.s3.amazonaws.com/public/land_shallow_topo_21600_NW_cog.tif</url>
|
||||
<metadata>
|
||||
<entry key="CogSettings.Key">
|
||||
<cogSettings>
|
||||
<rangeReaderSettings>HTTP</rangeReaderSettings>
|
||||
</cogSettings>
|
||||
</entry>
|
||||
</metadata>
|
||||
</coverageStore>"""
|
||||
|
||||
rest_client = geoserver.rest_service.rest_client
|
||||
endpoints = geoserver.rest_service.rest_endpoints
|
||||
|
||||
response = rest_client.post(
|
||||
endpoints.coveragestores(workspace),
|
||||
data=store_xml,
|
||||
headers={"Content-Type": "application/xml"},
|
||||
content, status = geoserver.create_coverage_store(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=store_name,
|
||||
type="GeoTIFF",
|
||||
url=f"cog://https://test-data-cog-public.s3.amazonaws.com/public/land_shallow_topo_21600_NW_cog.tif",
|
||||
metadata={"cogSettings": {"rangeReaderSettings": "HTTP"}},
|
||||
)
|
||||
assert response.status_code == 201
|
||||
assert status == 201
|
||||
assert content == store_name
|
||||
|
||||
# Create coverage
|
||||
coverage_xml = f"""<coverage>
|
||||
<name>{coverage_name}</name>
|
||||
<nativeName>{store_name}</nativeName>
|
||||
</coverage>"""
|
||||
|
||||
response = rest_client.post(
|
||||
endpoints.coverages(workspace, store_name),
|
||||
data=coverage_xml,
|
||||
headers={"Content-Type": "application/xml"},
|
||||
content, status = geoserver.create_coverage(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=store_name,
|
||||
coverage_name=coverage_name,
|
||||
native_name=store_name,
|
||||
)
|
||||
assert response.status_code == 201
|
||||
assert status == 201
|
||||
assert content == coverage_name
|
||||
|
||||
# Verify the coverage was created - try listing coverages first
|
||||
list_response = rest_client.get(endpoints.coverages(workspace, store_name))
|
||||
assert (
|
||||
list_response.status_code == 200
|
||||
), f"Failed to get coverages: {list_response.status_code} - {list_response.text}"
|
||||
content, status = geoserver.get_coverages(workspace, store_name)
|
||||
assert status == 200, f"Failed to get coverages: {status} - {content}"
|
||||
assert content[0].get("name") == store_name
|
||||
|
||||
# Check specific coverage
|
||||
response = rest_client.get(endpoints.coverage(workspace, store_name, coverage_name))
|
||||
assert response.status_code == 200
|
||||
content, status = geoserver.get_coverage(workspace, store_name, coverage_name)
|
||||
assert status == 200, f"Failed to get coverage: {status} - {content}"
|
||||
|
||||
# Verify coverage properties
|
||||
coverage_data = response.json()["coverage"]
|
||||
assert coverage_data["name"] == coverage_name
|
||||
assert coverage_data["nativeName"] == coverage_name
|
||||
assert coverage_data["enabled"] == True
|
||||
assert content.get("name") == coverage_name
|
||||
assert content.get("nativeName") == store_name
|
||||
assert content.get("enabled") is True
|
||||
|
||||
# Test WMS GetMap request
|
||||
wms_response = rest_client.get(
|
||||
|
||||
@ -7,8 +7,6 @@ from tests.conftest import (
|
||||
PGSCHEMA,
|
||||
)
|
||||
|
||||
WORKSPACE = "test_pg_datastore"
|
||||
|
||||
|
||||
def test_create_get_and_delete_datastore(geoserver_factory):
|
||||
workspace = "test_pg_datastore"
|
||||
|
||||
@ -10,6 +10,7 @@ Tests various workflows for creating ImageMosaic stores and layers:
|
||||
All tests use sample data from a shared mount volume at /opt/geoserver_data
|
||||
that is accessible to both the test environment and GeoServer containers.
|
||||
"""
|
||||
|
||||
import tempfile
|
||||
import zipfile
|
||||
from pathlib import Path
|
||||
@ -19,76 +20,34 @@ def test_create_imagemosaic_local_files(geoserver_factory):
|
||||
"""Test creating an ImageMosaic using local sample data files via direct directory approach"""
|
||||
workspace = "local_sampledata"
|
||||
store_name = "ne_pyramid_store"
|
||||
coverage_name = "pyramid"
|
||||
geoserver = geoserver_factory(workspace)
|
||||
|
||||
# Use direct directory approach (like web UI) instead of individual file URLs
|
||||
directory_path = "/opt/geoserver_data/sampledata/ne/pyramid/"
|
||||
|
||||
# Create ImageMosaic store directly from directory
|
||||
response = geoserver.rest_service.rest_client.put(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/external.imagemosaic",
|
||||
data=directory_path,
|
||||
headers={"Content-Type": "text/plain"},
|
||||
content, status = geoserver.create_imagemosaic_store_from_directory(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=store_name,
|
||||
directory_path=directory_path,
|
||||
)
|
||||
assert response.status_code in [
|
||||
201,
|
||||
202,
|
||||
], f"Failed to create ImageMosaic from directory: {response.text}"
|
||||
assert status == 201, f"Failed to create ImageMosaic from directory: {content}"
|
||||
assert content == store_name
|
||||
|
||||
# List available coverages (should be auto-discovered)
|
||||
response = geoserver.rest_service.rest_client.get(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/coverages.xml?list=all"
|
||||
content, status = geoserver.get_coverages(
|
||||
workspace_name=workspace, coveragestore_name=store_name
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to list coverages: {response.text}"
|
||||
|
||||
# Extract the auto-discovered coverage name
|
||||
response_text = response.text
|
||||
import re
|
||||
|
||||
coverage_match = re.search(r"<coverageName>([^<]+)</coverageName>", response_text)
|
||||
assert coverage_match, f"No coverage found in response: {response_text}"
|
||||
|
||||
coverage_name = coverage_match.group(1)
|
||||
assert status == 200, f"Failed to list coverages: {content}"
|
||||
assert content[0].get("name") == coverage_name
|
||||
|
||||
# Check if coverage was auto-created (likely scenario)
|
||||
coverage_response = geoserver.rest_service.rest_client.get(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/coverages/{coverage_name}.json"
|
||||
)
|
||||
|
||||
if coverage_response.status_code == 200:
|
||||
# Coverage was auto-created - this is the normal case
|
||||
coverage_data = coverage_response.json()["coverage"]
|
||||
assert coverage_data["name"] == coverage_name
|
||||
assert coverage_data["nativeName"] == coverage_name
|
||||
assert coverage_data["enabled"] == True
|
||||
else:
|
||||
# Coverage not auto-created, create it manually
|
||||
coverage_xml = f"""<coverage>
|
||||
<name>{coverage_name}</name>
|
||||
<title>Natural Earth Pyramid Mosaic</title>
|
||||
<nativeName>{coverage_name}</nativeName>
|
||||
<enabled>true</enabled>
|
||||
</coverage>"""
|
||||
|
||||
response = geoserver.rest_service.rest_client.post(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/coverages",
|
||||
data=coverage_xml,
|
||||
headers={"Content-Type": "text/xml"},
|
||||
)
|
||||
assert (
|
||||
response.status_code == 201
|
||||
), f"Failed to create coverage: {response.text}"
|
||||
|
||||
# Verify the coverage was created
|
||||
response = geoserver.rest_service.rest_client.get(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/coverages/{coverage_name}.json"
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
coverage_data = response.json()["coverage"]
|
||||
assert coverage_data["name"] == coverage_name
|
||||
assert coverage_data["nativeName"] == coverage_name
|
||||
assert coverage_data["enabled"] == True
|
||||
coverage_data, status = geoserver.get_coverage(workspace, store_name, coverage_name)
|
||||
assert status == 200
|
||||
assert coverage_data.get("name") == coverage_name
|
||||
assert coverage_data.get("nativeName") == coverage_name
|
||||
assert coverage_data.get("enabled") == True
|
||||
|
||||
# Test WMS GetMap request (verify local file mosaic works)
|
||||
wms_response = geoserver.rest_service.rest_client.get(
|
||||
@ -98,12 +57,9 @@ def test_create_imagemosaic_local_files(geoserver_factory):
|
||||
assert wms_response.headers.get("content-type").startswith("image/png")
|
||||
|
||||
# Delete coverage store
|
||||
response = geoserver.rest_service.rest_client.delete(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}?recurse=true"
|
||||
)
|
||||
assert (
|
||||
response.status_code == 200
|
||||
), f"Failed to delete coverage store: {response.text}"
|
||||
content, status = geoserver.delete_coverage_store(workspace, store_name)
|
||||
assert status == 200, f"Failed to delete coverage store: {content}"
|
||||
assert content == ""
|
||||
|
||||
|
||||
def test_create_imagemosaic_manual_granules(geoserver_factory):
|
||||
@ -155,14 +111,13 @@ preparedStatements=false
|
||||
with open(zip_file, "rb") as f:
|
||||
zip_data = f.read()
|
||||
|
||||
response = geoserver.rest_service.rest_client.put(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/file.imagemosaic?configure=none",
|
||||
data=zip_data,
|
||||
headers={"Content-Type": "application/zip"},
|
||||
content, status = geoserver.create_imagemosaic_store_from_properties_zip(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=store_name,
|
||||
properties_zip=zip_data,
|
||||
)
|
||||
assert (
|
||||
response.status_code == 201
|
||||
), f"Failed to create ImageMosaic store: {response.text}"
|
||||
assert status == 201, f"Failed to create ImageMosaic store: {content}"
|
||||
assert content == ""
|
||||
|
||||
# Manually add individual granules from the sample data
|
||||
granule_paths = [
|
||||
@ -174,55 +129,41 @@ preparedStatements=false
|
||||
|
||||
for granule_path in granule_paths:
|
||||
# Use direct file paths (without file:// protocol) for external.imagemosaic
|
||||
response = geoserver.rest_service.rest_client.post(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/external.imagemosaic",
|
||||
data=granule_path,
|
||||
headers={"Content-Type": "text/plain"},
|
||||
content, status = geoserver.publish_granule_to_coverage_store(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=store_name,
|
||||
method="external",
|
||||
granule_path=granule_path,
|
||||
)
|
||||
assert response.status_code in [
|
||||
assert status in [
|
||||
201,
|
||||
202,
|
||||
], f"Failed to add granule {granule_path}: {response.text}"
|
||||
], f"Failed to publish granule {granule_path}: {content}"
|
||||
|
||||
# Initialize the store (list available coverages)
|
||||
response = geoserver.rest_service.rest_client.get(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/coverages.xml?list=all"
|
||||
content, status = geoserver.get_coverages(
|
||||
workspace_name=workspace, coveragestore_name=store_name
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to list coverages: {response.text}"
|
||||
|
||||
# Verify coverage name is available
|
||||
response_text = response.text
|
||||
assert (
|
||||
f"<coverageName>{coverage_name}</coverageName>" in response_text
|
||||
), f"Coverage name '{coverage_name}' not found in response: {response_text}"
|
||||
assert status == 200, f"Failed to list coverages: {content}"
|
||||
assert content[0].get("name") == coverage_name
|
||||
|
||||
# Create layer/coverage
|
||||
coverage_xml = f"""<coverage>
|
||||
<name>{coverage_name}</name>
|
||||
<title>Manual Granules Test Coverage</title>
|
||||
<nativeName>{coverage_name}</nativeName>
|
||||
<enabled>true</enabled>
|
||||
</coverage>"""
|
||||
|
||||
response = geoserver.rest_service.rest_client.post(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/coverages",
|
||||
data=coverage_xml,
|
||||
headers={"Content-Type": "text/xml"},
|
||||
content, status = geoserver.create_coverage(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=store_name,
|
||||
coverage_name=coverage_name,
|
||||
title="Manual Granules Test Coverage",
|
||||
)
|
||||
assert response.status_code == 201, f"Failed to create coverage: {response.text}"
|
||||
assert status == 201, f"Failed to create coverage: {content}"
|
||||
assert content == coverage_name
|
||||
|
||||
# Verify the coverage was created successfully
|
||||
response = geoserver.rest_service.rest_client.get(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/coverages/{coverage_name}.json"
|
||||
)
|
||||
assert (
|
||||
response.status_code == 200
|
||||
), f"Failed to get coverage details: {response.text}"
|
||||
|
||||
coverage_data = response.json()["coverage"]
|
||||
assert coverage_data["name"] == coverage_name
|
||||
assert coverage_data["nativeName"] == coverage_name
|
||||
assert coverage_data["enabled"] == True
|
||||
coverage_data, status = geoserver.get_coverage(workspace, store_name, coverage_name)
|
||||
assert status == 200
|
||||
assert coverage_data.get("name") == coverage_name
|
||||
assert coverage_data.get("nativeName") == coverage_name
|
||||
assert coverage_data.get("enabled") == True
|
||||
assert coverage_data.get("title") == "Manual Granules Test Coverage"
|
||||
|
||||
# Test WMS GetMap request (verify manual granule addition works)
|
||||
wms_response = geoserver.rest_service.rest_client.get(
|
||||
@ -285,67 +226,52 @@ preparedStatements=false
|
||||
with open(zip_file, "rb") as f:
|
||||
zip_data = f.read()
|
||||
|
||||
response = geoserver.rest_service.rest_client.put(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/file.imagemosaic?configure=none",
|
||||
data=zip_data,
|
||||
headers={"Content-Type": "application/zip"},
|
||||
content, status = geoserver.create_imagemosaic_store_from_properties_zip(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=store_name,
|
||||
properties_zip=zip_data,
|
||||
)
|
||||
assert (
|
||||
response.status_code == 201
|
||||
), f"Failed to create ImageMosaic store: {response.text}"
|
||||
assert status == 201, f"Failed to create ImageMosaic store: {content}"
|
||||
assert content == ""
|
||||
|
||||
# Step 3: Harvest granules from directory
|
||||
harvest_path = "/opt/geoserver_data/sampledata/ne/pyramid/"
|
||||
|
||||
response = geoserver.rest_service.rest_client.post(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/external.imagemosaic",
|
||||
data=harvest_path,
|
||||
headers={"Content-Type": "text/plain"},
|
||||
content, status = geoserver.harvest_granules_to_coverage_store(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=store_name,
|
||||
directory_path=harvest_path,
|
||||
)
|
||||
assert response.status_code in [
|
||||
assert status in [
|
||||
201,
|
||||
202,
|
||||
], f"Failed to harvest directory {harvest_path}: {response.text}"
|
||||
], f"Failed to harvest directory {harvest_path}: {content}"
|
||||
assert content == ""
|
||||
|
||||
# Step 4: List available coverages
|
||||
response = geoserver.rest_service.rest_client.get(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/coverages.xml?list=all"
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to list coverages: {response.text}"
|
||||
content, code = geoserver.get_coverages(workspace, store_name)
|
||||
assert code == 200, f"Failed to list coverages: {content}"
|
||||
|
||||
# Verify coverage name is available
|
||||
response_text = response.text
|
||||
assert (
|
||||
f"<coverageName>{coverage_name}</coverageName>" in response_text
|
||||
), f"Coverage name '{coverage_name}' not found in response: {response_text}"
|
||||
assert content[0].get("name") == coverage_name
|
||||
|
||||
# Step 5: Create layer/coverage
|
||||
coverage_xml = f"""<coverage>
|
||||
<name>{coverage_name}</name>
|
||||
<title>Directory Harvest Test Coverage</title>
|
||||
<nativeName>{coverage_name}</nativeName>
|
||||
<enabled>true</enabled>
|
||||
</coverage>"""
|
||||
|
||||
response = geoserver.rest_service.rest_client.post(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/coverages",
|
||||
data=coverage_xml,
|
||||
headers={"Content-Type": "text/xml"},
|
||||
content, status = geoserver.create_coverage(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=store_name,
|
||||
coverage_name=coverage_name,
|
||||
title="Directory Harvest Test Coverage",
|
||||
)
|
||||
assert response.status_code == 201, f"Layer creation failed: {response.text}"
|
||||
assert status == 201, f"Failed to create coverage: {content}"
|
||||
assert content == coverage_name
|
||||
|
||||
# Step 6: Verify the coverage was created successfully
|
||||
response = geoserver.rest_service.rest_client.get(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/coverages/{coverage_name}.json"
|
||||
)
|
||||
assert (
|
||||
response.status_code == 200
|
||||
), f"Failed to get coverage details: {response.text}"
|
||||
|
||||
coverage_data = response.json()["coverage"]
|
||||
assert coverage_data["name"] == coverage_name
|
||||
assert coverage_data["nativeName"] == coverage_name
|
||||
assert coverage_data["enabled"] == True
|
||||
coverage_data, status = geoserver.get_coverage(workspace, store_name, coverage_name)
|
||||
assert status == 200
|
||||
assert coverage_data.get("name") == coverage_name
|
||||
assert coverage_data.get("nativeName") == coverage_name
|
||||
assert coverage_data.get("enabled") == True
|
||||
assert coverage_data.get("title") == "Directory Harvest Test Coverage"
|
||||
|
||||
# Step 7: Test WMS GetMap request
|
||||
wms_response = geoserver.rest_service.rest_client.get(
|
||||
@ -409,48 +335,46 @@ preparedStatements=false
|
||||
with open(zip_file, "rb") as f:
|
||||
zip_data = f.read()
|
||||
|
||||
response = geoserver.rest_service.rest_client.put(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/file.imagemosaic?configure=none",
|
||||
data=zip_data,
|
||||
headers={"Content-Type": "application/zip"},
|
||||
content, status = geoserver.create_imagemosaic_store_from_properties_zip(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=store_name,
|
||||
properties_zip=zip_data,
|
||||
)
|
||||
assert (
|
||||
response.status_code == 201
|
||||
), f"Failed to create ImageMosaic store: {response.text}"
|
||||
assert status == 201, f"Failed to create ImageMosaic store: {content}"
|
||||
assert content == ""
|
||||
|
||||
# Step 3: Harvest single file
|
||||
single_file_path = "/opt/geoserver_data/sampledata/ne/NE1_LR_LC_SR_W_DR.tif"
|
||||
|
||||
response = geoserver.rest_service.rest_client.post(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/external.imagemosaic",
|
||||
data=single_file_path,
|
||||
headers={"Content-Type": "text/plain"},
|
||||
content, status = geoserver.harvest_granules_to_coverage_store(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=store_name,
|
||||
directory_path=single_file_path,
|
||||
)
|
||||
assert response.status_code in [
|
||||
201,
|
||||
202,
|
||||
], f"Failed to harvest file {single_file_path}: {response.text}"
|
||||
assert status in [201, 202], f"Failed to harvest file {single_file_path}: {content}"
|
||||
assert content == ""
|
||||
|
||||
# Step 4: List and create layer
|
||||
response = geoserver.rest_service.rest_client.get(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/coverages.xml?list=all"
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to list coverages: {response.text}"
|
||||
content, code = geoserver.get_coverages(workspace, store_name)
|
||||
assert code == 200
|
||||
assert content[0].get("name") == coverage_name
|
||||
|
||||
# Create layer/coverage
|
||||
coverage_xml = f"""<coverage>
|
||||
<name>{coverage_name}</name>
|
||||
<title>Single File Harvest Test Coverage</title>
|
||||
<nativeName>{coverage_name}</nativeName>
|
||||
<enabled>true</enabled>
|
||||
</coverage>"""
|
||||
|
||||
response = geoserver.rest_service.rest_client.post(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/coverages",
|
||||
data=coverage_xml,
|
||||
headers={"Content-Type": "text/xml"},
|
||||
content, status = geoserver.create_coverage(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=store_name,
|
||||
coverage_name=coverage_name,
|
||||
title="Single File Harvest Test Coverage",
|
||||
)
|
||||
assert response.status_code == 201, f"Layer creation failed: {response.text}"
|
||||
assert status == 201, f"Failed to create coverage: {content}"
|
||||
assert content == coverage_name
|
||||
|
||||
content, status = geoserver.get_coverage(workspace, store_name, coverage_name)
|
||||
assert status == 200
|
||||
assert content.get("name") == coverage_name
|
||||
assert content.get("nativeName") == coverage_name
|
||||
assert content.get("enabled") == True
|
||||
assert content.get("title") == "Single File Harvest Test Coverage"
|
||||
|
||||
# Verify WMS works
|
||||
wms_response = geoserver.rest_service.rest_client.get(
|
||||
@ -461,65 +385,47 @@ preparedStatements=false
|
||||
assert wms_response.headers.get("content-type").startswith("image/png")
|
||||
|
||||
|
||||
def test_create_imagemosaic_via_xml_store_creation(geoserver_factory):
|
||||
def test_create_imagemosaic_via_store_definition(geoserver_factory):
|
||||
"""
|
||||
Test creating an ImageMosaic store via XML store creation (not file upload).
|
||||
Test creating an ImageMosaic store via JSON store definition (not file upload).
|
||||
This tests direct store creation pointing to a directory.
|
||||
"""
|
||||
workspace = "xml_store_creation"
|
||||
store_name = "xml_store_creation_store"
|
||||
workspace = "json_store_creation"
|
||||
store_name = "json_store_creation_store"
|
||||
geoserver = geoserver_factory(workspace)
|
||||
|
||||
# Step 2: Create ImageMosaic store via XML store creation
|
||||
store_xml = f"""<coverageStore>
|
||||
<name>{store_name}</name>
|
||||
<workspace>
|
||||
<name>{workspace}</name>
|
||||
</workspace>
|
||||
<type>ImageMosaic</type>
|
||||
<enabled>true</enabled>
|
||||
<url>/opt/geoserver_data/sampledata/ne/pyramid/</url>
|
||||
</coverageStore>"""
|
||||
|
||||
response = geoserver.rest_service.rest_client.post(
|
||||
f"/rest/workspaces/{workspace}/coveragestores",
|
||||
data=store_xml,
|
||||
headers={"Content-Type": "text/xml"},
|
||||
# Step 2: Create ImageMosaic store via JSON store definition
|
||||
content, status = geoserver.create_coverage_store(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=store_name,
|
||||
url="/opt/geoserver_data/sampledata/ne/pyramid/",
|
||||
)
|
||||
assert (
|
||||
response.status_code == 201
|
||||
), f"Store creation via XML failed: {response.text}"
|
||||
assert status == 201
|
||||
assert content == store_name
|
||||
|
||||
# Step 3: List available coverages
|
||||
response = geoserver.rest_service.rest_client.get(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/coverages.xml?list=all"
|
||||
)
|
||||
assert response.status_code == 200, f"Failed to list coverages: {response.text}"
|
||||
assert (
|
||||
"coverageName" in response.text
|
||||
), f"No coverage found in response: {response.text}"
|
||||
content, status = geoserver.get_coverages(workspace, store_name)
|
||||
assert status == 200
|
||||
assert len(content) == 1
|
||||
|
||||
# Extract coverage name
|
||||
import re
|
||||
|
||||
coverage_match = re.search(r"<coverageName>([^<]+)</coverageName>", response.text)
|
||||
assert coverage_match, f"Could not extract coverage name from: {response.text}"
|
||||
coverage_name = coverage_match.group(1)
|
||||
coverage_name = content[0].get("name")
|
||||
|
||||
# Create layer
|
||||
coverage_xml = f"""<coverage>
|
||||
<name>{coverage_name}</name>
|
||||
<title>XML Store Creation Test Coverage</title>
|
||||
<nativeName>{coverage_name}</nativeName>
|
||||
<enabled>true</enabled>
|
||||
</coverage>"""
|
||||
|
||||
response = geoserver.rest_service.rest_client.post(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{store_name}/coverages",
|
||||
data=coverage_xml,
|
||||
headers={"Content-Type": "text/xml"},
|
||||
content, status = geoserver.create_coverage(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=store_name,
|
||||
coverage_name=coverage_name,
|
||||
title="JSON Store Creation Test Coverage",
|
||||
)
|
||||
assert response.status_code == 201, f"Layer creation failed: {response.text}"
|
||||
assert status == 201
|
||||
assert content == coverage_name
|
||||
|
||||
content, status = geoserver.get_coverage(workspace, store_name, coverage_name)
|
||||
assert status == 200
|
||||
assert content.get("name") == coverage_name
|
||||
assert content.get("nativeName") == coverage_name
|
||||
assert content.get("enabled") == True
|
||||
assert content.get("title") == "JSON Store Creation Test Coverage"
|
||||
|
||||
# Verify WMS works
|
||||
wms_response = geoserver.rest_service.rest_client.get(
|
||||
|
||||
@ -58,56 +58,50 @@ preparedStatements=false
|
||||
with open(zip_file, "rb") as f:
|
||||
zip_data = f.read()
|
||||
|
||||
response = geoserver.rest_service.rest_client.put(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{coverage}/file.imagemosaic?configure=none",
|
||||
data=zip_data,
|
||||
headers={"Content-Type": "application/zip"},
|
||||
content, status = geoserver.create_imagemosaic_store_from_properties_zip(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=coverage,
|
||||
properties_zip=zip_data,
|
||||
)
|
||||
assert response.status_code == 201
|
||||
assert status == 201, f"Failed to create ImageMosaic store: {content}"
|
||||
assert content == ""
|
||||
|
||||
# Add granules
|
||||
for uri in granules:
|
||||
response = geoserver.rest_service.rest_client.post(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{coverage}/remote.imagemosaic",
|
||||
data=uri,
|
||||
headers={"Content-Type": "text/plain"},
|
||||
content, status = geoserver.publish_granule_to_coverage_store(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=coverage,
|
||||
method="remote",
|
||||
granule_path=uri,
|
||||
)
|
||||
# Accept both 202 (Accepted) and 201 (Created) as valid responses
|
||||
assert response.status_code in [201, 202]
|
||||
assert status in [
|
||||
201,
|
||||
202,
|
||||
], f"Failed to publish granule {uri}: {content}"
|
||||
|
||||
# Initialize the store (list available coverages)
|
||||
response = geoserver.rest_service.rest_client.get(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{coverage}/coverages.xml?list=all"
|
||||
content, status = geoserver.get_coverages(
|
||||
workspace_name=workspace, coveragestore_name=coverage
|
||||
)
|
||||
assert response.status_code == 200
|
||||
|
||||
assert status == 200
|
||||
# Verify coverage name in response
|
||||
response_text = response.text
|
||||
assert f"<coverageName>{coverage}</coverageName>" in response_text
|
||||
assert content[0].get("name") == coverage
|
||||
|
||||
# Configure the coverage
|
||||
coverage_xml = f"""<coverage>
|
||||
<name>{coverage}</name>
|
||||
<title>{title}</title>
|
||||
<nativeName>{coverage}</nativeName>
|
||||
<enabled>true</enabled>
|
||||
</coverage>"""
|
||||
|
||||
response = geoserver.rest_service.rest_client.post(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{coverage}/coverages",
|
||||
data=coverage_xml,
|
||||
headers={"Content-Type": "text/xml"},
|
||||
content, status = geoserver.create_coverage(
|
||||
workspace_name=workspace,
|
||||
coveragestore_name=coverage,
|
||||
coverage_name=coverage,
|
||||
title=title,
|
||||
)
|
||||
assert response.status_code == 201
|
||||
assert status == 201, f"Failed to create coverage: {content}"
|
||||
assert content == coverage
|
||||
|
||||
# Verify the coverage was created
|
||||
response = geoserver.rest_service.rest_client.get(
|
||||
f"/rest/workspaces/{workspace}/coveragestores/{coverage}/coverages/{coverage}.json"
|
||||
)
|
||||
assert response.status_code == 200
|
||||
coverage_data, code = geoserver.get_coverage(workspace, coverage, coverage)
|
||||
assert code == 200
|
||||
|
||||
# Verify coverage properties
|
||||
coverage_data = response.json()["coverage"]
|
||||
assert coverage_data["name"] == coverage
|
||||
assert coverage_data["nativeName"] == coverage
|
||||
assert coverage_data["enabled"] == True
|
||||
|
||||
Loading…
x
Reference in New Issue
Block a user