Solaria
b417626778
- tests/test_github_ingest.py - GitHub repository ingestion - tests/test_s3_ingest.py - S3/Backblaze ingestion - tests/test_generation.py - Document generation - tests/test_output_push.py - Output file and push handling - tests/test_e2e.py - End-to-end integration tests Closes #58
61 lines
2.2 KiB
Python
61 lines
2.2 KiB
Python
"""S3/Backblaze Ingestion Tests for Opus Orchestrator."""
|
|
|
|
import pytest
|
|
from unittest.mock import Mock, patch, MagicMock
|
|
|
|
|
|
class TestS3Ingestor:
|
|
"""Test S3-compatible storage ingestion."""
|
|
|
|
def test_s3_ingestor_initialization(self):
|
|
"""Test S3 ingestor can be initialized."""
|
|
try:
|
|
from opus_orchestrator.utils.s3_ingest import S3Ingestor
|
|
ingestor = S3Ingestor()
|
|
assert ingestor is not None
|
|
except ImportError:
|
|
pytest.skip("S3Ingestor not implemented yet")
|
|
|
|
def test_multisource_ingestor_s3_support(self):
|
|
"""Test MultiSourceIngestor has S3 support."""
|
|
from opus_orchestrator.utils.multi_source_ingest import MultiSourceIngestor, SourceType
|
|
assert hasattr(SourceType, 'S3')
|
|
assert SourceType.S3.value == "s3"
|
|
|
|
@pytest.mark.integration
|
|
@pytest.mark.skip(reason="Requires B2 credentials")
|
|
def test_ingest_backblaze(self):
|
|
import os
|
|
required = ["B2_ENDPOINT", "B2_BUCKET", "B2_KEY_ID", "B2_APP_KEY"]
|
|
missing = [v for v in required if not os.environ.get(v)]
|
|
if missing:
|
|
pytest.skip(f"Missing B2 credentials: {missing}")
|
|
|
|
def test_s3_credentials_env_vars(self):
|
|
required_vars = {
|
|
"B2_ENDPOINT": "Backblaze B2 endpoint URL",
|
|
"B2_BUCKET": "Bucket name",
|
|
"B2_KEY_ID": "Key ID",
|
|
"B2_APP_KEY": "Application key",
|
|
}
|
|
assert len(required_vars) > 0
|
|
|
|
|
|
class TestMultiSourceIngestor:
|
|
def test_multisource_initialization(self):
|
|
from opus_orchestrator.utils.multi_source_ingest import MultiSourceIngestor
|
|
ingestor = MultiSourceIngestor()
|
|
assert ingestor is not None
|
|
|
|
def test_content_source_dataclass(self):
|
|
from opus_orchestrator.utils.multi_source_ingest import ContentSource, SourceType
|
|
source = ContentSource(source_type=SourceType.GITHUB, repo="test/repo")
|
|
assert source.source_type == SourceType.GITHUB
|
|
|
|
def test_source_type_enum(self):
|
|
from opus_orchestrator.utils.multi_source_ingest import SourceType
|
|
assert SourceType.GITHUB.value == "github"
|
|
assert SourceType.S3.value == "s3"
|
|
assert SourceType.LOCAL.value == "local"
|
|
assert SourceType.URL.value == "url"
|