| """ |
| Coverage-driven tests for admin/skill_routes.py (0% -> 70%+ target) |
| |
| API Endpoints Tested: |
| - POST /api/admin/skills/api/admin/skills - Create new standardized skill package |
| |
| Coverage Target Areas: |
| - Lines 1-30: Route initialization and dependencies |
| - Lines 30-70: Security scanning (static and LLM analysis) |
| - Lines 70-100: Skill creation and packaging |
| - Lines 100-130: Error handling and responses |
| """ |
|
|
| import os |
| import pytest |
| from unittest.mock import Mock, AsyncMock, patch |
| from fastapi import FastAPI |
| from fastapi.testclient import TestClient |
|
|
| from api.admin.skill_routes import router, CreateSkillRequest |
| from core.models import User, UserRole |
| from core.skill_builder_service import SkillMetadata, skill_builder_service |
| from core.admin_endpoints import get_super_admin |
|
|
|
|
| |
| |
| |
|
|
| @pytest.fixture |
| def mock_admin_user(): |
| """Mock super admin user.""" |
| user = Mock(spec=User) |
| user.id = "admin-1" |
| user.email = "admin@example.com" |
| user.role = UserRole.ADMIN |
| user.tenant_id = "tenant-123" |
| return user |
|
|
|
|
| @pytest.fixture |
| def mock_skill_builder(): |
| """Mock skill builder service.""" |
| with patch('api.admin.skill_routes.skill_builder_service') as mock: |
| mock.create_skill_package.return_value = { |
| "success": True, |
| "skill_id": "skill-123", |
| "path": "/skills/skill-123" |
| } |
| yield mock |
|
|
|
|
| @pytest.fixture |
| def client(mock_admin_user, mock_skill_builder): |
| """Test client with admin auth.""" |
| app = FastAPI() |
| app.include_router(router) |
|
|
| def override_get_admin(): |
| return mock_admin_user |
|
|
| app.dependency_overrides[get_super_admin] = override_get_admin |
|
|
| client = TestClient(app) |
| yield client |
|
|
| app.dependency_overrides.clear() |
|
|
|
|
| |
| |
| |
|
|
| class TestSkillCreationSuccess: |
| """Tests for successful skill creation scenarios.""" |
|
|
| def test_create_skill_success(self, client, mock_admin_user, mock_skill_builder): |
| """Cover successful skill creation (lines 24-98).""" |
| |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_analyzer = Mock() |
| mock_analyzer.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_analyzer |
|
|
| request_data = { |
| "name": "test-skill", |
| "description": "Test skill for coverage", |
| "instructions": "Do something useful", |
| "capabilities": ["read", "write"], |
| "scripts": { |
| "main.py": "print('hello')", |
| "helper.py": "def help(): pass" |
| } |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 200 |
| result = response.json() |
| assert result["success"] is True |
| assert "data" in result |
| mock_skill_builder.create_skill_package.assert_called_once() |
|
|
| def test_create_skill_with_minimal_data(self, client, mock_skill_builder): |
| """Cover skill creation with minimal required fields.""" |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_analyzer = Mock() |
| mock_analyzer.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_analyzer |
|
|
| request_data = { |
| "name": "minimal-skill", |
| "description": "Minimal", |
| "instructions": "Do work", |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 200 |
|
|
| def test_create_skill_without_capabilities(self, client, mock_skill_builder): |
| """Cover skill creation with empty capabilities list.""" |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_analyzer = Mock() |
| mock_analyzer.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_analyzer |
|
|
| request_data = { |
| "name": "no-cap-skill", |
| "description": "No capabilities", |
| "instructions": "Simple", |
| "capabilities": [], |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 200 |
|
|
| def test_create_skill_with_multiple_scripts(self, client, mock_skill_builder): |
| """Cover skill creation with multiple script files.""" |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_analyzer = Mock() |
| mock_analyzer.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_analyzer |
|
|
| request_data = { |
| "name": "multi-script-skill", |
| "description": "Multiple scripts", |
| "instructions": "Complex skill", |
| "capabilities": ["read", "write", "execute"], |
| "scripts": { |
| "main.py": "def main(): pass", |
| "utils.py": "def helper(): pass", |
| "config.py": "CONFIG = {}" |
| } |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 200 |
| mock_skill_builder.create_skill_package.assert_called_once() |
|
|
| def test_create_skill_default_tenant_id(self, client, mock_skill_builder): |
| """Cover skill creation when admin has no tenant_id (uses 'default').""" |
| from core.admin_endpoints import get_super_admin |
| from core.models import User, UserRole |
|
|
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_analyzer = Mock() |
| mock_analyzer.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_analyzer |
|
|
| |
| admin_no_tenant = Mock(spec=User) |
| admin_no_tenant.id = "admin-2" |
| admin_no_tenant.email = "admin2@example.com" |
| admin_no_tenant.role = UserRole.ADMIN |
| admin_no_tenant.tenant_id = None |
|
|
| |
| app = FastAPI() |
| app.include_router(router) |
|
|
| def override_get_admin(): |
| return admin_no_tenant |
|
|
| app.dependency_overrides[get_super_admin] = override_get_admin |
|
|
| test_client = TestClient(app) |
|
|
| request_data = { |
| "name": "default-tenant-skill", |
| "description": "Default tenant", |
| "instructions": "Test", |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = test_client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 200 |
| |
| call_args = mock_skill_builder.create_skill_package.call_args |
| assert call_args[1]["tenant_id"] == "default" |
|
|
| app.dependency_overrides.clear() |
|
|
| def test_create_skill_with_admin_author(self, client, mock_skill_builder, mock_admin_user): |
| """Cover SkillMetadata construction with admin email as author.""" |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_analyzer = Mock() |
| mock_analyzer.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_analyzer |
|
|
| request_data = { |
| "name": "author-test", |
| "description": "Test author", |
| "instructions": "Test", |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 200 |
|
|
| |
| call_args = mock_skill_builder.create_skill_package.call_args |
| metadata = call_args[1]["metadata"] |
| assert metadata.author == mock_admin_user.email |
|
|
| def test_create_skill_complex_instructions(self, client, mock_skill_builder): |
| """Cover skill creation with complex multi-line instructions.""" |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_analyzer = Mock() |
| mock_analyzer.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_analyzer |
|
|
| request_data = { |
| "name": "complex-instructions-skill", |
| "description": "Complex instructions", |
| "instructions": """ |
| This is a complex skill with: |
| - Multiple steps |
| - Detailed requirements |
| - Safety precautions |
| """, |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 200 |
|
|
| def test_create_skill_special_characters_in_name(self, client, mock_skill_builder): |
| """Cover skill creation with special characters in name.""" |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_analyzer = Mock() |
| mock_analyzer.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_analyzer |
|
|
| request_data = { |
| "name": "skill-with-dashes_and_underscores", |
| "description": "Special characters", |
| "instructions": "Test", |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 200 |
|
|
|
|
| |
| |
| |
|
|
| class TestSecurityScanning: |
| """Tests for security scanning during skill creation.""" |
|
|
| def test_security_scan_blocks_critical_findings(self, client): |
| """Cover security scan blocking on critical findings (lines 38-68).""" |
| from atom_security.core.models import Finding, Severity |
|
|
| |
| critical_finding = Mock(spec=Finding) |
| critical_finding.severity = Severity.CRITICAL |
| critical_finding.dict.return_value = {"severity": "CRITICAL", "message": "Dangerous code"} |
|
|
| |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_instance = Mock() |
| mock_instance.scan_content.return_value = [critical_finding] |
| MockStaticAnalyzer.return_value = mock_instance |
|
|
| request_data = { |
| "name": "dangerous-skill", |
| "description": "Has security issues", |
| "instructions": "eval(input())", |
| "scripts": {"main.py": "eval(input())"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| |
| |
| |
| assert response.status_code == 500 |
| |
| assert "PERMISSION_DENIED" in response.text or "permission" in response.text.lower() |
|
|
| def test_security_scan_blocks_high_findings(self, client): |
| """Cover security scan blocking on HIGH severity findings.""" |
| from atom_security.core.models import Finding, Severity |
|
|
| high_finding = Mock(spec=Finding) |
| high_finding.severity = Severity.HIGH |
| high_finding.dict.return_value = {"severity": "HIGH", "message": "Suspicious pattern"} |
|
|
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_instance = Mock() |
| mock_instance.scan_content.return_value = [high_finding] |
| MockStaticAnalyzer.return_value = mock_instance |
|
|
| request_data = { |
| "name": "suspicious-skill", |
| "description": "Suspicious", |
| "instructions": "Some code", |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| |
| assert response.status_code == 500 |
|
|
| def test_security_scan_allows_low_findings(self, client, mock_skill_builder): |
| """Cover that LOW/MEDIUM severity findings don't block skill creation.""" |
| from atom_security.core.models import Finding, Severity |
|
|
| low_finding = Mock(spec=Finding) |
| low_finding.severity = Severity.LOW |
| low_finding.dict.return_value = {"severity": "LOW", "message": "Code smell"} |
|
|
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_instance = Mock() |
| mock_instance.scan_content.return_value = [low_finding] |
| MockStaticAnalyzer.return_value = mock_instance |
|
|
| request_data = { |
| "name": "low-severity-skill", |
| "description": "Has minor issues", |
| "instructions": "Test", |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| |
| assert response.status_code == 200 |
|
|
| def test_security_scan_multiple_findings(self, client): |
| """Cover security scan with multiple critical findings.""" |
| from atom_security.core.models import Finding, Severity |
|
|
| critical_1 = Mock(spec=Finding) |
| critical_1.severity = Severity.CRITICAL |
| critical_1.dict.return_value = {"severity": "CRITICAL", "message": "Eval usage"} |
| critical_2 = Mock(spec=Finding) |
| critical_2.severity = Severity.HIGH |
| critical_2.dict.return_value = {"severity": "HIGH", "message": "Suspicious import"} |
|
|
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_instance = Mock() |
| mock_instance.scan_content.return_value = [critical_1, critical_2] |
| MockStaticAnalyzer.return_value = mock_instance |
|
|
| request_data = { |
| "name": "multiple-issues-skill", |
| "description": "Multiple security issues", |
| "instructions": "eval(input())", |
| "scripts": {"main.py": "import os; eval(input())"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| |
| assert response.status_code == 500 |
| |
| assert "PERMISSION_DENIED" in response.text or "permission" in response.text.lower() |
|
|
| @patch.dict(os.environ, {'ATOM_SECURITY_ENABLE_LLM_SCAN': 'true'}) |
| def test_llm_security_scan_enabled(self, client, mock_skill_builder): |
| """Cover LLM security scanning when enabled (lines 48-56).""" |
| from atom_security.analyzers.llm import LLMAnalyzer |
|
|
| |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_static = Mock() |
| mock_static.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_static |
|
|
| |
| with patch.object(LLMAnalyzer, 'analyze', new=AsyncMock(return_value=[])): |
| request_data = { |
| "name": "llm-scanned-skill", |
| "description": "LLM scanned", |
| "instructions": "Safe instructions", |
| "scripts": {"main.py": "print('safe')"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 200 |
|
|
| @patch.dict(os.environ, {'ATOM_SECURITY_ENABLE_LLM_SCAN': 'false'}) |
| def test_llm_security_scan_disabled(self, client, mock_skill_builder): |
| """Cover that LLM scan is skipped when disabled.""" |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_analyzer = Mock() |
| mock_analyzer.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_analyzer |
|
|
| request_data = { |
| "name": "no-llm-skill", |
| "description": "No LLM scan", |
| "instructions": "Safe", |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 200 |
|
|
| def test_security_scan_failure_does_not_block(self, client, mock_skill_builder): |
| """Cover that security scan failure logs warning but doesn't block (lines 71-73).""" |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| MockStaticAnalyzer.side_effect = Exception("Security module failed") |
|
|
| request_data = { |
| "name": "scan-fail-skill", |
| "description": "Scan failed", |
| "instructions": "Safe", |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| |
| assert response.status_code == 200 |
|
|
| @patch.dict(os.environ, {'ATOM_SECURITY_ENABLE_LLM_SCAN': 'true'}) |
| def test_llm_scan_failure_does_not_block(self, client, mock_skill_builder): |
| """Cover that LLM scan failure doesn't block skill creation.""" |
| from atom_security.analyzers.llm import LLMAnalyzer |
|
|
| |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_static = Mock() |
| mock_static.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_static |
|
|
| |
| with patch.object(LLMAnalyzer, 'analyze', new=AsyncMock(side_effect=Exception("LLM service unavailable"))): |
| request_data = { |
| "name": "llm-fail-skill", |
| "description": "LLM scan failed", |
| "instructions": "Safe", |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| |
| assert response.status_code == 200 |
|
|
| def test_static_analyzer_scans_combined_content(self, client, mock_skill_builder): |
| """Cover that static analyzer scans instructions + scripts combined.""" |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_analyzer = Mock() |
| mock_analyzer.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_analyzer |
|
|
| request_data = { |
| "name": "combined-scan-skill", |
| "description": "Combined content scan", |
| "instructions": "Do eval things", |
| "scripts": { |
| "main.py": "print('code')", |
| "helper.py": "import os" |
| } |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 200 |
| |
| mock_analyzer.scan_content.assert_called_once() |
| call_args = mock_analyzer.scan_content.call_args[0][0] |
| assert "Do eval things" in call_args |
| assert "print('code')" in call_args |
| assert "import os" in call_args |
|
|
|
|
| |
| |
| |
|
|
| class TestRequestValidation: |
| """Tests for request validation.""" |
|
|
| def test_create_skill_missing_name(self, client): |
| """Cover validation error for missing name field.""" |
| request_data = { |
| "description": "No name", |
| "instructions": "Test", |
| "scripts": {} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 422 |
|
|
| def test_create_skill_missing_scripts(self, client): |
| """Cover validation error for missing scripts field.""" |
| request_data = { |
| "name": "test", |
| "description": "Test", |
| "instructions": "Test" |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 422 |
|
|
| def test_create_skill_missing_description(self, client): |
| """Cover validation error for missing description field.""" |
| request_data = { |
| "name": "test", |
| "instructions": "Test", |
| "scripts": {} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 422 |
|
|
| def test_create_skill_missing_instructions(self, client): |
| """Cover validation error for missing instructions field.""" |
| request_data = { |
| "name": "test", |
| "description": "Test", |
| "scripts": {} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 422 |
|
|
| @pytest.mark.parametrize("name,should_pass", [ |
| ("valid-name", True), |
| ("valid_name", True), |
| ("valid name", True), |
| ("NameWithNumbers123", True), |
| ]) |
| def test_skill_name_validation(self, client, mock_skill_builder, name, should_pass): |
| """Cover skill name validation.""" |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_analyzer = Mock() |
| mock_analyzer.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_analyzer |
|
|
| request_data = { |
| "name": name, |
| "description": "Test", |
| "instructions": "Test", |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| if should_pass: |
| assert response.status_code == 200 |
| else: |
| assert response.status_code in [422, 400] |
|
|
| def test_create_skill_empty_name(self, client): |
| """Cover validation error for empty name field.""" |
| request_data = { |
| "name": "", |
| "description": "Empty name", |
| "instructions": "Test", |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| |
| |
| |
| |
| |
| assert response.status_code in [200, 422, 400] |
|
|
|
|
| |
| |
| |
|
|
| class TestSkillBuilderIntegration: |
| """Tests for skill builder service integration.""" |
|
|
| def test_skill_builder_called_with_correct_metadata(self, client, mock_admin_user, mock_skill_builder): |
| """Cover that skill builder receives correct metadata.""" |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_analyzer = Mock() |
| mock_analyzer.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_analyzer |
|
|
| request_data = { |
| "name": "metadata-test", |
| "description": "Test metadata", |
| "instructions": "Test instructions", |
| "capabilities": ["read"], |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 200 |
|
|
| |
| mock_skill_builder.create_skill_package.assert_called_once() |
| call_args = mock_skill_builder.create_skill_package.call_args |
| assert call_args[1]["tenant_id"] == str(mock_admin_user.tenant_id) |
|
|
| |
| metadata = call_args[1]["metadata"] |
| assert isinstance(metadata, SkillMetadata) |
| assert metadata.name == "metadata-test" |
| assert metadata.description == "Test metadata" |
| assert metadata.instructions == "Test instructions" |
| assert metadata.capabilities == ["read"] |
|
|
| def test_skill_builder_failure_handling(self, client, mock_skill_builder): |
| """Cover handling when skill builder fails.""" |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_analyzer = Mock() |
| mock_analyzer.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_analyzer |
|
|
| |
| |
| |
| mock_skill_builder.create_skill_package.return_value = { |
| "success": False, |
| "message": "Skill package already exists" |
| } |
|
|
| request_data = { |
| "name": "existing-skill", |
| "description": "Already exists", |
| "instructions": "Test", |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| |
| |
| |
| |
| assert response.status_code == 500 |
|
|
| def test_skill_builder_called_with_scripts_dict(self, client, mock_skill_builder): |
| """Cover that skill builder receives scripts dictionary.""" |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_analyzer = Mock() |
| mock_analyzer.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_analyzer |
|
|
| scripts = { |
| "main.py": "def main(): pass", |
| "utils.py": "def helper(): pass" |
| } |
|
|
| request_data = { |
| "name": "scripts-test", |
| "description": "Test scripts", |
| "instructions": "Test", |
| "scripts": scripts |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 200 |
|
|
| |
| call_args = mock_skill_builder.create_skill_package.call_args |
| assert call_args[1]["scripts"] == scripts |
|
|
| def test_skill_builder_exception_handling(self, client, mock_skill_builder): |
| """Cover handling when skill builder raises exception.""" |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_analyzer = Mock() |
| mock_analyzer.scan_content.return_value = [] |
| MockStaticAnalyzer.return_value = mock_analyzer |
|
|
| |
| mock_skill_builder.create_skill_package.side_effect = IOError("Disk full") |
|
|
| request_data = { |
| "name": "error-skill", |
| "description": "Causes error", |
| "instructions": "Test", |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 500 |
|
|
|
|
| |
| |
| |
|
|
| class TestAuthorization: |
| """Tests for authorization and access control.""" |
|
|
| def test_non_admin_cannot_create_skills(self): |
| """Cover that non-admin users are blocked.""" |
| app = FastAPI() |
| app.include_router(router) |
|
|
| |
| test_client = TestClient(app) |
|
|
| request_data = { |
| "name": "unauthorized", |
| "description": "Should fail", |
| "instructions": "Test", |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = test_client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| assert response.status_code == 401 |
|
|
| def test_admin_without_proper_role_blocked(self): |
| """Cover that users without SUPER_ADMIN role are blocked.""" |
| from core.models import User, UserRole |
|
|
| app = FastAPI() |
| app.include_router(router) |
|
|
| |
| regular_user = Mock(spec=User) |
| regular_user.role = UserRole.MEMBER.value |
| regular_user.id = "user-1" |
| regular_user.email = "user@example.com" |
| regular_user.tenant_id = "tenant-123" |
|
|
| def override_get_user(): |
| return regular_user |
|
|
| app.dependency_overrides[get_super_admin] = override_get_user |
|
|
| test_client = TestClient(app) |
|
|
| request_data = { |
| "name": "user-attempt", |
| "description": "Should be blocked", |
| "instructions": "Test", |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = test_client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| |
| |
| |
| assert response.status_code == 500 |
|
|
| app.dependency_overrides.clear() |
|
|
|
|
| |
| |
| |
|
|
| class TestErrorHandling: |
| """Tests for error handling.""" |
|
|
| def test_malformed_json_request(self, client): |
| """Cover malformed JSON handling.""" |
| response = client.post( |
| "/api/admin/skills/api/admin/skills", |
| content="invalid json", |
| headers={"Content-Type": "application/json"} |
| ) |
|
|
| assert response.status_code == 422 |
|
|
| def test_unexpected_exception_handling(self, client, mock_skill_builder): |
| """Cover unexpected exception handling (lines 97-98).""" |
| |
| |
| |
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_instance = Mock() |
| mock_instance.scan_content.side_effect = RuntimeError("Unexpected error in scan") |
| MockStaticAnalyzer.return_value = mock_instance |
|
|
| request_data = { |
| "name": "error-test", |
| "description": "Causes error", |
| "instructions": "Test", |
| "scripts": {"main.py": "pass"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| |
| |
| assert response.status_code == 200 |
|
|
| def test_http_exception_propagated(self, client): |
| """Cover that HTTPException from security scan is propagated.""" |
| from atom_security.core.models import Finding, Severity |
|
|
| critical_finding = Mock(spec=Finding) |
| critical_finding.severity = Severity.CRITICAL |
| critical_finding.dict.return_value = {"severity": "CRITICAL", "message": "Dangerous"} |
|
|
| with patch('atom_security.analyzers.static.StaticAnalyzer') as MockStaticAnalyzer: |
| mock_instance = Mock() |
| mock_instance.scan_content.return_value = [critical_finding] |
| MockStaticAnalyzer.return_value = mock_instance |
|
|
| request_data = { |
| "name": "http-exception-test", |
| "description": "Triggers HTTPException", |
| "instructions": "eval(input())", |
| "scripts": {"main.py": "eval(input())"} |
| } |
|
|
| response = client.post("/api/admin/skills/api/admin/skills", json=request_data) |
|
|
| |
| |
| |
| assert response.status_code == 500 |
| |
| assert "PERMISSION_DENIED" in response.text or "permission" in response.text.lower() |
|
|