"""End-to-End Integration-Tests für den gesamten Komponentenstack. Testet die folgenden Workflows über mehrere Komponenten hinweg: 1. Projekt erstellen → Env verschlüsseln → Env laden → Secret-Zugriff prüfen 2. Artefakt erstellen → Indexieren → Suchen → Freigeben → Widerrufen 3. Repo einbinden → Sync → Read-Only-Schutz 4. Migration → Validierung → Rollback 5. Secret verschlüsseln → Maschinenkontext wechseln → Entschlüsselung fehlschlägt 6. Team_Repo vorbereiten → Isolation prüfen → Sync → Konflikt-Auflösung 7. Shared-Mirror in Team_Repo → Read-Only-Prüfung Requirements: 1.1, 2.1, 3.2, 4.1, 5.2, 6.1, 9.1, 9.4, 10.3, 10.5, 10.12 """ from __future__ import annotations import subprocess from pathlib import Path from unittest.mock import MagicMock, patch import pytest import yaml from monorepo.bridge import ContextBridge from monorepo.encryption import DecryptionResult, EncryptionResult, SecretEncryptionManager from monorepo.federation import FederationManager from monorepo.knowledge.etl import ETLPipeline from monorepo.knowledge.index import IndexEntry, YAMLIndex from monorepo.knowledge.sources.markdown import MarkdownSource from monorepo.knowledge.store import KnowledgeStore from monorepo.migration import MigrationEngine from monorepo.models import ( ConflictInfo, MachineContext, MigrationPlan, RepoEntry, ScopeConfig, SyncResult, ) from monorepo.repos import RepoManager from monorepo.security import ContextGuard from monorepo.structure import StructureManager # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture() def monorepo_root(tmp_path: Path) -> Path: """Creates a full monorepo structure for E2E integration tests.""" root = tmp_path / "monorepo" root.mkdir() # Context directories for ctx in ("privat", "dhive", "bahn", "shared"): (root / ctx).mkdir() # Shared subdirectories for sub in ("tools", "powers", "knowledge-store", "config", "mcp-servers"): (root / "shared" / sub).mkdir(parents=True, exist_ok=True) # .env files per context (root / "privat" / ".env").write_text( "PRIVAT_SECRET=privat_val_123\nPRIVAT_API=http://localhost:3000\n", encoding="utf-8", ) (root / "dhive" / ".env").write_text( "DHIVE_SECRET=dhive_val_456\nDHIVE_TOKEN=tok_dhive\n", encoding="utf-8", ) (root / "bahn" / ".env").write_text( "BAHN_SECRET=bahn_val_789\nBAHN_ENDPOINT=https://bahn.api\n", encoding="utf-8", ) (root / "shared" / ".env").write_text( "SHARED_KEY=shared_val\n", encoding="utf-8" ) # access-config.yaml config_dir = root / "shared" / "config" access_config = { "contexts": { "privat": { "env_file": "privat/.env", "allowed_shared": ["shared/tools/", "shared/config/"], }, "dhive": { "env_file": "dhive/.env", "allowed_shared": ["shared/tools/", "shared/config/"], }, "bahn": { "env_file": "bahn/.env", "allowed_shared": [ "shared/tools/", "shared/config/", "shared/knowledge-store/", ], }, "shared": { "env_file": "shared/.env", "allowed_shared": ["*"], }, } } (config_dir / "access-config.yaml").write_text( yaml.dump(access_config), encoding="utf-8" ) # machine-context.yaml (full access) machine_config = { "machine": { "name": "test-hauptrechner", "description": "Test machine with full access", "authorized_contexts": ["privat", "dhive", "bahn"], "key_source": "keyring", } } (config_dir / "machine-context.yaml").write_text( yaml.dump(machine_config), encoding="utf-8" ) # team-repos.yaml team_repos_config = { "version": "1.0", "federation": { "topology": "hub-and-spoke", "hub_owner": "andre", "conflict_strategy": "team-wins", }, "team_repos": [ { "context": "privat", "url": "https://github.com/test/privat-team.git", "branch": "main", "sync_direction": "bidirectional", "sync_frequency": "manual", "shared_mirror": { "enabled": True, "paths": ["shared/tools/common-scripts/"], "mode": "read-only", }, }, { "context": "dhive", "url": "https://gitlab.dhive.io/team/dhive-mono.git", "branch": "main", "sync_direction": "bidirectional", "sync_frequency": "on-push", "shared_mirror": { "enabled": True, "paths": ["shared/tools/"], "mode": "read-only", }, }, { "context": "bahn", "url": "https://gitlab.2700.2db.it/team/bahn-workspace.git", "branch": "main", "sync_direction": "bidirectional", "sync_frequency": "daily", "shared_mirror": { "enabled": True, "paths": ["shared/tools/"], "mode": "read-only", }, }, ], } (config_dir / "team-repos.yaml").write_text( yaml.dump(team_repos_config), encoding="utf-8" ) # repos.yaml (empty registry) repos_config = {"repos": []} (config_dir / "repos.yaml").write_text( yaml.dump(repos_config), encoding="utf-8" ) # Knowledge store index ks_path = root / "shared" / "knowledge-store" (ks_path / "_index.yaml").write_text( yaml.dump({"version": "1.0", "last_updated": "", "artifacts": []}), encoding="utf-8", ) return root @pytest.fixture() def full_machine_context() -> MachineContext: """MachineContext authorized for all contexts.""" return MachineContext( name="test-hauptrechner", description="Full access", authorized_contexts=["privat", "dhive", "bahn"], key_source="keyring", ) @pytest.fixture() def limited_machine_context() -> MachineContext: """MachineContext authorized only for dhive (simulates dhive-laptop).""" return MachineContext( name="dhive-laptop", description="Only dhive access", authorized_contexts=["dhive"], key_source="keyring", ) @pytest.fixture() def scope_config() -> ScopeConfig: """ScopeConfig for knowledge store.""" return ScopeConfig( scopes={ "privat": {"scope": "privat", "paths": ["privat/"]}, "dhive": {"scope": "dhive", "paths": ["dhive/"]}, "bahn": {"scope": "bahn", "paths": ["bahn/"]}, "shared": {"scope": "shared", "paths": ["shared/"]}, } ) # --------------------------------------------------------------------------- # Test 1: Projekt erstellen → Env verschlüsseln → Env laden → Secret-Zugriff # Requirements: 1.1, 2.1, 9.1, 9.4 # --------------------------------------------------------------------------- class TestProjectEncryptionSecretAccess: """E2E: Project creation → Encrypt env → Load env → Cross-context denial.""" def test_create_project_encrypt_env_load_and_deny_cross_context( self, monorepo_root: Path, full_machine_context: MachineContext ) -> None: """Full flow: create project, encrypt its env, load it, deny cross-access.""" # Step 1: Create a project in privat context structure_mgr = StructureManager(monorepo_root) project_path = structure_mgr.create_project("privat", "my-new-app") assert project_path.exists() assert project_path == monorepo_root / "privat" / "my-new-app" # Step 2: Create a project-level .env for the context env_content = "MY_APP_SECRET=super-secret-123\nDB_URL=postgres://localhost/mydb\n" env_file = monorepo_root / "privat" / ".env" env_file.write_text(env_content, encoding="utf-8") # Step 3: Create encryption manager and encrypt the file encryption_mgr = SecretEncryptionManager(monorepo_root, full_machine_context) result = encryption_mgr.encrypt_file(env_file, "privat") # Encryption may or may not succeed (git-crypt might not be installed) # but the flow should be exercised regardless # Step 4: Load env via ContextGuard (with encryption integration) guard = ContextGuard( root_path=monorepo_root, encryption_manager=encryption_mgr, ) # Loading env for own context should succeed env_vars = guard.load_env("privat") assert "MY_APP_SECRET" in env_vars assert env_vars["MY_APP_SECRET"] == "super-secret-123" # Step 5: Cross-context access should be denied access_allowed = guard.check_access("dhive", monorepo_root / "privat" / ".env") assert access_allowed is False # dhive should not be able to access privat's .env content via its own context access_allowed_bahn = guard.check_access("bahn", monorepo_root / "privat" / ".env") assert access_allowed_bahn is False def test_shared_tools_accessible_from_any_context( self, monorepo_root: Path, full_machine_context: MachineContext ) -> None: """Shared tools are accessible from all contexts.""" encryption_mgr = SecretEncryptionManager(monorepo_root, full_machine_context) guard = ContextGuard(root_path=monorepo_root, encryption_manager=encryption_mgr) # Create a shared tool file tool_file = monorepo_root / "shared" / "tools" / "helper.py" tool_file.write_text("# shared helper", encoding="utf-8") # All contexts can access shared tools for ctx in ("privat", "dhive", "bahn"): assert guard.check_access(ctx, tool_file) is True # --------------------------------------------------------------------------- # Test 2: Artefakt erstellen → Indexieren → Suchen → Freigeben → Widerrufen # Requirements: 3.2, 5.2 # --------------------------------------------------------------------------- class TestArtifactIngestSearchShareRevoke: """E2E: Create artifact → Ingest via ETL → Search → Share → Revoke.""" def test_full_knowledge_lifecycle( self, monorepo_root: Path, scope_config: ScopeConfig ) -> None: """Ingest markdown, search it, share it, then revoke the share.""" ks_path = monorepo_root / "shared" / "knowledge-store" # Step 1: Create a markdown artifact in bahn context source source_dir = monorepo_root / "bahn" / "docs" source_dir.mkdir(parents=True) artifact_md = source_dir / "api-patterns.md" artifact_md.write_text( "---\n" "type: decision\n" "title: REST API Patterns für InfraGO\n" "tags: [api, rest, patterns]\n" "---\n\n" "# REST API Patterns\n\n" "Wir verwenden RESTful API-Konventionen für alle neuen Services.\n", encoding="utf-8", ) # Step 2: Ingest via KnowledgeStore store = KnowledgeStore(ks_path, scope_config) source = MarkdownSource(directory=source_dir) ingest_result = store.ingest(source, "bahn") assert ingest_result.processed >= 1 assert ingest_result.updated >= 1 # Step 3: Search the index results = store.search("REST API", allowed_scopes=["bahn", "shared"]) assert len(results) >= 1 found_titles = [r.entry.title for r in results] assert any("REST API" in t or "api" in t.lower() for t in found_titles) # Step 4: Verify scope filtering - privat should not see bahn artifacts privat_results = store.search("REST API", allowed_scopes=["privat"]) bahn_artifact_in_privat = [ r for r in privat_results if r.entry.scope == "bahn" ] assert len(bahn_artifact_in_privat) == 0 # Step 5: Share the artifact via ContextBridge bridge = ContextBridge( knowledge_base_path=ks_path, index=store.index, ) # Get the artifact ID from the index index_entries = store.get_index(scope="bahn") assert len(index_entries) >= 1 artifact_id = index_entries[0].id # Share with user confirmation share_result = bridge.share_artifact(artifact_id, user_confirmed=True) assert share_result.success is True assert bridge.is_shared(artifact_id) is True # Other contexts can now access the shared content content = bridge.get_shared_content(artifact_id, "privat") assert content is not None assert "REST" in content or "api" in content.lower() # Step 6: Revoke the share bridge.revoke_share(artifact_id) assert bridge.is_shared(artifact_id) is False # After revocation, other contexts cannot access content_after = bridge.get_shared_content(artifact_id, "privat") assert content_after is None def test_share_blocked_for_sensitive_content( self, monorepo_root: Path, scope_config: ScopeConfig ) -> None: """Sensitive artifacts cannot be shared across contexts.""" ks_path = monorepo_root / "shared" / "knowledge-store" # Create artifact with sensitive content source_dir = monorepo_root / "dhive" / "secrets-docs" source_dir.mkdir(parents=True) sensitive_md = source_dir / "credentials.md" sensitive_md.write_text( "---\n" "type: note\n" "title: API Credentials\n" "tags: [credentials]\n" "---\n\n" "# Credentials\n\n" "api_key: sk-12345-abcdef\n", encoding="utf-8", ) # Ingest the sensitive artifact store = KnowledgeStore(ks_path, scope_config) source = MarkdownSource(directory=source_dir) store.ingest(source, "dhive") # Try to share - should be blocked bridge = ContextBridge(knowledge_base_path=ks_path, index=store.index) index_entries = store.get_index(scope="dhive") assert len(index_entries) >= 1 artifact_id = index_entries[0].id share_result = bridge.share_artifact(artifact_id, user_confirmed=True) assert share_result.success is False assert any("sensib" in e.lower() or "sensitive" in e.lower() for e in share_result.errors) # --------------------------------------------------------------------------- # Test 3: Repo einbinden → Sync → Read-Only-Schutz # Requirements: 4.1 # --------------------------------------------------------------------------- class TestRepoAddSyncReadonly: """E2E: Add repo → Sync → Protect read-only.""" def test_add_repo_sync_and_readonly_protection( self, monorepo_root: Path ) -> None: """Add an external repo, sync it, and verify read-only protection.""" config_path = monorepo_root / "shared" / "config" / "repos.yaml" # Initialize a git repo in monorepo_root for hook installation git_dir = monorepo_root / ".git" git_dir.mkdir() hooks_dir = git_dir / "hooks" hooks_dir.mkdir() repo_mgr = RepoManager(config_path=config_path, monorepo_root=monorepo_root) # Step 1: Add a read-only repo entry (mock git operations) entry = RepoEntry( name="symphony-spec", url="https://github.com/openai/symphony", mode="read-only", target="shared/references/symphony", pinned="v1.0.0", mechanism="subtree", ) # Mock git subprocess to avoid network calls with patch("monorepo.repos.subprocess.run") as mock_run: mock_run.return_value = MagicMock( returncode=0, stdout="", stderr="" ) repo_mgr.add_repo(entry) # Verify repo was registered assert repo_mgr._find_repo("symphony-spec") is not None # Step 2: Sync the repo (mocked) # Create the target directory to simulate existing subtree target_path = monorepo_root / "shared" / "references" / "symphony" target_path.mkdir(parents=True) (target_path / "README.md").write_text("# Symphony", encoding="utf-8") with patch("monorepo.repos.subprocess.run") as mock_run: mock_run.return_value = MagicMock( returncode=0, stdout="1 commit pulled", stderr="" ) sync_result = repo_mgr.sync("symphony-spec") # Sync should complete (success or graceful failure) assert sync_result is not None # Step 3: Protect read-only repo_mgr.protect_readonly(target_path) # Verify pre-commit hook was created/updated hook_path = hooks_dir / "pre-commit" assert hook_path.exists() hook_content = hook_path.read_text(encoding="utf-8") # The hook should reference the protected path assert "symphony" in hook_content or "references" in hook_content # --------------------------------------------------------------------------- # Test 4: Migration → Validierung → Rollback # Requirements: 6.1 # --------------------------------------------------------------------------- class TestMigrationValidateRollback: """E2E: Migrate repo → Validate → Rollback.""" def test_migrate_validate_and_rollback(self, monorepo_root: Path) -> None: """Migrate a repo, validate it, then rollback.""" # Initialize git in monorepo_root git_dir = monorepo_root / ".git" git_dir.mkdir(exist_ok=True) engine = MigrationEngine(monorepo_root=monorepo_root) plan = MigrationPlan( source_repo="https://github.com/test/old-project.git", target_context="dhive", target_name="old-project", mode="subtree", dependencies=[], order=1, ) # Step 1: Execute migration (mock git calls) with patch.object(engine, "_run_git") as mock_git: mock_git.return_value = MagicMock( returncode=0, stdout="", stderr="" ) # Also mock specific detection methods with patch.object(engine, "_get_remote_branches", return_value=["main"]): with patch.object(engine, "_get_local_branches", return_value=["main"]): result = engine.migrate(plan) # Migration should succeed or detect conflicts gracefully assert result is not None # Step 2: Validate the migration if result.success: # Create target path to simulate successful migration target = monorepo_root / "dhive" / "old-project" target.mkdir(parents=True, exist_ok=True) (target / "README.md").write_text("# Old Project", encoding="utf-8") # Mock git validation commands with patch.object(engine, "_run_git") as mock_git: mock_git.return_value = MagicMock( returncode=0, stdout="5", stderr="" ) validation = engine.validate("old-project") assert validation is not None # Step 3: Rollback with patch.object(engine, "_run_git") as mock_git: mock_git.return_value = MagicMock( returncode=0, stdout="", stderr="" ) engine.rollback("old-project") # After rollback, the project should no longer be in registry assert engine.is_migrated("old-project") is False # --------------------------------------------------------------------------- # Test 5: Secret verschlüsseln → Maschinenkontext wechseln → Entschlüsselung # fehlschlägt # Requirements: 9.1, 9.4 # --------------------------------------------------------------------------- class TestSecretEncryptionMachineContextSwitch: """E2E: Encrypt secret → Switch machine context → Decryption fails.""" def test_encrypt_then_switch_context_decrypt_fails( self, monorepo_root: Path, full_machine_context: MachineContext, limited_machine_context: MachineContext, ) -> None: """Encrypt with full access, then try to decrypt with limited access.""" # Step 1: Encrypt a secret file with full machine context full_mgr = SecretEncryptionManager(monorepo_root, full_machine_context) secret_file = monorepo_root / "privat" / ".env" encrypt_result = full_mgr.encrypt_file(secret_file, "privat") # The encryption manager should at least attempt to encrypt # (git-crypt may not be available but the flow is tested) # Step 2: Verify full context can still decrypt assert full_mgr.is_authorized("privat") is True assert full_mgr.is_authorized("dhive") is True assert full_mgr.is_authorized("bahn") is True # Step 3: Switch to limited machine context (only dhive) limited_mgr = SecretEncryptionManager(monorepo_root, limited_machine_context) # Step 4: Limited context should NOT be authorized for privat assert limited_mgr.is_authorized("privat") is False assert limited_mgr.is_authorized("bahn") is False assert limited_mgr.is_authorized("dhive") is True # Step 5: Decryption attempt for privat should fail/be denied decrypt_result = limited_mgr.decrypt_file(secret_file) # Either returns failure or raises PermissionError if decrypt_result is not None: assert decrypt_result.success is False def test_context_guard_denies_load_env_on_unauthorized_machine( self, monorepo_root: Path, limited_machine_context: MachineContext, ) -> None: """ContextGuard denies env loading for unauthorized context on limited machine.""" limited_mgr = SecretEncryptionManager(monorepo_root, limited_machine_context) guard = ContextGuard(root_path=monorepo_root, encryption_manager=limited_mgr) # dhive env should load fine (authorized) env = guard.load_env("dhive") assert "DHIVE_SECRET" in env # privat env should fail (not authorized on this machine) with pytest.raises(PermissionError): guard.load_env("privat") # --------------------------------------------------------------------------- # Test 6: Team_Repo vorbereiten → Isolation prüfen → Sync → Konflikt-Auflösung # Requirements: 10.3, 10.5 # --------------------------------------------------------------------------- class TestFederationPrepareIsolateSyncConflict: """E2E: Prepare team repo → Verify isolation → Sync → Resolve conflicts.""" def test_prepare_team_repo_verify_isolation( self, monorepo_root: Path, full_machine_context: MachineContext ) -> None: """Prepare team repo and verify it contains only its own context.""" encryption_mgr = SecretEncryptionManager(monorepo_root, full_machine_context) config_path = monorepo_root / "shared" / "config" / "team-repos.yaml" fed_mgr = FederationManager( config_path=config_path, monorepo_root=monorepo_root, encryption_manager=encryption_mgr, ) # Create some content in privat context (monorepo_root / "privat" / "my-project").mkdir(parents=True) (monorepo_root / "privat" / "my-project" / "main.py").write_text( "# My private project\nprint('hello')\n", encoding="utf-8" ) # Create content in other contexts (should NOT appear in privat team repo) (monorepo_root / "dhive" / "dhive-project").mkdir(parents=True) (monorepo_root / "dhive" / "dhive-project" / "app.py").write_text( "# dhive project", encoding="utf-8" ) # Step 1: Prepare the team repo for privat team_repo_path = fed_mgr.prepare_team_repo("privat") assert team_repo_path.exists() # Step 2: Verify isolation - no references to other contexts isolation_report = fed_mgr.verify_isolation(team_repo_path, "privat") assert isolation_report.is_isolated is True assert len(isolation_report.leaks) == 0 # Verify privat content is present privat_files = list(team_repo_path.rglob("*.py")) assert len(privat_files) >= 1 # Verify dhive/bahn content is NOT present dhive_files = list(team_repo_path.rglob("*dhive*")) # Filter out any path references in filenames only for f in dhive_files: assert "dhive-project" not in str(f) or not f.exists() def test_sync_and_conflict_resolution( self, monorepo_root: Path, full_machine_context: MachineContext ) -> None: """Sync from team repo and resolve conflicts with team-wins strategy.""" encryption_mgr = SecretEncryptionManager(monorepo_root, full_machine_context) config_path = monorepo_root / "shared" / "config" / "team-repos.yaml" fed_mgr = FederationManager( config_path=config_path, monorepo_root=monorepo_root, encryption_manager=encryption_mgr, ) # Mock the git subtree operations (no actual network calls) with patch.object( fed_mgr.sync_engine, "subtree_pull" ) as mock_pull: mock_pull.return_value = SyncResult( success=True, context="privat", direction="pull", commits_synced=3, conflicts=[], ) result = fed_mgr.sync_from_team("privat") assert result.success is True assert result.commits_synced == 3 # Test conflict resolution with team-wins strategy with patch.object( fed_mgr.sync_engine, "subtree_pull" ) as mock_pull: mock_pull.return_value = SyncResult( success=False, context="dhive", direction="pull", commits_synced=0, conflicts=[ ConflictInfo( file_path="dhive/config.yaml", conflict_type="content", source="team-repo", details="Conflicting changes in config.yaml", ) ], ) conflict_result = fed_mgr.sync_from_team("dhive") assert conflict_result.success is False assert len(conflict_result.conflicts) >= 1 # Resolve conflict with team-wins strategy (mock git operations) with patch.object(fed_mgr.sync_engine, "_run_git") as mock_git: # First call: git diff to find conflicted files mock_git.return_value = MagicMock( stdout="dhive/config.yaml\n", stderr="", returncode=0 ) resolved = fed_mgr.resolve_conflict("dhive", strategy="team-wins") assert resolved is not None assert resolved.context == "dhive" assert resolved.strategy == "team-wins" # --------------------------------------------------------------------------- # Test 7: Shared-Mirror in Team_Repo → Read-Only-Prüfung # Requirements: 10.12 # --------------------------------------------------------------------------- class TestSharedMirrorReadOnly: """E2E: Mirror shared files into team repo → Verify read-only.""" def test_mirror_shared_to_team_repo_readonly( self, monorepo_root: Path, full_machine_context: MachineContext ) -> None: """Shared files mirrored to team repo should be read-only.""" encryption_mgr = SecretEncryptionManager(monorepo_root, full_machine_context) config_path = monorepo_root / "shared" / "config" / "team-repos.yaml" fed_mgr = FederationManager( config_path=config_path, monorepo_root=monorepo_root, encryption_manager=encryption_mgr, ) # Create shared content to be mirrored shared_tools = monorepo_root / "shared" / "tools" / "common-scripts" shared_tools.mkdir(parents=True) (shared_tools / "deploy.sh").write_text( "#!/bin/bash\necho 'deploying...'\n", encoding="utf-8" ) (shared_tools / "lint.sh").write_text( "#!/bin/bash\necho 'linting...'\n", encoding="utf-8" ) # Step 1: Mirror shared files to privat team repo mirror_result = fed_mgr.mirror_shared( "privat", ["shared/tools/common-scripts/"] ) assert mirror_result is not None assert mirror_result.success is True # Step 2: Verify mirrored paths are recorded assert len(mirror_result.mirrored_paths) >= 1 # Step 3: Verify context is correct assert mirror_result.context == "privat" def test_mirror_shared_with_multiple_paths( self, monorepo_root: Path, full_machine_context: MachineContext ) -> None: """Multiple shared paths can be mirrored to team repo.""" encryption_mgr = SecretEncryptionManager(monorepo_root, full_machine_context) config_path = monorepo_root / "shared" / "config" / "team-repos.yaml" fed_mgr = FederationManager( config_path=config_path, monorepo_root=monorepo_root, encryption_manager=encryption_mgr, ) # Create multiple shared resources (monorepo_root / "shared" / "tools" / "script-a.py").write_text( "# script A", encoding="utf-8" ) (monorepo_root / "shared" / "tools" / "script-b.py").write_text( "# script B", encoding="utf-8" ) mirror_result = fed_mgr.mirror_shared("dhive", ["shared/tools/"]) assert mirror_result is not None assert mirror_result.success is True assert len(mirror_result.mirrored_paths) >= 1 # --------------------------------------------------------------------------- # Test 8: Full Knowledge Pipeline E2E # Requirements: 2.2, 2.3, 5.1, 6.1 # --------------------------------------------------------------------------- class TestKnowledgePipelineE2E: """E2E: Full ingestion pipeline with source configs, enrichment, and indexing.""" @pytest.fixture() def knowledge_root(self, tmp_path: Path) -> Path: """Creates a knowledge-ready monorepo structure.""" root = tmp_path / "monorepo" root.mkdir() # Context knowledge folders for ctx in ("bahn", "dhive", "privat"): kp = root / ctx / "knowledge" for sub in ("inbox", "meetings", "decisions", "projects", "people", "references", "links"): (kp / sub).mkdir(parents=True) # Empty YAML index (kp / "_index.yaml").write_text( "version: '1.0'\nlast_updated: ''\nartifacts: []\n", encoding="utf-8", ) return root def test_full_pipeline_with_markdown_source(self, knowledge_root: Path) -> None: """Full pipeline run: sources.yaml → MarkdownSource → enrich → store → index.""" from monorepo.knowledge.ingestion.pipeline import IngestionPipeline from monorepo.knowledge.ingestion.enrichment import EnrichmentAgent, EnrichmentResult ctx = "bahn" kp = knowledge_root / ctx / "knowledge" # Create a markdown file source source_dir = knowledge_root / ctx / "docs" source_dir.mkdir(parents=True) (source_dir / "api-design.md").write_text( "---\n" "type: decision\n" "title: API Design Principles\n" "tags: [api, design]\n" "---\n\n" "# API Design\n\nWe use RESTful conventions.\n", encoding="utf-8", ) # Create sources.yaml sources_yaml = { "sources": [ { "type": "markdown", "name": "bahn-docs", "enabled": True, "params": {"directory": str(source_dir)}, } ], "settings": { "create_tasks": False, "git": {"auto_commit": False}, }, } (kp / "sources.yaml").write_text( yaml.dump(sources_yaml), encoding="utf-8" ) # Mock enrichment to avoid LLM calls mock_enrichment = MagicMock(spec=EnrichmentAgent) mock_enrichment.enrich.return_value = EnrichmentResult( title="API Design Principles", category="decision", tags=["api", "design"], people=[], projects=[], action_items=[], confidence=0.9, ) pipeline = IngestionPipeline( monorepo_root=knowledge_root, context=ctx, no_commit=True, enrichment_agent=mock_enrichment, ) result = pipeline.run() # Assertions assert result.processed >= 1 assert result.context == ctx assert len(result.errors) == 0 # Index should be updated index_content = (kp / "_index.yaml").read_text(encoding="utf-8") assert "api-design" in index_content or "API" in index_content def test_quick_capture_to_inbox_to_ingest_categorization( self, knowledge_root: Path ) -> None: """Quick Capture → Inbox → pipeline.process_inbox() → categorization.""" from monorepo.knowledge.ingestion.capture import QuickCapture from monorepo.knowledge.ingestion.pipeline import IngestionPipeline from monorepo.knowledge.ingestion.enrichment import EnrichmentAgent, EnrichmentResult ctx = "dhive" kp = knowledge_root / ctx / "knowledge" # Step 1: Quick Capture a note capture = QuickCapture(monorepo_root=knowledge_root) captured_path = capture.capture( input_text="Meeting mit Team: Sprint Planning nächste Woche vorbereiten", context=ctx, tags=["sprint", "planning"], ) # Verify capture created file in inbox assert captured_path.exists() assert "inbox" in str(captured_path) content = captured_path.read_text(encoding="utf-8") assert "type: inbox" in content assert "sprint" in content # Step 2: Process inbox (enrichment categorizes as meeting) mock_enrichment = MagicMock(spec=EnrichmentAgent) mock_enrichment.enrich.return_value = EnrichmentResult( title="Sprint Planning Vorbereitung", category="meeting", tags=["sprint", "planning", "team"], people=["Team"], projects=[], action_items=[], confidence=0.85, ) pipeline = IngestionPipeline( monorepo_root=knowledge_root, context=ctx, no_commit=True, enrichment_agent=mock_enrichment, ) result = pipeline.process_inbox() # Step 3: Verify categorization assert result.processed >= 1 assert result.context == ctx # Index should contain the processed artifact index_content = (kp / "_index.yaml").read_text(encoding="utf-8") assert len(index_content) > 50 # non-empty index def test_confluence_source_with_mocked_api(self, knowledge_root: Path) -> None: """Confluence source extraction with mocked REST API responses.""" from monorepo.knowledge.sources.confluence import ConfluenceSource, ConfluencePage from monorepo.knowledge.sources.base import SourceConfig source = ConfluenceSource() config = SourceConfig( type="confluence", name="bahn-confluence", params={ "base_url": "https://confluence.bahn.de", "token": "fake-token", "pages": ["12345"], }, ) # Mock the HTTP client interactions mock_page = ConfluencePage( page_id="12345", title="Architecture Decision Records", space_key="TEAM", version=3, content_html="

