tegwick
ed2c8e2f4d
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 2m59s
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
168 lines
8 KiB
Python
168 lines
8 KiB
Python
"""Synthetic source bodies only; offline preparation must never imply acceptance."""
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from coulomb_social.apps.spaces.bubble_bundle import prepare_bundle
|
|
from coulomb_social.apps.spaces.import_tree import import_space_tree
|
|
from coulomb_social.apps.spaces.models import Space
|
|
from tests.test_space_permissions import make_member
|
|
|
|
|
|
def source(created, **kwargs):
|
|
return {"Created Date": created, "Modified Date": created, **kwargs}
|
|
|
|
|
|
def capture():
|
|
return ({"s": source(1700000000123, _chunk_custom_article="r")}, {
|
|
"r": source(1700000001456, content_text="\n\n Root\r\n\n", nomer_text="Synthetic",
|
|
chunks_list_custom_article=["c"], space_custom_space="s"),
|
|
"c": source(1700000002789, content_text="Child \n", context_custom_article="r"),
|
|
})
|
|
|
|
|
|
def test_bundle_deterministic_preserves_bodies_dates_and_ordered_relations():
|
|
spaces, articles = capture()
|
|
articles["r"]["chunks_list_custom_article"] = ["c", "c"]
|
|
report, files = prepare_bundle(spaces, articles)
|
|
reversed_report, reversed_files = prepare_bundle(
|
|
dict(reversed(list(spaces.items()))), dict(reversed(list(articles.items()))))
|
|
assert (report, files) == (reversed_report, reversed_files)
|
|
entry = report["spaces"][0]
|
|
assert entry["created_at_raw"] == 1700000000123
|
|
assert b"[Source root](?page=article-" in files[entry["tree"] + "/pages/index.md"]
|
|
for page in entry["articles"]:
|
|
raw = articles[page["id"]]["content_text"].encode()
|
|
assert files[page["file"]].endswith(b"\n---\n" + raw)
|
|
assert page["body_sha256"] == hashlib.sha256(raw).hexdigest()
|
|
assert f'source_created_at: {articles[page["id"]]["Created Date"]}'.encode() in (
|
|
files[page["file"]])
|
|
root = next(a for a in report["articles"] if a["id"] == "r")
|
|
assert root["relations"]["chunks_list_custom_article"] == ["c", "c"]
|
|
assert not entry["import_ready"]
|
|
assert not report["production_imported"]
|
|
|
|
|
|
def test_missing_dates_bodies_and_inventory_only_records_remain_queued_without_trees():
|
|
spaces, articles = capture()
|
|
del articles["c"]["Created Date"]
|
|
articles["orphan"] = {"content_text": "Orphan original"}
|
|
report, files = prepare_bundle(spaces, articles, inventory_spaces={"old": {}})
|
|
assert len(report["spaces"]) == 2
|
|
assert all(s["tree"] is None for s in report["spaces"])
|
|
assert not any(p.startswith("review-trees/") for p in files)
|
|
assert report["unassigned_article_ids"] == ["orphan"]
|
|
assert len([p for p in files if p.startswith("source-bodies/")]) == 3
|
|
|
|
|
|
def test_reference_images_html_relations_and_attachments_are_unresolved_and_private():
|
|
spaces, articles = capture()
|
|
body = ('[root](https://coulomb.social/pg_chunk/r?q=1#part)\n'
|
|
'![image][visual]\n\n[visual]: https://assets.example/synthetic.png\n\n'
|
|
'<img src="./local.png" srcset="large.png 2x">'
|
|
'<a href="https://coulomb.social.attacker.example/pg_chunk/r">lookalike</a>'
|
|
'<iframe src="https://example.invalid/embed"></iframe>')
|
|
articles["r"].update(content_text=body, contents_list_custom_article=["missing"],
|
|
logo_image="https://assets.example/logo.webp", hasattachments_boolean=True)
|
|
report, files = prepare_bundle(spaces, articles)
|
|
detail = next(a for a in report["articles"] if a["id"] == "r")
|
|
refs = detail["references"]
|
|
root = next(r for r in refs if r["classification"] == "bubble_article")
|
|
assert root["target_status"] == "unique"
|
|
assert root["query"] == "q=1" and root["fragment"] == "part"
|
|
assert {r["kind"] for r in refs} == {"asset", "visual", "link"}
|
|
assert next(r for r in refs if "attacker.example" in r["source"])["classification"] == (
|
|
"external")
|
|
assert all(not r["resolved"] for r in refs)
|
|
entry = report["spaces"][0]
|
|
assert "uncaptured_relation_records" in entry["blockers"]
|
|
assert "attachment_records_require_capture" in detail["review_flags"]
|
|
assert "srcset_requires_review" in detail["review_flags"]
|
|
assert "iframe_requires_review" in detail["review_flags"]
|
|
assert any(data.endswith(body.encode()) for data in files.values())
|
|
assert report["network_requests"] == 0
|
|
|
|
|
|
def test_destination_names_survive_label_changes_and_additional_records():
|
|
spaces, articles = capture()
|
|
report, _ = prepare_bundle(spaces, articles)
|
|
articles["r"]["nomer_text"] = "../../index"
|
|
articles["extra"] = source(1700000001000, space_custom_space="s", content_text="Extra")
|
|
later, files = prepare_bundle(spaces, articles)
|
|
assert report["spaces"][0]["slug"] == later["spaces"][0]["slug"]
|
|
paths = {p["id"]: p["file"] for p in later["spaces"][0]["articles"]}
|
|
assert all(p["file"] == paths[p["id"]] for p in report["spaces"][0]["articles"])
|
|
assert all(".." not in Path(path).parts for path in files)
|
|
|
|
|
|
def test_shared_source_reference_never_selects_an_arbitrary_destination():
|
|
spaces, articles = capture()
|
|
spaces["another"] = source(1700000000000, _chunk_custom_article="c")
|
|
articles["r"]["content_text"] = "[shared](/pg_chunk/c)"
|
|
report, _ = prepare_bundle(spaces, articles)
|
|
root = next(a for a in report["articles"] if a["id"] == "r")
|
|
reference = root["references"][0]
|
|
assert reference["target_status"] == "ambiguous"
|
|
assert len(reference["candidate_targets"]) == 2
|
|
assert not reference["resolved"]
|
|
|
|
|
|
def test_relation_id_placeholder_does_not_count_as_a_complete_source_record():
|
|
spaces, articles = capture()
|
|
articles["r"]["contents_list_custom_article"] = ["stub"]
|
|
articles["stub"] = {"_id": "stub"}
|
|
report, _ = prepare_bundle(spaces, articles)
|
|
entry = report["spaces"][0]
|
|
root = next(a for a in entry["articles"] if a["id"] == "r")
|
|
assert root["incomplete_relation_ids"] == ["stub"]
|
|
assert not root["missing_relation_ids"]
|
|
assert "incomplete_relation_records" in entry["blockers"]
|
|
|
|
|
|
@pytest.mark.django_db
|
|
def test_import_keeps_distinct_space_root_and_child_creation_dates(tmp_path, settings):
|
|
spaces, articles = capture()
|
|
report, files = prepare_bundle(spaces, articles)
|
|
for name, data in files.items():
|
|
path = tmp_path / name
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_bytes(data)
|
|
settings.CONTENT_ROOT = str(tmp_path / "content")
|
|
owner = make_member("synthetic-bundle-owner")
|
|
entry = report["spaces"][0]
|
|
import_space_tree(tmp_path / entry["tree"], tenant_id=owner.tenant_id, member=owner)
|
|
space = Space.objects.get(slug=entry["slug"])
|
|
assert space.created_at.isoformat() == "2023-11-14T22:13:20.123000+00:00"
|
|
for page in entry["articles"]:
|
|
actual = Path(settings.CONTENT_ROOT) / "spaces" / entry["slug"] / "pages" / (
|
|
Path(page["file"]).name)
|
|
assert actual.read_bytes() == files[page["file"]]
|
|
|
|
|
|
def test_cli_private_output_completion_hashes_no_overwrite_or_checkout_output(tmp_path):
|
|
repo = Path(__file__).resolve().parents[1]
|
|
capture_dir = tmp_path / "capture"
|
|
capture_dir.mkdir()
|
|
spaces, articles = capture()
|
|
for name, data in [("spaces", spaces), ("articles", articles)]:
|
|
(capture_dir / (name + ".json")).write_text(json.dumps(data))
|
|
output = tmp_path / "bundle"
|
|
cmd = [sys.executable, str(repo / "scripts/prepare-bubble-bundle.py"), str(capture_dir),
|
|
"--output", str(output)]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
assert result.returncode == 0, result.stderr
|
|
receipt = json.loads((output / "reconciliation.json").read_text())
|
|
assert receipt["status"] == "completed_offline_preparation"
|
|
assert "Synthetic" not in result.stdout
|
|
assert os.stat(output).st_mode & 0o777 == 0o700
|
|
assert all(os.stat(p).st_mode & 0o777 == 0o600 for p in output.rglob("*") if p.is_file())
|
|
assert subprocess.run(cmd, capture_output=True).returncode == 1
|
|
cmd[-1] = str(repo / "var/test-bundle-refusal")
|
|
assert subprocess.run(cmd, capture_output=True).returncode == 1
|
|
assert not Path(cmd[-1]).exists()
|