Nexus/tests/test_nexus_v6.py

"""Nexus — Step 0-5: Unit Tests (no external dependencies)

All tests use mocks — no MySQL, Redis, or SSH connections required.
Run with: python -m pytest tests/test_nexus_v6.py -v
"""

import pytest
import sys
from unittest.mock import MagicMock, patch


# ── Model Tests (no DB required) ──

class TestDomainModels:
    """D10: Model field validation"""

    def test_platform_model(self):
        from server.domain.models import Platform
        p = Platform(name="Linux", category="host", type="linux")
        assert p.name == "Linux"
        assert p.category == "host"
        # Note: Column default only applies on INSERT, not on object construction

    def test_node_model(self):
        from server.domain.models import Node
        n = Node(name="Web", sort_order=1)
        assert n.name == "Web"
        assert n.parent_id is None

    def test_server_model_new_fields(self):
        from server.domain.models import Server
        s = Server(name="test", domain="1.2.3.4")
        # New fields exist and are nullable (Column default only on INSERT)
        assert s.platform_id is None
        assert s.node_id is None

    def test_admin_jwt_fields(self):
        from server.domain.models import Admin
        a = Admin(username="admin", password_hash="xxx")
        assert (a.token_version or 0) == 0

    def test_ssh_session_model(self):
        from server.domain.models import SshSession
        s = SshSession(server_id=1)
        # Note: Column default only applies on INSERT
        assert s.closed_at is None

    def test_command_log_model(self):
        from server.domain.models import CommandLog
        c = CommandLog(server_id=1, command="ls")
        assert c.command == "ls"


# ── JWT Auth Tests ──

class TestJWTAuth:
    """A5: JWT authentication logic"""

    def test_public_path_detection(self):
        from server.api.auth_jwt import _is_public_path
        assert _is_public_path("/api/auth/login") is True
        assert _is_public_path("/api/auth/refresh") is True
        assert _is_public_path("/api/agent/heartbeat") is True
        assert _is_public_path("/health") is True
        assert _is_public_path("/ws/alerts") is True
        assert _is_public_path("/api/settings/bing-wallpapers", "GET") is True
        assert _is_public_path("/docs") is False
        assert _is_public_path("/health/detail") is False

    def test_protected_path_detection(self):
        from server.api.auth_jwt import _is_public_path
        assert _is_public_path("/api/servers/") is False
        assert _is_public_path("/api/settings/") is False
        assert _is_public_path("/api/sync/files") is False
        assert _is_public_path("/api/assets/platforms") is False
        assert _is_public_path("/api/auth/me") is False
        assert _is_public_path("/api/auth/totp/setup") is False


# ── AuthService Tests (no DB) ──

class TestAuthService:
    """A5: Auth service JWT/TOTP unit tests"""

    def _make_service(self):
        from server.application.services.auth_service import AuthService
        return AuthService(
            admin_repo=MagicMock(),
            attempt_repo=MagicMock(),
            audit_repo=MagicMock(),
        )

    def test_jwt_token_creation_and_decode(self):
        from server.domain.models import Admin
        from server.config import settings
        import jwt

        service = self._make_service()
        admin = Admin(id=42, username="testuser", password_hash="xxx")
        token = service._create_access_token(admin)

        payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
        assert payload["sub"] == "42"
        assert payload["username"] == "testuser"
        assert "exp" in payload

    def test_jwt_token_roundtrip(self):
        service = self._make_service()
        from server.domain.models import Admin
        admin = Admin(id=42, username="testuser", password_hash="xxx")
        token = service._create_access_token(admin)
        payload = service._decode_access_token(token)
        assert payload is not None
        assert payload["sub"] == "42"

    def test_invalid_token_returns_none(self):
        service = self._make_service()
        assert service._decode_access_token("invalid.token.here") is None

    def test_refresh_token_unique(self):
        from server.domain.models import Admin
        service = self._make_service()
        admin = Admin(id=1, username="admin", password_hash="xxx", token_version=0)
        t1 = service._create_refresh_token(admin)
        t2 = service._create_refresh_token(admin)
        assert t1 != t2
        assert len(t1) > 20

    def test_totp_invalid_code(self):
        service = self._make_service()
        import base64, secrets
        secret = base64.b32encode(secrets.token_bytes(20)).decode()
        assert service._verify_totp("000000", secret) is False


# ── WebSocket Manager Tests ──

class TestWebSocketManager:
    """R8: WebSocket ConnectionManager"""

    def test_initial_state(self):
        from server.api.websocket import ConnectionManager
        m = ConnectionManager()
        assert m.client_count == 0

    def test_disconnect_nonexistent(self):
        from server.api.websocket import ConnectionManager
        m = ConnectionManager()
        m.disconnect("nonexistent")
        assert m.client_count == 0

    def test_update_pong(self):
        from server.api.websocket import ConnectionManager
        import time
        m = ConnectionManager()
        m._connections["test"] = {"websocket": MagicMock(), "last_pong": time.monotonic() - 10}
        before = m._connections["test"]["last_pong"]
        m.update_pong("test")
        assert m._connections["test"]["last_pong"] > before


# ── AsyncSSH Pool Tests ──

class TestAsyncSSHPool:
    """W7: asyncssh connection pool"""

    def test_initial_state(self):
        from server.infrastructure.ssh.asyncssh_pool import AsyncSSHPool
        p = AsyncSSHPool()
        assert p.stats == {"total": 0, "active": 0, "idle": 0}

    def test_config_values(self):
        from server.infrastructure.ssh.asyncssh_pool import AsyncSSHPool
        p = AsyncSSHPool()
        assert p.MAX_CONNECTIONS == 100
        assert p.IDLE_TIMEOUT == 300
        assert p.CONNECT_TIMEOUT == 10


# ── Sync Engine Tests ──

class TestSyncEngineV2:
    """S6: Sync engine"""

    def test_max_concurrent(self):
        from server.application.services.sync_engine_v2 import MAX_CONCURRENT
        assert MAX_CONCURRENT == 10

    def test_engine_creation(self):
        from server.application.services.sync_engine_v2 import SyncEngineV2
        e = SyncEngineV2(
            server_repo=MagicMock(),
            sync_log_repo=MagicMock(),
            audit_repo=MagicMock(),
            retry_repo=MagicMock(),
        )
        assert e is not None


# ── Redis Client Tests (mocked) ──

class TestRedisClient:
    """R8: Redis client logic"""

    def test_get_redis_raises_before_init(self):
        from server.infrastructure.redis.client import get_redis
        import server.infrastructure.redis.client as mod
        mod._redis = None
        with pytest.raises(RuntimeError, match="Redis not initialized"):
            get_redis()

    def test_get_redis_sync_returns_none(self):
        from server.infrastructure.redis.client import get_redis_sync
        import server.infrastructure.redis.client as mod
        mod._redis = None
        assert get_redis_sync() is None