Bruno Charest
5bc12d0729
Some checks failed
Tests / Backend Tests (Python) (3.10) (push) Has been cancelled
Tests / Backend Tests (Python) (3.11) (push) Has been cancelled
Tests / Backend Tests (Python) (3.12) (push) Has been cancelled
Tests / Frontend Tests (JS) (push) Has been cancelled
Tests / Integration Tests (push) Has been cancelled
Tests / All Tests Passed (push) Has been cancelled
296 lines
10 KiB
Python
296 lines
10 KiB
Python
"""
|
|
Unit tests for Terminal Command History API endpoints.
|
|
|
|
Tests cover:
|
|
- GET /api/terminal/{host_id}/command-history
|
|
- GET /api/terminal/{host_id}/command-history/unique
|
|
- GET /api/terminal/command-history (global)
|
|
- DELETE /api/terminal/{host_id}/command-history
|
|
- POST /api/terminal/command-history/purge
|
|
"""
|
|
import pytest
|
|
from datetime import datetime, timezone
|
|
from unittest.mock import AsyncMock, MagicMock, patch
|
|
|
|
from httpx import AsyncClient
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.models.terminal_command_log import TerminalCommandLog
|
|
|
|
|
|
class TestGetHostCommandHistory:
|
|
"""Test GET /api/terminal/{host_id}/command-history endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_returns_command_history(self, client: AsyncClient, db_session: AsyncSession, auth_headers):
|
|
"""Test successful retrieval of command history."""
|
|
# Create test data
|
|
from app.crud.terminal_command_log import TerminalCommandLogRepository
|
|
from app.crud.host import HostRepository
|
|
|
|
# First ensure we have a host
|
|
host_repo = HostRepository(db_session)
|
|
hosts = await host_repo.list()
|
|
|
|
if hosts:
|
|
host_id = hosts[0].id
|
|
|
|
response = await client.get(
|
|
f"/api/terminal/{host_id}/command-history",
|
|
headers=auth_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "commands" in data
|
|
assert "total" in data
|
|
assert isinstance(data["commands"], list)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_returns_404_for_unknown_host(self, client: AsyncClient, auth_headers):
|
|
"""Test 404 response for non-existent host."""
|
|
response = await client.get(
|
|
"/api/terminal/nonexistent-host-id/command-history",
|
|
headers=auth_headers,
|
|
)
|
|
|
|
assert response.status_code == 404
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filters_by_query(self, client: AsyncClient, db_session: AsyncSession, auth_headers):
|
|
"""Test search query filtering."""
|
|
from app.crud.host import HostRepository
|
|
|
|
host_repo = HostRepository(db_session)
|
|
hosts = await host_repo.list()
|
|
|
|
if hosts:
|
|
host_id = hosts[0].id
|
|
|
|
response = await client.get(
|
|
f"/api/terminal/{host_id}/command-history?query=ls",
|
|
headers=auth_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["query"] == "ls"
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_respects_limit(self, client: AsyncClient, db_session: AsyncSession, auth_headers):
|
|
"""Test limit parameter."""
|
|
from app.crud.host import HostRepository
|
|
|
|
host_repo = HostRepository(db_session)
|
|
hosts = await host_repo.list()
|
|
|
|
if hosts:
|
|
host_id = hosts[0].id
|
|
|
|
response = await client.get(
|
|
f"/api/terminal/{host_id}/command-history?limit=10",
|
|
headers=auth_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert len(data["commands"]) <= 10
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_caps_limit_at_100(self, client: AsyncClient, db_session: AsyncSession, auth_headers):
|
|
"""Test that limit is capped at 100."""
|
|
from app.crud.host import HostRepository
|
|
|
|
host_repo = HostRepository(db_session)
|
|
hosts = await host_repo.list()
|
|
|
|
if hosts:
|
|
host_id = hosts[0].id
|
|
|
|
response = await client.get(
|
|
f"/api/terminal/{host_id}/command-history?limit=500",
|
|
headers=auth_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
# Should still work, just capped internally
|
|
|
|
|
|
class TestGetHostUniqueCommands:
|
|
"""Test GET /api/terminal/{host_id}/command-history/unique endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_returns_unique_commands(self, client: AsyncClient, db_session: AsyncSession, auth_headers):
|
|
"""Test retrieval of unique commands."""
|
|
from app.crud.host import HostRepository
|
|
|
|
host_repo = HostRepository(db_session)
|
|
hosts = await host_repo.list()
|
|
|
|
if hosts:
|
|
host_id = hosts[0].id
|
|
|
|
response = await client.get(
|
|
f"/api/terminal/{host_id}/command-history/unique",
|
|
headers=auth_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "commands" in data
|
|
assert "total" in data
|
|
assert "host_id" in data
|
|
|
|
|
|
class TestGetGlobalCommandHistory:
|
|
"""Test GET /api/terminal/command-history endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_returns_global_history(self, client: AsyncClient, auth_headers):
|
|
"""Test retrieval of global command history."""
|
|
response = await client.get(
|
|
"/api/terminal/command-history",
|
|
headers=auth_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "commands" in data
|
|
assert isinstance(data["commands"], list)
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_filters_by_host_id(self, client: AsyncClient, db_session: AsyncSession, auth_headers):
|
|
"""Test filtering by host_id."""
|
|
from app.crud.host import HostRepository
|
|
|
|
host_repo = HostRepository(db_session)
|
|
hosts = await host_repo.list()
|
|
|
|
if hosts:
|
|
host_id = hosts[0].id
|
|
|
|
response = await client.get(
|
|
f"/api/terminal/command-history?host_id={host_id}",
|
|
headers=auth_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["host_id"] == host_id
|
|
|
|
|
|
class TestClearHostCommandHistory:
|
|
"""Test DELETE /api/terminal/{host_id}/command-history endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_requires_admin_role(self, client: AsyncClient, db_session: AsyncSession, auth_headers):
|
|
"""Test that only admins can clear history."""
|
|
from app.crud.host import HostRepository
|
|
|
|
host_repo = HostRepository(db_session)
|
|
hosts = await host_repo.list()
|
|
|
|
if hosts:
|
|
host_id = hosts[0].id
|
|
|
|
# This test depends on the auth setup
|
|
# If the test user is admin, it should succeed
|
|
# If not admin, it should return 403
|
|
response = await client.delete(
|
|
f"/api/terminal/{host_id}/command-history",
|
|
headers=auth_headers,
|
|
)
|
|
|
|
# Either 200 (admin) or 403 (not admin) is acceptable
|
|
assert response.status_code in [200, 403]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_returns_404_for_unknown_host(self, client: AsyncClient, auth_headers):
|
|
"""Test 404 for non-existent host."""
|
|
response = await client.delete(
|
|
"/api/terminal/nonexistent-host-id/command-history",
|
|
headers=auth_headers,
|
|
)
|
|
|
|
# Either 404 (host not found) or 403 (not admin) is acceptable
|
|
assert response.status_code in [403, 404]
|
|
|
|
|
|
class TestPurgeCommandHistory:
|
|
"""Test POST /api/terminal/command-history/purge endpoint."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_requires_admin_role(self, client: AsyncClient, auth_headers):
|
|
"""Test that only admins can purge history."""
|
|
response = await client.post(
|
|
"/api/terminal/command-history/purge?days=30",
|
|
headers=auth_headers,
|
|
)
|
|
|
|
# Either 200 (admin) or 403 (not admin) is acceptable
|
|
assert response.status_code in [200, 403]
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_accepts_days_parameter(self, client: AsyncClient, auth_headers):
|
|
"""Test days parameter is accepted."""
|
|
response = await client.post(
|
|
"/api/terminal/command-history/purge?days=7",
|
|
headers=auth_headers,
|
|
)
|
|
|
|
# Either 200 (admin) or 403 (not admin) is acceptable
|
|
assert response.status_code in [200, 403]
|
|
|
|
|
|
class TestCommandHistorySchemas:
|
|
"""Test schema validation for command history responses."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_command_history_item_schema(self, client: AsyncClient, db_session: AsyncSession, auth_headers):
|
|
"""Test that response matches CommandHistoryItem schema."""
|
|
from app.crud.host import HostRepository
|
|
|
|
host_repo = HostRepository(db_session)
|
|
hosts = await host_repo.list()
|
|
|
|
if hosts:
|
|
host_id = hosts[0].id
|
|
|
|
response = await client.get(
|
|
f"/api/terminal/{host_id}/command-history",
|
|
headers=auth_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
for cmd in data["commands"]:
|
|
assert "id" in cmd
|
|
assert "command" in cmd
|
|
assert "created_at" in cmd
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_unique_command_item_schema(self, client: AsyncClient, db_session: AsyncSession, auth_headers):
|
|
"""Test that response matches UniqueCommandItem schema."""
|
|
from app.crud.host import HostRepository
|
|
|
|
host_repo = HostRepository(db_session)
|
|
hosts = await host_repo.list()
|
|
|
|
if hosts:
|
|
host_id = hosts[0].id
|
|
|
|
response = await client.get(
|
|
f"/api/terminal/{host_id}/command-history/unique",
|
|
headers=auth_headers,
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
|
|
for cmd in data["commands"]:
|
|
assert "command" in cmd
|
|
assert "command_hash" in cmd
|
|
assert "last_used" in cmd
|
|
assert "execution_count" in cmd
|