331 lines
11 KiB
Python
331 lines
11 KiB
Python
"""
|
|
Integration tests for deployment target assignment in batch generation
|
|
"""
|
|
|
|
import pytest
|
|
import json
|
|
import tempfile
|
|
from pathlib import Path
|
|
from unittest.mock import Mock, MagicMock, patch
|
|
from src.generation.batch_processor import BatchProcessor
|
|
from src.generation.service import ContentGenerator
|
|
from src.generation.job_config import JobConfig
|
|
from src.database.models import SiteDeployment, Project, GeneratedContent
|
|
|
|
|
|
class TestDeploymentTargetAssignment:
|
|
"""Integration tests for deployment target assignment"""
|
|
|
|
def test_job_config_parses_deployment_targets(self, tmp_path):
|
|
"""Test JobConfig parses deployment_targets field correctly"""
|
|
job_file = tmp_path / "test_job.json"
|
|
job_data = {
|
|
"jobs": [{
|
|
"project_id": 1,
|
|
"deployment_targets": [
|
|
"www.domain1.com",
|
|
"www.domain2.com",
|
|
"www.domain3.com"
|
|
],
|
|
"tiers": {
|
|
"tier1": {"count": 5}
|
|
}
|
|
}]
|
|
}
|
|
|
|
job_file.write_text(json.dumps(job_data))
|
|
|
|
config = JobConfig(str(job_file))
|
|
jobs = config.get_jobs()
|
|
|
|
assert len(jobs) == 1
|
|
assert jobs[0].deployment_targets == [
|
|
"www.domain1.com",
|
|
"www.domain2.com",
|
|
"www.domain3.com"
|
|
]
|
|
|
|
def test_job_config_handles_missing_deployment_targets(self, tmp_path):
|
|
"""Test JobConfig handles jobs without deployment_targets"""
|
|
job_file = tmp_path / "test_job.json"
|
|
job_data = {
|
|
"jobs": [{
|
|
"project_id": 1,
|
|
"tiers": {
|
|
"tier1": {"count": 5}
|
|
}
|
|
}]
|
|
}
|
|
|
|
job_file.write_text(json.dumps(job_data))
|
|
|
|
config = JobConfig(str(job_file))
|
|
jobs = config.get_jobs()
|
|
|
|
assert len(jobs) == 1
|
|
assert jobs[0].deployment_targets is None
|
|
|
|
def test_job_config_validates_deployment_targets_type(self, tmp_path):
|
|
"""Test JobConfig validates deployment_targets is an array"""
|
|
job_file = tmp_path / "test_job.json"
|
|
job_data = {
|
|
"jobs": [{
|
|
"project_id": 1,
|
|
"deployment_targets": "not_an_array",
|
|
"tiers": {
|
|
"tier1": {"count": 5}
|
|
}
|
|
}]
|
|
}
|
|
|
|
job_file.write_text(json.dumps(job_data))
|
|
|
|
with pytest.raises(ValueError, match="must be an array"):
|
|
JobConfig(str(job_file))
|
|
|
|
def test_job_config_validates_deployment_targets_elements(self, tmp_path):
|
|
"""Test JobConfig validates deployment_targets contains only strings"""
|
|
job_file = tmp_path / "test_job.json"
|
|
job_data = {
|
|
"jobs": [{
|
|
"project_id": 1,
|
|
"deployment_targets": ["www.domain1.com", 123, "www.domain2.com"],
|
|
"tiers": {
|
|
"tier1": {"count": 5}
|
|
}
|
|
}]
|
|
}
|
|
|
|
job_file.write_text(json.dumps(job_data))
|
|
|
|
with pytest.raises(ValueError, match="must be an array of strings"):
|
|
JobConfig(str(job_file))
|
|
|
|
def test_batch_processor_validates_targets_at_job_start(self, tmp_path):
|
|
"""Test BatchProcessor validates deployment targets at job start"""
|
|
job_file = tmp_path / "test_job.json"
|
|
job_data = {
|
|
"jobs": [{
|
|
"project_id": 1,
|
|
"deployment_targets": ["www.domain1.com", "invalid.com"],
|
|
"tiers": {
|
|
"tier1": {"count": 5}
|
|
}
|
|
}]
|
|
}
|
|
|
|
job_file.write_text(json.dumps(job_data))
|
|
|
|
mock_generator = Mock(spec=ContentGenerator)
|
|
mock_content_repo = Mock()
|
|
mock_project_repo = Mock()
|
|
|
|
mock_project = Mock(spec=Project)
|
|
mock_project.id = 1
|
|
mock_project.main_keyword = "test keyword"
|
|
mock_project_repo.get_by_id.return_value = mock_project
|
|
|
|
mock_site_repo = Mock()
|
|
|
|
def mock_get_by_hostname(hostname):
|
|
if hostname == "www.domain1.com":
|
|
return Mock(id=1)
|
|
return None
|
|
|
|
mock_site_repo.get_by_hostname.side_effect = mock_get_by_hostname
|
|
|
|
processor = BatchProcessor(
|
|
content_generator=mock_generator,
|
|
content_repo=mock_content_repo,
|
|
project_repo=mock_project_repo,
|
|
site_deployment_repo=mock_site_repo
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="invalid.com"):
|
|
processor.process_job(str(job_file), debug=False, continue_on_error=False)
|
|
|
|
def test_batch_processor_requires_site_repo_with_deployment_targets(self, tmp_path):
|
|
"""Test BatchProcessor requires SiteDeploymentRepository when deployment_targets specified"""
|
|
job_file = tmp_path / "test_job.json"
|
|
job_data = {
|
|
"jobs": [{
|
|
"project_id": 1,
|
|
"deployment_targets": ["www.domain1.com"],
|
|
"tiers": {
|
|
"tier1": {"count": 1}
|
|
}
|
|
}]
|
|
}
|
|
|
|
job_file.write_text(json.dumps(job_data))
|
|
|
|
mock_generator = Mock(spec=ContentGenerator)
|
|
mock_content_repo = Mock()
|
|
mock_project_repo = Mock()
|
|
|
|
mock_project = Mock(spec=Project)
|
|
mock_project.id = 1
|
|
mock_project.main_keyword = "test keyword"
|
|
mock_project_repo.get_by_id.return_value = mock_project
|
|
|
|
processor = BatchProcessor(
|
|
content_generator=mock_generator,
|
|
content_repo=mock_content_repo,
|
|
project_repo=mock_project_repo,
|
|
site_deployment_repo=None
|
|
)
|
|
|
|
with pytest.raises(ValueError, match="SiteDeploymentRepository not provided"):
|
|
processor.process_job(str(job_file), debug=False, continue_on_error=False)
|
|
|
|
def test_assignment_logic_with_ten_articles_three_targets(self):
|
|
"""Test 10 articles with 3 targets: first 3 assigned, rest null"""
|
|
from src.generation.deployment_assignment import assign_site_for_article
|
|
|
|
resolved_targets = {
|
|
"www.domain1.com": 5,
|
|
"www.domain2.com": 8,
|
|
"www.domain3.com": 12
|
|
}
|
|
|
|
assignments = [assign_site_for_article(i, resolved_targets) for i in range(10)]
|
|
|
|
assert assignments[0] == 5
|
|
assert assignments[1] == 8
|
|
assert assignments[2] == 12
|
|
assert assignments[3] is None
|
|
assert assignments[4] is None
|
|
assert assignments[5] is None
|
|
assert assignments[6] is None
|
|
assert assignments[7] is None
|
|
assert assignments[8] is None
|
|
assert assignments[9] is None
|
|
|
|
def test_content_repository_accepts_site_deployment_id(self):
|
|
"""Test GeneratedContentRepository.create() accepts site_deployment_id"""
|
|
from src.database.repositories import GeneratedContentRepository
|
|
|
|
mock_session = Mock()
|
|
mock_session.add = Mock()
|
|
mock_session.commit = Mock()
|
|
mock_session.refresh = Mock()
|
|
|
|
repo = GeneratedContentRepository(mock_session)
|
|
|
|
content = repo.create(
|
|
project_id=1,
|
|
tier="tier1",
|
|
keyword="test keyword",
|
|
title="Test Title",
|
|
outline={"outline": []},
|
|
content="<p>Test content</p>",
|
|
word_count=100,
|
|
status="generated",
|
|
site_deployment_id=5
|
|
)
|
|
|
|
assert content.site_deployment_id == 5
|
|
mock_session.add.assert_called_once()
|
|
mock_session.commit.assert_called_once()
|
|
|
|
def test_content_repository_defaults_site_deployment_id_to_none(self):
|
|
"""Test GeneratedContentRepository.create() defaults site_deployment_id to None"""
|
|
from src.database.repositories import GeneratedContentRepository
|
|
|
|
mock_session = Mock()
|
|
mock_session.add = Mock()
|
|
mock_session.commit = Mock()
|
|
mock_session.refresh = Mock()
|
|
|
|
repo = GeneratedContentRepository(mock_session)
|
|
|
|
content = repo.create(
|
|
project_id=1,
|
|
tier="tier1",
|
|
keyword="test keyword",
|
|
title="Test Title",
|
|
outline={"outline": []},
|
|
content="<p>Test content</p>",
|
|
word_count=100,
|
|
status="generated"
|
|
)
|
|
|
|
assert content.site_deployment_id is None
|
|
mock_session.add.assert_called_once()
|
|
mock_session.commit.assert_called_once()
|
|
|
|
def test_only_tier1_gets_deployment_targets(self, tmp_path):
|
|
"""Test that only tier1 articles get assigned to deployment targets, tier2+ get null"""
|
|
job_file = tmp_path / "test_job.json"
|
|
job_data = {
|
|
"jobs": [{
|
|
"project_id": 1,
|
|
"deployment_targets": ["www.domain1.com", "www.domain2.com"],
|
|
"tiers": {
|
|
"tier1": {"count": 3},
|
|
"tier2": {"count": 5}
|
|
}
|
|
}]
|
|
}
|
|
|
|
job_file.write_text(json.dumps(job_data))
|
|
|
|
mock_generator = Mock(spec=ContentGenerator)
|
|
mock_generator.generate_title.return_value = "Test Title"
|
|
mock_generator.generate_outline.return_value = {"outline": []}
|
|
mock_generator.generate_content.return_value = "<p>Test</p>"
|
|
mock_generator.count_words.return_value = 2000
|
|
|
|
mock_content_repo = Mock()
|
|
mock_project_repo = Mock()
|
|
|
|
mock_project = Mock(spec=Project)
|
|
mock_project.id = 1
|
|
mock_project.main_keyword = "test keyword"
|
|
mock_project_repo.get_by_id.return_value = mock_project
|
|
|
|
mock_site_repo = Mock()
|
|
mock_site_repo.get_by_hostname.side_effect = lambda h: Mock(id=1) if h == "www.domain1.com" else Mock(id=2)
|
|
|
|
created_contents = []
|
|
def mock_create(**kwargs):
|
|
content = Mock()
|
|
for k, v in kwargs.items():
|
|
setattr(content, k, v)
|
|
created_contents.append(content)
|
|
return content
|
|
|
|
mock_content_repo.create.side_effect = mock_create
|
|
|
|
processor = BatchProcessor(
|
|
content_generator=mock_generator,
|
|
content_repo=mock_content_repo,
|
|
project_repo=mock_project_repo,
|
|
site_deployment_repo=mock_site_repo
|
|
)
|
|
|
|
processor.process_job(str(job_file), debug=False, continue_on_error=False)
|
|
|
|
assert len(created_contents) == 8
|
|
|
|
tier1_contents = [c for c in created_contents if c.tier == "tier1"]
|
|
tier2_contents = [c for c in created_contents if c.tier == "tier2"]
|
|
|
|
assert len(tier1_contents) == 3
|
|
assert len(tier2_contents) == 5
|
|
|
|
assert tier1_contents[0].site_deployment_id == 1
|
|
assert tier1_contents[1].site_deployment_id == 2
|
|
assert tier1_contents[2].site_deployment_id is None
|
|
|
|
for content in tier2_contents:
|
|
assert content.site_deployment_id is None
|
|
|
|
|
|
@pytest.fixture
|
|
def tmp_path():
|
|
"""Create a temporary directory for test files"""
|
|
with tempfile.TemporaryDirectory() as tmpdir:
|
|
yield Path(tmpdir)
|
|
|