Increase test coverage and refactor tests (#160)

* Increase test coverage and refactor tests

* Avoid unnecessary os.path.normcase() when matching permissions

product_listings_manager/config.py

@@ -5,9 +5,6 @@
 
 
 def read_json_file(filename: str):
-    if not filename:
-        return None
-
     with open(filename) as f:
         return json.load(f)
 

product_listings_manager/permissions.py

@@ -1,13 +1,13 @@
 # SPDX-License-Identifier: GPL-2.0+
-from fnmatch import fnmatch
+from fnmatch import fnmatchcase
 from typing import Any
 
 from product_listings_manager.authorization import LdapConfig, get_user_groups
 
 
 def query_matches(query: str, permission: dict[str, Any]) -> bool:
     return any(
-        fnmatch(query.upper(), pattern.upper())
+        fnmatchcase(query.upper(), pattern.upper())
         for pattern in permission.get("queries", [])
     )
 

product_listings_manager/utils.py

@@ -1,3 +1,4 @@
+# SPDX-License-Identifier: GPL-2.0+
 import traceback
 
 from flask import current_app, request

tests/conftest.py

@@ -1,20 +1,35 @@
-# Flask < 1.0 does not have get_json.
+# SPDX-License-Identifier: GPL-2.0+
+import base64
 import json
 import os
+from unittest.mock import patch
 
-import pytest
-from flask import Response
+from pytest import fixture, mark
 
 from product_listings_manager.app import create_app
+from product_listings_manager.config import load_config_from_env
 from product_listings_manager.models import db
 
-if not hasattr(Response, "get_json"):
-
-    def get_json(self, force=False, silent=False, cache=True):
-        """Minimal implementation we can use this in the tests."""
-        return json.loads(self.data)
-
-    Response.get_json = get_json
+LDAP_HOST = "ldap://ldap.example.com"
+LDAP_BASE = "ou=Groups,dc=example,dc=com"
+LDAP_SEARCH = "(memberUid=test_user)"
+PERMISSIONS = [
+    {
+        "name": "test user permission",
+        "users": ["test_user"],
+        "queries": ["INSERT *", "ROLLBACK"],
+    },
+    {
+        "name": "test group1 permission",
+        "groups": ["group1"],
+        "queries": ["SELECT *"],
+    },
+    {
+        "name": "test group2 permission",
+        "groups": ["group2"],
+        "queries": ["DELETE *"],
+    },
+]
 
 
 def pytest_addoption(parser):
@@ -36,15 +51,15 @@ def pytest_configure(config):
 def pytest_collection_modifyitems(config, items):
     if config.getoption("--live"):
         return
-    skip_live = pytest.mark.skip(
+    skip_live = mark.skip(
         reason="use --live argument to query production servers"
     )
     for item in items:
         if "live" in item.keywords:
             item.add_marker(skip_live)
 
 
-@pytest.fixture
+@fixture
 def app():
     os.environ["SQLALCHEMY_DATABASE_URI"] = "sqlite://"
     app = create_app()
@@ -53,7 +68,50 @@ def app():
         yield app
 
 
-@pytest.fixture
+@fixture
 def client(app):
     client = app.test_client()
     yield client
+
+
+@fixture
+def gssapi_context(client):
+    with patch(
+        "product_listings_manager.auth.gssapi.SecurityContext", autospec=True
+    ) as context:
+        context().initiator_name = "test_user@example.com"
+        context().complete = True
+        context().step.return_value = b"SECRET"
+        yield context
+
+
+@fixture
+def ldap_connection(client):
+    with patch("ldap.initialize", autospec=True) as ldap_init:
+        ldap_connection = ldap_init(LDAP_HOST)
+        ldap_connection.search_s.return_value = [
+            ("ou=Groups,dc=example,dc=com", {"cn": [b"group1"]})
+        ]
+        yield ldap_connection
+
+
+@fixture
+def auth_client(
+    client, monkeypatch, tmp_path, gssapi_context, ldap_connection
+):
+    ldap_searches = [{"BASE": LDAP_BASE, "SEARCH_STRING": LDAP_SEARCH}]
+    monkeypatch.setenv("PLM_LDAP_HOST", "ldap://ldap.example.com")
+    monkeypatch.setenv("PLM_LDAP_SEARCHES", json.dumps(ldap_searches))
+
+    permissions_file = tmp_path / "permissions.json"
+    with open(permissions_file, "w") as f:
+        json.dump(PERMISSIONS, f)
+    monkeypatch.setenv("PLM_PERMISSIONS", str(permissions_file))
+
+    load_config_from_env(client.application)
+    yield client
+
+
+def auth_headers():
+    token = base64.b64encode(b"SECRET").decode("utf-8")
+    return {"Authorization": f"Negotiate {token}"}