workflow+updated confest+basic_auth_tests

.github/workflows/cicd.yml

@@ -61,43 +61,4 @@ jobs:
           MONGO_DB: stac
           MONGO_USER: root
           MONGO_PASS: example
-          MONGO_PORT: 27017
-
-      - name: Run test suite against Mongo w/ Basic Auth
-        run: |
-          pipenv run pytest -k "basic_auth" -svvv
-        env:
-          MONGO_HOST: 172.17.0.1
-          BACKEND: mongo
-          APP_HOST: 0.0.0.0
-          APP_PORT: 8084
-          ENVIRONMENT: testing
-          MONGO_DB: stac
-          MONGO_USER: root
-          MONGO_PASS: example
-          MONGO_PORT: 27017
-          BASIC_AUTH: >
-            {
-              "public_endpoints": [
-                {"path": "/","method": "GET"},
-                {"path": "/search","method": "GET"}
-              ],
-              "users": [
-                {"username": "admin", "password": "admin", "permissions": "*"},
-                {
-                  "username": "reader",
-                  "password": "reader",
-                  "permissions": [
-                    {"path": "/conformance","method": ["GET"]},
-                    {"path": "/collections/{collection_id}/items/{item_id}","method": ["GET"]},
-                    {"path": "/search","method": ["POST"]},
-                    {"path": "/collections","method": ["GET"]},
-                    {"path": "/collections/{collection_id}","method": ["GET"]},
-                    {"path": "/collections/{collection_id}/items","method": ["GET"]},
-                    {"path": "/queryables","method": ["GET"]},
-                    {"path": "/queryables/collections/{collection_id}/queryables","method": ["GET"]},
-                    {"path": "/_mgmt/ping","method": ["GET"]}
-                  ]
-                }
-              ]
-            }
+          MONGO_PORT: 27017

stac_fastapi/tests/basic_auth/test_basic_auth.py

@@ -2,65 +2,92 @@
 
 import pytest
 
-# - BASIC_AUTH={"public_endpoints":[{"path":"/","method":"GET"},{"path":"/search","method":"GET"}],"users":[{"username":"admin","password":"admin","permissions":"*"},{"username":"reader","password":"reader","permissions":[{"path":"/conformance","method":["GET"]},{"path":"/collections/{collection_id}/items/{item_id}","method":["GET"]},{"path":"/search","method":["POST"]},{"path":"/collections","method":["GET"]},{"path":"/collections/{collection_id}","method":["GET"]},{"path":"/collections/{collection_id}/items","method":["GET"]},{"path":"/queryables","method":["GET"]},{"path":"/queryables/collections/{collection_id}/queryables","method":["GET"]},{"path":"/_mgmt/ping","method":["GET"]}]}]}
-
 
 @pytest.mark.asyncio
-async def test_get_search_not_authenticated(app_client_basic_auth):
-    """Test public endpoint search without authentication"""
+async def test_get_search_not_authenticated(app_client_basic_auth, ctx):
+    """Test public endpoint [GET /search] without authentication"""
     if not os.getenv("BASIC_AUTH"):
         pytest.skip()
-    params = {"query": '{"gsd": {"gt": 14}}'}
+    params = {"id": ctx.item["id"]}
 
     response = await app_client_basic_auth.get("/search", params=params)
 
-    assert response.status_code == 200
-    assert response.json() == {
-        "type": "FeatureCollection",
-        "features": [],
-        "links": [],
-        "context": {"returned": 0, "limit": 10, "matched": 0},
-    }
+    assert response.status_code == 200, response
+    assert response.json()["features"][0]["geometry"] == ctx.item["geometry"]
 
 
 @pytest.mark.asyncio
-async def test_post_search_authenticated(app_client_basic_auth):
-    """Test protected post search with reader auhtentication"""
+async def test_post_search_authenticated(app_client_basic_auth, ctx):
+    """Test protected endpoint [POST /search] with reader auhtentication"""
     if not os.getenv("BASIC_AUTH"):
         pytest.skip()
-    params = {
-        "bbox": [97.504892, -45.254738, 174.321298, -2.431580],
-        "fields": {"exclude": ["properties"]},
-    }
+    params = {"id": ctx.item["id"]}
     headers = {"Authorization": "Basic cmVhZGVyOnJlYWRlcg=="}
 
     response = await app_client_basic_auth.post("/search", json=params, headers=headers)
 
