"""Phase 01 TDD — ``muse social`` CLI surface. RED → GREEN cycle. Run before implementation to confirm failures, then implement until all pass. Test tiers ---------- TestSocialCliShape — command registered, all subcommands in --help, all run_* importable, module docstring present. TestSocialStateHelpers — pure state-layer logic: write_post, write_follow, write_reaction, read_profile — no repo, no network. TestSocialCliPost — muse social post: JSON output, file written, body stored, reply_to wired, dedup by content. TestSocialCliFollow — follow/unfollow: file created/deleted, idempotent. TestSocialCliTimeline — timeline: sorted by created_at desc, --limit, reply_to threading, empty feed. TestSocialCliReact — react: file written, emoji stored. TestSocialCliProfile — profile show (no args) and profile set. TestSocialCliDocstrings — all public run_* and helpers carry docstrings. """ from __future__ import annotations import hashlib import json import pathlib import sys import time import types import pytest # --------------------------------------------------------------------------- # Invoke helper — same pattern as test_mist_cli.py # --------------------------------------------------------------------------- def invoke_social(args: list[str]) -> tuple[int, str, str]: """Run ``muse social `` in-process and capture stdout/stderr.""" from io import StringIO old_stdout, old_stderr = sys.stdout, sys.stderr sys.stdout = out = StringIO() sys.stderr = err = StringIO() exit_code = 0 try: from muse.cli.app import main main(["social"] + args) except SystemExit as exc: exit_code = int(exc.code) if exc.code is not None else 0 finally: sys.stdout = old_stdout sys.stderr = old_stderr return exit_code, out.getvalue(), err.getvalue() # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- @pytest.fixture() def state_dir(tmp_path: pathlib.Path) -> pathlib.Path: """Minimal social repo state directory.""" (tmp_path / "posts").mkdir() (tmp_path / "reactions").mkdir() (tmp_path / "graph" / "follows").mkdir(parents=True) return tmp_path # =========================================================================== # Shape — command registered, subcommands visible, symbols importable # =========================================================================== class TestSocialCliShape: def test_social_registered_in_app(self) -> None: code, out, err = invoke_social(["--help"]) # --help exits 0 and prints usage; if unregistered it exits non-zero assert code == 0 def test_help_lists_post_subcommand(self) -> None: _, out, _ = invoke_social(["--help"]) assert "post" in out def test_help_lists_follow_subcommand(self) -> None: _, out, _ = invoke_social(["--help"]) assert "follow" in out def test_help_lists_unfollow_subcommand(self) -> None: _, out, _ = invoke_social(["--help"]) assert "unfollow" in out def test_help_lists_timeline_subcommand(self) -> None: _, out, _ = invoke_social(["--help"]) assert "timeline" in out def test_help_lists_react_subcommand(self) -> None: _, out, _ = invoke_social(["--help"]) assert "react" in out def test_help_lists_profile_subcommand(self) -> None: _, out, _ = invoke_social(["--help"]) assert "profile" in out def test_run_functions_importable(self) -> None: from muse.cli.commands.social import ( run_post, run_follow, run_unfollow, run_timeline, run_react, run_profile, ) for fn in (run_post, run_follow, run_unfollow, run_timeline, run_react, run_profile): assert callable(fn) def test_register_importable(self) -> None: from muse.cli.commands.social import register assert callable(register) def test_module_has_docstring(self) -> None: import muse.cli.commands.social as mod assert mod.__doc__ and len(mod.__doc__.strip()) > 20 # =========================================================================== # State helpers — pure logic, no repo, no network # =========================================================================== class TestSocialStateHelpers: def test_build_post_returns_dict_with_required_keys(self) -> None: from muse.cli.commands.social import _build_post post = _build_post(body="hello muse") assert "body" in post assert "created_at" in post assert "post_id" in post def test_build_post_body_stored(self) -> None: from muse.cli.commands.social import _build_post post = _build_post(body="hello muse") assert post["body"] == "hello muse" def test_build_post_post_id_is_sha256_prefix(self) -> None: from muse.cli.commands.social import _build_post post = _build_post(body="hello muse") assert post["post_id"].startswith("sha256:") def test_build_post_reply_to_stored(self) -> None: from muse.cli.commands.social import _build_post post = _build_post(body="reply text", reply_to="sha256:abc123") assert post["reply_to"] == "sha256:abc123" def test_build_post_reply_to_defaults_none(self) -> None: from muse.cli.commands.social import _build_post post = _build_post(body="hello") assert post.get("reply_to") is None def test_build_post_deterministic(self) -> None: from muse.cli.commands.social import _build_post p1 = _build_post(body="same text", created_at="2026-05-01T00:00:00Z") p2 = _build_post(body="same text", created_at="2026-05-01T00:00:00Z") assert p1["post_id"] == p2["post_id"] def test_write_post_creates_file( self, state_dir: pathlib.Path ) -> None: from muse.cli.commands.social import _build_post, _write_post post = _build_post(body="hello muse", created_at="2026-05-01T00:00:00Z") path = _write_post(state_dir, post) assert path.exists() def test_write_post_file_is_valid_json( self, state_dir: pathlib.Path ) -> None: from muse.cli.commands.social import _build_post, _write_post post = _build_post(body="hello muse", created_at="2026-05-01T00:00:00Z") path = _write_post(state_dir, post) data = json.loads(path.read_text()) assert data["body"] == "hello muse" def test_write_post_file_under_posts_dir( self, state_dir: pathlib.Path ) -> None: from muse.cli.commands.social import _build_post, _write_post post = _build_post(body="hello", created_at="2026-05-01T00:00:00Z") path = _write_post(state_dir, post) assert path.parent.parent.name == "posts" def test_write_post_idempotent(self, state_dir: pathlib.Path) -> None: from muse.cli.commands.social import _build_post, _write_post post = _build_post(body="hello", created_at="2026-05-01T00:00:00Z") p1 = _write_post(state_dir, post) p2 = _write_post(state_dir, post) assert p1 == p2 assert len(list((state_dir / "posts").iterdir())) == 1 def test_write_follow_creates_file(self, state_dir: pathlib.Path) -> None: from muse.cli.commands.social import _write_follow path = _write_follow(state_dir, "alice") assert path.exists() def test_write_follow_filename_contains_handle( self, state_dir: pathlib.Path ) -> None: from muse.cli.commands.social import _write_follow path = _write_follow(state_dir, "alice") assert "alice" in path.name def test_write_follow_content_has_handle( self, state_dir: pathlib.Path ) -> None: from muse.cli.commands.social import _write_follow path = _write_follow(state_dir, "alice") data = json.loads(path.read_text()) assert data["handle"] == "alice" def test_delete_follow_removes_file(self, state_dir: pathlib.Path) -> None: from muse.cli.commands.social import _write_follow, _delete_follow _write_follow(state_dir, "alice") _delete_follow(state_dir, "alice") assert not (state_dir / "graph" / "follows" / "alice.json").exists() def test_delete_follow_nonexistent_is_noop( self, state_dir: pathlib.Path ) -> None: from muse.cli.commands.social import _delete_follow # Should not raise _delete_follow(state_dir, "nobody") def test_write_reaction_creates_file(self, state_dir: pathlib.Path) -> None: from muse.cli.commands.social import _write_reaction path = _write_reaction(state_dir, post_id="sha256:aaa", emoji="❤️") assert path.exists() def test_write_reaction_content_has_emoji( self, state_dir: pathlib.Path ) -> None: from muse.cli.commands.social import _write_reaction path = _write_reaction(state_dir, post_id="sha256:aaa", emoji="❤️") data = json.loads(path.read_text()) assert data["emoji"] == "❤️" def test_write_reaction_content_has_post_id( self, state_dir: pathlib.Path ) -> None: from muse.cli.commands.social import _write_reaction path = _write_reaction(state_dir, post_id="sha256:aaa", emoji="❤️") data = json.loads(path.read_text()) assert data["post_id"] == "sha256:aaa" def test_read_profile_returns_none_when_missing( self, state_dir: pathlib.Path ) -> None: from muse.cli.commands.social import _read_profile assert _read_profile(state_dir) is None def test_write_read_profile_roundtrip(self, state_dir: pathlib.Path) -> None: from muse.cli.commands.social import _write_profile, _read_profile _write_profile(state_dir, {"handle": "gabriel", "bio": "building the future"}) profile = _read_profile(state_dir) assert profile is not None assert profile["handle"] == "gabriel" assert profile["bio"] == "building the future" def test_load_posts_empty_dir(self, state_dir: pathlib.Path) -> None: from muse.cli.commands.social import _load_posts posts = _load_posts(state_dir) assert posts == [] def test_load_posts_returns_list(self, state_dir: pathlib.Path) -> None: from muse.cli.commands.social import _build_post, _write_post, _load_posts post = _build_post(body="hello", created_at="2026-05-01T00:00:00Z") _write_post(state_dir, post) posts = _load_posts(state_dir) assert len(posts) == 1 def test_load_posts_sorted_newest_first(self, state_dir: pathlib.Path) -> None: from muse.cli.commands.social import _build_post, _write_post, _load_posts older = _build_post(body="older", created_at="2026-01-01T00:00:00Z") newer = _build_post(body="newer", created_at="2026-06-01T00:00:00Z") _write_post(state_dir, older) _write_post(state_dir, newer) posts = _load_posts(state_dir) assert posts[0]["body"] == "newer" assert posts[1]["body"] == "older" # =========================================================================== # CLI: post # =========================================================================== class TestSocialCliPost: def test_post_help_exits_zero(self) -> None: code, _, _ = invoke_social(["post", "--help"]) assert code == 0 def test_post_json_output_has_post_id(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.social import _build_post, _write_post post = _build_post(body="test post", created_at="2026-05-01T00:00:00Z") (tmp_path / "posts").mkdir() (tmp_path / "reactions").mkdir() (tmp_path / "graph" / "follows").mkdir(parents=True) path = _write_post(tmp_path, post) data = json.loads(path.read_text()) assert "post_id" in data def test_post_json_has_body(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.social import _build_post, _write_post post = _build_post(body="hello world", created_at="2026-05-01T00:00:00Z") (tmp_path / "posts").mkdir() path = _write_post(tmp_path, post) data = json.loads(path.read_text()) assert data["body"] == "hello world" def test_post_with_reply_to_stored(self, tmp_path: pathlib.Path) -> None: from muse.cli.commands.social import _build_post, _write_post post = _build_post( body="great point", reply_to="sha256:abc", created_at="2026-05-01T00:00:00Z" ) (tmp_path / "posts").mkdir() path = _write_post(tmp_path, post) data = json.loads(path.read_text()) assert data["reply_to"] == "sha256:abc" def test_post_missing_body_arg_exits_nonzero(self) -> None: code, _, err = invoke_social(["post"]) assert code != 0 # =========================================================================== # CLI: follow / unfollow # =========================================================================== class TestSocialCliFollow: def test_follow_help_exits_zero(self) -> None: code, _, _ = invoke_social(["follow", "--help"]) assert code == 0 def test_unfollow_help_exits_zero(self) -> None: code, _, _ = invoke_social(["unfollow", "--help"]) assert code == 0 def test_write_follow_idempotent(self, state_dir: pathlib.Path) -> None: from muse.cli.commands.social import _write_follow p1 = _write_follow(state_dir, "alice") p2 = _write_follow(state_dir, "alice") assert p1 == p2 assert len(list((state_dir / "graph" / "follows").iterdir())) == 1 def test_follow_multiple_handles(self, state_dir: pathlib.Path) -> None: from muse.cli.commands.social import _write_follow _write_follow(state_dir, "alice") _write_follow(state_dir, "bob") _write_follow(state_dir, "carol") follows = list((state_dir / "graph" / "follows").iterdir()) assert len(follows) == 3 def test_unfollow_after_follow_removes_file( self, state_dir: pathlib.Path ) -> None: from muse.cli.commands.social import _write_follow, _delete_follow _write_follow(state_dir, "alice") _delete_follow(state_dir, "alice") assert not (state_dir / "graph" / "follows" / "alice.json").exists() def test_follow_missing_handle_exits_nonzero(self) -> None: code, _, _ = invoke_social(["follow"]) assert code != 0 def test_unfollow_missing_handle_exits_nonzero(self) -> None: code, _, _ = invoke_social(["unfollow"]) assert code != 0 # =========================================================================== # CLI: timeline # =========================================================================== class TestSocialCliTimeline: def test_timeline_help_exits_zero(self) -> None: code, _, _ = invoke_social(["timeline", "--help"]) assert code == 0 def test_load_posts_empty_returns_empty_list( self, state_dir: pathlib.Path ) -> None: from muse.cli.commands.social import _load_posts assert _load_posts(state_dir) == [] def test_timeline_sorted_newest_first( self, state_dir: pathlib.Path ) -> None: from muse.cli.commands.social import _build_post, _write_post, _load_posts for i, ts in enumerate( ["2026-01-01T00:00:00Z", "2026-03-01T00:00:00Z", "2026-06-01T00:00:00Z"] ): _write_post(state_dir, _build_post(body=f"post {i}", created_at=ts)) posts = _load_posts(state_dir) timestamps = [p["created_at"] for p in posts] assert timestamps == sorted(timestamps, reverse=True) def test_timeline_limit(self, state_dir: pathlib.Path) -> None: from muse.cli.commands.social import _build_post, _write_post, _load_posts for i in range(5): _write_post( state_dir, _build_post(body=f"post {i}", created_at=f"2026-0{i+1}-01T00:00:00Z"), ) posts = _load_posts(state_dir, limit=3) assert len(posts) == 3 def test_replies_have_reply_to_field(self, state_dir: pathlib.Path) -> None: from muse.cli.commands.social import _build_post, _write_post, _load_posts root = _build_post(body="root post", created_at="2026-01-01T00:00:00Z") reply = _build_post( body="reply", created_at="2026-01-01T01:00:00Z", reply_to=root["post_id"], ) _write_post(state_dir, root) _write_post(state_dir, reply) posts = _load_posts(state_dir) reply_post = next(p for p in posts if p["body"] == "reply") assert reply_post["reply_to"] == root["post_id"] # =========================================================================== # CLI: react # =========================================================================== class TestSocialCliReact: def test_react_help_exits_zero(self) -> None: code, _, _ = invoke_social(["react", "--help"]) assert code == 0 def test_react_creates_file_in_reactions_dir( self, state_dir: pathlib.Path ) -> None: from muse.cli.commands.social import _write_reaction _write_reaction(state_dir, post_id="sha256:aaa", emoji="❤️") reactions = list((state_dir / "reactions").iterdir()) assert len(reactions) == 1 def test_react_different_emojis_different_files( self, state_dir: pathlib.Path ) -> None: from muse.cli.commands.social import _write_reaction _write_reaction(state_dir, post_id="sha256:aaa", emoji="❤️") _write_reaction(state_dir, post_id="sha256:aaa", emoji="🔥") reactions = list((state_dir / "reactions").rglob("*.json")) assert len(reactions) == 2 def test_react_same_emoji_idempotent(self, state_dir: pathlib.Path) -> None: from muse.cli.commands.social import _write_reaction _write_reaction(state_dir, post_id="sha256:aaa", emoji="❤️") _write_reaction(state_dir, post_id="sha256:aaa", emoji="❤️") reactions = list((state_dir / "reactions").rglob("*.json")) assert len(reactions) == 1 def test_react_missing_args_exits_nonzero(self) -> None: code, _, _ = invoke_social(["react"]) assert code != 0 # =========================================================================== # CLI: profile # =========================================================================== class TestSocialCliProfile: def test_profile_help_exits_zero(self) -> None: code, _, _ = invoke_social(["profile", "--help"]) assert code == 0 def test_write_profile_creates_file(self, state_dir: pathlib.Path) -> None: from muse.cli.commands.social import _write_profile _write_profile(state_dir, {"handle": "gabriel", "bio": "hi"}) assert (state_dir / "profile.json").exists() def test_write_profile_content_stored(self, state_dir: pathlib.Path) -> None: from muse.cli.commands.social import _write_profile, _read_profile _write_profile(state_dir, {"handle": "gabriel", "bio": "building the future"}) profile = _read_profile(state_dir) assert profile["bio"] == "building the future" def test_profile_set_updates_bio(self, state_dir: pathlib.Path) -> None: from muse.cli.commands.social import _write_profile, _read_profile _write_profile(state_dir, {"handle": "gabriel", "bio": "old bio"}) existing = _read_profile(state_dir) existing["bio"] = "new bio" _write_profile(state_dir, existing) assert _read_profile(state_dir)["bio"] == "new bio" # =========================================================================== # Docstrings # =========================================================================== class TestSocialCliDocstrings: def test_module_has_docstring(self) -> None: import muse.cli.commands.social as mod assert mod.__doc__ and len(mod.__doc__.strip()) > 20 def test_run_functions_have_docstrings(self) -> None: from muse.cli.commands import social for name in ("run_post", "run_follow", "run_unfollow", "run_timeline", "run_react", "run_profile", "register"): fn = getattr(social, name) assert fn.__doc__ and len(fn.__doc__.strip()) > 10, \ f"{name}() missing docstring" def test_helper_functions_have_docstrings(self) -> None: from muse.cli.commands import social for name in ("_build_post", "_write_post", "_write_follow", "_delete_follow", "_write_reaction", "_read_profile", "_write_profile", "_load_posts"): fn = getattr(social, name) assert fn.__doc__ and len(fn.__doc__.strip()) > 10, \ f"{name}() missing docstring"