Nexus/test_p1_p2.py

"""Quick validation tests for P1+P2 features"""
import sys
sys.path.insert(0, ".")

import datetime
from datetime import timezone


def test_jwt_updated_at():
    """P2-6: JWT token includes updated_at claim"""
    from server.domain.models import Admin
    from server.application.services.auth_service import AuthService

    admin = Admin(
        id=1,
        username="testadmin",
        password_hash="fakehash",
        updated_at=datetime.datetime.now(timezone.utc),
    )

    service = AuthService(None, None, None)
    token = service._create_access_token(admin)

    import jwt
    from server.config import settings
    payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])

    assert "updated" in payload, "Missing 'updated' claim in JWT"
    admin_ts = int(admin.updated_at.timestamp())
    token_ts = payload["updated"]
    assert abs(admin_ts - token_ts) < 2, f"updated_at mismatch: {admin_ts} vs {token_ts}"
    print(f"  JWT updated_at claim: OK (ts={token_ts})")


def test_jwt_password_invalidation():
    """P2-6: Token invalidated when admin.updated_at changes"""
    from server.domain.models import Admin
    from server.application.services.auth_service import AuthService
    import datetime
    from datetime import timezone

    admin = Admin(
        id=1,
        username="testadmin",
        password_hash="fakehash",
        updated_at=datetime.datetime.now(timezone.utc),
    )

    service = AuthService(None, None, None)
    token = service._create_access_token(admin)

    import jwt
    from server.config import settings
    payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
    token_updated = payload["updated"]

    # Simulate password change (updated_at moves forward)
    import time
    time.sleep(1)
    admin.updated_at = datetime.datetime.now(timezone.utc)
    admin_ts = int(admin.updated_at.timestamp())

    # Token should be invalid if admin updated after token issued
    assert admin_ts > token_updated, "Password change should make token stale"
    print(f"  JWT password invalidation: OK (token_updated={token_updated}, admin_updated={admin_ts})")


def test_alert_dedup():
    """P2-1: Alert deduplication cooldown"""
    from server.api.websocket import _should_push_telegram

    # First call should allow
    assert _should_push_telegram("alert:1:cpu") is True, "First alert should be allowed"

    # Immediate second call should be suppressed
    assert _should_push_telegram("alert:1:cpu") is False, "Duplicate alert should be suppressed"

    # Different server/metric should be allowed
    assert _should_push_telegram("alert:2:mem") is True, "Different alert should be allowed"

    # Same server, different metric should be allowed
    assert _should_push_telegram("alert:1:mem") is True, "Different metric should be allowed"

    print("  Alert dedup cooldown: OK")


def test_server_repo_paginated():
    """P2-3: Server repository paginated query signature"""
    import inspect
    from server.infrastructure.database.server_repo import ServerRepositoryImpl

    # Check method exists and has correct signature
    assert hasattr(ServerRepositoryImpl, "get_paginated"), "Missing get_paginated method"

    sig = inspect.signature(ServerRepositoryImpl.get_paginated)
    params = list(sig.parameters.keys())
    assert "category" in params, "Missing category parameter"
    assert "offset" in params, "Missing offset parameter"
    assert "limit" in params, "Missing limit parameter"

    print(f"  Server repo get_paginated: OK (params={params})")


def test_config_dedup_command():
    """P1-7: Config dedup sed command uses escaped \\s"""
    # Just verify the string doesn't have invalid escape
    import re
    key = "net.ipv4.ip_forward"
    escaped_key = re.escape(key)
    cmd = f"sed -i '/^{escaped_key}\\s*=/d' /etc/sysctl.d/99-nexus.conf 2>/dev/null || true"
    # Should contain \\s* not \s*
    assert "\\s*" in cmd, f"Dedup command should use \\s* not \\s"
    print(f"  Config dedup command: OK")


def test_server_to_dict_no_password():
    """P2-7/P2-8: _server_to_dict never returns password/ssh_key_private"""
    from server.api.servers import _server_to_dict
    from server.domain.models import Server

    server = Server(
        id=1,
        name="test",
        domain="10.0.0.1",
        password="secret123",
        ssh_key_private="-----BEGIN KEY-----",
        agent_api_key="nxs-verylongkeystring1234567890",
    )
    d = _server_to_dict(server)

    assert "password" not in d, "password should not be in API response"
    assert "ssh_key_private" not in d, "ssh_key_private should not be in API response"
    assert d.get("password_set") is True, "password_set should be True"
    assert d.get("ssh_key_private_set") is True, "ssh_key_private_set should be True"
    assert d.get("agent_api_key") == "nxs-very...", "agent_api_key should be masked (first 8 chars + ...)"
    print("  Server dict no password: OK")


def test_admin_model_updated_at():
    """P2-6: Admin model has updated_at field"""
    from server.domain.models import Admin
    assert hasattr(Admin, "updated_at"), "Admin model missing updated_at column"
    print("  Admin.updated_at field: OK")


if __name__ == "__main__":
    print("=" * 60)
    print("Nexus P1+P2 Validation Tests")
    print("=" * 60)

    tests = [
        ("P2-6 JWT updated_at claim", test_jwt_updated_at),
        ("P2-6 JWT password invalidation", test_jwt_password_invalidation),
        ("P2-1 Alert dedup cooldown", test_alert_dedup),
        ("P2-3 Server repo pagination", test_server_repo_paginated),
        ("P1-7 Config dedup command", test_config_dedup_command),
        ("P2-7/P2-8 Server dict no password", test_server_to_dict_no_password),
        ("P2-6 Admin.updated_at field", test_admin_model_updated_at),
    ]

    passed = 0
    failed = 0
    for name, fn in tests:
        try:
            print(f"\n[{name}]")
            fn()
            passed += 1
        except Exception as e:
            print(f"  FAIL: {e}")
            failed += 1

    print("\n" + "=" * 60)
    print(f"Results: {passed} passed, {failed} failed, {passed + failed} total")
    print("=" * 60)