Orchestrata Ultimate › Plans › test_coverage_plan

Test Coverage Plan: Uncovered Methods

Source: plans/test_coverage_plan.md at 8a2cb42ad · last changed 2026-07-27 · first committed 2026-05-25 · 470 lines.

Prose pending.

Plan text

Test Coverage Plan: Uncovered Methods

Goal: Cover 104 concrete untested methods to push module-level coverage to ≥90%. Approach: 6 parallel agent workstreams, each with isolated scope. Standards: All tests must pass pytest -n auto with zero isolation violations.

Current State (Accurate)

How the count was derived: The AST test-to-method matcher was completely rewritten with deterministic import-aware logic:

• Class-qualified exact match (TestClass → Class) has highest priority
• Global exact match is restricted to imported classes only
• Word-level matching handles conceptual test names (test_stats_empty → get_session_stats)
• Compound test names are split and matched (test_save_and_load → save_thread + load_thread)
• Unresolved test class names (TestEnablePlugin) fall back to import-filtered word matching

Untested methods by subsystem:

Testing Standards (Mandatory)

Every new test MUST comply with these rules. Violations = reject.

Absolute Prohibitions

Required Patterns

Required Verification (per workstream)

# 1. Run tests in parallel
uv run pytest tests/<module>/ -n auto --timeout=120

# 2. Run isolation audit
python scripts/test_isolation_audit.py tests/<module>/

# 3. Run test review
python scripts/test-review.py tests/<module>/

# 4. Check coverage
uv run pytest tests/<module>/ --cov=src/orchestrata_ultimate/<module> --cov-report=term-missing --timeout=120

Workstream A: CXP Networking + Proto (Agent 1)

Scope: gRPC/WebSocket server lifecycle, proto servicers, dispatcher count, version compat. Target: 20 methods | Files to create/extend: tests/cxp/test_grpc_extended.py, tests/cxp/test_websocket_extended.py, tests/cxp/test_dispatcher_extended.py

Methods to Cover

cxp.dispatcher.CXPDispatcher.get_dispatch_count
cxp.version_compat.VersionChecker.check_compatibility
cxp.grpc_client.GRPCClient.acknowledge
cxp.grpc_client.GRPCClient.disconnect
cxp.grpc_client.GRPCClient.subscribe
cxp.grpc_server.CXPServer.serve_forever
cxp.grpc_server.CXPServer.start
cxp.grpc_server.CXPServicer.acknowledge
cxp.grpc_server.CXPServicer.receive_packet
cxp.grpc_server.CXPServicer.send_packet
cxp.grpc_server.CXPServicer.send_packet_stream
cxp.grpc_server.CXPServicer.subscribe
cxp.proto.cxp_pb2_grpc.CXPReceiver.acknowledge
cxp.proto.cxp_pb2_grpc.CXPReceiver.receive_packet
cxp.proto.cxp_pb2_grpc.CXPReceiverServicer.acknowledge
cxp.proto.cxp_pb2_grpc.CXPReceiverServicer.receive_packet
cxp.proto.cxp_pb2_grpc.CXPReceiverServicer.subscribe
cxp.proto.cxp_pb2_grpc.CXPReceiver.subscribe
cxp.proto.cxp_pb2_grpc.CXPServerServicer.send_packet
cxp.proto.cxp_pb2_grpc.CXPServerServicer.send_packet_stream
cxp.ws_server.CXPWebSocketHandler.handle
cxp.ws_server.WSServer.serve_forever
cxp.ws_server.WSServer.start
cxp.session_adapter.SessionAdapter.is_bound

Testing Strategy

Dynamic port allocation pattern (REQUIRED for all network tests):

import socket

async def test_grpc_server_start_and_serve():
    sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    sock.bind(("127.0.0.1", 0))
    port = sock.getsockname()[1]
    sock.close()

    server = CXPServer(port=port)
    await server.start()
    try:
        assert server._started is True
        # serve_forever blocks, so test it in a task with timeout
        task = asyncio.create_task(asyncio.wait_for(server.serve_forever(), timeout=0.5))
        await asyncio.sleep(0.1)
        task.cancel()
        try:
            await task
        except (asyncio.CancelledError, asyncio.TimeoutError):
            pass
    finally:
        await server.stop()

Key rules:

• Always use socket.bind(("127.0.0.1", 0)) for dynamic ports
• Always wrap server lifecycle in try/finally
• serve_forever tests must use asyncio.wait_for or cancel after brief delay
• Proto servicers are generated code — test them through gRPC server integration, not in isolation

Workstream B: Kernel Performance Manager (Agent 2)

Scope: PerformanceManager and sub-components. Target: 20 methods | Files to create/extend: tests/kernel/test_performance_extended.py

Methods to Cover

