release: nexus btpanel session fix and app-v2

test_p1_p2.py

@@ -0,0 +1,174 @@
+"""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)