167 lines
5.0 KiB
Python
167 lines
5.0 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import json
|
|
import logging
|
|
import os
|
|
from pathlib import Path
|
|
|
|
from tenbackward.logging_setup import JsonFormatter
|
|
from tenbackward.state import (
|
|
PostedStore,
|
|
is_posted,
|
|
load_posted,
|
|
mark_posted,
|
|
mark_posted_many,
|
|
posted_path,
|
|
)
|
|
|
|
|
|
def _store(tmp_path: Path) -> PostedStore:
|
|
return PostedStore(tmp_path)
|
|
|
|
|
|
def test_load_creates_empty_file_when_missing(tmp_path: Path) -> None:
|
|
store = _store(tmp_path)
|
|
result = store.load()
|
|
|
|
assert result == []
|
|
assert posted_path(tmp_path).exists()
|
|
document = json.loads(posted_path(tmp_path).read_text(encoding="utf-8"))
|
|
assert document == {"posted": []}
|
|
|
|
|
|
def test_load_returns_empty_when_file_missing_no_crash(tmp_path: Path) -> None:
|
|
assert _store(tmp_path).load() == []
|
|
|
|
|
|
def test_load_warns_and_returns_empty_on_malformed_json(
|
|
tmp_path: Path,
|
|
) -> None:
|
|
buf = io.StringIO()
|
|
handler = logging.StreamHandler(buf)
|
|
handler.setFormatter(JsonFormatter())
|
|
logger = logging.getLogger("tenbackward")
|
|
original_handlers = list(logger.handlers)
|
|
original_propagate = logger.propagate
|
|
logger.handlers = [handler]
|
|
logger.setLevel(logging.WARNING)
|
|
logger.propagate = False
|
|
try:
|
|
posted_path(tmp_path).parent.mkdir(parents=True, exist_ok=True)
|
|
posted_path(tmp_path).write_text("not-json", encoding="utf-8")
|
|
result = _store(tmp_path).load()
|
|
finally:
|
|
logger.handlers = original_handlers
|
|
logger.propagate = original_propagate
|
|
|
|
assert result == []
|
|
lines = [
|
|
json.loads(line) for line in buf.getvalue().splitlines() if line.strip()
|
|
]
|
|
events = [line.get("event") for line in lines]
|
|
assert "posted_store_corrupt" in events
|
|
|
|
|
|
def test_mark_posted_writes_relative_path_to_file(tmp_path: Path) -> None:
|
|
_store(tmp_path).mark_posted("2014/2014-08-04-foo.md")
|
|
|
|
document = json.loads(posted_path(tmp_path).read_text(encoding="utf-8"))
|
|
assert document == {"posted": ["2014/2014-08-04-foo.md"]}
|
|
|
|
|
|
def test_is_posted_round_trip(tmp_path: Path) -> None:
|
|
store = _store(tmp_path)
|
|
store.mark_posted("2014/2014-08-04-foo.md")
|
|
|
|
assert store.is_posted("2014/2014-08-04-foo.md") is True
|
|
assert store.is_posted("2015/2015-01-01-bar.md") is False
|
|
|
|
|
|
def test_mark_posted_is_idempotent(tmp_path: Path) -> None:
|
|
store = _store(tmp_path)
|
|
first = store.mark_posted("2014/2014-08-04-foo.md")
|
|
second = store.mark_posted("2014/2014-08-04-foo.md")
|
|
|
|
assert first is True
|
|
assert second is False
|
|
|
|
document = json.loads(posted_path(tmp_path).read_text(encoding="utf-8"))
|
|
assert document == {"posted": ["2014/2014-08-04-foo.md"]}
|
|
|
|
|
|
def test_mark_posted_many_persists_in_one_write(
|
|
tmp_path: Path, monkeypatch
|
|
) -> None:
|
|
store = _store(tmp_path)
|
|
replaces: list[tuple[object, object]] = []
|
|
real_replace = os.replace
|
|
|
|
def counting_replace(src, dst) -> None:
|
|
replaces.append((src, dst))
|
|
real_replace(src, dst)
|
|
|
|
monkeypatch.setattr("tenbackward.state.os.replace", counting_replace)
|
|
|
|
added = store.mark_posted_many(
|
|
["2014/2014-08-04-foo.md", "2015/2015-01-01-bar.md"]
|
|
)
|
|
|
|
assert added == ["2014/2014-08-04-foo.md", "2015/2015-01-01-bar.md"]
|
|
assert len(replaces) == 1
|
|
assert Path(replaces[0][1]) == posted_path(tmp_path)
|
|
|
|
|
|
def test_mark_posted_leaves_no_temp_files(tmp_path: Path) -> None:
|
|
_store(tmp_path).mark_posted("2014/2014-08-04-foo.md")
|
|
|
|
leftovers = [
|
|
p for p in tmp_path.iterdir() if p.name.startswith("posted.json.tmp.")
|
|
]
|
|
assert leftovers == []
|
|
|
|
|
|
def test_mark_posted_many_against_existing_is_no_op(tmp_path: Path) -> None:
|
|
store = _store(tmp_path)
|
|
store.mark_posted("2014/2014-08-04-foo.md")
|
|
|
|
added = store.mark_posted_many(
|
|
["2014/2014-08-04-foo.md", "2015/2015-01-01-bar.md"]
|
|
)
|
|
|
|
assert added == ["2015/2015-01-01-bar.md"]
|
|
document = json.loads(posted_path(tmp_path).read_text(encoding="utf-8"))
|
|
assert document == {
|
|
"posted": ["2014/2014-08-04-foo.md", "2015/2015-01-01-bar.md"]
|
|
}
|
|
|
|
|
|
def test_posted_path_defaults_via_DATA_DIR(
|
|
monkeypatch, tmp_path: Path
|
|
) -> None:
|
|
monkeypatch.setenv("DATA_DIR", str(tmp_path))
|
|
assert posted_path(None) == tmp_path / "posted.json"
|
|
|
|
|
|
def test_load_filters_non_string_entries(tmp_path: Path) -> None:
|
|
posted_path(tmp_path).parent.mkdir(parents=True, exist_ok=True)
|
|
posted_path(tmp_path).write_text(
|
|
json.dumps({"posted": ["good", 42, None, "also-good"]}),
|
|
encoding="utf-8",
|
|
)
|
|
|
|
assert _store(tmp_path).load() == ["good", "also-good"]
|
|
|
|
|
|
def test_module_helpers_round_trip(tmp_path: Path) -> None:
|
|
assert load_posted(tmp_path) == []
|
|
assert is_posted("2014/2014-08-04-foo.md", tmp_path) is False
|
|
assert mark_posted("2014/2014-08-04-foo.md", tmp_path) is True
|
|
added = mark_posted_many(
|
|
["2014/2014-08-04-foo.md", "2015/2015-01-01-bar.md"], tmp_path
|
|
)
|
|
assert added == ["2015/2015-01-01-bar.md"]
|
|
assert load_posted(tmp_path) == [
|
|
"2014/2014-08-04-foo.md",
|
|
"2015/2015-01-01-bar.md",
|
|
] |