kernel.performance_manager.AutoScaler.evaluate
kernel.performance_manager.BackpressureController.should_accept
kernel.performance_manager.BatchProcessor.add
kernel.performance_manager.ChaosController.get_stats
kernel.performance_manager.ConnectionPool.get_stats
kernel.performance_manager.PerformanceManager.add_chaos_fault
kernel.performance_manager.PerformanceManager.disable_chaos_testing
kernel.performance_manager.PerformanceManager.enable_chaos_testing
kernel.performance_manager.PerformanceManager.get_cache_stats
kernel.performance_manager.PerformanceManager.get_node_for_session
kernel.performance_manager.PerformanceManager.get_node_for_volk
kernel.performance_manager.PerformanceManager.get_policy_p95
kernel.performance_manager.PerformanceManager.get_policy_p99
kernel.performance_manager.PerformanceManager.get_routing_p95
kernel.performance_manager.PerformanceManager.get_routing_p99
kernel.performance_manager.PerformanceManager.get_scaling_decisions
kernel.performance_manager.PerformanceManager.mark_system_down
kernel.performance_manager.PerformanceManager.mark_system_up
kernel.performance_manager.PerformanceManager.should_accept_request
kernel.performance_manager.PerformanceManager.update_load

Testing Strategy

All in-memory metrics. No external services needed.

def test_performance_manager_chaos_testing():
    pm = PerformanceManager()
    pm.enable_chaos_testing()
    assert pm.is_chaos_enabled() is True
    pm.add_chaos_fault("network_delay")
    stats = pm.get_chaos_stats()
    assert "network_delay" in stats.faults
    pm.disable_chaos_testing()
    assert pm.is_chaos_enabled() is False

def test_performance_manager_mark_system_up_down():
    pm = PerformanceManager()
    pm.mark_system_down("node-1")
    assert pm.get_uptime_percent() < 100.0
    pm.mark_system_up("node-1")
    assert pm.get_uptime_percent() == 100.0

Workstream C: Plugin System (Agent 3)

Scope: PluginSDK, PluginContext, PluginManager, PluginLoader. Target: 14 methods | Files to create/extend: tests/plugin/test_plugin_extended.py

Methods to Cover

plugin.context.PluginContext.emit_event
plugin.context.PluginContext.register_channel
plugin.context.PluginContext.register_tool
plugin.context.PluginContext.register_volk
plugin.context.PluginContext.subscribe_event
plugin.loader.PluginLoader.topological_sort
plugin.manager.PluginManager.reload_plugin
plugin.manager.PluginManager.unload_plugin
plugin.sdk.PluginSDK.disable
plugin.sdk.PluginSDK.enable
plugin.sdk.PluginSDK.install
plugin.sdk.PluginSDK.load
plugin.sdk.PluginSDK.reload
plugin.sdk.PluginSDK.uninstall
plugin.sdk.PluginSDK.unload

Testing Strategy

Use isolation fixture for temp plugin directories.

def test_plugin_sdk_enable_disable(isolation):
    """Test PluginSDK.enable and PluginSDK.disable."""
    plugin_dir = isolation.create_test_directory("plugins")
    sdk = PluginSDK(plugin_dirs=[str(plugin_dir)])
    sdk.load("test_plugin")
    assert sdk.enable("test_plugin") is True
    assert sdk.disable("test_plugin") is True
    assert sdk.unload("test_plugin") is True

Workstream D: Knowledge Graph + Neo4j Sync (Agent 4)

Scope: Neo4jSyncer lifecycle, Neo4jClient operations. Target: 7 methods | Files to create/extend: tests/knowledge_graph/test_sync_extended.py

Methods to Cover

knowledge_graph.neo4j_client.Neo4jClient.close
knowledge_graph.sync.Neo4jSyncer.close
knowledge_graph.sync.Neo4jSyncer.connect
knowledge_graph.sync.Neo4jSyncer.full_sync
knowledge_graph.sync.Neo4jSyncer.get_stats
knowledge_graph.sync.Neo4jSyncer.map_classes
knowledge_graph.sync.Neo4jSyncer.verify

Testing Strategy

Neo4j tests must use test_id for all node labels/properties.

async def test_neo4j_syncer_connect_close(neo4j_client, test_id):
    syncer = Neo4jSyncer(neo4j_client)
    await syncer.connect()
    assert syncer.is_connected() is True
    stats = await syncer.get_stats()
    assert "nodes" in stats
    await syncer.close()
    assert syncer.is_connected() is False

Workstream E: Orchestrator Prolog Worker + Redis State (Agent 5)

Scope: PrologWorker lifecycle, PrologProxy, RedisStateManager. Target: 8 methods | Files to create/extend: tests/orchestrator/test_prolog_worker_extended.py

Methods to Cover

orchestrator.prolog_worker.PrologProxy.assertz
orchestrator.prolog_worker.PrologProxy.consult
orchestrator.prolog_worker.PrologProxy.retract
orchestrator.prolog_worker.PrologWorker.shutdown
orchestrator.prolog_worker.PrologWorker.start
orchestrator.prolog_worker.PrologWorker.submit
orchestrator.redis_state.RedisStateManager.aclose
orchestrator.redis_state.RedisStateManager.acquire_lock

Testing Strategy

