test_symlog_time_resolution.py
file-level
1
files
1
commits
0
hotspots
0
🧊 dead
0
💥 blast risk
| 1 | """muse#53 item 6 -- relative time syntax for symlog @{...} refs. |
| 2 | |
| 3 | `@{1.hour.ago}`, `@{2.days.ago}`, `@{yesterday}` resolve to the most |
| 4 | recent symlog entry at or before that point in time, exercised at the |
| 5 | `resolve_symlog_addr` core level (with fabricated timestamps written |
| 6 | directly to the symlog file for determinism) and through the CLI |
| 7 | (`muse symlog resolve`). |
| 8 | """ |
| 9 | |
| 10 | from __future__ import annotations |
| 11 | |
| 12 | import datetime |
| 13 | import json |
| 14 | import os |
| 15 | import pathlib |
| 16 | import textwrap |
| 17 | |
| 18 | import pytest |
| 19 | |
| 20 | from tests.cli_test_helper import CliRunner |
| 21 | from muse.core.symlog import resolve_symlog_addr, symlog_path, NULL_CONTENT_ID |
| 22 | |
| 23 | runner = CliRunner() |
| 24 | |
| 25 | |
| 26 | def _invoke(repo: pathlib.Path, args: list[str]): |
| 27 | saved = os.getcwd() |
| 28 | try: |
| 29 | os.chdir(repo) |
| 30 | return runner.invoke(None, args) |
| 31 | finally: |
| 32 | os.chdir(saved) |
| 33 | |
| 34 | |
| 35 | def _init_repo(repo: pathlib.Path) -> None: |
| 36 | repo.mkdir(parents=True, exist_ok=True) |
| 37 | _invoke(repo, ["init"]) |
| 38 | |
| 39 | |
| 40 | def _write_file(repo: pathlib.Path, rel: str, content: str) -> None: |
| 41 | p = repo / rel |
| 42 | p.parent.mkdir(parents=True, exist_ok=True) |
| 43 | p.write_text(textwrap.dedent(content)) |
| 44 | |
| 45 | |
| 46 | def _commit(repo: pathlib.Path, msg: str) -> str: |
| 47 | _invoke(repo, ["code", "add", "."]) |
| 48 | result = _invoke(repo, [ |
| 49 | "commit", "-m", msg, |
| 50 | "--agent-id", "claude-code", |
| 51 | "--model-id", "claude-sonnet-4-6", |
| 52 | "--author", "claude-code", |
| 53 | "--json", |
| 54 | ]) |
| 55 | data = json.loads(result.stdout) |
| 56 | return data["commit_id"] |
| 57 | |
| 58 | |
| 59 | def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path: |
| 60 | _init_repo(tmp_path) |
| 61 | return tmp_path |
| 62 | |
| 63 | |
| 64 | def _write_raw_entry(path: pathlib.Path, old_id: str, new_id: str, commit_id: str, when: datetime.datetime, author: str, operation: str) -> None: |
| 65 | path.parent.mkdir(parents=True, exist_ok=True) |
| 66 | ts = int(when.timestamp()) |
| 67 | line = f"{old_id} {new_id} {commit_id} {author} {ts} +0000\t{operation}\n" |
| 68 | with path.open("a", encoding="utf-8") as f: |
| 69 | f.write(line) |
| 70 | |
| 71 | |
| 72 | def _fake_id(n: int) -> str: |
| 73 | return f"sha256:{n:064x}" |
| 74 | |
| 75 | |
| 76 | class TestSymlogTimeAgoResolutionCore: |
| 77 | def test_hour_ago_finds_entry_before_cutoff(self, tmp_path: pathlib.Path) -> None: |
| 78 | repo = _make_repo(tmp_path) |
| 79 | addr = "src/billing.py::compute_total" |
| 80 | log_path = symlog_path(repo, addr) |
| 81 | now = datetime.datetime.now(datetime.timezone.utc) |
| 82 | |
| 83 | c1 = _fake_id(1) |
| 84 | c2 = _fake_id(2) |
| 85 | _write_raw_entry(log_path, NULL_CONTENT_ID, c1, _fake_id(101), now - datetime.timedelta(hours=3), "gabriel", "symbol-created: compute_total") |
| 86 | _write_raw_entry(log_path, c1, c2, _fake_id(102), now - datetime.timedelta(minutes=10), "gabriel", "symbol-modified: fix") |
| 87 | |
| 88 | resolution = resolve_symlog_addr(f"{addr}@{{1.hour.ago}}", repo) |
| 89 | assert resolution is not None |
| 90 | assert resolution.content_id == c1 |
| 91 | |
| 92 | def test_minutes_ago_finds_most_recent(self, tmp_path: pathlib.Path) -> None: |
| 93 | repo = _make_repo(tmp_path) |
| 94 | addr = "src/billing.py::compute_total" |
| 95 | log_path = symlog_path(repo, addr) |
| 96 | now = datetime.datetime.now(datetime.timezone.utc) |
| 97 | |
| 98 | c1 = _fake_id(1) |
| 99 | c2 = _fake_id(2) |
| 100 | _write_raw_entry(log_path, NULL_CONTENT_ID, c1, _fake_id(101), now - datetime.timedelta(hours=2), "gabriel", "symbol-created: compute_total") |
| 101 | _write_raw_entry(log_path, c1, c2, _fake_id(102), now - datetime.timedelta(hours=1), "gabriel", "symbol-modified: fix") |
| 102 | |
| 103 | resolution = resolve_symlog_addr(f"{addr}@{{30.minute.ago}}", repo) |
| 104 | assert resolution is not None |
| 105 | assert resolution.content_id == c2 |
| 106 | |
| 107 | def test_yesterday_finds_entry_before_todays_midnight(self, tmp_path: pathlib.Path) -> None: |
| 108 | repo = _make_repo(tmp_path) |
| 109 | addr = "src/billing.py::compute_total" |
| 110 | log_path = symlog_path(repo, addr) |
| 111 | now = datetime.datetime.now(datetime.timezone.utc) |
| 112 | today_midnight = now.replace(hour=0, minute=0, second=0, microsecond=0) |
| 113 | |
| 114 | c1 = _fake_id(1) |
| 115 | c2 = _fake_id(2) |
| 116 | _write_raw_entry(log_path, NULL_CONTENT_ID, c1, _fake_id(101), today_midnight - datetime.timedelta(days=1), "gabriel", "symbol-created: compute_total") |
| 117 | _write_raw_entry(log_path, c1, c2, _fake_id(102), now, "gabriel", "symbol-modified: fix") |
| 118 | |
| 119 | resolution = resolve_symlog_addr(f"{addr}@{{yesterday}}", repo) |
| 120 | assert resolution is not None |
| 121 | assert resolution.content_id == c1 |
| 122 | |
| 123 | def test_no_entry_old_enough_raises_index_error(self, tmp_path: pathlib.Path) -> None: |
| 124 | repo = _make_repo(tmp_path) |
| 125 | addr = "src/billing.py::compute_total" |
| 126 | log_path = symlog_path(repo, addr) |
| 127 | now = datetime.datetime.now(datetime.timezone.utc) |
| 128 | c1 = _fake_id(1) |
| 129 | _write_raw_entry(log_path, NULL_CONTENT_ID, c1, _fake_id(101), now, "gabriel", "symbol-created: compute_total") |
| 130 | |
| 131 | with pytest.raises(IndexError): |
| 132 | resolve_symlog_addr(f"{addr}@{{1.week.ago}}", repo) |
| 133 | |
| 134 | def test_unrecognized_spec_raises_value_error(self, tmp_path: pathlib.Path) -> None: |
| 135 | repo = _make_repo(tmp_path) |
| 136 | addr = "src/billing.py::compute_total" |
| 137 | log_path = symlog_path(repo, addr) |
| 138 | now = datetime.datetime.now(datetime.timezone.utc) |
| 139 | c1 = _fake_id(1) |
| 140 | _write_raw_entry(log_path, NULL_CONTENT_ID, c1, _fake_id(101), now, "gabriel", "symbol-created: compute_total") |
| 141 | |
| 142 | with pytest.raises(ValueError): |
| 143 | resolve_symlog_addr(f"{addr}@{{not.a.real.spec}}", repo) |
| 144 | |
| 145 | def test_plain_index_still_works(self, tmp_path: pathlib.Path) -> None: |
| 146 | """Non-regression: the pre-existing @{N} integer form must still work |
| 147 | after broadening the regex to accept time specs.""" |
| 148 | repo = _make_repo(tmp_path) |
| 149 | addr = "src/billing.py::compute_total" |
| 150 | log_path = symlog_path(repo, addr) |
| 151 | now = datetime.datetime.now(datetime.timezone.utc) |
| 152 | c1 = _fake_id(1) |
| 153 | _write_raw_entry(log_path, NULL_CONTENT_ID, c1, _fake_id(101), now, "gabriel", "symbol-created: compute_total") |
| 154 | |
| 155 | resolution = resolve_symlog_addr(f"{addr}@{{0}}", repo) |
| 156 | assert resolution is not None |
| 157 | assert resolution.content_id == c1 |
| 158 | |
| 159 | |
| 160 | class TestSymlogTimeAgoViaCli: |
| 161 | def test_resolve_accepts_time_spec(self, tmp_path: pathlib.Path) -> None: |
| 162 | repo = _make_repo(tmp_path) |
| 163 | _write_file(repo, "src/billing.py", "def compute_total(items):\n return sum(items)\n") |
| 164 | _commit(repo, "add compute_total") |
| 165 | |
| 166 | result = _invoke(repo, ["symlog", "resolve", "src/billing.py::compute_total@{0.second.ago}", "--json"]) |
| 167 | assert result.exit_code == 0, result.stdout |
| 168 | data = json.loads(result.stdout) |
| 169 | assert data["symbol"] == "src/billing.py::compute_total" |