Test against a pgstac database

.github/workflows/cicd.yml

@@ -16,6 +16,19 @@ jobs:
     env:
       AWS_DEFAULT_REGION: us-west-2
 
+    services:
+      pgstac:
+        image: ghcr.io/stac-utils/pgstac:v0.7.10
+        env:
+          POSTGRES_USER: username
+          POSTGRES_PASSWORD: password
+          POSTGRES_DB: postgis
+          PGUSER: username
+          PGPASSWORD: password
+          PGDATABASE: postgis
+        ports:
+          - 5432:5432
+
     steps:
       - uses: actions/checkout@v3
 

README.md

@@ -66,6 +66,20 @@ This script is also available at `scripts/sync_env.sh`, which can be invoked wit
 . scripts/sync_env.sh stac-ingestor-env-secret-<stage>
 ```
 
+## Testing
+
+```shell
+pytest
+```
+
+Some tests require a locally-running **pgstac** database, and will be skipped if there isn't one at `postgresql://username:password@localhost:5432/postgis`.
+To run the **pgstac** tests:
+
+```shell
+docker compose up -d
+pytest
+docker compose down
+```
 
 ## License
 

api/requirements.txt

@@ -10,6 +10,7 @@ psycopg[binary,pool]>=3.0.15
 pydantic_ssm_settings>=0.2.0
 pydantic>=1.9.0,<2
 pypgstac==0.7.10
+pystac[jsonschema]>=1.8.4
 python-multipart==0.0.5
 requests>=2.27.1
 s3fs==2023.3.0

api/src/collection.py

@@ -1,10 +1,11 @@
 import os
-from typing import Union
+from typing import Optional, Union
 
 import fsspec
 import xarray as xr
 import xstac
 from pypgstac.db import PgstacDB
+
 from src.schemas import (
     COGDataset,
     DashboardCollection,
@@ -13,6 +14,7 @@
     ZarrDataset,
 )
 from src.utils import (
+    DbCreds,
     IngestionType,
     convert_decimals_to_float,
     get_db_credentials,
@@ -40,8 +42,10 @@ class Publisher:
         "type": "Collection",
         "stac_version": "1.0.0",
     }
+    db_creds: Optional[DbCreds]
 
-    def __init__(self) -> None:
+    def __init__(self, db_creds: Optional[DbCreds] = None) -> None:
+        self.db_creds = db_creds
         self.func_map = {
             DataType.zarr: self.create_zarr_collection,
             DataType.cog: self.create_cog_collection,
@@ -147,9 +151,9 @@ def ingest(self, collection: DashboardCollection):
         does necessary preprocessing,
         and loads into the PgSTAC collection table
         """
-        creds = get_db_credentials(os.environ["DB_SECRET_ARN"])
+        db_creds = self._get_db_credentials()
         collection = [convert_decimals_to_float(collection.dict(by_alias=True))]
-        with PgstacDB(dsn=creds.dsn_string, debug=True) as db:
+        with PgstacDB(dsn=db_creds.dsn_string, debug=True) as db:
             load_into_pgstac(
                 db=db, ingestions=collection, table=IngestionType.collections
             )
@@ -158,7 +162,13 @@ def delete(self, collection_id: str):
         """
         Deletes the collection from the database
         """
-        creds = get_db_credentials(os.environ["DB_SECRET_ARN"])
-        with PgstacDB(dsn=creds.dsn_string, debug=True) as db:
+        db_creds = self._get_db_credentials()
+        with PgstacDB(dsn=db_creds.dsn_string, debug=True) as db:
             loader = VEDALoader(db=db)
             loader.delete_collection(collection_id)
+
+    def _get_db_credentials(self) -> DbCreds:
+        if self.db_creds:
+            return self.db_creds
+        else:
+            return get_db_credentials(os.environ["DB_SECRET_ARN"])

api/tests/conftest.py

@@ -1,9 +1,15 @@
+import datetime
 import os
+from typing import Generator
 
 import boto3
+import psycopg
 import pytest
 from fastapi.testclient import TestClient
 from moto import mock_dynamodb, mock_ssm
+from pypgstac.db import PgstacDB
+from pystac import Collection, Extent, SpatialExtent, TemporalExtent
+from src.schemas import DashboardCollection
 from stac_pydantic import Item
 
 
@@ -145,6 +151,21 @@ def example_stac_item():
     }
 
 
+@pytest.fixture
+def dashboard_collection() -> DashboardCollection:
+    collection = Collection(
+        "test-collection",
+        "A test collection",
+        Extent(
+            SpatialExtent(
+                [[-180, -90, 180, 90]],
+            ),
+            TemporalExtent([[datetime.datetime.utcnow(), None]]),
+        ),
+    )
+    return DashboardCollection.parse_obj(collection.to_dict())
+
+
 @pytest.fixture
 def example_ingestion(example_stac_item):
     from src import schemas
@@ -155,3 +176,14 @@ def example_ingestion(example_stac_item):
         status=schemas.Status.queued,
         item=Item.parse_obj(example_stac_item),
     )
+
+
+@pytest.fixture
+def pgstac() -> Generator[PgstacDB, None, None]:
+    dsn = "postgresql://username:password@localhost:5432/postgis"
+    try:
+        psycopg.connect(dsn)
+    except Exception:
+        pytest.skip(f"could not connect to pgstac database: {dsn}")
+    with PgstacDB(dsn, commit_on_exit=False) as db:
+        yield db

api/tests/test_collection.py

@@ -0,0 +1,32 @@
+import pytest
+from pypgstac.db import PgstacDB
+from pystac import Collection
+from src.collection import Publisher
+from src.schemas import DashboardCollection
+from src.utils import DbCreds
+
+
+@pytest.fixture
+def publisher() -> Publisher:
+    return Publisher(
+        DbCreds(
+            username="username",
+            password="password",
+            host="localhost",
+            port=5432,
+            dbname="postgis",
+            engine="postgresql",
+        )
+    )
+
+
+def test_ingest(
+    pgstac: PgstacDB, publisher: Publisher, dashboard_collection: DashboardCollection
+) -> None:
+    publisher.ingest(dashboard_collection)
+    collection = Collection.from_dict(
+        pgstac.query_one(
+            r"SELECT * FROM pgstac.get_collection(%s)", [dashboard_collection.id]
+        )
+    )
+    collection.validate()

docker-compose.yml

@@ -0,0 +1,15 @@
+version: '3'
+services:
+  database:
+    container_name: pgstac
+    image: ghcr.io/stac-utils/pgstac:v0.7.10
+    environment:
+      - POSTGRES_USER=username
+      - POSTGRES_PASSWORD=password
+      - POSTGRES_DB=postgis
+      - PGUSER=username
+      - PGPASSWORD=password
+      - PGDATABASE=postgis
+    ports:
+      - "5432:5432"
+    command: postgres -N 500