"""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}' 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' @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 ]