test cases for opapluginfilter

Signed-off-by: Shriti Priya <[email protected]>

Author: monshri

plugins/external/opa/tests/server/opa_server.py

@@ -0,0 +1,50 @@
+# -*- coding: utf-8 -*-
+"""Test cases for OPA plugin
+
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+Authors: Shriti Priya
+
+This module mocks up an opa server for testing.
+"""
+
+
+# Standard
+import json
+import threading
+
+# Third-Party
+from http.server import BaseHTTPRequestHandler, HTTPServer
+
+
+# This class mocks up the post request for OPA server to evaluate policies.
+class MockOPAHandler(BaseHTTPRequestHandler):
+    def do_POST(self):
+        if self.path == "/v1/data/example/allow":
+            content_length = int(self.headers.get('Content-Length', 0))
+            post_body = self.rfile.read(content_length).decode('utf-8')
+        try:
+            data = json.loads(post_body)
+            if "IBM" in data["input"]["payload"]["args"]["repo_path"]:
+                self.send_response(200)
+                self.send_header("Content-Type", "application/json")
+                self.end_headers()
+                self.wfile.write(b'{"result": true}')
+            else:
+                self.send_response(200)
+                self.send_header("Content-Type", "application/json")
+                self.end_headers()
+                self.wfile.write(b'{"result": false}')
+            # Process data dictionary...
+        except json.JSONDecodeError:
+            # Handle invalid JSON 
+            self.send_response(400)
+            self.end_headers()
+            self.wfile.write(b"Invalid JSON")
+            return
+
+# This creates a mock up server for OPA at port 8181
+def run_mock_opa():
+    server = HTTPServer(('localhost', 8181), MockOPAHandler)
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    return server

plugins/external/opa/tests/test_opapluginfilter.py

@@ -1,5 +1,13 @@
 # -*- coding: utf-8 -*-
-"""Tests for plugin."""
+"""Test cases for OPA plugin
+
+Copyright 2025
+SPDX-License-Identifier: Apache-2.0
+Authors: Shriti Priya
+
+This module contains test cases for running opa plugin.
+"""
+
 
 # Third-Party
 import pytest
@@ -13,23 +21,71 @@
     GlobalContext
 )
 
+from tests.server.opa_server import run_mock_opa
+
 
 @pytest.mark.asyncio
-async def test_opapluginfilter():
+# Test for when opaplugin is not applied to a tool
+async def test_benign_opapluginfilter():
     """Test plugin prompt prefetch hook."""
     config = PluginConfig(
         name="test",
         kind="opapluginfilter.OPAPluginFilter",
         hooks=["tool_pre_invoke"],
-        config={"setting_one": "test_value"},
+        applied_to = {"tools" : [{"tool_name": "fast-time-git-status", "context": ["global.opa_policy_context.git_context"], "extensions": {"policy": "example", "policy_endpoint" : "allow"}}]},
+        config={"opa_base_url": "http://127.0.0.1:8181/v1/data/"}
     )
+    mock_server = run_mock_opa()
+    
 
     plugin = OPAPluginFilter(config)
 
     # Test your plugin logic
-    payload = ToolPreInvokePayload(name="test_tool", args={"repo_path": "This is an argument"})
+    payload = ToolPreInvokePayload(name="fast-time-git-status", args={"repo_path": "/path/IBM"})
     context = PluginContext(global_context=GlobalContext(request_id="1", server_id="2"))
     result = await plugin.tool_pre_invoke(payload, context)
-    import pdb
-    pdb.set_trace()
+    mock_server.shutdown()
     assert result.continue_processing
+
+
+@pytest.mark.asyncio
+# Test for when opaplugin is not applied to a tool
+async def test_malign_opapluginfilter():
+    """Test plugin prompt prefetch hook."""
+    config = PluginConfig(
+        name="test",
+        kind="opapluginfilter.OPAPluginFilter",
+        hooks=["tool_pre_invoke"],
+        applied_to = {"tools" : [{"tool_name": "fast-time-git-status", "context": ["global.opa_policy_context.git_context"], "extensions": {"policy": "example", "policy_endpoint" : "allow"}}]},
+        config={"opa_base_url": "http://127.0.0.1:8181/v1/data/"}
+    )
+    mock_server = run_mock_opa()
+    plugin = OPAPluginFilter(config)
+
+    # Test your plugin logic
+    payload = ToolPreInvokePayload(name="fast-time-git-status", args={"repo_path": "/path/IM"})
+    context = PluginContext(global_context=GlobalContext(request_id="1", server_id="2"))
+    result = await plugin.tool_pre_invoke(payload, context)
+    mock_server.shutdown()
+    assert not result.continue_processing and result.violation.code == "deny"
+    
+@pytest.mark.asyncio
+# Test for opa plugin not applied to any of the tools
+async def test_applied_to_opaplugin():
+    """Test plugin prompt prefetch hook."""
+    config = PluginConfig(
+        name="test",
+        kind="opapluginfilter.OPAPluginFilter",
+        hooks=["tool_pre_invoke"],
+        applied_to = {},
+        config={"opa_base_url": "http://127.0.0.1:8181/v1/data/"}
+    )
+    mock_server = run_mock_opa()
+    plugin = OPAPluginFilter(config)
+
+    # Test your plugin logic
+    payload = ToolPreInvokePayload(name="fast-time-git-status", args={"repo_path": "/path/IM"})
+    context = PluginContext(global_context=GlobalContext(request_id="1", server_id="2"))
+    result = await plugin.tool_pre_invoke(payload, context)
+    mock_server.shutdown()
+    assert result.continue_processing