ADR-001

We use microservices architecture.

", url="https://confluence.bahn.de/pages/viewpage.action?pageId=12345", last_modified="2024-03-15T10:30:00Z", attachments=[], ) with patch.object(source, "_create_client") as mock_client_factory: mock_client = MagicMock() mock_client.get_page.return_value = mock_page mock_client_factory.return_value = mock_client with patch.object(source, "_collect_page_ids", return_value=["12345"]): extraction = source.extract(config, context="bahn") # Verify extraction result assert len(extraction.artifacts) == 1 assert len(extraction.errors) == 0 artifact = extraction.artifacts[0] assert "Architecture Decision Records" in artifact.metadata.title assert artifact.metadata.source.get("type") == "confluence" assert "confluence.bahn.de" in artifact.metadata.source.get("url", "") def test_confluence_source_api_error_graceful(self, knowledge_root: Path) -> None: """Confluence source handles API errors gracefully without crashing.""" import urllib.error from monorepo.knowledge.sources.confluence import ConfluenceSource from monorepo.knowledge.sources.base import SourceConfig source = ConfluenceSource() config = SourceConfig( type="confluence", name="bahn-confluence", params={ "base_url": "https://confluence.bahn.de", "token": "fake-token", "pages": ["99999"], }, ) with patch.object(source, "_create_client") as mock_client_factory: mock_client = MagicMock() mock_client.get_page.side_effect = urllib.error.URLError("Connection refused") mock_client_factory.return_value = mock_client with patch.object(source, "_collect_page_ids", return_value=["99999"]): extraction = source.extract(config, context="bahn") # Should not crash, should record error assert len(extraction.artifacts) == 0 assert len(extraction.errors) >= 1 assert extraction.errors[0].error_type == "connection" def test_jira_source_with_mocked_api(self, knowledge_root: Path) -> None: """Jira source extraction with mocked REST API responses.""" from monorepo.knowledge.sources.jira import JiraSource from monorepo.knowledge.sources.base import SourceConfig source = JiraSource() config = SourceConfig( type="jira", name="bahn-jira", params={ "base_url": "https://jira.bahn.de", "token": "fake-token", "username": "test-user", "jql": "project = ACV2 AND updatedDate > -7d", }, ) # Mock issue data as Jira REST API v2 format mock_issues = [ { "key": "ACV2-101", "fields": { "summary": "Implement OAuth2 login flow", "description": "Implement the OAuth2 flow for SSO integration.", "status": {"name": "In Progress"}, "assignee": {"displayName": "Max Mustermann"}, "reporter": {"displayName": "Anna Schmidt"}, "project": {"key": "ACV2"}, "comment": {"comments": [ {"body": "Started working on this", "author": {"displayName": "Max Mustermann"}}, ]}, "updated": "2024-03-20T14:00:00.000+0000", "created": "2024-03-18T09:00:00.000+0000", }, }, { "key": "ACV2-102", "fields": { "summary": "Fix logout redirect", "description": "Logout should redirect to landing page.", "status": {"name": "Done"}, "assignee": {"displayName": "Anna Schmidt"}, "reporter": {"displayName": "Max Mustermann"}, "project": {"key": "ACV2"}, "comment": {"comments": []}, "updated": "2024-03-19T16:30:00.000+0000", "created": "2024-03-17T11:00:00.000+0000", }, }, ] with patch.object(source, "_fetch_issues", return_value=mock_issues): with patch.object(source, "_create_session", return_value=MagicMock()): extraction = source.extract(config, context="bahn") # Verify extraction result assert len(extraction.artifacts) == 2 assert len(extraction.errors) == 0 # Verify first artifact metadata art1 = extraction.artifacts[0] assert "ACV2-101" in art1.metadata.source.get("issue_key", "") assert art1.metadata.source.get("type") == "jira" assert "OAuth2" in art1.metadata.title or "OAuth2" in art1.content def test_jira_source_api_error_graceful(self, knowledge_root: Path) -> None: """Jira source handles API errors gracefully without crashing.""" from monorepo.knowledge.sources.jira import JiraSource, JiraAPIError from monorepo.knowledge.sources.base import SourceConfig source = JiraSource() config = SourceConfig( type="jira", name="bahn-jira", params={ "base_url": "https://jira.bahn.de", "token": "fake-token", "jql": "project = ACV2", }, ) with patch.object( source, "_fetch_issues", side_effect=JiraAPIError("Connection timeout", error_type="connection", retry=True), ): with patch.object(source, "_create_session", return_value=MagicMock()): extraction = source.extract(config, context="bahn") # Should not crash, should record error assert len(extraction.artifacts) == 0 assert len(extraction.errors) >= 1 assert extraction.errors[0].error_type == "connection" assert extraction.errors[0].retry is True def test_pipeline_with_confluence_and_jira_sources(self, knowledge_root: Path) -> None: """Full pipeline E2E with both Confluence and Jira sources configured.""" from monorepo.knowledge.ingestion.pipeline import IngestionPipeline from monorepo.knowledge.ingestion.enrichment import EnrichmentAgent, EnrichmentResult from monorepo.knowledge.sources.confluence import ConfluenceSource, ConfluencePage from monorepo.knowledge.sources.jira import JiraSource from monorepo.knowledge.sources.base import SourceStrategy ctx = "bahn" kp = knowledge_root / ctx / "knowledge" # Create sources.yaml with both sources sources_yaml = { "sources": [ { "type": "confluence", "name": "bahn-confluence", "enabled": True, "params": { "base_url": "https://confluence.bahn.de", "token": "fake-token", "pages": ["100"], }, }, { "type": "jira", "name": "bahn-jira", "enabled": True, "params": { "base_url": "https://jira.bahn.de", "token": "fake-token", "username": "test", "jql": "project = ACV2", }, }, ], "settings": { "create_tasks": False, "git": {"auto_commit": False}, }, } (kp / "sources.yaml").write_text( yaml.dump(sources_yaml), encoding="utf-8" ) # Mock enrichment mock_enrichment = MagicMock(spec=EnrichmentAgent) mock_enrichment.enrich.return_value = EnrichmentResult( title="Test Artifact", category="reference", tags=["test"], people=[], projects=[], action_items=[], confidence=0.8, ) # Create mock strategies that return pre-built artifacts from monorepo.knowledge.artifact import Artifact, ArtifactMetadata from monorepo.knowledge.sources.base import ExtractionResult from datetime import date mock_confluence = MagicMock(spec=ConfluenceSource) mock_confluence.extract.return_value = ExtractionResult( artifacts=[ Artifact( metadata=ArtifactMetadata( type="reference", title="Confluence Page: Architecture", tags=["architecture"], source_context=ctx, source={"type": "confluence", "url": "https://confluence.bahn.de/page/100"}, created=date(2024, 3, 15), ), content="# Architecture\n\nMicroservices approach.\n", file_path=Path("confluence-architecture.md"), ), ], errors=[], skipped=0, ) mock_jira = MagicMock(spec=JiraSource) mock_jira.extract.return_value = ExtractionResult( artifacts=[ Artifact( metadata=ArtifactMetadata( type="reference", title="ACV2-200: Feature Request", tags=["jira", "feature"], source_context=ctx, source={"type": "jira", "issue_key": "ACV2-200"}, created=date(2024, 3, 20), ), content="# ACV2-200: Feature Request\n\nDetails here.\n", file_path=Path("acv2-200-feature-request.md"), ), ], errors=[], skipped=0, ) pipeline = IngestionPipeline( monorepo_root=knowledge_root, context=ctx, no_commit=True, enrichment_agent=mock_enrichment, source_strategies={ "confluence": mock_confluence, "jira": mock_jira, }, ) result = pipeline.run() # Both sources should contribute artifacts assert result.processed == 2 assert result.context == ctx assert len(result.errors) == 0 # Index should be non-trivial index_content = (kp / "_index.yaml").read_text(encoding="utf-8") assert "artifacts" in index_content def test_pipeline_partial_source_failure(self, knowledge_root: Path) -> None: """Pipeline continues processing when one source fails (fail-forward).""" from monorepo.knowledge.ingestion.pipeline import IngestionPipeline from monorepo.knowledge.ingestion.enrichment import EnrichmentAgent, EnrichmentResult from monorepo.knowledge.sources.confluence import ConfluenceSource from monorepo.knowledge.sources.jira import JiraSource from monorepo.knowledge.sources.base import ExtractionResult, SourceError from monorepo.knowledge.artifact import Artifact, ArtifactMetadata from datetime import date ctx = "bahn" kp = knowledge_root / ctx / "knowledge" # sources.yaml with two sources – first will fail sources_yaml = { "sources": [ { "type": "confluence", "name": "failing-source", "enabled": True, "params": {"base_url": "https://broken.example.com", "token": "x"}, }, { "type": "jira", "name": "working-source", "enabled": True, "params": { "base_url": "https://jira.bahn.de", "token": "x", "jql": "project = X", }, }, ], "settings": { "create_tasks": False, "git": {"auto_commit": False}, }, } (kp / "sources.yaml").write_text( yaml.dump(sources_yaml), encoding="utf-8" ) # Mock enrichment mock_enrichment = MagicMock(spec=EnrichmentAgent) mock_enrichment.enrich.return_value = EnrichmentResult( title="Working Artifact", category="reference", tags=[], people=[], projects=[], action_items=[], confidence=0.8, ) # Confluence raises an exception mock_confluence = MagicMock(spec=ConfluenceSource) mock_confluence.extract.side_effect = ConnectionError("API unreachable") # Jira returns a valid artifact mock_jira = MagicMock(spec=JiraSource) mock_jira.extract.return_value = ExtractionResult( artifacts=[ Artifact( metadata=ArtifactMetadata( type="reference", title="Surviving Artifact", tags=["jira"], source_context=ctx, source={"type": "jira", "issue_key": "X-1"}, created=date(2024, 3, 20), ), content="# Surviving Artifact\n\nStill processed.\n", file_path=Path("x-1-surviving.md"), ), ], errors=[], skipped=0, ) pipeline = IngestionPipeline( monorepo_root=knowledge_root, context=ctx, no_commit=True, enrichment_agent=mock_enrichment, source_strategies={ "confluence": mock_confluence, "jira": mock_jira, }, ) result = pipeline.run() # Should still process the working source assert result.processed >= 1 # Should log the failure assert len(result.errors) >= 1 assert any("unreachable" in e.message.lower() or "API" in e.message for e in result.errors) # --------------------------------------------------------------------------- # Test: Integration Stack Factory (create_integrated_stack) # Requirements: 2.2, 9.3, 9.4, 10.10 # --------------------------------------------------------------------------- class TestIntegratedStackFactory: """E2E: Verify create_integrated_stack wires all components correctly.""" def test_create_integrated_stack_full(self, monorepo_root: Path) -> None: """Full stack creation with all components wired.""" from monorepo.integration import create_integrated_stack stack = create_integrated_stack(monorepo_root) # All components should be created assert stack["context_guard"] is not None assert stack["audit_logger"] is not None assert stack["orchestrator_adapter"] is not None assert stack["encryption_manager"] is not None assert stack["federation_manager"] is not None assert stack["machine_context_manager"] is not None # Verify the ContextGuard has encryption wired guard = stack["context_guard"] assert guard.encryption_manager is not None # Verify the guard can load env with encryption env = guard.load_env("privat") assert "PRIVAT_SECRET" in env def test_create_integrated_stack_cross_context_denied( self, monorepo_root: Path ) -> None: """Integrated stack correctly denies cross-context access.""" from monorepo.integration import create_integrated_stack stack = create_integrated_stack(monorepo_root) guard = stack["context_guard"] # Cross-context access should be denied assert guard.check_access("privat", monorepo_root / "dhive" / ".env") is False assert guard.check_access("dhive", monorepo_root / "bahn" / ".env") is False assert guard.check_access("bahn", monorepo_root / "privat" / ".env") is False # Same-context access should be allowed assert guard.check_access("privat", monorepo_root / "privat" / ".env") is True assert guard.check_access("dhive", monorepo_root / "dhive" / ".env") is True