ankn
a5f8fb49ab
Bahn: aisupport, Analyse-O2C-C2S, awesome-bahn-mcp-servers, beam-mcp,
Confluence_Bot, db-planet-mcp-server, O2C-Harness, project-audit,
Projekt-KIQ-HP, teamlandkarte-mcp
Dhive: Jury-Voting
Privat: CV, NoteGraph (NOTE: NoteGraph needs complete redo after consolidation)
Shared: AI-Orchestrator, OrgMyLife, power_skills_and_more
Shared/references: symphony (read-only)
Bahn repos remain available as independent remotes - this monorepo
pulls them in via subtree, the originals are untouched.
94 lines
2.4 KiB
Python
94 lines
2.4 KiB
Python
"""Tests for workflow loader."""
|
|
|
|
import pytest
|
|
|
|
from ai_orchestrator.workflow import (
|
|
MissingWorkflowFileError,
|
|
WorkflowFrontMatterNotAMapError,
|
|
WorkflowParseError,
|
|
load_workflow,
|
|
parse_workflow,
|
|
)
|
|
|
|
|
|
class TestParseWorkflow:
|
|
"""Tests for parse_workflow function."""
|
|
|
|
def test_full_workflow_with_front_matter(self):
|
|
content = """---
|
|
tracker:
|
|
kind: linear
|
|
project_slug: my-project
|
|
polling:
|
|
interval_ms: 15000
|
|
---
|
|
You are working on {{ issue.identifier }}: {{ issue.title }}
|
|
"""
|
|
result = parse_workflow(content)
|
|
assert result.config["tracker"]["kind"] == "linear"
|
|
assert result.config["tracker"]["project_slug"] == "my-project"
|
|
assert result.config["polling"]["interval_ms"] == 15000
|
|
assert "{{ issue.identifier }}" in result.prompt_template
|
|
|
|
def test_no_front_matter(self):
|
|
content = "Just a prompt template with {{ issue.title }}"
|
|
result = parse_workflow(content)
|
|
assert result.config == {}
|
|
assert result.prompt_template == content
|
|
|
|
def test_empty_front_matter(self):
|
|
content = """---
|
|
---
|
|
Hello {{ issue.identifier }}"""
|
|
result = parse_workflow(content)
|
|
assert result.config == {}
|
|
assert "Hello" in result.prompt_template
|
|
|
|
def test_front_matter_not_a_map(self):
|
|
content = """---
|
|
- item1
|
|
- item2
|
|
---
|
|
prompt body"""
|
|
with pytest.raises(WorkflowFrontMatterNotAMapError):
|
|
parse_workflow(content)
|
|
|
|
def test_invalid_yaml(self):
|
|
content = """---
|
|
invalid: yaml: content: [
|
|
---
|
|
prompt"""
|
|
with pytest.raises(WorkflowParseError):
|
|
parse_workflow(content)
|
|
|
|
def test_prompt_is_trimmed(self):
|
|
content = """---
|
|
key: value
|
|
---
|
|
|
|
Some prompt with whitespace
|
|
|
|
"""
|
|
result = parse_workflow(content)
|
|
assert result.prompt_template == "Some prompt with whitespace"
|
|
|
|
|
|
class TestLoadWorkflow:
|
|
"""Tests for load_workflow function."""
|
|
|
|
def test_missing_file(self, tmp_path):
|
|
with pytest.raises(MissingWorkflowFileError):
|
|
load_workflow(tmp_path / "nonexistent.md")
|
|
|
|
def test_valid_file(self, tmp_path):
|
|
workflow_file = tmp_path / "WORKFLOW.md"
|
|
workflow_file.write_text("""---
|
|
tracker:
|
|
kind: linear
|
|
---
|
|
Hello {{ issue.title }}""")
|
|
|
|
result = load_workflow(workflow_file)
|
|
assert result.config["tracker"]["kind"] == "linear"
|
|
assert "Hello" in result.prompt_template
|