band-aider-pub/tests/band_aider/test_global_context_json.py
2026-10-17 11:57:37 -07:00

142 lines
5 KiB
Python

# SPDX-FileCopyrightText: 2026 Art Sea LLC
#
# SPDX-License-Identifier: AGPL-3.0-only
from __future__ import annotations
from pathlib import Path
from typing import TextIO
import pytest
from band_aider.context.class_meta import ClassMeta, ContentType
from band_aider.context.global_context import GlobalContext, StoryCursor
from band_aider.context.cot_task_generator import CotTaskRecord
from band_aider.judgement.code_parser import CodeParser
from band_aider.judgement.intent_check import IntentCheckFactory
from band_aider.judgement.prompt_formatter import MarkdownPromptFormatterBuilder
from tests.support.completion_clients import StaticOpenAICompatibleClient
from tests.support.memory_path import MemoryPath
from tests.support.memory_repo_exec import MemoryRepoExec
class _ContextParser(CodeParser):
"""Provide GlobalContext's parser contract without filesystem or endpoint access."""
def parse_file(
self,
file_name: str,
prompt: str,
content_type: ContentType = ContentType.Source,
) -> ClassMeta:
"""Return one minimal source record for the supplied lexical path."""
del prompt
return ClassMeta(file_name, "", [], {}, content_type=content_type)
def parse_stream(
self,
file_name: str,
stream: TextIO,
prompt: str,
content_type: ContentType = ContentType.Source,
) -> ClassMeta:
"""Return one minimal source record without reading a host stream."""
del stream
return self.parse_file(file_name, prompt, content_type)
def parse_files(
self,
file_names: list[str],
prompt: str,
content_type: ContentType = ContentType.Source,
) -> list[ClassMeta]:
"""Return one minimal source record per supplied lexical path."""
return [self.parse_file(file_name, prompt, content_type) for file_name in file_names]
class _RecordingGlobalContext(GlobalContext):
"""Run GlobalContext state transitions without checkpoint filesystem writes."""
def __init__(self, root: MemoryPath, records: list[CotTaskRecord] | None = None) -> None:
"""Create a fully nominal global context over test-owned memory boundaries."""
client = StaticOpenAICompatibleClient('{"intent": "existing_files"}')
formatter = MarkdownPromptFormatterBuilder()
super().__init__(
goal="goal",
parser=_ContextParser(),
client=client,
repo_root=root,
src_root=root / "src",
test_root=root / "tests",
actor="test",
intent_check_factory=IntentCheckFactory(client, formatter),
prompt_formatter_builder=formatter,
original_prompt="original",
)
self.story_cursor = StoryCursor(cot_task_records=list(records or []))
self.recorded_files: list[str] = []
def persist(self, keep: int | None = None) -> Path:
"""Return a lexical checkpoint path without writing a checkpoint."""
del keep
return self.repo_root / "checkpoints" / "context.pickle"
def record_stabilization_result(self, files: list[ClassMeta], summary: str) -> None:
"""Record stabilized identities while avoiding unrelated FileManager behavior."""
del summary
self.recorded_files = [file.location for file in files]
@pytest.fixture
def memory_root() -> MemoryPath:
"""Return a fresh memory-only project root for context state tests."""
return MemoryPath.from_repo(MemoryRepoExec())
def test_global_context_records_are_stored_as_generated(memory_root: MemoryPath) -> None:
"""Verify generated COT records are stored without a parallel string batch."""
records = [
CotTaskRecord("new two", "two.py", ClassMeta("two.py", "", [], {})),
CotTaskRecord("new one", "one.py", ClassMeta("one.py", "", [], {})),
]
context = _RecordingGlobalContext(memory_root, records)
assert context.story_cursor.cot_task_records == records
def test_global_context_records_bootstrap_involved_locations_and_story_roots(
memory_root: MemoryPath,
) -> None:
"""Verify bootstrap structural identities and surrogate story roots are durable strings."""
root = ClassMeta(
"src/root.py",
"root",
[],
{},
)
sibling = ClassMeta(
"src/sibling.py",
"sibling",
[],
{},
)
context = _RecordingGlobalContext(
memory_root,
[CotTaskRecord("task", "src/root.py", root)],
)
context.record_bootstrap_involved_locations(["src/root.py", "", "src/sibling.py"])
context.mark_root_stabilized(root, "summary", story_roots=[root, sibling])
assert context.story_cursor.bootstrap_involved_locations == {"src/root.py", "src/sibling.py"}
assert context.story_cursor.task_cursor.stabilized_locations_by_cot["0"] == {"src/root.py", "src/sibling.py"}
assert context.story_cursor.claimed_locations_this_recursive_pass == {"src/root.py", "src/sibling.py"}
assert context.recorded_files == ["src/root.py", "src/sibling.py"]