!/usr/bin/env python3
"""
Test the tiered content generation pipeline structure.
Does NOT actually call models — just validates imports and structure.
"""
import sys
sys.path.insert(0, '/home/hoffmann_admin/.openclaw/workspace-socrates/hoffdesk-api')
def test_imports():
"""Test that all modules import correctly."""
print("Testing imports...")
from content.models import (
ContentType, PipelineStage, JobStatus, GenerateRequest,
StrategyOutput, StructureOutput, DraftOutput, SEOOutput,
ComplianceReport, PipelineResult
)
print(" ✓ Models imported")
from content.compliance_filter import (
ComplianceFilter, SOVEREIGN_STACK_CONTEXT, filter_content
)
print(" ✓ Compliance filter imported")
from content.pipeline import (
PipelineOrchestrator, get_orchestrator,
stage_strategy, stage_structure, stage_draft, stage_seo, stage_compliance
)
print(" ✓ Pipeline imported")
from content.router import router
print(" ✓ Router imported")
return True
def test_compliance_filter():
"""Test the compliance filter on sample text."""
from content.compliance_filter import filter_content
print("\nTesting compliance filter...")
test_text = """
# My Journey to Self-Hosting
Last Tuesday morning, I decided to embark on a transformative project.
We needed to leverage our holistic infrastructure paradigm.
My wife Aundrea suggested we streamline our calendar setup.
The solution was robust and seamless.
"""
report = filter_content(test_text)
print(f" Banned words found: {report.banned_found}")
print(f" Real names flagged: {report.names_found}")
print(f" Dates flagged: {report.dates_found}")
print(f" Is compliant: {report.is_compliant}")
print(f" Replacements made: {report.replacements_made}")
# Should find banned words (leverage, holistic, embark, transformative, robust, seamless)
# Should find real name (aundrea)
# Should find dates (Last Tuesday morning)
assert len(report.banned_found) > 0, "Should detect banned words"
assert len(report.names_found) > 0, "Should detect real names"
assert len(report.dates_found) > 0, "Should detect dates"
assert not report.is_compliant, "Should not be compliant"
print(" ✓ Compliance filter working")
return True
def test_sovereign_stack_context():
"""Verify the sovereign stack context exists."""
from content.compliance_filter import SOVEREIGN_STACK_CONTEXT
print("\nTesting sovereign stack context...")
assert "Radicale" in SOVEREIGN_STACK_CONTEXT
assert "Tailscale" in SOVEREIGN_STACK_CONTEXT
assert "Beelink" in SOVEREIGN_STACK_CONTEXT
assert "Gaming PC" in SOVEREIGN_STACK_CONTEXT
print(" ✓ Sovereign stack context contains local tools")
return True
def test_pipeline_structure():
"""Test that pipeline orchestrator can be created."""
from content.pipeline import get_orchestrator
from content.models import ContentType
print("\nTesting pipeline structure...")
orchestrator = get_orchestrator()
job_id = orchestrator.create_job("Test topic", ContentType.HOW_I_SOLVED)
print(f" Created job: {job_id}")
job = orchestrator.get_job(job_id)
assert job is not None
assert job.status.value == "queued"
print(" ✓ Pipeline orchestrator structure working")
return True
def test_models():
"""Test model instantiation."""
from content.models import (
ContentType, PipelineStage, StrategyOutput, StructureOutput,
DraftOutput, SEOOutput, ComplianceReport, PipelineResult
)
print("\nTesting model instantiation...")
# Strategy
strategy = StrategyOutput(
title="Test Title",
angle="A hook",
target_audience="Technical readers",
key_takeaways=["One", "Two"],
tone_notes="Direct"
)
assert strategy.title == "Test Title"
# Structure
structure = StructureOutput(
validated=True,
structure=["Intro", "Body", "Conclusion"],
estimated_word_count=1000
)
assert structure.validated
# Draft
draft = DraftOutput(
content="# Hello\n\nThis is content.",
word_count=50,
reading_time_minutes=1
)
assert draft.word_count == 50
# SEO
seo = SEOOutput(
excerpt="A summary",
tags=["homelab", "test"],
meta_description="A description",
keywords=["test"]
)
assert len(seo.tags) == 2
# Compliance
compliance = ComplianceReport(
replacements_made=2,
warnings=["Found: 'leverage'"],
is_compliant=False,
banned_words_found=["leverage"],
dates_flagged=["last tuesday"],
names_flagged=[]
)
assert compliance.replacements_made == 2
# Full result
result = PipelineResult(
title="Final Title",
content="# Final\n\nContent",
excerpt="Summary",
tags=["tag1"],
meta_description="Meta",
word_count=100,
reading_time_minutes=1,
compliance=compliance
)
assert result.word_count == 100
print(" ✓ All models instantiate correctly")
return True
def main():
"""Run all tests."""
print("=" * 60)
print("TIERED CONTENT PIPELINE STRUCTURE TESTS")
print("=" * 60)
tests = [
test_imports,
test_compliance_filter,
test_sovereign_stack_context,
test_pipeline_structure,
test_models,
]
passed = 0
failed = 0
for test in tests:
try:
if test():
passed += 1
except Exception as e:
print(f" ✗ {test.__name__} failed: {e}")
import traceback
traceback.print_exc()
failed += 1
print("\n" + "=" * 60)
print(f"RESULTS: {passed} passed, {failed} failed")
print("=" * 60)
return failed == 0
if name == "main":
success = main()
sys.exit(0 if success else 1)