clu
27205022a5
Add missing test modules for previously untested core areas: - console: ForegroundColorEnum, BackgroundColorEnum, Console methods - errors: dependency_error, module_dependency_error - log: LogLevel ordering/values, LogSettings, Logger (should_log, format, file write, fatal) - service: HostedService, StartupTask, CronjobABC (start/stop/loop/task cancellation) - time: TimeFormatSettings properties and setters - utils: Benchmark.time / .memory / .all call-count and output Also fix existing test files: environment cleanup, cron exception specificity, json_processor kwargs bug doc, configuration_model_abc to_dict bug doc. All 199 tests pass, black clean. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
119 lines
2.6 KiB
Python
119 lines
2.6 KiB
Python
import asyncio
|
|
import pytest
|
|
from cpl.core.service.hosted_service import HostedService
|
|
from cpl.core.service.startup_task import StartupTask
|
|
from cpl.core.service.cronjob import CronjobABC
|
|
from cpl.core.time.cron import Cron
|
|
|
|
# --- HostedService ---
|
|
|
|
|
|
def test_hosted_service_cannot_instantiate():
|
|
with pytest.raises(TypeError):
|
|
HostedService()
|
|
|
|
|
|
def test_hosted_service_concrete_can_start_stop():
|
|
class MyService(HostedService):
|
|
def __init__(self):
|
|
self.started = False
|
|
self.stopped = False
|
|
|
|
async def start(self):
|
|
self.started = True
|
|
|
|
async def stop(self):
|
|
self.stopped = True
|
|
|
|
async def run():
|
|
svc = MyService()
|
|
await svc.start()
|
|
assert svc.started is True
|
|
await svc.stop()
|
|
assert svc.stopped is True
|
|
|
|
asyncio.run(run())
|
|
|
|
|
|
# --- StartupTask ---
|
|
|
|
|
|
def test_startup_task_cannot_instantiate():
|
|
with pytest.raises(TypeError):
|
|
StartupTask()
|
|
|
|
|
|
def test_startup_task_concrete_runs():
|
|
class MyTask(StartupTask):
|
|
def __init__(self):
|
|
self.ran = False
|
|
|
|
async def run(self):
|
|
self.ran = True
|
|
|
|
async def execute():
|
|
task = MyTask()
|
|
await task.run()
|
|
assert task.ran is True
|
|
|
|
asyncio.run(execute())
|
|
|
|
|
|
# --- CronjobABC ---
|
|
|
|
|
|
def test_cronjob_cannot_instantiate():
|
|
with pytest.raises(TypeError):
|
|
CronjobABC(Cron("* * * * *"))
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cronjob_loop_is_called():
|
|
loop_calls = []
|
|
|
|
class MyCronJob(CronjobABC):
|
|
def __init__(self):
|
|
CronjobABC.__init__(self, Cron("* * * * *"))
|
|
|
|
async def loop(self):
|
|
loop_calls.append(1)
|
|
|
|
job = MyCronJob()
|
|
assert job._running is False
|
|
await job.start()
|
|
assert job._running is True
|
|
# Give it a moment then stop
|
|
await asyncio.sleep(0.05)
|
|
await job.stop()
|
|
assert job._running is False
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cronjob_stop_cancels_task():
|
|
class MyCronJob(CronjobABC):
|
|
def __init__(self):
|
|
CronjobABC.__init__(self, Cron("* * * * *"))
|
|
|
|
async def loop(self):
|
|
pass
|
|
|
|
job = MyCronJob()
|
|
await job.start()
|
|
assert job._task is not None
|
|
await job.stop()
|
|
assert job._task.cancelled() or job._task.done()
|
|
|
|
|
|
@pytest.mark.asyncio
|
|
async def test_cronjob_stop_without_start():
|
|
class MyCronJob(CronjobABC):
|
|
def __init__(self):
|
|
CronjobABC.__init__(self, Cron("* * * * *"))
|
|
|
|
async def loop(self):
|
|
pass
|
|
|
|
job = MyCronJob()
|
|
# stop without start should not raise
|
|
await job.stop()
|