feat(re org tests):

tests/test_dependencies.py

@@ -0,0 +1,15 @@
+from agentic_security.dependencies import InMemorySecrets, get_in_memory_secrets
+
+
+def test_in_memory_secrets():
+    secrets = InMemorySecrets()
+    secrets.set_secret("api_key", "12345")
+    assert secrets.get_secret("api_key") == "12345"
+    assert secrets.get_secret("non_existent_key") is None
+
+
+def test_get_in_memory_secrets():
+    secrets = get_in_memory_secrets()
+    assert isinstance(secrets, InMemorySecrets)
+    secrets.set_secret("token", "abcde")
+    assert secrets.get_secret("token") == "abcde"

tests/test_lib.py

@@ -0,0 +1,209 @@
+import importlib
+import os
+import signal
+import subprocess
+import tempfile
+import time
+
+import pytest
+
+import agentic_security.test_spec_assets as test_spec_assets
+from agentic_security.lib import AgenticSecurity
+
+
+def has_module(module_name):
+    module_obj = importlib.util.find_spec(module_name)
+    return module_obj is not None
+
+
+@pytest.fixture(scope="module")
+def test_server(request):
+    # Start server process
+    server = subprocess.Popen(
+        ["uvicorn", "agentic_security.app:app", "--host", "0.0.0.0", "--port", "9094"],
+        preexec_fn=lambda: signal.signal(signal.SIGINT, signal.SIG_IGN),
+    )
+
+    # Give the server time to start
+    time.sleep(2)
+
+    def cleanup():
+        server.terminate()
+        server.wait()
+
+    request.addfinalizer(cleanup)
+    return server
+
+
+def make_test_registry():
+    return [
+        {
+            "dataset_name": "rubend18/ChatGPT-Jailbreak-Prompts",
+            "num_prompts": 79,
+            "tokens": 26971,
+            "approx_cost": 0.0,
+            "source": "Hugging Face Datasets",
+            "selected": True,
+            "dynamic": False,
+            "url": "https://huggingface.co/rubend18/ChatGPT-Jailbreak-Prompts",
+        },
+    ]
+
+
+class TestLibraryLevel:
+    # Handles an empty dataset list.
+    def test_class(self, test_server):
+        llmSpec = test_spec_assets.SAMPLE_SPEC
+        maxBudget = 1000000
+        max_th = 0.3
+        datasets = make_test_registry()
+        result = AgenticSecurity.scan(llmSpec, maxBudget, datasets, max_th)
+        assert isinstance(result, dict)
+        print(result)
+        assert len(result) in [0, 1]
+
+    @pytest.mark.slow
+    def test_class_msj(self, test_server):
+        llmSpec = test_spec_assets.SAMPLE_SPEC
+        maxBudget = 1000
+        max_th = 0.3
+        datasets = make_test_registry()
+        result = AgenticSecurity.scan(
+            llmSpec, maxBudget, datasets, max_th, enableMultiStepAttack=True
+        )
+        assert isinstance(result, dict)
+        print(result)
+        assert len(result) in [0, 1]
+
+    @pytest.mark.skipif(not has_module("garak"), reason="Garak module not installed")
+    def _test_garak(self, test_server):
+        llmSpec = test_spec_assets.SAMPLE_SPEC
+        maxBudget = 1000000
+        max_th = 0.3
+        datasets = [
+            {
+                "dataset_name": "Garak",
+                "num_prompts": 10,
+                "tokens": 0,
+                "approx_cost": 0.0,
+                "source": "Github: https://github.com/leondz/garak#v0.9.0.1",
+                "selected": True,
+                "url": "https://github.com/leondz/garak2",
+                "dynamic": True,
+                "opts": {"port": 9094},
+            },
+        ]
+        result = AgenticSecurity.scan(llmSpec, maxBudget, datasets, max_th)
+        assert isinstance(result, dict)
+        print(result)
+        assert len(result) in [0, 1]
+
+    @pytest.mark.slow
+    def test_backend(self, test_server):
+        llmSpec = test_spec_assets.SAMPLE_SPEC
+        maxBudget = 1000000
+        max_th = 0.3
+        datasets = [
+            {
+                "dataset_name": "AgenticBackend",
+                "num_prompts": 0,
+                "tokens": 0,
+                "approx_cost": 0.0,
+                "source": "Fine-tuned cloud hosted model",
+                "selected": True,
+                "url": "",
+                "dynamic": True,
+                "opts": {
+                    "port": 9094,
+                    "modules": ["encoding"],
+                },
+                "modality": "text",
+            },
+        ]
+        result = AgenticSecurity.scan(llmSpec, maxBudget, datasets, max_th)
+        assert isinstance(result, dict)
+        print(result)
+        assert len(result) in [0, 1]
+
+    def test_image_modality(self):
+        llmSpec = test_spec_assets.IMAGE_SPEC
+        maxBudget = 2
+        max_th = 0.3
+        datasets = [
+            {
+                "dataset_name": "AgenticBackend",
+                "num_prompts": 0,
+                "tokens": 0,
+                "approx_cost": 0.0,
+                "source": "Fine-tuned cloud hosted model",
+                "selected": True,
+                "url": "",
+                "dynamic": True,
+                "opts": {
+                    # "port": 8718,
+                    "port": 9094,
+                    "modules": ["encoding"],
+                    "max_prompts": 2,
+                },
+                "modality": "text",
+            },
+        ]
+        result = AgenticSecurity.scan(llmSpec, maxBudget, datasets, max_th)
+        assert isinstance(result, dict)
+        print(result)
+        assert len(result) in [0, 1]
+
+
+class TestEntrypointCI:
+    def test_generate_default_cfg_to_tmp_path(self):
+        """
+        Test that the `generate_default_settings` method generates a valid default config file in a temporary path.
+        """
+        # Create a temporary directory
+        with tempfile.TemporaryDirectory() as tmpdir:
+            temp_path = os.path.join(tmpdir, "custom_agesec.toml")
+
+            # Override default_path to the temporary path
+            AgenticSecurity.default_path = temp_path
+
+            # Generate the default configuration
+            security = AgenticSecurity()
+            security.generate_default_settings()
+
+            # Check that the config file was created at the temporary path
+            assert os.path.exists(temp_path), f"{temp_path} file should be generated."
+
+            # Validate the contents of the generated config file
+            with open(temp_path) as f:
+                generated_content = f.read()
+                assert (
+                    "maxBudget = 1000000" in generated_content
+                ), "maxBudget should be 1000000"
+
+    def test_load_generated_tmp_config(self):
+        """
+        Test that the configuration generated in a temporary path can be loaded successfully.
+        """
+        # Create a temporary directory
+        with tempfile.TemporaryDirectory() as tmpdir:
+            temp_path = os.path.join(tmpdir, "custom_agesec.toml")
+
+            # Override default_path to the temporary path
+            AgenticSecurity.default_path = temp_path
+
+            # Generate the default configuration
+            security = AgenticSecurity()
+            security.generate_default_settings()
+
+            # Load the generated configuration
+            AgenticSecurity.load_config(temp_path)
+
+            # Validate loaded configuration
+            config = AgenticSecurity.config
+            assert (
+                config["general"]["maxBudget"] == 1000000
+            ), "maxBudget should be 1000000"
+            assert config["general"]["max_th"] == 0.3, "max_th should be 0.3"
+            assert (
+                config["modules"]["AgenticBackend"]["dataset_name"] == "AgenticBackend"
+            ), "Dataset name should be 'AgenticBackend'"

