huncode/content-factory
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Task 003: add postgres schema and seed data

Browse Source
This commit is contained in:
huncode
2026-05-21 18:11:29 +03:00
parent a0ae06adfa
commit 7a11d50aa7
18 changed files with 1961 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/backend/tests/integration/__init__.py
+1
View File
@@ -0,0 +1 @@
apps/backend/tests/integration/test_postgres_alembic_migration.py
+43
View File
@@ -0,0 +1,43 @@
from __future__ import annotations
import os
import subprocess
import sys
import unittest
from importlib.util import find_spec
from pathlib import Path
REPO_ROOT = Path(__file__).resolve().parents[4]
BACKEND_ROOT = REPO_ROOT / "apps" / "backend"
sys.path.insert(0, str(BACKEND_ROOT))
class PostgresAlembicMigrationIntegrationTest(unittest.TestCase):
def test_alembic_upgrade_head_creates_schema_on_configured_postgres(self) -> None:
dsn = os.environ.get("PIPELINE_TEST_DATABASE_DSN")
if not dsn:
self.skipTest("PIPELINE_TEST_DATABASE_DSN is not set")
if not dsn.startswith(("postgresql://", "postgres://")):
self.skipTest("Postgres DSN is required for Alembic migration test")
if find_spec("alembic") is None:
self.skipTest("alembic is not installed")
env = os.environ.copy()
env["DATABASE_URL"] = dsn
subprocess.run(
[sys.executable, "-m", "alembic", "upgrade", "head"],
cwd=REPO_ROOT,
env=env,
check=True,
)
from src.domain.schema import CORE_TABLES
from src.infrastructure.repositories import open_backend_repository
repository = open_backend_repository(dsn)
self.assertTrue(set(CORE_TABLES).issubset(repository.schema.list_tables()))
if __name__ == "__main__":
unittest.main()
apps/backend/tests/integration/test_schema_storage_contracts.py
+136
View File
@@ -0,0 +1,136 @@
from __future__ import annotations
import os
import sys
import tempfile
import unittest
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
BACKEND_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(BACKEND_ROOT))
@contextmanager
def repository_dsn() -> Iterator[str]:
configured_dsn = os.environ.get("PIPELINE_TEST_DATABASE_DSN")
if configured_dsn:
yield configured_dsn
return
with tempfile.TemporaryDirectory() as tmp_dir:
yield f"sqlite:///{Path(tmp_dir) / 'pipeline-schema.db'}"
class SchemaStorageContractsIntegrationTest(unittest.TestCase):
def test_schema_contains_task_tables_and_operational_indexes(self) -> None:
from src.domain.schema import CORE_TABLES
from src.infrastructure.repositories import open_backend_repository
with repository_dsn() as dsn:
repository = open_backend_repository(dsn)
repository.setup()
self.assertTrue(set(CORE_TABLES).issubset(repository.schema.list_tables()))
self.assertIn(
"idx_target_sites_slug",
repository.schema.list_indexes("target_sites"),
)
self.assertIn(
"idx_articles_dashboard",
repository.schema.list_indexes("articles"),
)
self.assertIn(
"idx_agent_jobs_queue",
repository.schema.list_indexes("agent_jobs"),
)
self.assertIn(
"idx_workflow_events_timeline",
repository.schema.list_indexes("workflow_events"),
)
def test_required_domain_storage_fields_exist(self) -> None:
from src.infrastructure.repositories import open_backend_repository
with repository_dsn() as dsn:
repository = open_backend_repository(dsn)
repository.setup()
self.assertTrue(
{"status", "publishing_status"}.issubset(
repository.schema.list_columns("articles")
)
)
self.assertIn("status", repository.schema.list_columns("article_drafts"))
self.assertTrue(
{
"created_by",
"created_at",
"diff",
"rollback_target_version_id",
"activated_at",
"publishing_yaml",
"transform_script",
}.issubset(repository.schema.list_columns("script_config_versions"))
)
self.assertTrue(
{
"s3_prefix",
"source_urls",
"object_keys",
"content_hashes",
"artifact_types",
"metadata_references",
"artifacts",
}.issubset(repository.schema.list_columns("research_run_manifests"))
)
self.assertTrue(
{
"repository_url",
"branch",
"commit_sha",
"content_bundle_manifest",
"status",
}.issubset(repository.schema.list_columns("publish_commits"))
)
def test_seed_data_is_idempotent_for_users_site_and_script_version(self) -> None:
from src.application.seed_data import seed_reference_data
from src.domain.schema import (
SEEDED_ADMIN_EMAIL,
SEEDED_EDITOR_EMAIL,
SEEDED_TARGET_SITE_SLUG,
)
from src.infrastructure.repositories import open_backend_repository
with repository_dsn() as dsn:
repository = open_backend_repository(dsn)
repository.setup()
seed_reference_data(repository)
seed_reference_data(repository)
site = repository.target_sites.get_by_slug(SEEDED_TARGET_SITE_SLUG)
seeded_sites = [
candidate
for candidate in repository.target_sites.list()
if candidate.slug == SEEDED_TARGET_SITE_SLUG
]
script_versions = repository.script_config_versions.list_for_site(site.id)
self.assertEqual(1, repository.users.count_by_email(SEEDED_ADMIN_EMAIL))
self.assertEqual(1, repository.users.count_by_email(SEEDED_EDITOR_EMAIL))
self.assertEqual([site.id], [candidate.id for candidate in seeded_sites])
self.assertEqual(1, len(script_versions))
self.assertEqual("ACTIVE", script_versions[0]["status"])
self.assertEqual(
str(site.active_script_config_version_id), str(script_versions[0]["id"])
)
self.assertIn("repository:", script_versions[0]["publishing_yaml"])
self.assertIn("transformArticle", script_versions[0]["transform_script"])
if __name__ == "__main__":
unittest.main()
apps/backend/tests/integration/test_seeded_target_site_lookup.py
+74
View File
@@ -0,0 +1,74 @@
from __future__ import annotations
import os
import sys
import tempfile
import unittest
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
BACKEND_ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(BACKEND_ROOT))
SEEDED_SITE_SLUG = "b2b_saas_blog"
@contextmanager
def repository_dsn() -> Iterator[str]:
configured_dsn = os.environ.get("PIPELINE_TEST_DATABASE_DSN")
if configured_dsn:
yield configured_dsn
return
with tempfile.TemporaryDirectory() as tmp_dir:
yield f"sqlite:///{Path(tmp_dir) / 'pipeline-seed-red.db'}"
class SeededTargetSiteLookupIntegrationTest(unittest.TestCase):
def test_seeded_git_next_target_site_lookup_by_slug_is_idempotent(self) -> None:
from src.application.seed_data import seed_reference_data
from src.infrastructure.repositories import open_backend_repository
with repository_dsn() as dsn:
repository = open_backend_repository(dsn)
repository.setup()
seed_reference_data(repository)
seed_reference_data(repository)
site = repository.target_sites.get_by_slug(SEEDED_SITE_SLUG)
seeded_sites = [
candidate
for candidate in repository.target_sites.list()
if candidate.slug == SEEDED_SITE_SLUG
]
from src.domain.contracts import TargetSiteConfig
self.assertIsInstance(site, TargetSiteConfig)
self.assertEqual([site.id], [candidate.id for candidate in seeded_sites])
self.assertEqual("B2B SaaS Blog", site.name)
self.assertEqual("git_next", site.publishing_type)
self.assertEqual("en", site.default_language)
self.assertEqual(
"git@github.com:example/site.git",
site.publishing_rules.repository_url,
)
self.assertEqual("main", site.publishing_rules.production_branch)
self.assertEqual("mdx", site.publishing_rules.content_format)
self.assertEqual(
"content/articles/{slug}.mdx",
site.publishing_rules.content_path_template,
)
self.assertEqual(
"public/articles/{slug}/{filename}",
site.publishing_rules.asset_path_template,
)
self.assertIsNotNone(site.active_script_config_version_id)
if __name__ == "__main__":
unittest.main()