"""Synthetic source identities, timestamps and nested chunks only.""" from datetime import UTC, datetime import pytest from coulomb_social.apps.spaces.bubble_transfer import ( normalize_ref, source_timestamp, transfer_plan, ) from coulomb_social.apps.spaces.import_tree import import_space_tree from coulomb_social.apps.spaces.models import Space from coulomb_social.apps.spaces.services import put_space_page from tests.test_import_tree import _write_sample_tree from tests.test_space_permissions import make_member def record(created, **fields): return {"Created Date": created, "Modified Date": created, **fields} def test_numeric_prefixed_lookup_references_use_the_record_id_after_lookup(): identity = "1750000000000x123456789012345678" assert normalize_ref("1234567890__LOOKUP__" + identity) == identity def test_lifo_is_deterministic_and_uses_creation_not_modification(): spaces = {"old": record(1700000000000), "b": record(1750000000123), "a": record(1750000000123), "unknown": {}} spaces["old"]["Modified Date"] = 1800000000000 plan = transfer_plan(spaces, {}) assert [s["id"] for s in plan["spaces"]] == ["a", "b", "old", "unknown"] assert plan == transfer_plan(dict(reversed(list(spaces.items()))), {}) assert plan["spaces"][0]["created_at_raw"] == 1750000000123 assert plan["spaces"][0]["created_at"] == "2025-06-15T15:06:40.123000+00:00" assert not plan["spaces"][-1]["captured_graph_complete"] assert not plan["production_imported"] def test_recursive_chunks_preserve_every_date_and_report_missing_records(): spaces = {"space": record(1750000000000)} articles = { "page": record(1750000001000, space_custom_space="space", content_text="Page", chunks_list_custom_article=["child"]), "child": record(1750000002000, content_text="Nested", context_custom_article="page", chunks_list_custom_article=["grandchild"]), "grandchild": record(1750000003000, content_text="Leaf", chunks_list_custom_article=["page", "missing"]), } plan = transfer_plan(spaces, articles) entry = plan["spaces"][0] assert len(entry["articles"]) == 3 assert {p["created_at_raw"] for p in entry["articles"]} == { 1750000001000, 1750000002000, 1750000003000, } assert {g["reason"] for g in entry["gaps"]} == {"missing_chunk_record"} assert not plan["unassigned_article_ids"] del articles["grandchild"]["Created Date"] assert "missing_or_invalid_created_at" in { g["reason"] for g in transfer_plan(spaces, articles)["spaces"][0]["gaps"] } def test_space_root_and_its_chunks_are_included_without_article_space_fields(): spaces = {"space": record(1750000000000, _chunk_custom_article="root")} articles = {"root": record(1750000001000, content_text="Root", chunks_list_custom_article=["child"]), "child": record(1750000002000, content_text="Child")} plan = transfer_plan(spaces, articles) assert plan["spaces"][0]["captured_graph_complete"] assert plan["spaces"][0]["root_article_id"] == "root" assert len(plan["spaces"][0]["articles"]) == 2 assert not plan["unassigned_article_ids"] del articles["root"] assert not transfer_plan(spaces, articles)["spaces"][0]["captured_graph_complete"] def test_fresh_capture_does_not_silently_drop_retained_inventory_or_certify_stale_dates(): inventory = {"missing": record(1750000000000), "fresh": record(1700000000000)} plan = transfer_plan({"fresh": record(1700000000000)}, {}, inventory_spaces=inventory) assert [s["id"] for s in plan["spaces"]] == ["missing", "fresh"] assert plan["inventory_only_space_ids"] == ["missing"] assert plan["spaces"][0]["source_record_fresh"] is False assert plan["spaces"][0]["gaps"] == [{"id": "missing", "reason": "space_not_refreshed"}] assert not plan["spaces"][0]["captured_graph_complete"] assert plan["spaces"][1]["captured_graph_complete"] @pytest.mark.parametrize("value", [None, True, "2026-10-27T12:00:00", "invalid"]) def test_missing_or_timezone_ambiguous_source_dates_are_not_fabricated(value): with pytest.raises(ValueError): source_timestamp(value) @pytest.mark.django_db def test_import_preserves_creation_date_in_db_pages_and_chunks_after_edit(tmp_path, settings): settings.CONTENT_ROOT = str(tmp_path / "content") tree = _write_sample_tree(tmp_path / "export") owner = make_member("date-owner") for filename, timestamp in [("index", "2020-01-02T03:04:05.123Z"), ("notes", "2021-02-03T04:05:06.456Z"), ("chunk", "2022-03-04T05:06:07.789Z")]: (tree / "pages" / (filename + ".md")).write_text( f'---\ntitle: "{filename}"\nsource_created_at: "{timestamp}"\n' 'parent: "bubble:page"\n---\n\nOriginal\n' ) import_space_tree(tree, tenant_id=owner.tenant_id, member=owner) space = Space.objects.get(slug="sample-space") assert space.created_at == datetime(2020, 1, 2, 3, 4, 5, 123000, tzinfo=UTC) from coulomb_social.apps.spaces import pageops for slug in ["notes", "chunk"]: before = pageops.read_page(space.slug, slug) put_space_page(owner, space, slug, title=slug, body="Edited") assert pageops.read_page(space.slug, slug).meta["source_created_at"] == ( before.meta["source_created_at"] ) import_space_tree(tree, tenant_id=owner.tenant_id, member=owner, replace=True) space.refresh_from_db() assert space.created_at.year == 2020 @pytest.mark.django_db def test_invalid_import_date_refuses_before_content_or_database_writes(tmp_path, settings): from coulomb_social.apps.spaces import pageops settings.CONTENT_ROOT = str(tmp_path / "content") tree = _write_sample_tree(tmp_path / "export") with (tree / "pages/index.md").open("w") as stream: stream.write('---\nsource_created_at: "unknown"\n---\nBody\n') with pytest.raises(pageops.PageOpsError, match="source creation timestamp"): import_space_tree(tree, tenant_id="tenant:a") assert not Space.objects.exists() assert not pageops.content_root().exists() @pytest.mark.django_db @pytest.mark.parametrize("missing", ["index", "notes", "index-file"]) def test_bubble_import_requires_every_original_date_before_any_writes(tmp_path, settings, missing): from coulomb_social.apps.spaces import pageops settings.CONTENT_ROOT = str(tmp_path / "content") tree = _write_sample_tree(tmp_path / "export") path = tree / "pages" / ("index.md" if missing == "index-file" else missing + ".md") if missing == "index-file": path.unlink() else: path.write_text("\n".join(line for line in path.read_text().splitlines() if not line.startswith("source_created_at:")) + "\n") with pytest.raises(pageops.PageOpsError, match="original creation dates"): import_space_tree(tree, tenant_id="tenant:a") assert not Space.objects.exists() assert not pageops.content_root().exists()