test(core): extend coverage — console, errors, log, service, time, benchmark
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>
This commit is contained in:
0
test/core/service/__init__.py
Normal file
0
test/core/service/__init__.py
Normal file
118
test/core/service/service_test.py
Normal file
118
test/core/service/service_test.py
Normal file
@@ -0,0 +1,118 @@
|
||||
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()
|
||||
Reference in New Issue
Block a user