feat: Project Scaffold and Docker Packaging
This commit is contained in:
@@ -0,0 +1,37 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from tenbackward.config import Config
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def env_setup(monkeypatch: pytest.MonkeyPatch) -> None:
|
||||
"""Provide a fully populated environment for Config success paths.
|
||||
|
||||
Tests that exercise the missing-required-key path should call
|
||||
`monkeypatch.delenv(...)` explicitly.
|
||||
"""
|
||||
monkeypatch.setenv("MASTODON_BASE_URL", "https://mastodon.example")
|
||||
monkeypatch.setenv("MASTODON_ACCESS_TOKEN", "test-token")
|
||||
monkeypatch.setenv("SITE_URL", "https://blog.example.com")
|
||||
monkeypatch.setenv("HASHTAGS", "#throwback,#10backward")
|
||||
monkeypatch.setenv("RUN_AT", "09:00")
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def data_dir(tmp_path: Path) -> Path:
|
||||
return tmp_path / "data"
|
||||
|
||||
|
||||
@pytest.fixture()
|
||||
def full_config(env_setup, data_dir) -> Config:
|
||||
"""Build a Config object that points at an isolated data dir."""
|
||||
from tenbackward.config import load_config
|
||||
|
||||
import os
|
||||
|
||||
os.environ["DATA_DIR"] = str(data_dir)
|
||||
return load_config()
|
||||
@@ -0,0 +1,103 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from tenbackward.config import ConfigError, load_config, validate_config
|
||||
|
||||
|
||||
def _populate(env_setup) -> dict[str, str]:
|
||||
return {k: os.environ[k] for k in os.environ if k.startswith(("MASTODON_", "SITE_", "HASHTAGS", "RUN_AT", "THROWBACK_", "RETRY_", "TZ"))}
|
||||
|
||||
|
||||
def test_load_config_succeeds_with_complete_env(env_setup) -> None:
|
||||
config = load_config()
|
||||
assert config.mastodon_base_url == "https://mastodon.example"
|
||||
assert config.run_at == "09:00"
|
||||
assert config.retry_count == 3
|
||||
assert config.throwback_prefix == "Throwback:"
|
||||
|
||||
|
||||
def test_validate_config_lists_every_missing_key(env_setup, monkeypatch) -> None:
|
||||
monkeypatch.delenv("MASTODON_ACCESS_TOKEN")
|
||||
monkeypatch.delenv("SITE_URL")
|
||||
|
||||
values = {k: os.environ.get(k, "") for k in [
|
||||
"MASTODON_BASE_URL",
|
||||
"MASTODON_ACCESS_TOKEN",
|
||||
"SITE_URL",
|
||||
"HASHTAGS",
|
||||
"RUN_AT",
|
||||
]}
|
||||
|
||||
with pytest.raises(ConfigError) as excinfo:
|
||||
validate_config(values)
|
||||
|
||||
message = str(excinfo.value)
|
||||
assert "MASTODON_ACCESS_TOKEN" in message
|
||||
assert "SITE_URL" in message
|
||||
|
||||
|
||||
def test_validate_config_rejects_bad_run_at() -> None:
|
||||
values = {
|
||||
"MASTODON_BASE_URL": "https://mastodon.example",
|
||||
"MASTODON_ACCESS_TOKEN": "x",
|
||||
"SITE_URL": "https://blog.example.com",
|
||||
"HASHTAGS": "#x",
|
||||
"RUN_AT": "25:99",
|
||||
"MASTODON_VISIBILITY": "public",
|
||||
"THROWBACK_PREFIX": "Throwback:",
|
||||
"RETRY_COUNT": "3",
|
||||
"TZ": "Europe/Berlin",
|
||||
}
|
||||
with pytest.raises(ConfigError):
|
||||
validate_config(values)
|
||||
|
||||
|
||||
def test_load_config_applies_optional_defaults(env_setup, monkeypatch) -> None:
|
||||
monkeypatch.delenv("THROWBACK_PREFIX", raising=False)
|
||||
monkeypatch.delenv("RETRY_COUNT", raising=False)
|
||||
|
||||
config = load_config()
|
||||
assert config.throwback_prefix == "Throwback:"
|
||||
assert config.retry_count == 3
|
||||
assert config.tz == "Europe/Berlin"
|
||||
assert config.mastodon_visibility == "public"
|
||||
|
||||
|
||||
def test_load_config_reads_dotenv_file(tmp_path: Path, monkeypatch) -> None:
|
||||
dotenv = tmp_path / ".env"
|
||||
dotenv.write_text(
|
||||
"MASTODON_BASE_URL=https://from-file.example\n"
|
||||
"MASTODON_ACCESS_TOKEN=file-token\n"
|
||||
"SITE_URL=https://blog.example.com\n"
|
||||
"HASHTAGS=#throwback\n"
|
||||
"RUN_AT=12:34\n"
|
||||
)
|
||||
|
||||
for key in ("MASTODON_BASE_URL", "MASTODON_ACCESS_TOKEN", "SITE_URL", "HASHTAGS", "RUN_AT"):
|
||||
monkeypatch.delenv(key, raising=False)
|
||||
|
||||
config = load_config(dotenv_path=dotenv)
|
||||
assert config.mastodon_base_url == "https://from-file.example"
|
||||
assert config.mastodon_access_token == "file-token"
|
||||
assert config.run_at == "12:34"
|
||||
|
||||
|
||||
def test_env_overrides_dotenv(tmp_path: Path, monkeypatch) -> None:
|
||||
dotenv = tmp_path / ".env"
|
||||
dotenv.write_text("RUN_AT=01:00\n")
|
||||
|
||||
monkeypatch.setenv("RUN_AT", "23:00")
|
||||
for key in (
|
||||
"MASTODON_BASE_URL",
|
||||
"MASTODON_ACCESS_TOKEN",
|
||||
"SITE_URL",
|
||||
"HASHTAGS",
|
||||
):
|
||||
monkeypatch.setenv(key, "x")
|
||||
|
||||
config = load_config(dotenv_path=dotenv)
|
||||
assert config.run_at == "23:00"
|
||||
@@ -0,0 +1,193 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
from tenbackward.config import _is_valid_hhmm
|
||||
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
ENTRYPOINT = REPO_ROOT / "entrypoint.sh"
|
||||
|
||||
|
||||
def test_valid_hhmm() -> None:
|
||||
assert _is_valid_hhmm("00:00")
|
||||
assert _is_valid_hhmm("09:30")
|
||||
assert _is_valid_hhmm("23:59")
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", ["24:00", "9:00", "09:60", "9", "9:00:00", "ab:cd", "", None])
|
||||
def test_invalid_hhmm(value) -> None:
|
||||
assert not _is_valid_hhmm(value) # type: ignore[arg-type]
|
||||
|
||||
|
||||
def _entrypoint_env(extra: dict[str, str]) -> dict[str, str]:
|
||||
env = {
|
||||
"PATH": "/usr/bin:/bin:/usr/sbin:/sbin",
|
||||
"DEBIAN_FRONTEND": "noninteractive",
|
||||
"HOME": "/tmp",
|
||||
"LANG": "C.UTF-8",
|
||||
}
|
||||
env.update(extra)
|
||||
return env
|
||||
|
||||
|
||||
def _run_entrypoint_cron_render(extra: dict[str, str]) -> str:
|
||||
raise NotImplementedError
|
||||
|
||||
|
||||
def _extract_rendered_cron(env: dict[str, str]) -> tuple[int, str, str]:
|
||||
"""Run the entrypoint with a fake `install` that records the cron file,
|
||||
and a stub `cron` that exits immediately so the entrypoint doesn't hang.
|
||||
Returns (returncode, stderr, installed-cron-contents).
|
||||
"""
|
||||
import tempfile
|
||||
|
||||
with tempfile.TemporaryDirectory() as fake_bin_dir, tempfile.TemporaryDirectory() as tmp:
|
||||
fake_cron = Path(tmp) / "tenbackward.cron"
|
||||
|
||||
# Fake `install`: real syntax is `install [-m mode] [-o owner] [-g group] SRC DST`.
|
||||
# We extract the last positional arg as DST and the last existing source before it.
|
||||
fake_install = Path(fake_bin_dir) / "install"
|
||||
fake_install.write_text(
|
||||
"#!/bin/bash\n"
|
||||
"src=\"\"\n"
|
||||
"dst=\"\"\n"
|
||||
"while [ $# -gt 0 ]; do\n"
|
||||
" case \"$1\" in\n"
|
||||
" -m|-o|-g) shift; shift;;\n"
|
||||
" *) src=\"${src:-$1}\"; dst=\"$1\"; shift;;\n"
|
||||
" esac\n"
|
||||
"done\n"
|
||||
"cp \"$src\" \"" + str(fake_cron) + "\"\n"
|
||||
)
|
||||
fake_install.chmod(0o755)
|
||||
|
||||
# Fake `cron` exits immediately so we never hang waiting on cron.
|
||||
fake_cron_bin = Path(fake_bin_dir) / "cron"
|
||||
fake_cron_bin.write_text("#!/bin/bash\nexit 0\n")
|
||||
fake_cron_bin.chmod(0o755)
|
||||
|
||||
env = dict(env)
|
||||
env["PATH"] = f"{fake_bin_dir}:/usr/bin:/bin"
|
||||
|
||||
proc = subprocess.run(
|
||||
["/bin/bash", str(ENTRYPOINT)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
timeout=10,
|
||||
check=False,
|
||||
)
|
||||
|
||||
rendered = fake_cron.read_text() if fake_cron.exists() else ""
|
||||
return proc.returncode, proc.stderr, rendered
|
||||
|
||||
|
||||
@pytest.mark.skipif(not shutil.which("bash"), reason="bash not available")
|
||||
def test_entrypoint_renders_run_at_into_cron() -> None:
|
||||
"""Drive the entrypoint with a hermetic PATH so we can observe the rendered cron line.
|
||||
|
||||
The entrypoint hard-codes `/app/crontab/install-cron.sh` (correct in the
|
||||
Docker image). For the host-side test we override PATH so `install` and
|
||||
`cron` are stubbed, but we still need `/app/crontab/install-cron.sh` — we
|
||||
bind-mount it via a bind-mount emulation by overriding the env var
|
||||
`PATH` and having the install-cron.sh check fall back: instead, we patch
|
||||
the entrypoint with an in-place substitution for the test.
|
||||
"""
|
||||
tmp_cron_dest = _write_substituted_entrypoint()
|
||||
try:
|
||||
env = _entrypoint_env({
|
||||
"MASTODON_BASE_URL": "https://mastodon.example",
|
||||
"MASTODON_ACCESS_TOKEN": "x",
|
||||
"SITE_URL": "https://blog.example.com",
|
||||
"HASHTAGS": "#throwback",
|
||||
"RUN_AT": "09:00",
|
||||
"TZ": "Europe/Berlin",
|
||||
})
|
||||
|
||||
returncode, stderr, rendered = _extract_rendered_cron(env, tmp_cron_dest)
|
||||
|
||||
assert returncode == 0, (returncode, stderr)
|
||||
assert "00 09 * * *" in rendered, rendered
|
||||
assert "python -m tenbackward" in rendered
|
||||
assert "TZ=Europe/Berlin" in rendered
|
||||
finally:
|
||||
tmp_cron_dest.unlink(missing_ok=True)
|
||||
|
||||
|
||||
def _write_substituted_entrypoint():
|
||||
"""Copy entrypoint.sh to a temp file with `/app/crontab/install-cron.sh`
|
||||
rewritten to the repo-relative path so the host can execute it."""
|
||||
import tempfile
|
||||
src = ENTRYPOINT.read_text()
|
||||
sub = src.replace("/app/crontab/install-cron.sh", str(REPO_ROOT / "crontab" / "install-cron.sh"))
|
||||
tmp = Path(tempfile.mkstemp(prefix="entrypoint-", suffix=".sh")[1])
|
||||
tmp.write_text(sub)
|
||||
tmp.chmod(0o755)
|
||||
return tmp
|
||||
|
||||
|
||||
def _extract_rendered_cron(env: dict[str, str], entrypoint_path: Path) -> tuple[int, str, str]:
|
||||
"""Run the (substituted) entrypoint with stubs for `install` and `cron`."""
|
||||
import tempfile
|
||||
|
||||
with tempfile.TemporaryDirectory() as fake_bin_dir, tempfile.TemporaryDirectory() as tmp:
|
||||
sidecar = Path(tmp) / "tenbackward.cron"
|
||||
|
||||
fake_install = Path(fake_bin_dir) / "install"
|
||||
fake_install.write_text(
|
||||
"#!/bin/bash\n"
|
||||
"src=\"\"; dst=\"\"\n"
|
||||
"while [ $# -gt 0 ]; do\n"
|
||||
" case \"$1\" in -m|-o|-g) shift; shift;; *) src=\"${src:-$1}\"; dst=\"$1\"; shift;; esac\n"
|
||||
"done\n"
|
||||
"cp \"$src\" \"" + str(sidecar) + "\"\n"
|
||||
"mkdir -p \"$(dirname \"$dst\")\" 2>/dev/null\n"
|
||||
"cp \"$src\" \"$dst\" 2>/dev/null || true\n"
|
||||
)
|
||||
fake_install.chmod(0o755)
|
||||
|
||||
fake_cron_bin = Path(fake_bin_dir) / "cron"
|
||||
fake_cron_bin.write_text("#!/bin/bash\nexit 0\n")
|
||||
fake_cron_bin.chmod(0o755)
|
||||
|
||||
env = dict(env)
|
||||
env["PATH"] = f"{fake_bin_dir}:/usr/bin:/bin"
|
||||
|
||||
proc = subprocess.run(
|
||||
["/bin/bash", str(entrypoint_path)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
timeout=10,
|
||||
check=False,
|
||||
)
|
||||
|
||||
rendered = sidecar.read_text() if sidecar.exists() else ""
|
||||
return proc.returncode, proc.stderr, rendered
|
||||
|
||||
|
||||
@pytest.mark.skipif(not shutil.which("bash"), reason="bash not available")
|
||||
def test_install_cron_rejects_unresolved_placeholder(tmp_path: Path) -> None:
|
||||
script = REPO_ROOT / "crontab" / "install-cron.sh"
|
||||
|
||||
bad = tmp_path / "bad.cron"
|
||||
bad.write_text("__RUN_AT__ * * * * /bin/false\n")
|
||||
proc = subprocess.run(
|
||||
["/bin/bash", str(script), str(bad)],
|
||||
capture_output=True, text=True, timeout=10,
|
||||
)
|
||||
assert proc.returncode != 0
|
||||
assert "__RUN_AT__" in proc.stderr
|
||||
|
||||
good = tmp_path / "good.cron"
|
||||
good.write_text("SHELL=/bin/bash\n5 9 * * * /bin/true\n")
|
||||
proc2 = subprocess.run(
|
||||
["/bin/bash", str(script), str(good)],
|
||||
capture_output=True, text=True, timeout=10,
|
||||
)
|
||||
assert proc2.returncode == 0, proc2.stderr
|
||||
@@ -0,0 +1,59 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
|
||||
|
||||
def test_dockerignore_blocks_secrets_and_data() -> None:
|
||||
dockerignore = (REPO_ROOT / ".dockerignore").read_text().splitlines()
|
||||
normalized = {line.strip() for line in dockerignore if line.strip() and not line.startswith("#")}
|
||||
|
||||
for required in (".env", "data/", ".git/", "tests/", ".kilo/", ".venv/"):
|
||||
assert required in normalized, f"missing {required} from .dockerignore"
|
||||
|
||||
|
||||
def test_gitignore_blocks_dotenv_and_data() -> None:
|
||||
gitignore = (REPO_ROOT / ".gitignore").read_text().splitlines()
|
||||
normalized = {line.strip() for line in gitignore if line.strip() and not line.startswith("#")}
|
||||
assert ".env" in normalized
|
||||
assert "data/" in normalized
|
||||
assert ".kilo/" in normalized
|
||||
|
||||
|
||||
def test_env_example_has_no_real_tokens() -> None:
|
||||
text = (REPO_ROOT / ".env.example").read_text()
|
||||
assert "MASTODON_ACCESS_TOKEN=replace-me" in text
|
||||
|
||||
for token in ("ghp_", "gho_", "xoxb-", "Bearer ey", "AKIA"):
|
||||
assert token not in text, f"found suspicious token prefix {token!r} in .env.example"
|
||||
|
||||
mastodon_line = next(
|
||||
line for line in text.splitlines() if line.startswith("MASTODON_ACCESS_TOKEN=")
|
||||
)
|
||||
value = mastodon_line.split("=", 1)[1].strip()
|
||||
assert not re.fullmatch(r"[A-Za-z0-9_\-]{40,}", value), (
|
||||
"MASTODON_ACCESS_TOKEN placeholder should not look like a real token"
|
||||
)
|
||||
|
||||
|
||||
def test_dockerfile_does_not_copy_env_or_data() -> None:
|
||||
dockerfile = (REPO_ROOT / "Dockerfile").read_text()
|
||||
for blocked in (".env", "data/", "./data"):
|
||||
for line in dockerfile.splitlines():
|
||||
if line.lstrip().upper().startswith("COPY"):
|
||||
assert blocked not in line, f"Dockerfile COPY must not reference {blocked}: {line!r}"
|
||||
|
||||
|
||||
def test_compose_mounts_data_and_env_readonly() -> None:
|
||||
compose = (REPO_ROOT / "docker-compose.yml").read_text()
|
||||
assert "./.env:/.env:ro" in compose
|
||||
assert "./data:/app/data" in compose
|
||||
assert "build:" in compose
|
||||
|
||||
|
||||
def test_cron_template_has_env_header() -> None:
|
||||
cron = (REPO_ROOT / "crontab" / "tenbackward.cron").read_text()
|
||||
assert "SHELL=/bin/bash" in cron
|
||||
assert "PATH=" in cron
|
||||
@@ -0,0 +1,88 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import re
|
||||
import shutil
|
||||
import subprocess
|
||||
from pathlib import Path
|
||||
|
||||
import pytest
|
||||
|
||||
REPO_ROOT = Path(__file__).resolve().parents[1]
|
||||
ENTRYPOINT = REPO_ROOT / "entrypoint.sh"
|
||||
|
||||
|
||||
def _has_bash() -> bool:
|
||||
return shutil.which("bash") is not None
|
||||
|
||||
|
||||
def _minimal_path() -> str:
|
||||
parts = ["/usr/bin", "/bin", "/usr/sbin", "/sbin"]
|
||||
for p in parts:
|
||||
if Path(p).exists():
|
||||
return ":".join(parts)
|
||||
return os.environ.get("PATH", "")
|
||||
|
||||
|
||||
def _clean_env() -> dict[str, str]:
|
||||
env = {
|
||||
"PATH": _minimal_path(),
|
||||
"DEBIAN_FRONTEND": "noninteractive",
|
||||
"HOME": "/tmp",
|
||||
"LANG": "C.UTF-8",
|
||||
}
|
||||
return env
|
||||
|
||||
|
||||
@pytest.mark.skipif(not _has_bash(), reason="bash not available")
|
||||
def test_entrypoint_rejects_bad_run_at(tmp_path: Path, monkeypatch) -> None:
|
||||
"""The entrypoint must exit non-zero with a clear message on a bad RUN_AT."""
|
||||
env = _clean_env()
|
||||
env["MASTODON_BASE_URL"] = "https://mastodon.example"
|
||||
env["MASTODON_ACCESS_TOKEN"] = "x"
|
||||
env["SITE_URL"] = "https://blog.example.com"
|
||||
env["HASHTAGS"] = "#throwback"
|
||||
env["RUN_AT"] = "25:99"
|
||||
env["TZ"] = "Europe/Berlin"
|
||||
|
||||
proc = subprocess.run(
|
||||
["/bin/bash", str(ENTRYPOINT)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
timeout=10,
|
||||
check=False,
|
||||
)
|
||||
assert proc.returncode != 0
|
||||
assert "RUN_AT" in (proc.stdout + proc.stderr)
|
||||
|
||||
|
||||
@pytest.mark.skipif(not _has_bash(), reason="bash not available")
|
||||
def test_entrypoint_fails_fast_when_token_missing(tmp_path: Path, monkeypatch) -> None:
|
||||
"""A missing required var must abort before cron is started."""
|
||||
env = _clean_env()
|
||||
env["MASTODON_BASE_URL"] = "https://mastodon.example"
|
||||
env["SITE_URL"] = "https://blog.example.com"
|
||||
env["HASHTAGS"] = "#throwback"
|
||||
env["RUN_AT"] = "09:00"
|
||||
env["TZ"] = "Europe/Berlin"
|
||||
env.pop("MASTODON_ACCESS_TOKEN", None)
|
||||
|
||||
proc = subprocess.run(
|
||||
["/bin/bash", str(ENTRYPOINT)],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
env=env,
|
||||
timeout=10,
|
||||
check=False,
|
||||
)
|
||||
assert proc.returncode != 0
|
||||
combined = (proc.stdout + proc.stderr).lower()
|
||||
assert "missing" in combined
|
||||
assert "mastodon_access_token" in combined.lower()
|
||||
|
||||
|
||||
def test_entrypoint_references_cron_in_foreground() -> None:
|
||||
text = ENTRYPOINT.read_text()
|
||||
assert re.search(r"exec\s+cron\s+-f", text), "entrypoint must exec cron in the foreground"
|
||||
cron = (REPO_ROOT / "crontab" / "tenbackward.cron").read_text()
|
||||
assert "SHELL=/bin/bash" in cron
|
||||
@@ -0,0 +1,31 @@
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from tenbackward.state import load_posted, posted_path, save_posted
|
||||
|
||||
|
||||
def test_load_posted_creates_empty_when_missing(tmp_path: Path) -> None:
|
||||
state = load_posted(tmp_path)
|
||||
assert state == {}
|
||||
|
||||
|
||||
def test_round_trip_persists(tmp_path: Path) -> None:
|
||||
state = {"post-1": {"posted_at": "2024-01-01T00:00:00Z"}}
|
||||
save_posted(tmp_path, state)
|
||||
|
||||
assert posted_path(tmp_path).exists()
|
||||
again = load_posted(tmp_path)
|
||||
assert again == state
|
||||
|
||||
|
||||
def test_load_posted_handles_corrupt_file(tmp_path: Path) -> None:
|
||||
posted_path(tmp_path).parent.mkdir(parents=True, exist_ok=True)
|
||||
posted_path(tmp_path).write_text("not-json")
|
||||
assert load_posted(tmp_path) == {}
|
||||
|
||||
|
||||
def test_save_posted_creates_parent_dirs(tmp_path: Path) -> None:
|
||||
nested = tmp_path / "deep" / "data"
|
||||
save_posted(nested, {"x": 1})
|
||||
assert posted_path(nested).exists()
|
||||
Reference in New Issue
Block a user