tegwick
789e69902d
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 7m5s
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
151 lines
7 KiB
Python
151 lines
7 KiB
Python
"""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()
|