coulomb-social/tests/test_backup_capacity.py

"""Capacity checks happen inside writer locks; actual file limits run in children."""

import importlib.util
import resource
import subprocess
import sys
from pathlib import Path

import pytest

from tests.test_paired_backup import NOW, setup
from tests.test_paired_backup import backup as backup


@pytest.fixture
def capacity(monkeypatch):
    source = Path(__file__).resolve().parents[1] / "scripts/backup_capacity.py"
    spec = importlib.util.spec_from_file_location("capacity_cases", source)
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    monkeypatch.setattr(module, "active_file_limit", lambda: module.FILE_LIMIT)
    monkeypatch.setattr(module, "disk_free", lambda path: 128 * 1024**3)
    monkeypatch.setattr(module, "free_inodes", lambda path: 1000000)
    return module


@pytest.mark.parametrize("phase,slots", [("capture", 7), ("resume", 3), ("mirror", 1)])
def test_reserve_includes_complete_phase_scratch_and_preserves_state(
    capacity, tmp_path, phase, slots
):
    tmp_path.chmod(0o700)
    original = tmp_path / "retained.age"
    original.write_bytes(b"retained synthetic ciphertext")
    original.chmod(0o600)
    before = original.read_bytes()
    plan = capacity.require_capacity(tmp_path, phase=phase)
    assert plan["status"] == "ready"
    assert plan["state_reserve_bytes"] == slots * capacity.FILE_LIMIT + capacity.CONTROL_MARGIN
    assert plan["disk_reserve_bytes"] == plan["state_reserve_bytes"] + capacity.FILE_LIMIT
    assert not plan["filesystem_blocks_reserved"] and not plan["limits_owner_admitted"]
    assert list(tmp_path.iterdir()) == [original] and original.read_bytes() == before


@pytest.mark.parametrize("condition", ["disk", "budget", "inodes", "unbounded"])
def test_limits_refuse_before_capture_or_pending_creation(backup, monkeypatch, tmp_path, condition):
    state, calls, publications = setup(backup, monkeypatch, tmp_path)
    if condition == "disk":
        monkeypatch.setattr(backup.capacity, "disk_free", lambda path: 1)
    elif condition == "budget":
        monkeypatch.setattr(backup.capacity, "state_usage", lambda path: 8 * 1024**3)
    elif condition == "inodes":
        monkeypatch.setattr(backup.capacity, "free_inodes", lambda path: 0)
    else:
        monkeypatch.setattr(backup.capacity, "active_file_limit", lambda: resource.RLIM_INFINITY)
    with pytest.raises(ValueError, match="capacity_review_required"):
        backup.run_backup(state, tmp_path / "primary.py", "age1-synthetic", NOW)
    assert not calls and not publications and not list(state.glob("pending-*"))


def test_read_only_preflight_distinguishes_unbounded_executor(capacity, tmp_path, monkeypatch):
    tmp_path.chmod(0o700)
    monkeypatch.setattr(capacity, "active_file_limit", lambda: resource.RLIM_INFINITY)
    report = capacity.reserve_plan(tmp_path, require_file_limit=False)
    assert report["status"] == "reviewed_unbounded_executor"
    assert not report["bounded_file_executor"] and not list(tmp_path.iterdir())
    with pytest.raises(ValueError, match="capacity_review_required"):
        capacity.require_capacity(tmp_path)


def test_state_scan_counts_allocated_blocks_and_refuses_symlinks(capacity, tmp_path):
    tmp_path.chmod(0o700)
    path = tmp_path / "metadata.json"
    path.write_bytes(b"{}")
    path.chmod(0o600)
    assert capacity.state_usage(tmp_path) >= max(path.stat().st_size, path.stat().st_blocks * 512)
    (tmp_path / "pending-link").symlink_to(path)
    with pytest.raises(ValueError, match="unsafe_state_entry"):
        capacity.state_usage(tmp_path)


def test_resume_capacity_refusal_keeps_exact_pending_evidence(backup, monkeypatch, tmp_path):
    state, _, _ = setup(backup, monkeypatch, tmp_path, "upload")
    with pytest.raises(backup.Refused):
        backup.run_backup(state, tmp_path / "primary.py", "age1-public-test-key", NOW)
    before = {str(p): p.read_bytes() for p in state.rglob("*") if p.is_file()}
    _, calls, publications = setup(backup, monkeypatch, tmp_path)
    monkeypatch.setattr(backup.capacity, "disk_free", lambda path: 1)
    with pytest.raises(ValueError, match="capacity_review_required"):
        backup.run_backup(state, tmp_path / "primary.py", "age1-public-test-key", NOW, resume=True)
    assert {str(p): p.read_bytes() for p in state.rglob("*") if p.is_file()} == before
    assert not calls and not publications


def test_real_inherited_file_limit_raises_write_error_and_runs_cleanup(tmp_path):
    root = Path(__file__).resolve().parents[1]
    code = '''
import errno,json,subprocess,sys
from pathlib import Path
sys.path.insert(0,sys.argv[1])
import backup_capacity as capacity
assert capacity.enforce_file_limit(65536)==65536
assert capacity.enforce_file_limit(1073741824)==65536
child = r"""
import errno,json,resource,sys
from pathlib import Path
root=Path(sys.argv[1])
try:
    (root/'bounded.bin').write_bytes(b'x'*131072)
except OSError as error:
    assert error.errno==errno.EFBIG
    print(json.dumps({'bounded':resource.getrlimit(resource.RLIMIT_FSIZE)[0]==65536}))
else:
    raise AssertionError('file write escaped inherited cap')
finally:
    (root/'cleanup-ran').write_text('cleanup')
"""
result=subprocess.run([sys.executable,'-c',child,sys.argv[2]],capture_output=True,text=True)
assert result.returncode==0
assert json.loads(result.stdout)['bounded']
assert (Path(sys.argv[2])/'cleanup-ran').read_text()=='cleanup'
assert (Path(sys.argv[2])/'bounded.bin').stat().st_size<=65536
print('inherited limit and finally cleanup verified')
'''
    result = subprocess.run(
        [sys.executable, "-c", code, str(root / "scripts"), str(tmp_path)],
        capture_output=True,
        text=True,
    )
    assert result.returncode == 0, result.stderr
    assert "cleanup verified" in result.stdout


def test_service_applies_file_limit_to_capture_and_mirror_commands():
    root = Path(__file__).resolve().parents[1]
    service = (root / "deploy/backup/coulomb-social-paired-backup.service").read_text()
    assert "\nLimitFSIZE=1G\n" in service