OrgTJWater/TJWaterServerBinary
2
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

refactor(admin): remove geoserver config

Browse Source
This commit is contained in:
jiang
2026-06-12 15:28:14 +08:00
parent a6e7a2e75c
commit d99f4cec6a
14 changed files with 14 additions and 420 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.env.example
+3 -9
View File
@@ -4,18 +4,12 @@
ENVIRONMENT="production"
NETWORK_NAME="tjwater"
# ============================================
# 安全配置 (必填)
# 敏感配置加密 (必填)
# ============================================
# JWT 密钥 - 用于生成和验证 Token
# 生成方式: openssl rand -hex 32
SECRET_KEY=your-secret-key-here-change-in-production-use-openssl-rand-hex-32
# 数据加密密钥 - Fernet 格式，生产环境必须替换为独立密钥
# Fernet 格式，生产环境必须替换为独立密钥
# 生成方式: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
# ENCRYPTION_KEY 用于 GeoServer 管理密码等通用敏感配置
ENCRYPTION_KEY="replace-with-generated-fernet-key"
# DATABASE_ENCRYPTION_KEY 专用于 project_databases.dsn_encrypted
# 用于项目数据库 DSN、GeoServer 管理密码等敏感配置
DATABASE_ENCRYPTION_KEY="replace-with-generated-fernet-key"
# ============================================
app/api/v1/endpoints/admin_metadata.py
+1 -110
View File
@@ -26,8 +26,6 @@ from app.domain.schemas.admin_metadata import (
ProjectDatabaseResponse,
ProjectDatabaseUpsertRequest,
ProjectDbRole,
ProjectGeoServerConfigResponse,
ProjectGeoServerConfigUpsertRequest,
ProjectMemberCreateRequest,
ProjectMemberResponse,
ProjectMemberUpdateRequest,
@@ -69,23 +67,6 @@ def _project_database_response(
)
def _geoserver_config_response(
record: models.ProjectGeoServerConfig,
) -> ProjectGeoServerConfigResponse:
return ProjectGeoServerConfigResponse(
id=record.id,
project_id=record.project_id,
gs_base_url=record.gs_base_url,
gs_admin_user=record.gs_admin_user,
gs_datastore_name=record.gs_datastore_name,
default_extent=record.default_extent,
srid=record.srid,
configured=True,
has_password=bool(record.gs_admin_password_encrypted),
updated_at=record.updated_at,
)
def _database_audit_payload(payload: ProjectDatabaseUpsertRequest) -> dict:
return {
"db_role": payload.db_role,
@@ -96,19 +77,6 @@ def _database_audit_payload(payload: ProjectDatabaseUpsertRequest) -> dict:
}
def _geoserver_audit_payload(
payload: ProjectGeoServerConfigUpsertRequest,
) -> dict:
return {
"gs_base_url": payload.gs_base_url,
"gs_admin_user": payload.gs_admin_user,
"gs_datastore_name": payload.gs_datastore_name,
"default_extent": payload.default_extent,
"srid": payload.srid,
"password_updated": "gs_admin_password" in payload.model_fields_set,
}
def _to_async_sqlalchemy_url(dsn: str) -> str:
parsed = make_url(dsn)
if parsed.drivername in {"postgresql", "postgres"}:
@@ -123,7 +91,7 @@ def _db_type_for_role(db_role: str) -> str:
def _status_for_config_value_error(exc: ValueError) -> int:
if "ENCRYPTION_KEY" in str(exc):
if "DATABASE_ENCRYPTION_KEY" in str(exc):
return status.HTTP_503_SERVICE_UNAVAILABLE
return status.HTTP_400_BAD_REQUEST
@@ -535,83 +503,6 @@ async def check_project_database_health(
)
@router.get(
"/projects/{project_id}/geoserver",
response_model=ProjectGeoServerConfigResponse,
)
async def get_project_geoserver_config(
project_id: UUID = Path(...),
current_user=Depends(get_current_metadata_admin),
metadata_repo: MetadataRepository = Depends(get_metadata_repository),
) -> ProjectGeoServerConfigResponse:
project = await metadata_repo.get_project_by_id(project_id)
if project is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
record = await metadata_repo.get_geoserver_config_record(project_id)
if record is None:
return ProjectGeoServerConfigResponse(
id=None,
project_id=project_id,
gs_base_url=None,
gs_admin_user=None,
gs_datastore_name="ds_postgis",
default_extent=None,
srid=4326,
configured=False,
has_password=False,
updated_at=None,
)
return _geoserver_config_response(record)
@router.put(
"/projects/{project_id}/geoserver",
response_model=ProjectGeoServerConfigResponse,
)
async def upsert_project_geoserver_config(
payload: ProjectGeoServerConfigUpsertRequest,
project_id: UUID = Path(...),
current_user=Depends(get_current_metadata_admin),
metadata_repo: MetadataRepository = Depends(get_metadata_repository),
) -> ProjectGeoServerConfigResponse:
project = await metadata_repo.get_project_by_id(project_id)
if project is None:
raise HTTPException(status_code=status.HTTP_404_NOT_FOUND, detail="Project not found")
try:
record = await metadata_repo.upsert_geoserver_config(
project_id,
gs_base_url=payload.gs_base_url,
gs_admin_user=payload.gs_admin_user,
gs_admin_password=payload.gs_admin_password,
password_update_requested="gs_admin_password" in payload.model_fields_set,
gs_datastore_name=payload.gs_datastore_name,
default_extent=payload.default_extent,
srid=payload.srid,
)
except ValueError as exc:
raise HTTPException(
status_code=_status_for_config_value_error(exc),
detail=str(exc),
) from exc
except SQLAlchemyError as exc:
raise HTTPException(
status_code=status.HTTP_503_SERVICE_UNAVAILABLE,
detail=f"Metadata database error: {exc}",
) from exc
await log_audit_event(
action=AuditAction.CONFIG_CHANGE,
user_id=current_user.id,
project_id=project_id,
resource_type="project_geoserver",
resource_id=str(project_id),
request_data=_geoserver_audit_payload(payload),
response_status=status.HTTP_200_OK,
session=metadata_repo.session,
)
return _geoserver_config_response(record)
@router.get("/users/{user_id}", response_model=MetadataUserResponse)
async def get_metadata_user(
user_id: UUID = Path(...),
app/api/v1/endpoints/meta.py
+1 -15
View File
@@ -16,7 +16,6 @@ from app.auth.project_dependencies import (
from app.auth.metadata_dependencies import get_current_metadata_user
from app.core.config import settings
from app.domain.schemas.metadata import (
GeoServerConfigResponse,
ProjectMetaResponse,
ProjectSummaryResponse,
)
@@ -34,25 +33,13 @@ async def get_project_metadata(
"""
获取项目元数据
返回当前项目的完整元数据，包括项目基本信息和GeoServer配置
返回当前项目的完整元数据，包括项目基本信息和项目权限
"""
project = await metadata_repo.get_project_by_id(ctx.project_id)
if not project:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND, detail="Project not found"
)
geoserver = await metadata_repo.get_geoserver_config(ctx.project_id)
geoserver_payload = (
GeoServerConfigResponse(
gs_base_url=geoserver.gs_base_url,
gs_admin_user=geoserver.gs_admin_user,
gs_datastore_name=geoserver.gs_datastore_name,
default_extent=geoserver.default_extent,
srid=geoserver.srid,
)
if geoserver
else None
)
return ProjectMetaResponse(
project_id=project.id,
name=project.name,
@@ -62,7 +49,6 @@ async def get_project_metadata(
map_extent=project.map_extent,
status=project.status,
project_role=ctx.project_role,
geoserver=geoserver_payload,
)
app/api/v1/endpoints/project.py
+1 -13
View File
@@ -4,7 +4,7 @@ from fastapi.responses import PlainTextResponse
from typing import Any, Dict, List
from app.infra.db.metadb.repositories.metadata_repository import MetadataRepository
from app.auth.project_dependencies import get_metadata_repository
from app.domain.schemas.metadata import ProjectMetaResponse, GeoServerConfigResponse
from app.domain.schemas.metadata import ProjectMetaResponse
import app.services.project_info as project_info
from app.infra.db.postgresql.database import get_database_instance as get_pg_db
from app.infra.db.timescaledb.database import get_database_instance as get_ts_db
@@ -55,17 +55,6 @@ async def get_project_info_endpoint(
project_detail = await metadata_repo.get_project_detail_by_code(network)
if not project_detail:
raise HTTPException(status_code=404, detail=f"Project {network} not found")
geoserver_payload = None
if project_detail.geoserver:
geoserver_payload = GeoServerConfigResponse(
gs_base_url=project_detail.geoserver.gs_base_url,
gs_admin_user=project_detail.geoserver.gs_admin_user,
gs_datastore_name=project_detail.geoserver.gs_datastore_name,
default_extent=project_detail.geoserver.default_extent,
srid=project_detail.geoserver.srid,
)
return ProjectMetaResponse(
project_id=project_detail.project_id,
name=project_detail.name,
@@ -75,7 +64,6 @@ async def get_project_info_endpoint(
map_extent=project_detail.map_extent,
status=project_detail.status,
project_role="viewer", # Default role for public access
geoserver=geoserver_payload
)
@router.get("/listprojects/", summary="获取项目列表", description="获取服务器上所有可用的供水管网项目名称列表。")
app/core/config.py
+2 -3
View File
@@ -11,9 +11,8 @@ class Settings(BaseSettings):
NETWORK_NAME: str = "default_network"
# 数据加密密钥 (使用 Fernet)
ENCRYPTION_KEY: str = "" # 必须从环境变量设置
DATABASE_ENCRYPTION_KEY: str = "" # project_databases.dsn_encrypted 专用
# 敏感配置加密密钥 (Fernet)
DATABASE_ENCRYPTION_KEY: str = ""
# Database Config (PostgreSQL)
DB_NAME: str = "tjwater"
app/core/encryption.py
+3 -7
View File
@@ -20,10 +20,10 @@ class Encryptor:
key: 加密密钥，如果为 None 则从环境变量读取
"""
if key is None:
key_str = os.getenv("ENCRYPTION_KEY") or settings.ENCRYPTION_KEY
key_str = os.getenv("DATABASE_ENCRYPTION_KEY") or settings.DATABASE_ENCRYPTION_KEY
if not key_str:
raise ValueError(
"ENCRYPTION_KEY not found in environment variables or .env. "
"DATABASE_ENCRYPTION_KEY not found in environment variables or .env. "
"Generate one using: Encryptor.generate_key()"
)
key = key_str.encode()
@@ -80,15 +80,13 @@ _database_encryptor: Optional[Encryptor] = None
def is_encryption_configured() -> bool:
return bool(os.getenv("ENCRYPTION_KEY") or settings.ENCRYPTION_KEY)
return is_database_encryption_configured()
def is_database_encryption_configured() -> bool:
return bool(
os.getenv("DATABASE_ENCRYPTION_KEY")
or settings.DATABASE_ENCRYPTION_KEY
or os.getenv("ENCRYPTION_KEY")
or settings.ENCRYPTION_KEY
)
@@ -107,8 +105,6 @@ def get_database_encryptor() -> Encryptor:
key_str = (
os.getenv("DATABASE_ENCRYPTION_KEY")
or settings.DATABASE_ENCRYPTION_KEY
or os.getenv("ENCRYPTION_KEY")
or settings.ENCRYPTION_KEY
)
if not key_str:
raise ValueError(
app/domain/schemas/admin_metadata.py
-22
View File
@@ -132,25 +132,3 @@ class ProjectDatabaseHealthResponse(BaseModel):
db_type: str
ok: bool
detail: str
class ProjectGeoServerConfigUpsertRequest(BaseModel):
gs_base_url: str | None = None
gs_admin_user: str | None = Field(default=None, max_length=50)
gs_admin_password: str | None = Field(default=None, min_length=1)
gs_datastore_name: str = Field(default="ds_postgis", min_length=1, max_length=100)
default_extent: dict | None = None
srid: int = Field(default=4326, ge=1)
class ProjectGeoServerConfigResponse(BaseModel):
id: UUID | None = None
project_id: UUID
gs_base_url: str | None = None
gs_admin_user: str | None = None
gs_datastore_name: str
default_extent: dict | None = None
srid: int
configured: bool = True
has_password: bool
updated_at: datetime | None = None
app/domain/schemas/metadata.py
-9
View File
@@ -4,14 +4,6 @@ from uuid import UUID
from pydantic import BaseModel
class GeoServerConfigResponse(BaseModel):
gs_base_url: Optional[str] = None
gs_admin_user: Optional[str] = None
gs_datastore_name: str
default_extent: Optional[dict] = None
srid: int
class ProjectMetaResponse(BaseModel):
project_id: UUID
name: str
@@ -21,7 +13,6 @@ class ProjectMetaResponse(BaseModel):
map_extent: Optional[dict] = None
status: str
project_role: str
geoserver: Optional[GeoServerConfigResponse] = None
class ProjectSummaryResponse(BaseModel):
app/infra/db/metadb/models.py
-20
View File
@@ -64,26 +64,6 @@ class ProjectDatabase(Base):
pool_max_size: Mapped[int] = mapped_column(Integer, default=10)
class ProjectGeoServerConfig(Base):
__tablename__ = "project_geoserver_configs"
id: Mapped[UUID] = mapped_column(PGUUID(as_uuid=True), primary_key=True)
project_id: Mapped[UUID] = mapped_column(
PGUUID(as_uuid=True), unique=True, index=True
)
gs_base_url: Mapped[str | None] = mapped_column(Text, nullable=True)
gs_admin_user: Mapped[str | None] = mapped_column(String(50), nullable=True)
gs_admin_password_encrypted: Mapped[str | None] = mapped_column(
Text, nullable=True
)
gs_datastore_name: Mapped[str] = mapped_column(String(100), default="ds_postgis")
default_extent: Mapped[dict | None] = mapped_column(JSONB, nullable=True)
srid: Mapped[int] = mapped_column(Integer, default=4326)
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True), default=datetime.utcnow
)
class UserProjectMembership(Base):
__tablename__ = "user_project_membership"
app/infra/db/metadb/repositories/metadata_repository.py
-106
View File
@@ -9,9 +9,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
from app.core.encryption import (
get_database_encryptor,
get_encryptor,
is_database_encryption_configured,
is_encryption_configured,
)
from app.infra.db.metadb import models
@@ -44,17 +42,6 @@ class ProjectDbRouting:
pool_max_size: int
@dataclass(frozen=True)
class ProjectGeoServerInfo:
project_id: UUID
gs_base_url: Optional[str]
gs_admin_user: Optional[str]
gs_admin_password: Optional[str]
gs_datastore_name: str
default_extent: Optional[dict]
srid: int
@dataclass(frozen=True)
class ProjectSummary:
project_id: UUID
@@ -76,7 +63,6 @@ class ProjectDetail:
gs_workspace: str
map_extent: Optional[dict]
status: str
geoserver: Optional[ProjectGeoServerInfo]
@dataclass(frozen=True)
@@ -100,12 +86,6 @@ def _encrypt_database_secret(value: str) -> str:
return get_database_encryptor().encrypt(value)
def _encrypt_general_secret(value: str) -> str:
if not is_encryption_configured():
raise ValueError("ENCRYPTION_KEY is not configured")
return get_encryptor().encrypt(value)
class MetadataRepository:
"""元数据访问层（system_hub）"""
@@ -276,9 +256,6 @@ class MetadataRepository:
project = await self.get_project_by_code(code)
if not project:
return None
geoserver = await self.get_geoserver_config(project.id)
return ProjectDetail(
project_id=project.id,
name=project.name,
@@ -287,7 +264,6 @@ class MetadataRepository:
gs_workspace=project.gs_workspace,
map_extent=project.map_extent,
status=project.status,
geoserver=geoserver
)
async def get_membership_role(
@@ -469,88 +445,6 @@ class MetadataRepository:
pool_max_size=record.pool_max_size,
)
async def get_geoserver_config(
self, project_id: UUID
) -> Optional[ProjectGeoServerInfo]:
result = await self.session.execute(
select(models.ProjectGeoServerConfig).where(
models.ProjectGeoServerConfig.project_id == project_id
)
)
record = result.scalar_one_or_none()
if not record:
return None
if record.gs_admin_password_encrypted:
if is_encryption_configured():
encryptor = get_encryptor()
password = encryptor.decrypt(record.gs_admin_password_encrypted)
else:
password = record.gs_admin_password_encrypted
else:
password = None
return ProjectGeoServerInfo(
project_id=record.project_id,
gs_base_url=record.gs_base_url,
gs_admin_user=record.gs_admin_user,
gs_admin_password=password,
gs_datastore_name=record.gs_datastore_name,
default_extent=record.default_extent,
srid=record.srid,
)
async def get_geoserver_config_record(
self, project_id: UUID
) -> Optional[models.ProjectGeoServerConfig]:
result = await self.session.execute(
select(models.ProjectGeoServerConfig).where(
models.ProjectGeoServerConfig.project_id == project_id
)
)
return result.scalar_one_or_none()
async def upsert_geoserver_config(
self,
project_id: UUID,
*,
gs_base_url: str | None,
gs_admin_user: str | None,
gs_admin_password: str | None,
password_update_requested: bool,
gs_datastore_name: str,
default_extent: dict | None,
srid: int,
) -> models.ProjectGeoServerConfig:
record = await self.get_geoserver_config_record(project_id)
encrypted_password: str | None = None
if password_update_requested and gs_admin_password is not None:
encrypted_password = _encrypt_general_secret(gs_admin_password)
if record is None:
record = models.ProjectGeoServerConfig(
id=uuid4(),
project_id=project_id,
gs_base_url=gs_base_url,
gs_admin_user=gs_admin_user,
gs_admin_password_encrypted=encrypted_password,
gs_datastore_name=gs_datastore_name,
default_extent=default_extent,
srid=srid,
updated_at=_utcnow(),
)
self.session.add(record)
else:
record.gs_base_url = gs_base_url
record.gs_admin_user = gs_admin_user
if password_update_requested:
record.gs_admin_password_encrypted = encrypted_password
record.gs_datastore_name = gs_datastore_name
record.default_extent = default_extent
record.srid = srid
record.updated_at = _utcnow()
await self.session.commit()
await self.session.refresh(record)
return record
async def list_projects_for_user(self, user_id: UUID) -> List[ProjectSummary]:
stmt = (
select(models.Project, models.UserProjectMembership.project_role)
resources/sql/005_metadata_project_configuration.sql
-23
View File
@@ -53,26 +53,3 @@ CREATE INDEX IF NOT EXISTS idx_project_databases_project_id
ON project_databases(project_id);
CREATE INDEX IF NOT EXISTS idx_project_databases_role
ON project_databases(db_role);
CREATE TABLE IF NOT EXISTS project_geoserver_configs (
id UUID PRIMARY KEY DEFAULT gen_random_uuid(),
project_id UUID UNIQUE NOT NULL REFERENCES projects(id) ON DELETE CASCADE,
gs_base_url TEXT,
gs_admin_user VARCHAR(50),
gs_admin_password_encrypted TEXT,
gs_datastore_name VARCHAR(100) DEFAULT 'ds_postgis' NOT NULL,
default_extent JSONB,
srid INTEGER DEFAULT 4326 NOT NULL,
updated_at TIMESTAMP WITH TIME ZONE DEFAULT CURRENT_TIMESTAMP NOT NULL,
CONSTRAINT project_geoserver_configs_srid_check CHECK (srid >= 1)
);
CREATE INDEX IF NOT EXISTS idx_project_geoserver_configs_project_id
ON project_geoserver_configs(project_id);
DROP TRIGGER IF EXISTS update_project_geoserver_configs_updated_at
ON project_geoserver_configs;
CREATE TRIGGER update_project_geoserver_configs_updated_at
BEFORE UPDATE ON project_geoserver_configs
FOR EACH ROW
EXECUTE FUNCTION update_updated_at_column();
tests/api/test_admin_metadata_endpoints.py
-69
View File
@@ -15,7 +15,6 @@ from app.domain.schemas.admin_metadata import (
MetadataUserSyncRequest,
MetadataUserUpdateRequest,
ProjectDatabaseUpsertRequest,
ProjectGeoServerConfigUpsertRequest,
ProjectMemberCreateRequest,
ProjectMemberUpdateRequest,
)
@@ -75,22 +74,6 @@ def _database_config(**overrides):
return SimpleNamespace(**data)
def _geoserver_config(**overrides):
data = {
"id": uuid4(),
"project_id": uuid4(),
"gs_base_url": "http://geoserver",
"gs_admin_user": "admin",
"gs_admin_password_encrypted": "encrypted-password",
"gs_datastore_name": "ds_postgis",
"default_extent": {"bbox": [1, 2, 3, 4]},
"srid": 4326,
"updated_at": datetime(2026, 1, 1, tzinfo=timezone.utc),
}
data.update(overrides)
return SimpleNamespace(**data)
def test_to_async_sqlalchemy_url_preserves_password():
url = admin_metadata._to_async_sqlalchemy_url(
"postgresql://tjwater:secret@192.168.1.114:5433/tjwater"
@@ -429,58 +412,6 @@ async def test_project_database_health_sanitizes_password_failures(monkeypatch):
assert "psycopg" not in response.detail
@pytest.mark.anyio
async def test_upsert_geoserver_config_hides_password_and_audits_without_plaintext(monkeypatch):
project_id = uuid4()
record = _geoserver_config(project_id=project_id)
repo = SimpleNamespace(
session=object(),
get_project_by_id=AsyncMock(return_value=_project(id=project_id)),
upsert_geoserver_config=AsyncMock(return_value=record),
)
monkeypatch.setattr(admin_metadata, "log_audit_event", AsyncMock())
response = await admin_metadata.upsert_project_geoserver_config(
ProjectGeoServerConfigUpsertRequest(
gs_base_url="http://geoserver",
gs_admin_user="admin",
gs_admin_password="secret-password",
gs_datastore_name="ds_postgis",
default_extent={"bbox": [1, 2, 3, 4]},
srid=4326,
),
project_id=project_id,
current_user=_user(role="admin", is_superuser=True),
metadata_repo=repo,
)
assert response.has_password is True
assert "password" not in response.model_dump()
request_data = admin_metadata.log_audit_event.await_args.kwargs["request_data"]
assert request_data["password_updated"] is True
assert "secret-password" not in str(request_data)
@pytest.mark.anyio
async def test_get_geoserver_config_returns_empty_state_when_unconfigured():
project_id = uuid4()
repo = SimpleNamespace(
get_project_by_id=AsyncMock(return_value=_project(id=project_id)),
get_geoserver_config_record=AsyncMock(return_value=None),
)
response = await admin_metadata.get_project_geoserver_config(
project_id=project_id,
current_user=_user(role="admin", is_superuser=True),
metadata_repo=repo,
)
assert response.project_id == project_id
assert response.configured is False
assert response.has_password is False
assert response.gs_datastore_name == "ds_postgis"
@pytest.mark.anyio
async def test_metadata_admin_dependency_rejects_non_admin_user():
with pytest.raises(HTTPException) as exc:
tests/api/test_meta_endpoints.py
-5
View File
@@ -36,7 +36,6 @@ def test_meta_project_returns_map_extent(monkeypatch):
project_id = uuid4()
repo = SimpleNamespace(
get_project_by_id=lambda _project_id: None,
get_geoserver_config=lambda _project_id: None,
)
async def get_project_by_id(_project_id):
@@ -50,11 +49,7 @@ def test_meta_project_returns_map_extent(monkeypatch):
status="active",
)
async def get_geoserver_config(_project_id):
return None
repo.get_project_by_id = get_project_by_id
repo.get_geoserver_config = get_geoserver_config
app = build_test_app(module.router, "/api/v1")
app.dependency_overrides[module.get_project_context] = lambda: SimpleNamespace(
tests/api/test_project_endpoints.py
+3 -9
View File
@@ -84,7 +84,7 @@ def test_project_info_returns_404_when_missing(monkeypatch):
assert response.json()["detail"] == "Project missing not found"
def test_project_info_returns_geoserver_payload(monkeypatch):
def test_project_info_returns_project_workspace(monkeypatch):
module = _load_project_module(monkeypatch)
detail = SimpleNamespace(
project_id=uuid4(),
@@ -94,13 +94,6 @@ def test_project_info_returns_geoserver_payload(monkeypatch):
gs_workspace="ws",
map_extent={"xmin": 1, "ymin": 2, "xmax": 3, "ymax": 4},
status="active",
geoserver=SimpleNamespace(
gs_base_url="http://gs",
gs_admin_user="admin",
gs_datastore_name="store",
default_extent={"xmin": 1, "ymin": 2, "xmax": 3, "ymax": 4},
srid=4326,
),
)
repo = SimpleNamespace(get_project_detail_by_code=AsyncMock(return_value=detail))
app = build_test_app(module.router, "/api/v1")
@@ -112,7 +105,8 @@ def test_project_info_returns_geoserver_payload(monkeypatch):
assert response.status_code == 200
payload = response.json()
assert payload["code"] == "demo"
assert payload["geoserver"]["gs_base_url"] == "http://gs"
assert payload["gs_workspace"] == "ws"
assert "geoserver" not in payload
def test_open_project_returns_network_even_when_db_connection_fails(monkeypatch):