164 lines
5 KiB
Python
164 lines
5 KiB
Python
"""Tests for daemon enable/disable functionality."""
|
|
|
|
import asyncio
|
|
|
|
import pytest
|
|
|
|
from .conftest import TestDaemon
|
|
|
|
|
|
class TestEnableDisable:
|
|
"""Test daemon enable/disable without stopping."""
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initial_enabled_state(self) -> None:
|
|
"""Test daemon starts enabled by default."""
|
|
daemon = TestDaemon(auto_enable=True)
|
|
assert daemon.is_enabled
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_initial_disabled_state(self) -> None:
|
|
"""Test daemon can start disabled."""
|
|
daemon = TestDaemon(auto_enable=False)
|
|
assert not daemon.is_enabled
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enable_disable_when_stopped(self, test_daemon: TestDaemon) -> None:
|
|
"""Test enable/disable when daemon is stopped."""
|
|
assert test_daemon.is_enabled
|
|
|
|
test_daemon.disable()
|
|
assert not test_daemon.is_enabled
|
|
|
|
test_daemon.enable()
|
|
assert test_daemon.is_enabled
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enable_disable_when_running(self, test_daemon: TestDaemon) -> None:
|
|
"""Test enable/disable while daemon is running."""
|
|
task = asyncio.create_task(test_daemon.start())
|
|
await asyncio.sleep(0.1)
|
|
|
|
# Daemon should be running and executing cycles
|
|
assert test_daemon.is_running
|
|
assert test_daemon.is_enabled
|
|
cycles_before_disable = test_daemon.total_cycles
|
|
assert cycles_before_disable > 0
|
|
|
|
# Disable daemon
|
|
test_daemon.disable()
|
|
assert not test_daemon.is_enabled
|
|
|
|
# Wait and verify no new cycles are executed
|
|
await asyncio.sleep(1.5)
|
|
assert test_daemon.total_cycles == cycles_before_disable
|
|
|
|
# Re-enable daemon
|
|
test_daemon.enable()
|
|
assert test_daemon.is_enabled
|
|
|
|
# Wait and verify cycles resume
|
|
await asyncio.sleep(1.5)
|
|
assert test_daemon.total_cycles > cycles_before_disable
|
|
|
|
await test_daemon.stop()
|
|
await task
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_start_disabled_daemon(self) -> None:
|
|
"""Test starting a daemon that is disabled."""
|
|
daemon = TestDaemon(auto_enable=False, interval_seconds=1)
|
|
task = asyncio.create_task(daemon.start())
|
|
|
|
# Wait and verify no cycles are executed
|
|
await asyncio.sleep(1.5)
|
|
assert daemon.is_running
|
|
assert not daemon.is_enabled
|
|
assert daemon.total_cycles == 0
|
|
|
|
# Enable and verify cycles start
|
|
daemon.enable()
|
|
await asyncio.sleep(1.5)
|
|
assert daemon.total_cycles > 0
|
|
|
|
await daemon.stop()
|
|
await task
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_enable_calls(self, test_daemon: TestDaemon) -> None:
|
|
"""Test that calling enable() multiple times is safe."""
|
|
assert test_daemon.is_enabled
|
|
|
|
test_daemon.enable()
|
|
test_daemon.enable()
|
|
|
|
assert test_daemon.is_enabled
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_multiple_disable_calls(self, test_daemon: TestDaemon) -> None:
|
|
"""Test that calling disable() multiple times is safe."""
|
|
test_daemon.disable()
|
|
test_daemon.disable()
|
|
|
|
assert not test_daemon.is_enabled
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_disable_stop_start_enable(self) -> None:
|
|
"""Test complex enable/disable/stop/start sequence."""
|
|
daemon = TestDaemon(interval_seconds=1)
|
|
|
|
# Start enabled
|
|
task = asyncio.create_task(daemon.start())
|
|
await asyncio.sleep(1.5)
|
|
assert daemon.total_cycles > 0
|
|
cycles_1 = daemon.total_cycles
|
|
|
|
# Disable
|
|
daemon.disable()
|
|
await asyncio.sleep(1.5)
|
|
assert daemon.total_cycles == cycles_1 # No new cycles
|
|
|
|
# Stop
|
|
await daemon.stop()
|
|
await task
|
|
|
|
# Start again (still disabled)
|
|
task = asyncio.create_task(daemon.start())
|
|
await asyncio.sleep(1.5)
|
|
assert daemon.total_cycles == cycles_1 # Still no new cycles
|
|
|
|
# Enable
|
|
daemon.enable()
|
|
await asyncio.sleep(1.5)
|
|
assert daemon.total_cycles > cycles_1 # Cycles resume
|
|
|
|
await daemon.stop()
|
|
await task
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_enabled_state_persists_across_restarts(self) -> None:
|
|
"""Test that enabled/disabled state persists when restarting."""
|
|
daemon = TestDaemon(interval_seconds=1)
|
|
|
|
# Start and disable immediately
|
|
task = asyncio.create_task(daemon.start())
|
|
await asyncio.sleep(0.05) # Just enough to start
|
|
daemon.disable()
|
|
assert not daemon.is_enabled
|
|
|
|
# Track cycles at disable
|
|
cycles_at_disable = daemon.total_cycles
|
|
|
|
# Stop
|
|
await daemon.stop()
|
|
await task
|
|
|
|
# Start again - should still be disabled
|
|
task = asyncio.create_task(daemon.start())
|
|
await asyncio.sleep(1.5)
|
|
assert not daemon.is_enabled
|
|
# No new cycles should have been added
|
|
assert daemon.total_cycles == cycles_at_disable
|
|
|
|
await daemon.stop()
|
|
await task
|