mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-10-24 17:35:19 +08:00
test(core): strengthen handler contracts and report source quality (#2632)
This commit is contained in:
huanghuoguoguo
@@ -172,6 +172,30 @@ class MockApplication:
|
||||
self.sess_mgr.get_conversation = AsyncMock()
|
||||
|
||||
|
||||
def make_handler_query(text):
|
||||
"""Build SDK input entities; only the adapter and orchestration are fakes."""
|
||||
from tests.factories import text_query
|
||||
from langbot_plugin.api.entities.builtin.provider.message import Message
|
||||
from langbot_plugin.api.entities.builtin.provider.prompt import Prompt
|
||||
from langbot_plugin.api.entities.builtin.provider.session import Conversation, Session
|
||||
|
||||
query = text_query(text)
|
||||
query.user_message = Message(role='user', content=text)
|
||||
query.adapter.is_stream_output_supported = AsyncMock(return_value=True)
|
||||
query.adapter.create_message_card = AsyncMock()
|
||||
query.session = Session(
|
||||
launcher_type=query.launcher_type,
|
||||
launcher_id=query.launcher_id,
|
||||
using_conversation=Conversation(
|
||||
prompt=Prompt(name='test', messages=[]),
|
||||
messages=[],
|
||||
pipeline_uuid=query.pipeline_uuid,
|
||||
bot_uuid=query.bot_uuid,
|
||||
),
|
||||
)
|
||||
return query
|
||||
|
||||
|
||||
class TestErrorHandling:
|
||||
"""Tests for orchestrator error handling."""
|
||||
|
||||
@@ -242,139 +266,95 @@ class TestChatHandlerAsyncBehavior:
|
||||
"""Real async tests for ChatMessageHandler.handle() with mocked orchestrator."""
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_streaming_single_resp_message_id(self):
|
||||
"""Streaming mode: all chunks should have same resp_message_id."""
|
||||
async def test_interleaved_streams_keep_response_ids_and_messages_isolated(self):
|
||||
"""Two turns on one handler must never update each other's response card."""
|
||||
from langbot.pkg.pipeline.process.handlers.chat import ChatMessageHandler
|
||||
from langbot.pkg.pipeline import entities
|
||||
from langbot_plugin.api.entities.builtin.provider.message import Message, MessageChunk
|
||||
|
||||
# Create chunks for streaming
|
||||
chunks = [
|
||||
MockMessageChunk('Hello'),
|
||||
MockMessageChunk('Hello World'),
|
||||
MockMessageChunk('Hello World!'),
|
||||
class InterleavedOrchestrator(MockAgentRunOrchestrator):
|
||||
async def run_from_query(self, query):
|
||||
text = query.user_message.content
|
||||
yield MessageChunk(role='assistant', content=f'{text}: par', all_content=f'{text}: par')
|
||||
yield MessageChunk(role='assistant', content='tial', all_content=f'{text}: partial')
|
||||
yield Message(role='assistant', content=f'{text}: completed')
|
||||
|
||||
handler = ChatMessageHandler(MockApplication(orchestrator=InterleavedOrchestrator()))
|
||||
queries = [make_handler_query('first'), make_handler_query('second')]
|
||||
streams = [handler.handle(query) for query in queries]
|
||||
observed = [[], []]
|
||||
saved_messages = [None, None]
|
||||
|
||||
try:
|
||||
for index in [0, 1, 0, 1, 0, 1]:
|
||||
result = await anext(streams[index])
|
||||
assert result.result_type == entities.ResultType.CONTINUE
|
||||
assert result.new_query is queries[index]
|
||||
assert len(queries[index].resp_messages) == 1
|
||||
current = queries[index].resp_messages[0]
|
||||
assert isinstance(current, MessageChunk)
|
||||
observed[index].append((current.content, current.is_final, current.resp_message_id))
|
||||
saved_messages[index] = current.model_dump()
|
||||
|
||||
other = 1 - index
|
||||
if saved_messages[other] is not None:
|
||||
assert [message.model_dump() for message in queries[other].resp_messages] == [saved_messages[other]]
|
||||
|
||||
for stream in streams:
|
||||
with pytest.raises(StopAsyncIteration):
|
||||
await anext(stream)
|
||||
finally:
|
||||
for stream in streams:
|
||||
await stream.aclose()
|
||||
|
||||
first_id, second_id = (messages[0][2] for messages in observed)
|
||||
assert first_id and second_id and first_id != second_id
|
||||
assert observed == [
|
||||
[
|
||||
('first: par', False, first_id),
|
||||
('first: partial', False, first_id),
|
||||
('first: completed', True, first_id),
|
||||
],
|
||||
[
|
||||
('second: par', False, second_id),
|
||||
('second: partial', False, second_id),
|
||||
('second: completed', True, second_id),
|
||||
],
|
||||
]
|
||||
|
||||
orchestrator = MockAgentRunOrchestrator(chunks=chunks)
|
||||
mock_ap = MockApplication(orchestrator=orchestrator)
|
||||
|
||||
# Mock event context to not prevent default
|
||||
event_ctx = MockEventContext(prevented=False)
|
||||
mock_ap.plugin_connector.emit_event = AsyncMock(return_value=event_ctx)
|
||||
|
||||
query = MockQuery()
|
||||
query.adapter.is_stream = True # Enable streaming mode
|
||||
|
||||
handler = ChatMessageHandler(mock_ap)
|
||||
|
||||
# Mock event creation and StageProcessResult to bypass pydantic validation
|
||||
mock_event = MagicMock()
|
||||
mock_event.return_value = MagicMock()
|
||||
|
||||
def make_result(*args, **kwargs):
|
||||
return MagicMock(result_type=kwargs.get('result_type', entities.ResultType.CONTINUE))
|
||||
|
||||
with (
|
||||
patch('langbot.pkg.pipeline.process.handlers.chat.events') as mock_events_module,
|
||||
patch('langbot.pkg.pipeline.entities.StageProcessResult', side_effect=make_result),
|
||||
):
|
||||
mock_events_module.PersonNormalMessageReceived = mock_event
|
||||
mock_events_module.GroupNormalMessageReceived = mock_event
|
||||
|
||||
results = []
|
||||
async for result in handler.handle(query):
|
||||
results.append(result)
|
||||
|
||||
# Verify single resp_message_id
|
||||
resp_ids = [msg.resp_message_id for msg in query.resp_messages if hasattr(msg, 'resp_message_id')]
|
||||
assert len(set(resp_ids)) == 1 # All same ID
|
||||
|
||||
# Verify pop/append pattern: only last chunk remains
|
||||
assert len(query.resp_messages) == 1
|
||||
assert query.resp_messages[0].content == 'Hello World!'
|
||||
for query, response_id in zip(queries, [first_id, second_id]):
|
||||
query.adapter.create_message_card.assert_awaited_once_with(response_id, query.message_event)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_streaming_renders_accumulated_deltas_and_completes_same_message(self):
|
||||
"""Incremental runner deltas and message.completed update one response stream."""
|
||||
async def test_non_streaming_keeps_prior_and_runner_messages_and_yields_once(self):
|
||||
"""Non-streaming delivery preserves earlier messages and exposes only the completed turn."""
|
||||
from langbot.pkg.pipeline.process.handlers.chat import ChatMessageHandler
|
||||
from langbot.pkg.pipeline import entities
|
||||
from langbot_plugin.api.entities.builtin.provider import message as provider_message
|
||||
from langbot_plugin.api.entities.builtin.provider.message import Message, MessageChunk
|
||||
|
||||
prior = Message(role='assistant', content='earlier reply', resp_message_id='earlier-id')
|
||||
prior_snapshot = prior.model_dump()
|
||||
chunks = [
|
||||
provider_message.MessageChunk(role='assistant', content='Hel', all_content='Hel'),
|
||||
provider_message.MessageChunk(role='assistant', content='lo', all_content='Hello'),
|
||||
provider_message.Message(role='assistant', content='Hello'),
|
||||
MessageChunk(role='assistant', content='working'),
|
||||
Message(role='assistant', content='completed'),
|
||||
]
|
||||
orchestrator = MockAgentRunOrchestrator(chunks=chunks)
|
||||
mock_ap = MockApplication(orchestrator=orchestrator)
|
||||
mock_ap.plugin_connector.emit_event = AsyncMock(return_value=MockEventContext(prevented=False))
|
||||
query = MockQuery()
|
||||
query.adapter.is_stream = True
|
||||
handler = ChatMessageHandler(mock_ap)
|
||||
mock_event = MagicMock()
|
||||
mock_event.return_value = MagicMock()
|
||||
|
||||
def make_result(*args, **kwargs):
|
||||
return MagicMock(result_type=kwargs.get('result_type', entities.ResultType.CONTINUE))
|
||||
|
||||
handler = ChatMessageHandler(MockApplication(orchestrator=MockAgentRunOrchestrator(chunks=chunks)))
|
||||
query = make_handler_query('question')
|
||||
query.adapter.is_stream_output_supported.return_value = False
|
||||
query.resp_messages = [prior]
|
||||
observed = []
|
||||
with (
|
||||
patch('langbot.pkg.pipeline.process.handlers.chat.events') as mock_events_module,
|
||||
patch('langbot.pkg.pipeline.entities.StageProcessResult', side_effect=make_result),
|
||||
):
|
||||
mock_events_module.PersonNormalMessageReceived = mock_event
|
||||
mock_events_module.GroupNormalMessageReceived = mock_event
|
||||
|
||||
async for _ in handler.handle(query):
|
||||
current = query.resp_messages[-1]
|
||||
observed.append((current.content, current.is_final, current.resp_message_id))
|
||||
async for result in handler.handle(query):
|
||||
assert result.result_type == entities.ResultType.CONTINUE
|
||||
assert result.new_query is query
|
||||
observed.append([message.content for message in result.new_query.resp_messages])
|
||||
|
||||
assert [content for content, _, _ in observed] == ['Hel', 'Hello', 'Hello']
|
||||
assert [is_final for _, is_final, _ in observed] == [False, False, True]
|
||||
assert len({response_id for _, _, response_id in observed}) == 1
|
||||
assert all(isinstance(message, provider_message.MessageChunk) for message in query.resp_messages)
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_non_streaming_keeps_results_and_yields_once(self):
|
||||
"""Non-streaming mode keeps runner results but runs downstream stages once."""
|
||||
from langbot.pkg.pipeline.process.handlers.chat import ChatMessageHandler
|
||||
from langbot.pkg.pipeline import entities
|
||||
|
||||
chunks = [
|
||||
MockMessageChunk('Response 1'),
|
||||
MockMessageChunk('Response 2'),
|
||||
]
|
||||
|
||||
orchestrator = MockAgentRunOrchestrator(chunks=chunks)
|
||||
mock_ap = MockApplication(orchestrator=orchestrator)
|
||||
mock_ap.plugin_connector.emit_event = AsyncMock(return_value=MockEventContext(prevented=False))
|
||||
|
||||
query = MockQuery()
|
||||
query.adapter.is_stream = False # Disable streaming mode
|
||||
|
||||
handler = ChatMessageHandler(mock_ap)
|
||||
|
||||
mock_event = MagicMock()
|
||||
mock_event.return_value = MagicMock()
|
||||
|
||||
def make_result(*args, **kwargs):
|
||||
return MagicMock(result_type=kwargs.get('result_type', entities.ResultType.CONTINUE))
|
||||
|
||||
with (
|
||||
patch('langbot.pkg.pipeline.process.handlers.chat.events') as mock_events_module,
|
||||
patch('langbot.pkg.pipeline.entities.StageProcessResult', side_effect=make_result),
|
||||
):
|
||||
mock_events_module.PersonNormalMessageReceived = mock_event
|
||||
mock_events_module.GroupNormalMessageReceived = mock_event
|
||||
|
||||
results = []
|
||||
async for result in handler.handle(query):
|
||||
results.append(result)
|
||||
|
||||
# No pop: all chunks should remain
|
||||
assert len(query.resp_messages) == 2
|
||||
assert query.resp_messages[0].content == 'Response 1'
|
||||
assert query.resp_messages[1].content == 'Response 2'
|
||||
assert len(results) == 1
|
||||
assert observed == [['earlier reply', 'working', 'completed']]
|
||||
assert query.resp_messages == [prior, *chunks]
|
||||
assert prior.model_dump() == prior_snapshot
|
||||
assert chunks[0].resp_message_id
|
||||
assert chunks[0].resp_message_id == chunks[1].resp_message_id
|
||||
assert chunks[0].resp_message_id != prior.resp_message_id
|
||||
query.adapter.create_message_card.assert_not_awaited()
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_agent_turn_recreates_conversation_if_tool_resets_it(self):
|
||||
|
||||
@@ -205,45 +205,45 @@ class TestChatMessageHandlerReal:
|
||||
assert query.resp_messages[0] == reply_chain
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_user_message_alter_string(self, fake_app, mock_event_ctx, set_runner):
|
||||
"""user_message_alter as string updates query.user_message."""
|
||||
@pytest.mark.parametrize('alteration_kind', ['string', 'element', 'list'])
|
||||
async def test_user_message_alter_reaches_orchestrator(
|
||||
self, fake_app, mock_event_ctx, chat_session, alteration_kind
|
||||
):
|
||||
"""The runner receives the exact plugin alteration before it starts."""
|
||||
from langbot_plugin.api.entities.builtin.provider.message import ContentElement, Message
|
||||
|
||||
from tests.factories import text_query
|
||||
from langbot_plugin.api.entities.builtin.provider.message import Message, ContentElement
|
||||
|
||||
chat = get_chat_handler()
|
||||
|
||||
mock_event_ctx.is_prevented_default.return_value = False
|
||||
mock_event_ctx.event.user_message_alter = 'altered text'
|
||||
handler_query = text_query('original')
|
||||
handler_query.user_message = Message(role='user', content='original')
|
||||
handler_query.session = chat_session
|
||||
expected = [ContentElement.from_text('altered text')]
|
||||
if alteration_kind == 'string':
|
||||
alteration = 'altered text'
|
||||
elif alteration_kind == 'element':
|
||||
alteration = expected[0]
|
||||
else:
|
||||
expected.append(ContentElement.from_image_url('https://example.com/input.png'))
|
||||
alteration = expected
|
||||
mock_event_ctx.event.user_message_alter = alteration
|
||||
fake_app.plugin_connector.emit_event = AsyncMock(return_value=mock_event_ctx)
|
||||
received = []
|
||||
|
||||
query = text_query('original')
|
||||
query.adapter = Mock()
|
||||
query.adapter.is_stream_output_supported = AsyncMock(return_value=False)
|
||||
query.user_message = Message(role='user', content=[ContentElement.from_text('original')])
|
||||
received_contents = []
|
||||
async def run_from_query(query):
|
||||
received.append(query.user_message.model_copy(deep=True))
|
||||
yield Message(role='assistant', content='ok')
|
||||
|
||||
class QuickRunner:
|
||||
name = 'local-agent'
|
||||
fake_app.agent_run_orchestrator.run_from_query = run_from_query
|
||||
handler = get_chat_handler().ChatMessageHandler(fake_app)
|
||||
|
||||
def __init__(self, app, config):
|
||||
self.app = app
|
||||
self.config = config
|
||||
results = [result async for result in handler.handle(handler_query)]
|
||||
|
||||
async def run(self, query):
|
||||
received_contents.append(query.user_message.content)
|
||||
yield Message(role='assistant', content='ok')
|
||||
|
||||
set_runner(QuickRunner)
|
||||
|
||||
handler = chat.ChatMessageHandler(fake_app)
|
||||
|
||||
results = []
|
||||
async for result in handler.handle(query):
|
||||
results.append(result)
|
||||
|
||||
expected_content = [ContentElement.from_text('altered text')]
|
||||
assert query.user_message.content == expected_content
|
||||
assert received_contents == [expected_content]
|
||||
assert received == [Message(role='user', content=expected)]
|
||||
assert handler_query.user_message.content == expected
|
||||
assert len(results) == 1
|
||||
assert results[0].result_type == get_entities().ResultType.CONTINUE
|
||||
assert results[0].new_query is handler_query
|
||||
assert [message.content for message in handler_query.resp_messages] == ['ok']
|
||||
|
||||
@pytest.mark.asyncio
|
||||
async def test_adapter_without_stream_method_defaults_non_stream(
|
||||
|
||||
@@ -0,0 +1,258 @@
|
||||
"""Keep complexity reporting honest about metric values and baseline changes."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import importlib.util
|
||||
import json
|
||||
from pathlib import Path
|
||||
import subprocess
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def reporter():
|
||||
path = Path(__file__).resolve().parents[2] / 'scripts/test-complexity.py'
|
||||
spec = importlib.util.spec_from_file_location('complexity_report', path)
|
||||
module = importlib.util.module_from_spec(spec)
|
||||
spec.loader.exec_module(module)
|
||||
return module
|
||||
|
||||
|
||||
def source(root, text, name='sample.py'):
|
||||
path = root / 'src/langbot' / name
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
path.write_text(text)
|
||||
return path
|
||||
|
||||
|
||||
def commit(root):
|
||||
def git(*args):
|
||||
return subprocess.run(['git', *args], cwd=root, check=True, capture_output=True).stdout.decode().strip()
|
||||
|
||||
git('init', '-q')
|
||||
git('add', 'src')
|
||||
git('-c', 'user.name=Test', '-c', '[email protected]', 'commit', '-qm', 'baseline')
|
||||
return git('rev-parse', 'HEAD')
|
||||
|
||||
|
||||
def test_real_ruff_counts_branches_nested_functions_and_ignores_noqa(reporter, tmp_path):
|
||||
source(
|
||||
tmp_path,
|
||||
"""\
|
||||
class Worker:
|
||||
def run(self, value): # noqa: C901
|
||||
if value:
|
||||
return value
|
||||
return None
|
||||
|
||||
async def outer(value):
|
||||
def inner():
|
||||
return 1
|
||||
if value:
|
||||
return inner()
|
||||
return None
|
||||
""",
|
||||
)
|
||||
functions = {item['name']: item for item in reporter.measure(tmp_path)}
|
||||
assert functions['Worker.run']['complexity'] == 2
|
||||
assert functions['outer']['complexity'] == 3 # Ruff also includes the nested definition.
|
||||
assert functions['outer.inner']['complexity'] == 1
|
||||
assert functions['Worker.run']['path'] == 'src/langbot/sample.py'
|
||||
|
||||
|
||||
def test_all_source_includes_libs_and_overrides_repo_lint_exclusions(reporter, tmp_path):
|
||||
source(tmp_path, 'def vendored():\n pass\n', 'libs/helper.py')
|
||||
(tmp_path / 'pyproject.toml').write_text('[tool.ruff]\nexclude = ["src"]\n')
|
||||
(tmp_path / '.gitignore').write_text('src/\n')
|
||||
assert [item['name'] for item in reporter.measure(tmp_path)] == ['vendored']
|
||||
|
||||
|
||||
def test_invalid_source_is_an_analysis_failure(reporter, tmp_path):
|
||||
source(tmp_path, 'def broken(:\n')
|
||||
with pytest.raises(ValueError, match='Unexpected Ruff diagnostic'):
|
||||
reporter.measure(tmp_path)
|
||||
|
||||
|
||||
def test_duplicate_definitions_have_distinct_stable_occurrences(reporter, tmp_path):
|
||||
path = source(
|
||||
tmp_path,
|
||||
"""\
|
||||
class Worker:
|
||||
@property
|
||||
def value(self):
|
||||
return 1
|
||||
@value.setter
|
||||
def value(self, value):
|
||||
pass
|
||||
""",
|
||||
)
|
||||
definitions = list(reporter.index_functions(path).values())
|
||||
assert [item['name'] for item in definitions] == ['Worker.value', 'Worker.value']
|
||||
assert [item['occurrence'] for item in definitions] == [1, 2]
|
||||
|
||||
|
||||
def test_changed_function_deltas_additions_deletions_and_formatting(reporter, tmp_path):
|
||||
path = source(
|
||||
tmp_path,
|
||||
"""\
|
||||
def changed(value):
|
||||
return value
|
||||
|
||||
def removed():
|
||||
pass
|
||||
|
||||
def unchanged():
|
||||
return 1
|
||||
""",
|
||||
)
|
||||
before = reporter.measure(tmp_path)
|
||||
path.write_text("""\
|
||||
# Moving lines and adding comments does not change a function.
|
||||
def unchanged():
|
||||
return 1
|
||||
|
||||
def changed(value):
|
||||
if value:
|
||||
return value
|
||||
return None
|
||||
|
||||
def added():
|
||||
pass
|
||||
""")
|
||||
changes = reporter.compare(before, reporter.measure(tmp_path), {})
|
||||
assert {item['status'] for item in changes} == {'modified', 'added', 'deleted'}
|
||||
modified = next(item for item in changes if item['status'] == 'modified')
|
||||
assert modified['after']['name'] == 'changed'
|
||||
assert modified['delta'] == 1
|
||||
assert all(item['delta'] is None for item in changes if item['status'] != 'modified')
|
||||
|
||||
|
||||
def test_function_renames_are_added_deleted_not_false_improvements(reporter, tmp_path):
|
||||
path = source(tmp_path, 'def old_name():\n pass\n')
|
||||
before = reporter.measure(tmp_path)
|
||||
path.write_text('def new_name():\n pass\n')
|
||||
changes = reporter.compare(before, reporter.measure(tmp_path), {})
|
||||
assert {item['status'] for item in changes} == {'added', 'deleted'}
|
||||
assert all(item['delta'] is None for item in changes)
|
||||
|
||||
|
||||
def test_git_baseline_and_file_rename_preserve_identity(reporter, tmp_path):
|
||||
path = source(tmp_path, 'def sample():\n return 1\n')
|
||||
revision = commit(tmp_path)
|
||||
path.rename(path.with_name('renamed.py'))
|
||||
# Stage the destination so git can detect its similarity to the deleted file.
|
||||
subprocess.run(['git', 'add', 'src'], cwd=tmp_path, check=True)
|
||||
resolved, before = reporter.baseline(tmp_path, revision)
|
||||
assert resolved == revision
|
||||
assert before[0]['path'] == 'src/langbot/sample.py'
|
||||
renames = reporter.file_renames(tmp_path, revision)
|
||||
assert renames == {'src/langbot/renamed.py': 'src/langbot/sample.py'}
|
||||
changes = reporter.compare(before, reporter.measure(tmp_path), renames)
|
||||
assert len(changes) == 1
|
||||
assert changes[0]['status'] == 'renamed_file'
|
||||
assert changes[0]['delta'] == 0
|
||||
|
||||
|
||||
def test_invalid_base_ref_fails_rather_than_using_head(reporter, tmp_path):
|
||||
source(tmp_path, 'def sample():\n pass\n')
|
||||
commit(tmp_path)
|
||||
with pytest.raises(ValueError):
|
||||
reporter.baseline(tmp_path, 'does-not-exist')
|
||||
|
||||
|
||||
def coverage_document(branch=True):
|
||||
return {
|
||||
'meta': {'branch_coverage': branch},
|
||||
'files': {
|
||||
'src/langbot/sample.py': {
|
||||
'executed_lines': [1, 2, 3],
|
||||
'missing_lines': [4],
|
||||
'executed_branches': [[2, 3]],
|
||||
'missing_branches': [[2, 4]],
|
||||
'summary': {
|
||||
'covered_lines': 3,
|
||||
'num_statements': 4,
|
||||
'covered_branches': 1,
|
||||
'num_branches': 2,
|
||||
'percent_covered': 66.666,
|
||||
},
|
||||
},
|
||||
'tests/unrelated.py': {
|
||||
'executed_lines': [1],
|
||||
'missing_lines': [],
|
||||
'summary': {'covered_lines': 1, 'num_statements': 1},
|
||||
},
|
||||
},
|
||||
}
|
||||
|
||||
|
||||
def test_line_and_branch_coverage_are_separate_not_combined(reporter, tmp_path):
|
||||
source(tmp_path, 'def sample(value):\n if value:\n return 1\n return 0\n')
|
||||
functions = reporter.measure(tmp_path)
|
||||
report = reporter.attach_coverage(functions, coverage_document(), tmp_path)
|
||||
assert report['line_percent'] == 75
|
||||
assert report['branch_percent'] == 50
|
||||
assert report['matched_source_files'] == 1
|
||||
assert functions[0]['coverage']['line_percent'] == 75
|
||||
assert functions[0]['coverage']['branch_percent'] == 50
|
||||
|
||||
|
||||
def test_unknown_coverage_is_not_reported_as_zero(reporter, tmp_path):
|
||||
source(tmp_path, 'def unknown():\n pass\n', 'unknown.py')
|
||||
functions = reporter.measure(tmp_path)
|
||||
report = reporter.attach_coverage(functions, coverage_document(branch=False), tmp_path)
|
||||
assert functions[0]['coverage'] is None
|
||||
assert report['functions_without_coverage'] == 1
|
||||
assert report['branch_percent'] is None
|
||||
|
||||
|
||||
def test_no_branch_function_has_no_artificial_branch_percent(reporter, tmp_path):
|
||||
source(tmp_path, '\n\n\n\ndef sample():\n return 1\n')
|
||||
functions = reporter.measure(tmp_path)
|
||||
reporter.attach_coverage(functions, coverage_document(), tmp_path)
|
||||
assert functions[0]['coverage']['branch_percent'] is None
|
||||
assert functions[0]['coverage']['line_percent'] is None
|
||||
|
||||
|
||||
def test_empty_scope_statistics_have_no_artificial_averages(reporter):
|
||||
stats = reporter.statistics([])
|
||||
assert stats['function_count'] == 0
|
||||
assert stats['maximum'] is None
|
||||
assert stats['mean'] is None
|
||||
assert sum(stats['distribution'].values()) == 0
|
||||
|
||||
|
||||
def test_cli_writes_reports_and_high_complexity_is_not_failure(reporter, tmp_path):
|
||||
# 60 independent decisions produce CC 61, intentionally above ordinary limits.
|
||||
source(tmp_path, 'def high(value):\n' + ' if value:\n value -= 1\n' * 60)
|
||||
json_path = tmp_path / 'reports/complexity.json'
|
||||
summary_path = tmp_path / 'reports/complexity.md'
|
||||
assert reporter.main(['--root', str(tmp_path), '--json', str(json_path), '--summary', str(summary_path)]) == 0
|
||||
document = json.loads(json_path.read_text())
|
||||
assert document['current']['maximum'] == 61
|
||||
assert document['current']['distribution']['51+'] == 1
|
||||
assert 'CC=61' in summary_path.read_text()
|
||||
assert 'report only; no complexity gate' in summary_path.read_text()
|
||||
|
||||
|
||||
def test_missing_source_fails_without_publishing_misleading_empty_report(reporter, tmp_path):
|
||||
destination = tmp_path / 'complexity.json'
|
||||
assert reporter.main(['--root', str(tmp_path), '--json', str(destination)]) == 1
|
||||
assert not destination.exists()
|
||||
|
||||
|
||||
def test_notebooks_and_stubs_are_outside_python_source_scope(reporter, tmp_path):
|
||||
source(tmp_path, 'def sample():\n pass\n')
|
||||
source(tmp_path, 'def annotation_only(): ...\n', 'types.pyi')
|
||||
source(tmp_path, 'invalid notebook content', 'notebook.ipynb')
|
||||
assert [item['name'] for item in reporter.measure(tmp_path)] == ['sample']
|
||||
|
||||
|
||||
def test_excluded_executed_lines_do_not_inflate_function_coverage(reporter, tmp_path):
|
||||
source(tmp_path, 'def sample(value):\n if value:\n return 1\n return 0\n')
|
||||
document = coverage_document()
|
||||
document['files']['src/langbot/sample.py']['excluded_lines'] = [3]
|
||||
functions = reporter.measure(tmp_path)
|
||||
reporter.attach_coverage(functions, document, tmp_path)
|
||||
assert functions[0]['coverage']['line_percent'] == 66.67
|
||||
Reference in New Issue
Block a user