spawner/tests/test_spawner.py

"""Test GLiCID spawner module."""

from collections import namedtuple
from itertools import repeat

import glicid_spawner.spawner
import pytest
from batchspawner import SlurmSpawner
from glicid_spawner import GlicidSpawner
from glicid_spawner.spawner import asyncio

User = namedtuple('User', 'name')


def test_spawner_config():
    """Test spawner configuration."""
    spawner = GlicidSpawner(progress_rate=10)

    assert isinstance(spawner, GlicidSpawner)
    assert isinstance(spawner, SlurmSpawner)

    assert spawner.batchspawner_singleuser_cmd == 'glicid-spawner-singleuser'

    assert spawner.disable_user_config
    assert spawner.notebook_dir == '/'
    assert spawner.default_url == '/lab/tree/home/{username}?reset'
    assert spawner.req_job_name == 'jupyterhub_glicid'
    assert spawner.req_qos == 'short'
    assert spawner.progress_rate == 10


def test_spawner_parse_job_id():
    """Test spawner job id parser."""
    spawner = GlicidSpawner()

    assert spawner.parse_job_id('123') == 123
    assert spawner.parse_job_id('456;nautilus') == '456 -M nautilus'

    assert spawner.parse_job_id('') is None


def test_spawner_options_form(monkeypatch):
    """Test spawner options form."""
    monkeypatch.setattr(
        glicid_spawner.spawner, 'options_form', lambda username: f"options_form('{username}')"
    )
    monkeypatch.setattr(
        glicid_spawner.spawner,
        'options_from_form',
        lambda formdata: f'options_from_form({formdata})',
    )

    spawner = GlicidSpawner(user=User('john-doe'))

    assert spawner.user.name == 'john-doe'
    assert spawner.options_form == "options_form('john-doe')"
    assert spawner.options_from_form({'foo': 123}) == "options_from_form({'foo': 123})"

    spawner = GlicidSpawner(user_options={'workdir': '/LAB-DATA'})

    assert spawner.default_url == '/lab/tree/LAB-DATA?reset'


@pytest.mark.asyncio
async def test_spawner_progress(monkeypatch):
    """Test spawner progress messages."""

    def isrunning():
        """Running generator values."""
        yield False  # NOTFOUND
        yield False  # PENDING
        yield from repeat(True)  # RUNNING

    def ispending():
        """Pending generator values."""
        yield False  # NOTFOUND
        yield True  # PENDING
        yield from repeat(False)  # RUNNING (never used)

    # Mock Job status generator
    iter_ispending = iter(ispending())
    iter_isrunning = iter(isrunning())

    monkeypatch.setattr(
        glicid_spawner.GlicidSpawner, 'state_isrunning', lambda _: next(iter_isrunning)
    )
    monkeypatch.setattr(
        glicid_spawner.GlicidSpawner, 'state_ispending', lambda _: next(iter_ispending)
    )

    async def mock_sleep(_):
        """Mock asyncio sleep."""

    monkeypatch.setattr(asyncio, 'sleep', mock_sleep)

    spawner = GlicidSpawner(progress_rate=20)

    progress = [msg['progress'] async for msg in spawner.progress()]

    assert progress == [
        10,  # submit
        20,  # pending
        60,  # running | elapse time = 20 -> setup
        80,  # running | elapse time = 40 -> connect
        95,  # running | elapse time = 60 -> too long
    ]