Added tests for keycloak, and a few more for the cli

esgf-generator/esgf_generator/cli.py

@@ -15,6 +15,7 @@
 NODE_PORTS = {"east": 9050, "west": 9051}
 ENV_FILE = find_dotenv()
 
+
 if ENV_FILE is None:
     raise Exception("No .env file found, please create one in the root directory")
 
@@ -26,6 +27,7 @@
 
 
 def validate_token() -> bool:
+    load_dotenv(ENV_FILE)
     token = os.getenv("TOKEN")
     if not token:
         return False
@@ -43,6 +45,7 @@ def validate_token() -> bool:
 
 
 def authenticate() -> str:
+    load_dotenv(ENV_FILE)
 
     token = os.getenv("TOKEN")
     if token and validate_token():
@@ -153,8 +156,11 @@ def esgf_generator(
                     click.echo("You are not Authorised")
                 elif result.status_code == 403:
                     click.echo("Not enough permissions")
+                elif result.status_code == 409:
+                    click.echo("Item already exists")
                 elif result.status_code >= 300:
                     raise Exception(result.content)
+
                 else:
                     click.echo(instance.model_dump_json(indent=2))
 
@@ -210,10 +216,10 @@ def esgf_update(
     if publish:
         with httpx.Client() as client:
             if partial_update_data:
+                click.echo()
                 click.echo(
                     f"Partially updating item {item_id} in collection {collection_id}"
                 )
-                click.echo()
 
                 result = client.patch(
                     f"http://localhost:{NODE_PORTS[node]}/{collection_id}/items/{item_id}",
@@ -222,19 +228,23 @@ def esgf_update(
                 )
 
             else:
-                click.echo(f"Updating item {item_id} in collection {collection_id}")
                 click.echo()
+                click.echo(f"Updating item {item_id} in collection {collection_id}")
                 result = client.put(
                     f"http://localhost:{NODE_PORTS[node]}/{collection_id}/items/{item_id}",
                     headers={"Authorization": f"Bearer {token}"},
                     content=item.model_dump_json(),
                 )
+
             if result.status_code == 401:
                 click.echo("You are not Authorised")
             elif result.status_code == 403:
                 click.echo("Not enough permissions")
+            elif result.status_code == 409:
+                click.echo("Cannot update non-existent item")
             elif result.status_code >= 300:
                 raise Exception(result.content)
+
             else:
                 click.echo()
                 click.echo("Done")
@@ -289,12 +299,16 @@ def esgf_delete(
                     headers={"Authorization": f"Bearer {token}"},
                     content=json.dumps(content),
                 )
+
             if result.status_code == 401:
                 click.echo("You are not Authorised")
             elif result.status_code == 403:
                 click.echo("Not enough permissions")
+            elif result.status_code == 409:
+                click.echo("Cannot delete non-existent item")
             elif result.status_code >= 300:
                 raise Exception(result.content)
+
             else:
                 click.echo()
                 click.echo("Done")

esgf-generator/esgf_generator/test_cli.py

@@ -3,17 +3,31 @@
 
 import pytest
 from click.testing import CliRunner, Result
+from dotenv import find_dotenv, load_dotenv, unset_key
 from elasticsearch import Elasticsearch
 
 from .cli import esgf_delete, esgf_generator, esgf_update
 
 es = Elasticsearch(["http://localhost:9200"])
 
+ENV_FILE = find_dotenv()
+load_dotenv(ENV_FILE)
 
 collection_id: str = ""
 item_id: str = ""
 
 
+@pytest.fixture(autouse=True)
+def clean_env() -> Generator[None, None, None]:
+    unset_key(ENV_FILE, "TOKEN")
+    load_dotenv(ENV_FILE)
+
+    yield
+
+    unset_key(ENV_FILE, "TOKEN")
+    load_dotenv(ENV_FILE)
+
+
 @pytest.fixture
 def runner() -> CliRunner:
     return CliRunner()
@@ -27,11 +41,9 @@ def get_item_details(result: Result) -> None:
             parts = line.split(", ")
             item_id = parts[0].split()[1]
             collection_id = parts[1].split()[1]
-    print(f"Item ID: {item_id}, Collection ID: {collection_id}")
 
 
 def check_elasticsearch_index(expected_properties: dict[str, Any]) -> None:
-    global collection_id, item_id
     time.sleep(8)
     response = es.get(
         index=f"items_{collection_id}-000001", id=f"{item_id}|{collection_id}"
@@ -50,9 +62,12 @@ def check_elasticsearch_index(expected_properties: dict[str, Any]) -> None:
 
 
 def test_add_new_item(runner: CliRunner) -> None:
-    global collection_id, item_id
-    result = runner.invoke(esgf_generator, ["1", "--node", "east", "--publish"])
-    time.sleep(20)
+    user_input = "test_user\ntest_user"
+
+    result = runner.invoke(
+        esgf_generator, ["1", "--node", "east", "--publish"], input=user_input
+    )
+    time.sleep(15)
 
     if result.exit_code != 0:
         raise RuntimeError(f"Command failed with exit code {result.exit_code}")
@@ -61,7 +76,8 @@ def test_add_new_item(runner: CliRunner) -> None:
 
 
 def test_add_replica(runner: CliRunner) -> None:
-    global collection_id, item_id
+    user_input = "test_user\ntest_user"
+
     result = runner.invoke(
         esgf_update,
         [
@@ -73,14 +89,16 @@ def test_add_replica(runner: CliRunner) -> None:
             "--partial",
             '{"properties": {"Replica": "Node 1"}}',
         ],
+        input=user_input,
     )
     if result.exit_code != 0:
         raise RuntimeError(f"Command failed with exit code {result.exit_code}")
     check_elasticsearch_index({"properties.Replica": "Node 1"})
 
 
 def test_update_item(runner: CliRunner) -> None:
-    global collection_id, item_id
+    user_input = "test_user\ntest_user"
+
     result = runner.invoke(
         esgf_update,
         [
@@ -92,14 +110,16 @@ def test_update_item(runner: CliRunner) -> None:
             "--partial",
             '{"properties": {"description": "Test Description"}}',
         ],
+        input=user_input,
     )
     if result.exit_code != 0:
         raise RuntimeError(f"Command failed with exit code {result.exit_code}")
     check_elasticsearch_index({"properties.description": "Test Description"})
 
 
 def test_remove_replica(runner: CliRunner) -> None:
-    global collection_id, item_id
+    user_input = "test_user\ntest_user"
+
     result = runner.invoke(
         esgf_delete,
         [
@@ -110,20 +130,54 @@ def test_remove_replica(runner: CliRunner) -> None:
             "--soft",
             "--publish",
         ],
+        input=user_input,
     )
     if result.exit_code != 0:
         raise RuntimeError(f"Command failed with exit code {result.exit_code}")
     check_elasticsearch_index({"properties.retracted": True})
 
 
 def test_remove_item(runner: CliRunner) -> None:
-    global collection_id, item_id
+    user_input = "test_admin\ntest_admin"
+
     result = runner.invoke(
         esgf_delete,
         [collection_id, item_id, "--node", "east", "--hard", "--publish"],
+        input=user_input,
     )
     if result.exit_code != 0:
         raise RuntimeError(f"Command failed with exit code {result.exit_code}")
     response = es.exists(index="item_{collection_id}-000001", id=item_id)
     if response:
         raise ValueError("Document still exists after deletion")
+
+
+def test_delete_non_existent_item(runner: CliRunner) -> None:
+    user_input = "test_admin\ntest_admin"
+
+    result = runner.invoke(
+        esgf_delete,
+        [collection_id, item_id, "--node", "east", "--hard", "--publish"],
+        input=user_input,
+    )
+
+    assert "Cannot delete non-existent item" in result.output
+
+
+def test_update_non_existent_item(runner: CliRunner) -> None:
+    user_input = "test_user\ntest_user"
+
+    result = runner.invoke(
+        esgf_update,
+        [
+            collection_id,
+            item_id,
+            "--node",
+            "east",
+            "--publish",
+            "--partial",
+            '{"properties": {"description": "Test Description"}}',
+        ],
+        input=user_input,
+    )
+    assert "Cannot update non-existent item" in result.output

esgf-generator/esgf_generator/test_keycloak.py

@@ -0,0 +1,97 @@
+import os
+from typing import Generator
+
+import pytest
+from click.testing import CliRunner
+from dotenv import find_dotenv, load_dotenv, unset_key
+
+from esgf_generator.cli import esgf_delete, esgf_generator, validate_token
+
+ENV_FILE = find_dotenv()
+
+if ENV_FILE is None:
+    raise Exception("No .env file found, please create one in the root directory")
+
+
+@pytest.fixture(autouse=True)
+def clean_env() -> Generator[None, None, None]:
+    unset_key(ENV_FILE, "TOKEN")
+    load_dotenv(ENV_FILE)
+
+    yield
+
+    unset_key(ENV_FILE, "TOKEN")
+    load_dotenv(ENV_FILE)
+
+
+@pytest.fixture
+def runner() -> CliRunner:
+    return CliRunner()
+
+
+def test_invalid_credentials(runner: CliRunner) -> None:
+
+    user_input = "invalid_user\ninvalid_user"
+
+    result = runner.invoke(
+        esgf_generator, ["1", "--node", "east", "--publish"], input=user_input
+    )
+
+    load_dotenv(ENV_FILE)
+    token = os.getenv("TOKEN")
+
+    assert "Authentication Failed" in result.output
+
+    assert not token
+
+
+def test_validate_token(runner: CliRunner) -> None:
+
+    user_input = "test_user\ntest_user"
+
+    result = runner.invoke(
+        esgf_generator, ["1", "--node", "east", "--publish"], input=user_input
+    )
+
+    assert result.exit_code == 0
+
+    assert validate_token()
+
+
+def test_invalid_token(runner: CliRunner, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setenv("TOKEN", "invalid_token")
+
+    result = validate_token()
+
+    assert not result
+
+
+def test_get_token(runner: CliRunner) -> None:
+
+    user_input = "test_user\ntest_user"
+
+    result = runner.invoke(
+        esgf_generator, ["1", "--node", "east", "--publish"], input=user_input
+    )
+
+    load_dotenv(ENV_FILE)
+    token = os.getenv("TOKEN")
+
+    assert result.exit_code == 0
+
+    assert token is not None
+
+
+def test_user_scope(runner: CliRunner) -> None:
+
+    user_input = "test_user\ntest_user"
+
+    result = runner.invoke(
+        esgf_delete,
+        ["collection_id", "item_id", "--node", "east", "--hard", "--publish"],
+        input=user_input,
+    )
+
+    assert result.exit_code == 0
+
+    assert "Not enough permissions" in result.output

esgf-transaction-api/esgf_transaction_api/dependencies.py → esgf-transaction-api/esgf_transaction_api/keycloak.py

@@ -47,7 +47,7 @@ async def get_current_user(token: str = Depends(oauth2_scheme)) -> TokenData:
         token_data = TokenData(username=username, roles=roles)
 
     except jwt.PyJWTError:
-        raise HTTPException(status_code=401, detail="Invalid token")
+        raise HTTPException(status_code=401, detail="Error decoding token")
 
     return token_data