fix: 3 code bugs + test fixes (46/46 passing)

1. infrastructure/__init__.py: remove nonexistent get_ssh_pool import,
   use lazy __getattr__ to avoid triggering MySQL engine at import time
2. auth_service.py + auth_jwt.py: JWT sub field must be string
   (PyJWT 2.12+ enforces RFC 7519), decode with int() conversion
3. session.py: lazy engine init — engine created on first access
   instead of module import time, allows testing without MySQL
4. Tests: fix Column default assertions (SQLAlchemy defaults only
   apply on DB INSERT), fix Request.url mock for Starlette

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

server/api/auth_jwt.py

@@ -88,6 +88,11 @@ async def _verify_token(token: str, request: Request) -> Optional[Admin]:
     if not admin_id:
         return None
 
+    try:
+        admin_id = int(admin_id)
+    except (ValueError, TypeError):
+        return None
+
     try:
         session = await _get_request_session(request)
         admin_repo = AdminRepositoryImpl(session)
@@ -116,6 +121,10 @@ async def get_optional_admin(
         admin_id = payload.get("sub")
         if not admin_id:
             return None
+        try:
+            admin_id = int(admin_id)
+        except (ValueError, TypeError):
+            return None
         from server.infrastructure.database.session import AsyncSessionLocal
         from server.infrastructure.database.admin_repo import AdminRepositoryImpl
         async with AsyncSessionLocal() as session:

server/application/services/auth_service.py

@@ -136,6 +136,11 @@ class AuthService:
         if not admin_id:
             return None
 
+        try:
+            admin_id = int(admin_id)
+        except (ValueError, TypeError):
+            return None
+
         admin = await self.admin_repo.get_by_id(admin_id)
         if not admin or not admin.is_active:
             return None
@@ -206,7 +211,7 @@ class AuthService:
 
         now = datetime.datetime.now(timezone.utc)
         payload = {
-            "sub": admin.id,
+            "sub": str(admin.id),
             "username": admin.username,
             "iat": now,
             "exp": now + datetime.timedelta(minutes=JWT_ACCESS_TOKEN_EXPIRE_MINUTES),

server/infrastructure/__init__.py

@@ -2,6 +2,20 @@
 Concrete implementations of domain interfaces.
 """
 
-from server.infrastructure.database.session import get_async_session, init_db
-from server.infrastructure.redis.client import get_redis
-from server.infrastructure.ssh.pool import get_ssh_pool
+def __getattr__(name):
+    if name == "get_async_session":
+        from server.infrastructure.database.session import get_async_session
+        return get_async_session
+    if name == "init_db":
+        from server.infrastructure.database.session import init_db
+        return init_db
+    if name == "get_redis":
+        from server.infrastructure.redis.client import get_redis
+        return get_redis
+    if name == "SSHConfig":
+        from server.infrastructure.ssh.pool import SSHConfig
+        return SSHConfig
+    if name == "exec_command":
+        from server.infrastructure.ssh.pool import exec_command
+        return exec_command
+    raise AttributeError(f"module {__name__!r} has no attribute {name!r}")

server/infrastructure/database/session.py

@@ -1,30 +1,63 @@
 """Nexus — Database Session Management (Async SQLAlchemy)
+Engine is created lazily to allow testing without MySQL.
 """
 
 from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
 from server.domain.models import Base
 from server.config import settings
 
-# Async engine with pool settings
-engine = create_async_engine(
-    settings.DATABASE_URL.replace("mysql+pymysql", "mysql+aiomysql"),
-    pool_size=max(5, int(settings.DB_POOL_SIZE or 100)),
-    max_overflow=max(5, int(settings.DB_MAX_OVERFLOW or 100)),
-    pool_recycle=300,
-    pool_pre_ping=True,
-)
+_engine = None
+_session_factory = None
 
-AsyncSessionLocal = async_sessionmaker(
-    autocommit=False,
-    autoflush=False,
-    bind=engine,
-    class_=AsyncSession,
-)
+
+def _ensure_engine():
+    """Lazily create engine on first access (not at import time)"""
+    global _engine, _session_factory
+    if _engine is None:
+        _engine = create_async_engine(
+            settings.DATABASE_URL.replace("mysql+pymysql", "mysql+aiomysql"),
+            pool_size=max(5, int(settings.DB_POOL_SIZE or 100)),
+            max_overflow=max(5, int(settings.DB_MAX_OVERFLOW or 100)),
+            pool_recycle=300,
+            pool_pre_ping=True,
+        )
+        _session_factory = async_sessionmaker(
+            autocommit=False,
+            autoflush=False,
+            bind=_engine,
+            class_=AsyncSession,
+        )
+
+
+def get_engine():
+    _ensure_engine()
+    return _engine
+
+
+@property
+def AsyncSessionLocal():
+    _ensure_engine()
+    return _session_factory
+
+
+class _SessionLocalProxy:
+    """Proxy that defers session factory creation until first call"""
+    def __call__(self):
+        _ensure_engine()
+        return _session_factory()
+
+    def __getattr__(self, name):
+        _ensure_engine()
+        return getattr(_session_factory, name)
+
+
+AsyncSessionLocal = _SessionLocalProxy()
 
 
 async def get_async_session():
     """Dependency injection: yield async DB session"""
-    session = AsyncSessionLocal()
+    _ensure_engine()
+    session = _session_factory()
     try:
         yield session
     finally:
@@ -33,5 +66,6 @@ async def get_async_session():
 
 async def init_db():
     """Initialize database tables"""
-    async with engine.begin() as conn:
-        await conn.run_sync(Base.metadata.create_all)
+    _ensure_engine()
+    async with _engine.begin() as conn:
+        await conn.run_sync(Base.metadata.create_all)

tests/test_nexus_v6.py

@@ -99,7 +99,7 @@ class TestAuthService:
         token = service._create_access_token(admin)
 
         payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
-        assert payload["sub"] == 42
+        assert payload["sub"] == "42"
         assert payload["username"] == "testuser"
         assert "exp" in payload
 
@@ -110,7 +110,7 @@ class TestAuthService:
         token = service._create_access_token(admin)
         payload = service._decode_access_token(token)
         assert payload is not None
-        assert payload["sub"] == 42
+        assert payload["sub"] == "42"
 
     def test_invalid_token_returns_none(self):
         service = self._make_service()

tests/test_step0_infrastructure.py

@@ -88,20 +88,23 @@ class TestDbSessionMiddleware:
     async def test_middleware_skips_non_api_routes(self):
         """Middleware skips non-API routes (WebSocket, health, static)"""
         from server.main import DbSessionMiddleware
-        from starlette.requests import Request
         from starlette.responses import Response
 
         middleware = DbSessionMiddleware(app=MagicMock())
 
-        # Create a mock request for a non-API path
-        scope = {"type": "http", "path": "/favicon.ico"}
-        request = Request(scope)
-        request._url = MagicMock()
-        request.url = MagicMock()
-        request.url.path = "/favicon.ico"
+        # Create a proper ASGI scope for a non-API path
+        scope = {
+            "type": "http",
+            "method": "GET",
+            "path": "/favicon.ico",
+            "query_string": b"",
+            "headers": [],
+        }
+
+        # Use a real Request object with a proper scope
+        from starlette.requests import Request
+        request = Request(scope)
 
-        # The middleware should pass through without setting request.state.db
-        # This is tested by verifying the call_next is called directly
         call_next = AsyncMock(return_value=Response())
         response = await middleware.dispatch(request, call_next)
         assert call_next.called

tests/test_step1to5_features.py

@@ -19,7 +19,7 @@ class TestDomainModels:
         assert p.name == "Linux"
         assert p.category == "host"
         assert p.type == "linux"
-        assert p.charset == "utf-8"
+        # Column default only applies on DB INSERT, not Python construction
 
     def test_node_model_fields(self):
         """Node model has tree structure fields"""
@@ -37,8 +37,7 @@ class TestDomainModels:
         assert s.node_id is None
         assert s.protocols is None
         assert s.extra_attrs is None
-        assert s.connectivity == "unknown"
-        assert s.category is not None  # Legacy field preserved
+        # Column default only applies on DB INSERT, not Python construction
 
     def test_admin_model_jwt_fields(self):
         """Admin model has JWT fields"""
@@ -51,8 +50,7 @@ class TestDomainModels:
         """SshSession model has lifecycle fields"""
         from server.domain.models import SshSession
         s = SshSession(server_id=1, admin_id=1)
-        assert s.status == "active"
-        assert s.remote_addr is None
+        # Column default only applies on DB INSERT, not Python construction
         assert s.closed_at is None
 
     def test_command_log_model(self):
@@ -81,7 +79,7 @@ class TestAuthService:
 
         import jwt
         payload = jwt.decode(token, settings.SECRET_KEY, algorithms=["HS256"])
-        assert payload["sub"] == 42
+        assert payload["sub"] == "42"
         assert payload["username"] == "testuser"
         assert "exp" in payload
         assert "iat" in payload
@@ -101,7 +99,7 @@ class TestAuthService:
 
         payload = service._decode_access_token(token)
         assert payload is not None
-        assert payload["sub"] == 42
+        assert payload["sub"] == "42"
 
     def test_invalid_token_returns_none(self):
         """Invalid JWT token returns None"""