120 lines
3.5 KiB
Python
120 lines
3.5 KiB
Python
"""Tests for dashboard API endpoints."""
|
|
import pytest
|
|
from aiohttp import web
|
|
from ayn_antivirus.dashboard.api import setup_routes, _safe_int
|
|
from ayn_antivirus.dashboard.store import DashboardStore
|
|
from ayn_antivirus.dashboard.collector import MetricsCollector
|
|
|
|
|
|
@pytest.fixture
|
|
def store(tmp_path):
|
|
s = DashboardStore(str(tmp_path / "test_api.db"))
|
|
yield s
|
|
s.close()
|
|
|
|
|
|
@pytest.fixture
|
|
def app(store, tmp_path):
|
|
application = web.Application()
|
|
application["store"] = store
|
|
application["collector"] = MetricsCollector(store, interval=9999)
|
|
from ayn_antivirus.config import Config
|
|
cfg = Config()
|
|
cfg.db_path = str(tmp_path / "sigs.db")
|
|
application["config"] = cfg
|
|
setup_routes(application)
|
|
return application
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# _safe_int unit tests
|
|
# ------------------------------------------------------------------
|
|
|
|
def test_safe_int_valid():
|
|
assert _safe_int("50", 10) == 50
|
|
assert _safe_int("0", 10, min_val=1) == 1
|
|
assert _safe_int("9999", 10, max_val=100) == 100
|
|
|
|
|
|
def test_safe_int_invalid():
|
|
assert _safe_int("abc", 10) == 10
|
|
assert _safe_int("", 10) == 10
|
|
assert _safe_int(None, 10) == 10
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# API endpoint tests (async, require aiohttp_client)
|
|
# ------------------------------------------------------------------
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_health_endpoint(app, aiohttp_client):
|
|
client = await aiohttp_client(app)
|
|
resp = await client.get("/api/health")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert "cpu_percent" in data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_status_endpoint(app, aiohttp_client):
|
|
client = await aiohttp_client(app)
|
|
resp = await client.get("/api/status")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert "hostname" in data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_threats_endpoint(app, store, aiohttp_client):
|
|
store.record_threat("/tmp/evil", "TestVirus", "malware", "HIGH")
|
|
client = await aiohttp_client(app)
|
|
resp = await client.get("/api/threats")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert data["count"] >= 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_scans_endpoint(app, store, aiohttp_client):
|
|
store.record_scan("quick", "/tmp", 100, 5, 0, 2.5)
|
|
client = await aiohttp_client(app)
|
|
resp = await client.get("/api/scans")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert data["count"] >= 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_logs_endpoint(app, store, aiohttp_client):
|
|
store.log_activity("Test log", "INFO", "test")
|
|
client = await aiohttp_client(app)
|
|
resp = await client.get("/api/logs")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert data["count"] >= 1
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_containers_endpoint(app, aiohttp_client):
|
|
client = await aiohttp_client(app)
|
|
resp = await client.get("/api/containers")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert "runtimes" in data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_definitions_endpoint(app, aiohttp_client):
|
|
client = await aiohttp_client(app)
|
|
resp = await client.get("/api/definitions")
|
|
assert resp.status == 200
|
|
data = await resp.json()
|
|
assert "total_hashes" in data
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_invalid_query_params(app, aiohttp_client):
|
|
client = await aiohttp_client(app)
|
|
resp = await client.get("/api/threats?limit=abc")
|
|
assert resp.status == 200 # Should not crash, uses default
|