-    assert response.status_code == 200
-    assert response.json() == {
-        "type": "FeatureCollection",
-        "features": [],
-        "links": [],
-        "context": {"returned": 0, "limit": 10, "matched": 0},
-    }
+    assert response.status_code == 200, response
+    assert response.json()["features"][0]["geometry"] == ctx.item["geometry"]
+
+
+@pytest.mark.asyncio
+async def test_delete_resource_anonymous(
+    app_client_basic_auth,
+):
+    """Test protected endpoint [DELETE /collections/{collection_id}] without auhtentication"""
+    if not os.getenv("BASIC_AUTH"):
+        pytest.skip()
+
+    response = await app_client_basic_auth.delete("/collections/test-collection")
+
+    assert response.status_code == 401
+    assert response.json() == {"detail": "Not authenticated"}
 
 
 @pytest.mark.asyncio
-async def test_delete_resource_insufficient_permissions(app_client_basic_auth):
-    """Test protected delete collection with reader auhtentication"""
+async def test_delete_resource_invalid_credentials(app_client_basic_auth, ctx):
+    """Test protected endpoint [DELETE /collections/{collection_id}] with invalid credentials"""
     if not os.getenv("BASIC_AUTH"):
         pytest.skip()
-    headers = {
-        "Authorization": "Basic cmVhZGVyOnJlYWRlcg=="
-    }  # Assuming this is a valid authorization token
+
+    headers = {"Authorization": "Basic YWRtaW46cGFzc3dvcmQ="}
 
     response = await app_client_basic_auth.delete(
-        "/collections/test-collection", headers=headers
+        f"/collections/{ctx.collection['id']}", headers=headers
     )
 
-    assert (
-        response.status_code == 403
-    )  # Expecting a 403 status code for insufficient permissions
+    assert response.status_code == 401
+    assert response.json() == {"detail": "Incorrect username or password"}
+
+
+@pytest.mark.asyncio
+async def test_delete_resource_insufficient_permissions(app_client_basic_auth, ctx):
+    """Test protected endpoint [DELETE /collections/{collection_id}] with reader user which has insufficient permissions"""
+    if not os.getenv("BASIC_AUTH"):
+        pytest.skip()
+
+    headers = {"Authorization": "Basic cmVhZGVyOnJlYWRlcg=="}
+
+    response = await app_client_basic_auth.delete(
+        f"/collections/{ctx.collection['id']}", headers=headers
+    )
+
+    assert response.status_code == 403
     assert response.json() == {
         "detail": "Insufficient permissions for [DELETE /collections/test-collection]"
     }
+
+
+@pytest.mark.asyncio
+async def test_delete_resource_sufficient_permissions(app_client_basic_auth, ctx):
+    """Test protected endpoint [DELETE /collections/{collection_id}] with admin user which has sufficient permissions"""
+    if not os.getenv("BASIC_AUTH"):
+        pytest.skip()
+
+    headers = {"Authorization": "Basic YWRtaW46YWRtaW4="}
+
+    response = await app_client_basic_auth.delete(
+        f"/collections/{ctx.collection['id']}", headers=headers
+    )
+
+    assert response.status_code == 204

stac_fastapi/tests/conftest.py

@@ -227,7 +227,7 @@ async def app_client(app):
 
 
 @pytest_asyncio.fixture(scope="session")
-async def app_auth():
+async def app_basic_auth():
     settings = AsyncSettings()
     extensions = [
         TransactionExtension(
@@ -259,14 +259,39 @@ async def app_auth():
         search_post_request_model=post_request_model,
     )
 
+    os.environ[
+        "BASIC_AUTH"
+    ] = """{
+        "public_endpoints": [
+            {"path": "/", "method": "GET"},
+            {"path": "/search", "method": "GET"}
+        ],
+        "users": [
+            {"username": "admin", "password": "admin", "permissions": "*"},
+            {
+                "username": "reader", "password": "reader",
+                "permissions": [
+                    {"path": "/conformance", "method": ["GET"]},
+                    {"path": "/collections/{collection_id}/items/{item_id}", "method": ["GET"]},
+                    {"path": "/search", "method": ["POST"]},
+                    {"path": "/collections", "method": ["GET"]},
+                    {"path": "/collections/{collection_id}", "method": ["GET"]},
+                    {"path": "/collections/{collection_id}/items", "method": ["GET"]},
+                    {"path": "/queryables", "method": ["GET"]},
+                    {"path": "/queryables/collections/{collection_id}/queryables", "method": ["GET"]},
+                    {"path": "/_mgmt/ping", "method": ["GET"]}
+                ]
+            }
+        ]
+    }"""
     apply_basic_auth(stac_api)
 
     return stac_api.app
 
 
 @pytest_asyncio.fixture(scope="session")
-async def app_client_basic_auth(app_auth):
+async def app_client_basic_auth(app_basic_auth):
     await create_collection_index()
 
-    async with AsyncClient(app=app_auth, base_url="http://test-server") as c:
+    async with AsyncClient(app=app_basic_auth, base_url="http://test-server") as c:
         yield c