async def test_prolog_worker_lifecycle():
    worker = PrologWorker()
    await worker.start()
    try:
        result = await worker.submit("assertz(foo(bar))")
        assert result.success is True
    finally:
        await worker.shutdown()

Workstream F: Drivers + Providers + Misc (Agent 6)

Scope: Code generation drivers, LLM provider adapters, misc uncovered methods. Target: 18 methods | Files to create/extend: tests/drivers/test_drivers_extended.py, tests/providers/test_adapters_extended.py

Methods to Cover

Drivers (5):

drivers.aider_driver.AiderDriver.execute
drivers.claude_api_driver.ClaudeAPIDriver.execute
drivers.claude_code_driver.ClaudeCodeDriver.execute
drivers.deepseek_coder_driver.DeepSeekCoderDriver.execute
drivers.qwen_driver.QwenDriver.execute

Providers (6):

providers.adapters.claude_cli_adapter.ClaudeCLIAdapter.complete
providers.adapters.deepseek_adapter.DeepSeekAdapter.complete
providers.adapters.ollama_adapter.OllamaAdapter.complete
providers.adapters.openai_adapter.OpenAIAdapter.complete
providers.adapters.openrouter_adapter.OpenRouterAdapter.complete
providers.registry.ProviderRegistry.get_config

Misc (7):

agents.volks.pm.goal_report.GoalReportGenerator.generate
agents.volks.ux.ux_js_scaffold.JSScaffoldBuilder.build_architecture_prompt
core.prolog_adapter.PrologEngine.consult
core.threading.thread_pool.CustomFuture.get
gateway.server.WebSocketGateway.start
governance.escalation_handler.LogNotificationChannel.send
governance.event_emitting_policy_engine.EventEmittingPolicyEngine.evaluate
kernel.dispatcher.MessageDispatcher.dispatch_websocket
kernel.pipeline_escalation.PipelineEscalationMixin.approve
kernel.pipeline.PipelineRunner.respond
kernel.pipeline.PipelineRunner.submit_intent
kernel.pipeline_runner.PipelineRunner.respond
kernel.pipeline_runner.PipelineRunner.submit_intent
kernel.resource_monitor.ResourceMonitor.register_alert_callback
orchestrator.fact_store.RealFactStore.close
orchestrator.impact.dependency_analyzer.DependencyAnalyzer.find_affected_scope
orchestrator.intent_lifecycle.IntentLifecycleManager.close
orchestrator.intent_template.TemplateRegistry.get_composite
orchestrator.nlp.SessionContext.add_recent_command

Testing Strategy

Drivers: Most drivers call external APIs. Test configuration/initialization paths instead of actual execution, or use environment-based skipping (but no @pytest.mark.skip!). Pass None for config and verify error handling.

def test_driver_execute_without_config():
    driver = AiderDriver()
    with pytest.raises(DriverConfigError):
        asyncio.run(driver.execute("test"))

Provider adapters: Same pattern — test config validation and error paths.

JSScaffoldBuilder: Filesystem test with isolation fixture.

def test_js_scaffold_builder_build_architecture_prompt(isolation):
    project_dir = isolation.create_test_directory("js_project")
    builder = JSScaffoldBuilder(project_dir=str(project_dir), project_name="test")
    prompt = builder.build_architecture_prompt()
    assert "test" in prompt
    assert "JavaScript" in prompt

Agent Dispatch Instructions

Pre-Dispatch Checklist

1. Start session:

   from orchestrata_ultimate.session.session import SessionController
   ctrl = SessionController(".", ai_model="Claude Opus 4.5")
   session = ctrl.begin(
       purpose="Test coverage workstream X",
       objectives=["Cover N untested methods", "Achieve 90% module coverage"],
   )
   

2. Create workstream branch:

   git checkout -b coverage/workstream-X
   

Per-Agent Deliverables

Each agent must produce:

1. Test file(s) in tests/<module>/test_*_extended.py
2. Coverage report showing ≥90% for targeted module
3. Isolation audit passing with 0 violations
4. Test review passing with 0 ERRORs

Post-Agent Integration

1. Run full suite:

   uv run pytest tests/ -n auto --timeout=120
   

2. If passing, merge workstream:

   git checkout main
   git merge coverage/workstream-X
   

3. Update AST and re-run comparison:

   python scripts/serialize_codebase_ast.py serialize --source src/orchestrata_ultimate --name orchestrata-ultimate --output plans/ast-ultimate.json
   python scripts/ast_explorer.py discover --untested --test-ast plans/ast-tests.json --limit 1000
   

Coverage Targets by Module

Success Criteria

• [ ] All 104 untested methods have tests OR documented justification for why they can't/shouldn't be tested
• [ ] pytest tests/ -n auto --timeout=120 passes with 0 failures
• [ ] scripts/test_isolation_audit.py tests/ reports 0 violations
• [ ] scripts/test-review.py tests/ reports 0 ERRORs
• [ ] Per-module coverage ≥90% for targeted modules
• [ ] AST re-serialized and gap analysis updated

Related

• Plans
• Orchestrata Ultimate — project index
• Workflow
• Tasks