You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
99 lines
3.0 KiB
99 lines
3.0 KiB
"""Thread root matching via ``e``/``E``/``a``/``A``/``q`` (Jumble-style).""" |
|
|
|
import tempfile |
|
from pathlib import Path |
|
from typing import Any, cast |
|
|
|
from imwald.core.database import Database, thread_root_link_targets |
|
from imwald.core.nostr_crypto import build_signed_event, pubkey_hex_from_secret |
|
|
|
|
|
def _sk() -> bytes: |
|
return bytes.fromhex("3501454135014541350145413501453fefb02227e449e57cf4d3a3ce05378683") |
|
|
|
|
|
def test_thread_root_link_targets_kind1() -> None: |
|
root = cast(dict[str, Any], {"id": "a" * 64, "kind": 1, "pubkey": "b" * 64, "tags": []}) |
|
assert thread_root_link_targets(root) == ["a" * 64] |
|
|
|
|
|
def test_thread_root_link_targets_addressable_adds_a_coordinate() -> None: |
|
pk = "c" * 64 |
|
root = cast( |
|
dict[str, Any], |
|
{"id": "d" * 64, "kind": 30023, "pubkey": pk, "tags": [["d", "slug-x"]]}, |
|
) |
|
t = thread_root_link_targets(root) |
|
assert "d" * 64 in t |
|
assert f"30023:{pk}:slug-x" in t |
|
|
|
|
|
def test_list_replies_to_matches_q_tag() -> None: |
|
sk = _sk() |
|
root = build_signed_event(sk, created_at=1, kind=1, tags=[], content="root") |
|
rid = root["id"] |
|
rep = build_signed_event( |
|
sk, |
|
created_at=2, |
|
kind=1, |
|
tags=[["q", rid]], |
|
content="quote", |
|
) |
|
with tempfile.TemporaryDirectory() as td: |
|
db = Database(Path(td) / "t.sqlite") |
|
db.connect() |
|
db.upsert_event(root) |
|
db.upsert_event(rep) |
|
got = db.list_replies_to(root, limit=20) |
|
assert len(got) == 1 and got[0]["id"] == rep["id"] |
|
|
|
|
|
def test_list_replies_to_nested_reply_tags_only_parent() -> None: |
|
"""Transitive closure: reply-to-reply often tags only the parent id, not the root.""" |
|
sk = _sk() |
|
root = build_signed_event(sk, created_at=1, kind=1, tags=[], content="root") |
|
rid = root["id"] |
|
direct = build_signed_event( |
|
sk, |
|
created_at=2, |
|
kind=1, |
|
tags=[["e", rid, "", "root"]], |
|
content="direct", |
|
) |
|
did = direct["id"] |
|
nested = build_signed_event( |
|
sk, |
|
created_at=3, |
|
kind=1, |
|
tags=[["e", did, "", "reply"]], |
|
content="nested", |
|
) |
|
with tempfile.TemporaryDirectory() as td: |
|
db = Database(Path(td) / "t3.sqlite") |
|
db.connect() |
|
db.upsert_event(root) |
|
db.upsert_event(direct) |
|
db.upsert_event(nested) |
|
got = db.list_replies_to(root, limit=20) |
|
ids = {str(x["id"]) for x in got} |
|
assert ids == {did, nested["id"]} |
|
|
|
|
|
def test_list_replies_to_matches_uppercase_e() -> None: |
|
sk = _sk() |
|
root = build_signed_event(sk, created_at=1, kind=1, tags=[], content="root") |
|
rid = root["id"] |
|
rep = build_signed_event( |
|
sk, |
|
created_at=2, |
|
kind=1111, |
|
tags=[["E", rid, "", "1" * 64], ["p", pubkey_hex_from_secret(sk)]], |
|
content="c", |
|
) |
|
with tempfile.TemporaryDirectory() as td: |
|
db = Database(Path(td) / "t2.sqlite") |
|
db.connect() |
|
db.upsert_event(root) |
|
db.upsert_event(rep) |
|
got = db.list_replies_to(root, limit=20) |
|
assert len(got) == 1 and got[0]["id"] == rep["id"]
|
|
|