tegwick
1af8740de4
Some checks failed
Use-case regression gate / usecase-regressions (push) Failing after 1m59s
Assistant: codex Assistant-Model: gpt-6.1-sol Assistant-Session: 01a17f16-d99c-7ea3-95f0-8212897fba6b
192 lines
8.8 KiB
Python
192 lines
8.8 KiB
Python
"""Synthetic observations only; date plans must preserve local edits/copies."""
|
|
|
|
import copy
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
from coulomb_social.apps.spaces.date_backfill import plan_date_backfill, prepare_index_date_patch
|
|
|
|
|
|
def fixtures():
|
|
digest = hashlib.sha256(b"Synthetic edited bytes").hexdigest()
|
|
selection = {"space_id": "destination-id", "space_slug": "accepted",
|
|
"tenant_id": "tenant:synthetic", "source_space_id": "source"}
|
|
pages = [
|
|
{"slug": "index", "sha256": digest,
|
|
"meta": {"id": "bubble-space:source", "bubble_type": "space"}},
|
|
{"slug": "notes", "sha256": digest,
|
|
"meta": {"id": "bubble:article", "source_created_at": 1700000001000}},
|
|
{"slug": "copy", "sha256": digest,
|
|
"meta": {"id": "bubble:article", "source_created_at": 1700000001000}},
|
|
{"slug": "local-notes", "sha256": digest, "meta": {}},
|
|
]
|
|
observation = {"schema": "coulomb.source-date-observation.v1", "spaces": [{
|
|
"id": "destination-id", "slug": "accepted", "tenant_id": "tenant:synthetic",
|
|
"created_at": "2026-10-06T12:00:00+00:00", "pages": pages,
|
|
}]}
|
|
spaces = {"source": {"Created Date": 1700000000123}}
|
|
articles = {"article": {"Created Date": 1700000001000}}
|
|
return observation, spaces, articles, selection
|
|
|
|
|
|
def test_plan_only_changes_index_field_and_sql_date_keeps_copies_edits_and_local_pages():
|
|
data = fixtures()
|
|
before = copy.deepcopy(data)
|
|
plan = plan_date_backfill(*data)
|
|
assert data == before
|
|
assert plan["prepared_change_set_valid"]
|
|
assert [c["operation"] for c in plan["proposed_changes"]] == [
|
|
"update_space_created_at", "insert_frontmatter_field"]
|
|
assert plan["proposed_changes"][1]["preserve_all_existing_metadata_and_body_bytes"]
|
|
assert plan["proposed_changes"][0]["value"] == "2023-11-14T22:13:20.123000+00:00"
|
|
assert sum(p["duplicate_provenance"] for p in plan["pages"]) == 2
|
|
assert all(not p["proposed_changes"] for p in plan["pages"])
|
|
assert plan["pages"][-1]["status"] == "local_page_preserved"
|
|
assert not plan["ready_to_apply"] and not plan["production_writes"]
|
|
assert plan["requires_fresh_paired_backup"] and plan["requires_live_hash_and_sql_preconditions"]
|
|
|
|
|
|
@pytest.mark.parametrize("field", ["space_id", "space_slug", "tenant_id", "source_space_id"])
|
|
def test_ambiguous_or_wrong_destination_is_not_inferred_from_provenance(field):
|
|
observation, spaces, articles, selection = fixtures()
|
|
selection[field] = "different"
|
|
with pytest.raises(ValueError):
|
|
plan_date_backfill(observation, spaces, articles, selection)
|
|
|
|
|
|
def test_existing_index_conflict_requires_review_and_never_proposes_overwriting_it():
|
|
observation, spaces, articles, selection = fixtures()
|
|
observation["spaces"][0]["pages"][0]["meta"]["source_created_at"] = 1600000000000
|
|
plan = plan_date_backfill(observation, spaces, articles, selection)
|
|
assert plan["blockers"] == ["existing_index_date_conflicts_with_source"]
|
|
assert not plan["prepared_change_set_valid"]
|
|
assert all(c["operation"] != "insert_frontmatter_field" for c in plan["proposed_changes"])
|
|
|
|
|
|
def test_article_date_gap_is_reported_without_rewriting_missing_source_or_local_copies():
|
|
observation, spaces, articles, selection = fixtures()
|
|
del observation["spaces"][0]["pages"][1]["meta"]["source_created_at"]
|
|
plan = plan_date_backfill(observation, spaces, articles, selection)
|
|
assert "article_date_missing" in plan["blockers"]
|
|
assert all(not p["proposed_changes"] for p in plan["pages"])
|
|
|
|
|
|
def test_numeric_prefixed_original_index_reference_matches_normalized_source():
|
|
observation, _, articles, selection = fixtures()
|
|
identity = "1700000000000x12345678"
|
|
selection["source_space_id"] = identity
|
|
observation["spaces"][0]["pages"][0]["meta"]["id"] = (
|
|
"bubble-space:1234__LOOKUP__" + identity)
|
|
plan = plan_date_backfill(observation, {identity: {"Created Date": 1700000000123}},
|
|
articles, selection)
|
|
assert plan["prepared_change_set_valid"]
|
|
|
|
|
|
def test_cli_private_plan_hashes_and_no_overwrite_or_checkout_output(tmp_path):
|
|
observation, spaces, articles, selection = fixtures()
|
|
capture = tmp_path / "capture"
|
|
capture.mkdir()
|
|
for name, data in [("spaces", spaces), ("articles", articles)]:
|
|
(capture / (name + ".json")).write_text(json.dumps(data))
|
|
observed = tmp_path / "observation.json"
|
|
observed.write_text(json.dumps(observation))
|
|
selected = tmp_path / "selection.json"
|
|
selected.write_text(json.dumps(selection))
|
|
output = tmp_path / "plan.json"
|
|
root = Path(__file__).resolve().parents[1]
|
|
cmd = [sys.executable, str(root / "scripts/plan-source-date-backfill.py"), str(capture),
|
|
"--observation", str(observed), "--selection", str(selected), "--output", str(output)]
|
|
result = subprocess.run(cmd, capture_output=True, text=True)
|
|
assert result.returncode == 0, result.stderr
|
|
assert "destination-id" not in result.stdout
|
|
plan = json.loads(output.read_text())
|
|
assert plan["source_sha256"]["observation"] == hashlib.sha256(observed.read_bytes()).hexdigest()
|
|
assert os.stat(output).st_mode & 0o777 == 0o600
|
|
assert subprocess.run(cmd, capture_output=True).returncode == 1
|
|
cmd[-1] = str(root / "var/refused-date-plan.json")
|
|
assert subprocess.run(cmd, capture_output=True).returncode == 1
|
|
assert not Path(cmd[-1]).exists()
|
|
|
|
|
|
def patch_fixture():
|
|
data = fixtures()
|
|
original = ('---\n# Keep comment\nid: "bubble-space:source"\n'
|
|
'title: "Locally edited — title"\n---\n\n\n# Local edit\n\n'
|
|
' whitespace and [alias](copy)\n\n').encode()
|
|
data[0]["spaces"][0]["pages"][0]["sha256"] = hashlib.sha256(original).hexdigest()
|
|
return plan_date_backfill(*data), original
|
|
|
|
|
|
def test_index_date_patch_preserves_every_original_byte_and_does_not_mutate_plan():
|
|
plan, original = patch_fixture()
|
|
before = copy.deepcopy(plan)
|
|
patch = prepare_index_date_patch(plan, original)
|
|
inserted = b"source_created_at: 1700000000123\n"
|
|
assert patch == original[:4] + inserted + original[4:]
|
|
assert patch.replace(inserted, b"", 1) == original
|
|
assert plan == before
|
|
with pytest.raises(ValueError, match="bytes changed"):
|
|
prepare_index_date_patch(plan, patch)
|
|
|
|
|
|
@pytest.mark.parametrize("change", ["sql_target", "sql_value", "file", "extra", "blocker"])
|
|
def test_patch_refuses_expanded_or_conflicting_change_sets(change):
|
|
plan, original = patch_fixture()
|
|
if change == "sql_target":
|
|
plan["proposed_changes"][0]["tenant_id"] = "other"
|
|
elif change == "sql_value":
|
|
plan["proposed_changes"][0]["value"] = "2020-01-01T00:00:00Z"
|
|
elif change == "file":
|
|
plan["proposed_changes"][1]["file"] = "../other/index.md"
|
|
elif change == "extra":
|
|
plan["proposed_changes"].append({"operation": "replace_content"})
|
|
else:
|
|
plan["blockers"] = ["article_date_conflict"]
|
|
with pytest.raises(ValueError):
|
|
prepare_index_date_patch(plan, original)
|
|
|
|
|
|
@pytest.mark.parametrize("prefix", [
|
|
b"source_created_at: 1600000000000\n",
|
|
b'id: "bubble-space:other"\n',
|
|
b'title: "duplicate"\n',
|
|
])
|
|
def test_patch_refuses_existing_dates_ambiguous_keys_and_wrong_provenance(prefix):
|
|
plan, original = patch_fixture()
|
|
original = original[:4] + prefix + original[4:]
|
|
plan["index_sha256"] = hashlib.sha256(original).hexdigest()
|
|
plan["proposed_changes"][1]["expected_sha256"] = plan["index_sha256"]
|
|
with pytest.raises(ValueError):
|
|
prepare_index_date_patch(plan, original)
|
|
|
|
|
|
@pytest.mark.parametrize("sql_already_correct", [False, True])
|
|
def test_existing_matching_index_is_preserved_for_sql_only_or_noop_plan(sql_already_correct):
|
|
data = fixtures()
|
|
original = b'---\nid: "bubble-space:source"\nsource_created_at: 1700000000123\n---\nBODY\n'
|
|
target = data[0]["spaces"][0]
|
|
index = target["pages"][0]
|
|
index["sha256"] = hashlib.sha256(original).hexdigest()
|
|
index["meta"]["source_created_at"] = 1700000000123
|
|
if sql_already_correct:
|
|
target["created_at"] = "2023-11-14T22:13:20.123000+00:00"
|
|
plan = plan_date_backfill(*data)
|
|
assert prepare_index_date_patch(plan, original) == original
|
|
assert len(plan["proposed_changes"]) == int(not sql_already_correct)
|
|
|
|
|
|
def test_patch_supports_raw_iso_date_without_normalizing_body_or_other_metadata():
|
|
data = fixtures()
|
|
raw_date = "2023-11-15T00:13:20.123+02:00"
|
|
data[1]["source"]["Created Date"] = raw_date
|
|
original = b'---\nid: "bubble-space:source"\n---\n\nbody without terminal newline'
|
|
data[0]["spaces"][0]["pages"][0]["sha256"] = hashlib.sha256(original).hexdigest()
|
|
patched = prepare_index_date_patch(plan_date_backfill(*data), original)
|
|
inserted = ('source_created_at: "' + raw_date + '"\n').encode()
|
|
assert patched == original[:4] + inserted + original[4:]
|