tests/test_spec.py

@@ -0,0 +1,118 @@
+import pytest
+
+from agentic_security.http_spec import LLMSpec, parse_http_spec
+
+
+class TestParseHttpSpec:
+    # Should correctly parse a simple HTTP spec with headers and body
+    def test_parse_simple_http_spec(self):
+        http_spec = (
+            'GET http://example.com\nContent-Type: application/json\n\n{"key": "value"}'
+        )
+        expected_spec = LLMSpec(
+            method="GET",
+            url="http://example.com",
+            headers={"Content-Type": "application/json"},
+            body='{"key": "value"}',
+        )
+        assert parse_http_spec(http_spec) == expected_spec
+
+    # Should correctly parse a HTTP spec with headers containing special characters
+    def test_parse_http_spec_with_special_characters(self):
+        http_spec = 'POST http://example.com\nX-Auth-Token: abcdefg1234567890!@#$%^&*\n\n{"key": "value"}'
+        expected_spec = LLMSpec(
+            method="POST",
+            url="http://example.com",
+            headers={"X-Auth-Token": "abcdefg1234567890!@#$%^&*"},
+            body='{"key": "value"}',
+        )
+        assert parse_http_spec(http_spec) == expected_spec
+
+    # Should correctly parse a spec with no headers and no body
+    def test_parse_http_spec_with_no_headers_and_no_body(self):
+        # Arrange
+        http_spec = "GET http://example.com"
+
+        # Act
+        result = parse_http_spec(http_spec)
+
+        # Assert
+        assert result.method == "GET"
+        assert result.url == "http://example.com"
+        assert result.headers == {}
+        assert result.body == ""
+
+    def test_parse_http_spec_with_headers_no_body(self):
+        # Arrange
+        http_spec = "GET http://example.com\nContent-Type: application/json\n\n"
+
+        # Act
+        result = parse_http_spec(http_spec)
+
+        # Assert
+        assert result.method == "GET"
+        assert result.url == "http://example.com"
+        assert result.headers == {"Content-Type": "application/json"}
+        assert result.body == ""
+
+
+class TestLLMSpec:
+    def test_validate_raises_error_for_missing_files(self):
+        spec = LLMSpec(
+            method="POST", url="http://example.com", headers={}, body="", has_files=True
+        )
+        with pytest.raises(ValueError, match="Files are required for this request."):
+            spec.validate(prompt="", encoded_image="", encoded_audio="", files={})
+
+    def test_validate_raises_error_for_missing_image(self):
+        spec = LLMSpec(
+            method="POST", url="http://example.com", headers={}, body="", has_image=True
+        )
+        with pytest.raises(ValueError, match="An image is required for this request."):
+            spec.validate(prompt="", encoded_image="", encoded_audio="", files={})
+
+    @pytest.mark.asyncio
+    async def test_probe_sends_request(self, httpx_mock):
+        httpx_mock.add_response(
+            method="POST", url="http://example.com", status_code=200
+        )
+        spec = LLMSpec(
+            method="POST",
+            url="http://example.com",
+            headers={},
+            body='{"prompt": "<<PROMPT>>"}',
+        )
+        response = await spec.probe(prompt="test")
+        assert response.status_code == 200
+
+    @pytest.mark.asyncio
+    async def test_probe_with_files(self, httpx_mock):
+        httpx_mock.add_response(
+            method="POST", url="http://example.com", status_code=200
+        )
+        spec = LLMSpec(
+            method="POST",
+            url="http://example.com",
+            headers={"Content-Type": "multipart/form-data"},
+            body='{"prompt": "<<PROMPT>>"}',
+            has_files=True,
+        )
+        files = {"file": ("filename.txt", "file content")}
+        response = await spec.probe(prompt="test", files=files)
+        assert response.status_code == 200
+
+    @pytest.mark.asyncio
+    async def test_probe_with_image(self, httpx_mock):
+        httpx_mock.add_response(
+            method="POST", url="http://example.com", status_code=200
+        )
+        spec = LLMSpec(
+            method="POST",
+            url="http://example.com",
+            headers={},
+            body='{"image": "<<BASE64_IMAGE>>"}',
+            has_image=True,
+        )
+        encoded_image = "base64encodedstring"
+        response = await spec.probe(prompt="test", encoded_image=encoded_image)
+        assert response.status_code == 200