#!/bin/bash
#   BAREOS® - Backup Archiving REcovery Open Sourced
#
#   Copyright (C) 2026-2026 Bareos GmbH & Co. KG
#
#   This program is Free Software; you can redistribute it and/or
#   modify it under the terms of version three of the GNU Affero General Public
#   License as published by the Free Software Foundation and included
#   in the file LICENSE.
#
#   This program is distributed in the hope that it will be useful, but
#   WITHOUT ANY WARRANTY; without even the implied warranty of
#   MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU
#   Affero General Public License for more details.
#
#   You should have received a copy of the GNU Affero General Public License
#   along with this program; if not, write to the Free Software
#   Foundation, Inc., 51 Franklin Street, Fifth Floor, Boston, MA
#   02110-1301, USA.

set -e
set -o pipefail
set -u
#
# Test the output of "status subscriptions", the real,
# File.LStat-based accounting report, against a backup of a handful of
# files with known, exact sizes -- including one whose size is not a
# multiple of 512 bytes, to exercise the block-rounding path -- plus a
# directory and a symlink to confirm non-regular catalog rows are
# excluded from the file count and byte total. Also runs a second,
# Incremental job with one file resized to confirm the multi-job chain
# de-duplication (latest JobTDate wins per catalog path) still works.
#
record_file_stats()
{
  local phase="$1" f
  if on_windows_host; then
    return
  fi

  {
    echo "=== source stat: ${phase} ==="
    date
    for f in "$tmp"/acct-data/file-*.bin; do
      if [ "$(uname -s)" == FreeBSD ]; then
        stat -f \
          'path=%N size=%z blocks=%b blksize=%k inode=%i mtime=%m ctime=%c' \
          "$f"
      else
        stat -c \
          'path=%n size=%s blocks=%b block-unit=%B inode=%i mtime=%Y ctime=%Z' \
          "$f"
      fi
    done
  } >>"$tmp/accounting-source-stats.log" 2>&1
}

record_filesystem_info()
{
  local dataset
  if on_windows_host; then
    return
  fi

  {
    echo "=== filesystem ==="
    uname -a
    df -T "$tmp/acct-data" || df "$tmp/acct-data"
    dataset="$(df -P "$tmp/acct-data" | awk 'END { print $1 }')"
    if command -v zfs >/dev/null 2>&1; then
      echo "=== ZFS dataset: ${dataset} ==="
      zfs get -H -o name,property,value \
        compression,recordsize,dedup,sync,logicalused,used "$dataset" \
        || true
    fi
  } >>"$tmp/accounting-source-stats.log" 2>&1
}

record_catalog_stats()
{
  local phase="$1"
  run_bconsole - <<END_OF_DATA || true
@$out $tmp/accounting-catalog-${phase}.log
.sql query="SELECT 'CATALOG_ROW jobid=' || j.JobId || \
' level=' || j.Level || ' jobtdate=' || j.JobTDate || \
' pathid=' || f.PathId || ' file=' || f.Name || \
' fileindex=' || f.FileIndex || ' size=' || s.st_size || \
' blocks=' || s.st_blocks || ' mtime=' || s.st_mtime \
FROM File f JOIN Job j USING (JobId) \
JOIN FileSet fs USING (FileSetId) \
CROSS JOIN LATERAL decode_lstat( \
f.LStat, ARRAY['st_size','st_blocks','st_mtime']) s \
WHERE j.Name='$JobName' AND fs.FileSet='AcctTest' \
AND f.Name IN ( \
'file-1byte.bin','file-512bytes.bin','file-1000bytes.bin','hardlink','sparse') \
ORDER BY j.JobTDate, j.JobId, f.PathId, f.Name"
.sql query="SELECT 'SELECTED_ROW jobid=' || recent.JobId || \
' jobtdate=' || recent.JobTDate || ' pathid=' || recent.PathId || \
' file=' || recent.Name || ' fileindex=' || recent.FileIndex || \
' size=' || s.st_size || ' blocks=' || s.st_blocks || \
' mtime=' || s.st_mtime \
FROM (SELECT DISTINCT ON (f.PathId, f.Name) \
f.JobId, j.JobTDate, f.PathId, f.Name, f.FileIndex, f.LStat \
FROM File f JOIN Job j USING (JobId) \
JOIN FileSet fs USING (FileSetId) \
WHERE j.Name='$JobName' AND fs.FileSet='AcctTest' \
AND f.Name IN ( \
'file-1byte.bin','file-512bytes.bin','file-1000bytes.bin','hardlink','sparse') \
ORDER BY f.PathId, f.Name, j.JobTDate DESC, f.JobId DESC, f.FileId DESC) recent \
CROSS JOIN LATERAL decode_lstat( \
recent.LStat, ARRAY['st_size','st_blocks','st_mtime']) s \
WHERE recent.FileIndex<>0 \
ORDER BY recent.PathId, recent.Name"
END_OF_DATA
}

catalog_accounted_totals()
{
  local phase="$1" row size blocks files=0 bytes=0 logical_bytes=0
  while IFS= read -r row; do
    if [[ "$row" =~ SELECTED_ROW.*size=([0-9]+)\ blocks=([0-9]+)\ mtime= ]]; then
      size="${BASH_REMATCH[1]}"
      blocks="${BASH_REMATCH[2]}"
      files=$((files + 1))
      logical_bytes=$((logical_bytes + size))
      if on_windows_host; then
        bytes=$((bytes + size))
      else
        bytes=$((bytes + blocks * 512))
      fi
    fi
  done <"$tmp/accounting-catalog-${phase}.log"
  echo "$files $bytes $logical_bytes"
}

check_logical_size()
{
  local report="$1" expected="$2"
  run_python_check "$report.json" "$expected" <<'PY' || estat=1
row = find_acct_test_row(load_result(sys.argv[1]))
assert row["logical_bytes"] == int(sys.argv[2]), row
PY
}

refresh_accounting_snapshot()
{
  local before_generation after_generation
  cat <<END_OF_DATA >"$tmp/bconcmds"
@$out $tmp/accounting-refresh-before.out
status subscriptions state
END_OF_DATA
  run_bconsole
  before_generation="$(
    sed -nE 's/^Subscription accounting generation: ([0-9]+);.*/\1/p' \
      "$tmp/accounting-refresh-before.out"
  )"
  if [[ -z "$before_generation" ]]; then
    echo "could not read subscription accounting generation before refresh" >&2
    cat "$tmp/accounting-refresh-before.out" >&2
    return 1
  fi

  cat <<END_OF_DATA >"$tmp/bconcmds"
@$out $tmp/accounting-refresh.out
status subscriptions recalculate
END_OF_DATA
  run_bconsole

  for _ in {1..60}; do
    cat <<END_OF_DATA >"$tmp/bconcmds"
@$out $tmp/accounting-refresh-status.out
status subscriptions state
END_OF_DATA
    run_bconsole
    if grep -q "worker: idle." \
      "$tmp/accounting-refresh-status.out"; then
      after_generation="$(
        sed -nE 's/^Subscription accounting generation: ([0-9]+);.*/\1/p' \
          "$tmp/accounting-refresh-status.out"
      )"
      if [[ -z "$after_generation" ]] \
        || ((after_generation <= before_generation)); then
        echo "subscription accounting generation did not advance after refresh" >&2
        cat "$tmp/accounting-refresh-status.out" >&2
        return 1
      fi
      return 0
    fi
    sleep 1
  done
  echo "timed out waiting for explicit subscription accounting refresh" >&2
  cat "$tmp/accounting-refresh.out" >&2
  return 1
}

wait_for_accounting_result()
{
  local report="$1" expected="$2"
  for _ in {1..60}; do
    cat <<END_OF_DATA >"$tmp/bconcmds"
@$out $report
status subscriptions client=bareos-fd
status subscriptions client=does-not-exist
@$out ${NULL_DEV}
.api 2
@$out $report.json
status subscriptions client=bareos-fd
END_OF_DATA
    run_bconsole
    if grep -qE "$expected" "$report"; then
      return 0
    fi
    sleep 1
  done
  echo "timed out waiting for subscription accounting snapshot containing ${expected}" >&2
  if [[ -f "$tmp/accounting-refresh.out" ]]; then
    cat "$tmp/accounting-refresh.out" >&2
  fi
  return 1
}

# Request a full refresh, then write "status subscriptions" as
# JSON to $tmp/status-subscriptions-<name>.json. With "all", the "all" and
# "all legacy" reports are written as <name>-all and <name>-legacy.
query_accounting()
{
  local name="$1" all="${2:-}"
  refresh_accounting_snapshot
  {
    echo ".api 2"
    echo "@$out $tmp/status-subscriptions-${name}.json"
    echo "status subscriptions"
    echo "@$out $tmp/status-subscriptions-${name}-revision.json"
    echo "status subscriptions state"
    if [[ -n "$all" ]]; then
      echo "@$out $tmp/status-subscriptions-${name}-all.json"
      echo "status subscriptions all"
      echo "@$out $tmp/status-subscriptions-${name}-legacy.json"
      echo "status subscriptions all legacy"
    fi
  } >"$tmp/bconcmds"
  run_bconsole
  run_python_check "$tmp/status-subscriptions-${name}-revision.json" <<'PY'
status = load_result(sys.argv[1])["accounting_status"]
assert isinstance(status["generation"], int), status
assert isinstance(status["queued"], bool), status
assert isinstance(status["running"], bool), status
assert status["refresh_thread_state"] in ("unavailable", "queued", "running", "idle"), status
PY
}

# Apply an UPDATE assignment list to the newest catalog row of the resized
# file in the AcctTest chain.
update_resized_file()
{
  run_bconsole - <<END_OF_DATA
.sql query="UPDATE File SET $1 WHERE FileId=(SELECT MAX(f.FileId) FROM File f JOIN Job j USING (JobId) JOIN FileSet fs USING (FileSetId) WHERE j.Name='$JobName' AND fs.FileSet='AcctTest' AND f.Name='file-1000bytes.bin')"
END_OF_DATA
}

# Run the Python check read from stdin, with the helpers of
# accounting_fixture_common available.
run_python_check()
{
  {
    cat <<'PY'
import sys

sys.path.insert(0, ".")
from accounting_fixture_common import find_acct_test_row, load_result

PY
    cat
  } | "${PYTHON_EXECUTABLE}" - "$@"
}

# Verify the catalog selection (latest row per path) after a backup.
verify_catalog()
{
  local phase="$1" label="$2"
  record_catalog_stats "$phase"
  read -r catalog_files catalog_bytes catalog_logical_bytes \
    < <(catalog_accounted_totals "$phase")
  if [[ "$catalog_files" -ne 3 ]]; then
    echo "catalog contained ${catalog_files} selected regular files" \
      "after the ${label}, expected 3!" >&2
    estat=1
  fi
  if [[ "$catalog_logical_bytes" -ne "$raw_bytes" ]]; then
    echo "catalog logical size was ${catalog_logical_bytes} bytes after the ${label}," \
      "expected ${raw_bytes}!" >&2
    estat=1
  fi
}

# The accounting rule differs per client platform (see ua_acct.cc): a
# Windows client's st_blocks is a purely synthetic value derived from
# st_size, so the report uses st_size directly there (exact, deterministic
# match against the known raw byte total). On other platforms st_blocks
# can change as the filesystem commits or compresses data, so compare the
# report to the LStat values stored in the catalog by that backup instead
# of sampling the live files before or after it.
verify_report()
{
  local log="$1" jobs="$2" label="$3" line files bytes actual_rule
  local expected_bytes expected_rule
  line="$(
    run_python_check "$log.json" <<'PY'
row = find_acct_test_row(load_result(sys.argv[1]))
print(row["files"], row["bytes"], row["rule"], row["jobs_in_chain"], end="")
PY
  )"
  if [[ -z "$line" ]]; then
    echo "status subscriptions produced no report line for bareos-fd / AcctTest ${label}!" >&2
    estat=1
    return
  fi
  if [[ ! "$line" =~ ^([0-9]+)\ ([0-9]+)\ ([^[:space:]]+)\ ${jobs}$ ]]; then
    echo "status subscriptions report line ${label} did not match" \
      "the expected format (expected exactly ${jobs} job(s) in chain): ${line}" >&2
    estat=1
    return
  fi
  files="${BASH_REMATCH[1]}"
  bytes="${BASH_REMATCH[2]//,/}"
  actual_rule="${BASH_REMATCH[3]}"
  if [[ "$files" -ne 3 ]]; then
    echo "status subscriptions reported ${files} files ${label}, expected 3!" >&2
    estat=1
  fi
  check_logical_size "$log" "$raw_bytes"
  if on_windows_host; then
    expected_bytes="$raw_bytes"
    expected_rule="st_size"
  else
    expected_bytes="$catalog_bytes"
    expected_rule='st_blocks*512'
  fi
  if [[ "$actual_rule" != "$expected_rule" || "$bytes" -ne "$expected_bytes" ]]; then
    echo "status subscriptions reported '${bytes} bytes (rule: ${actual_rule})' ${label}," \
      "expected '${expected_bytes} bytes (rule: ${expected_rule})'!" >&2
    estat=1
  fi
}

TestName="$(basename "$(pwd)")"
export TestName

#shellcheck source=../../environment.in
. ./environment

JobName=backup-acct-test
#shellcheck source=../../scripts/functions
. "${BAREOS_SCRIPTS_DIR}"/functions

start_test

rm -rf "$tmp/acct-data"
mkdir -p "$tmp/acct-data/edge"

# Known, exact sizes (bytes): one below one block, one exactly one block,
# one spanning into a second (non-512-aligned -- exercises rounding up).
generate_testdata "$tmp/acct-data/file-1byte.bin" --size=1
generate_testdata "$tmp/acct-data/file-512bytes.bin" --size=512
generate_testdata "$tmp/acct-data/file-1000bytes.bin" --size=1000
raw_bytes=$((1 + 512 + 1000))
record_filesystem_info
record_file_stats "after Full test-data creation"

# An (empty) subdirectory and a symlink, both explicitly included in the
# AcctTest fileset alongside the three regular files above. Their catalog
# File rows (FT_DIREND / FT_LNK) are not regular files, so they must not
# add to the reported file count or byte total (see ua_acct.cc's
# S_ISREG() check). Symlinks aren't created on Windows hosts here; a
# missing path is simply skipped by the backup (job status Warning, not
# Error), so the exclusion is only actively exercised on POSIX hosts.
mkdir -p "$tmp/acct-data/subdir"
if ! on_windows_host; then
  ln -sf file-1byte.bin "$tmp/acct-data/symlink-to-file"
fi

cat <<END_OF_DATA >"$tmp/bconcmds"
@$out ${NULL_DEV}
messages
@$out $tmp/status-subscriptions-accounting.log
run job=$JobName level=Full yes
wait
messages
END_OF_DATA

record_file_stats "immediately before Full"
run_bconsole
check_for_zombie_jobs storage=File
record_file_stats "immediately after Full"
wait_for_accounting_result "$tmp/status-subscriptions-accounting.log" \
  "AcctTest@bareos-fd[[:space:]]+[1-9][0-9,]*"
verify_catalog full Full
verify_report "$tmp/status-subscriptions-accounting.log" 1 "after the Full"
if ! grep -Eq 'Grand total: .*B \([0-9,]+ bytes\) across' "$tmp/status-subscriptions-accounting.log"; then
  echo "Accounting totals must include readable sizes followed by exact bytes." >&2
  estat=1
fi
if grep -q 'Logical size' "$tmp/status-subscriptions-accounting.log"; then
  echo "Logical size must be hidden unless explicitly requested." >&2
  estat=1
fi
run_bconsole - <<END_OF_DATA
@$out $tmp/status-subscriptions-logical.log
status subscriptions logical client=bareos-fd
@$out $tmp/status-subscriptions-logical-all.log
status subscriptions all anonymize logical
.api 2
@$out $tmp/status-subscriptions-logical.json
status subscriptions logical client=bareos-fd
END_OF_DATA
for report in "$tmp/status-subscriptions-logical.log" "$tmp/status-subscriptions-logical-all.log"; do
  if ! grep -Eq 'Accounted size[[:space:]]+Logical size[[:space:]]+Rule' "$report" \
    || ! grep -Eq 'Logical size total \(st_size\): .*B \([0-9,]+ bytes\)' "$report"; then
    echo "The logical option must show the logical column and total." >&2
    estat=1
  fi
done
run_python_check "$tmp/status-subscriptions-accounting.log.json" "$tmp/status-subscriptions-logical.json" <<'PY'
default = load_result(sys.argv[1])
logical = load_result(sys.argv[2])
assert default["accounting"] == logical["accounting"]
assert default["summary"] == logical["summary"]
assert "total_logical_bytes" in default["summary"]
assert "logical_bytes" in find_acct_test_row(default)
PY
run_bconsole - <<END_OF_DATA
.sql query="UPDATE SubscriptionAccountingSnapshot SET LastFullSuccess='2026-01-01 00:00:00', LastFullDurationMs=123 WHERE SnapshotId=1"
END_OF_DATA

if ! grep -q "No matching Client/FileSet combinations found." \
  "$tmp/status-subscriptions-accounting.log"; then
  echo "status subscriptions did not exclude a non-existing client as expected!" >&2
  estat=1
fi

# Now grow the chain to two jobs: rewrite one file with a much larger
# size and run an Incremental. This exercises the "latest JobTDate wins
# per (PathId, Name)" de-duplication across multiple jobs in the chain
# (now done server-side via a DISTINCT ON query, see ua_acct.cc), instead
# of only ever seeing a single-job chain where every path is trivially
# unique. The file count must stay at 3 (same catalog paths, not new
# ones. The catalog check below confirms that the selected row for the
# resized file comes from the Incremental. Its st_blocks may not increase
# on filesystems that compress data, so the expected total is computed
# from the selected catalog rows rather than compared to the Full total.
# Ensure the resized file's mtime is unambiguously newer than the Full
# backup's "since" reference time (whole-second granularity), otherwise
# the Incremental may not detect it as changed.
sleep 2
generate_testdata "$tmp/acct-data/file-1000bytes.bin" --size=65536
raw_bytes=$((1 + 512 + 65536))
record_file_stats \
  "after Incremental test-data creation"

cat <<END_OF_DATA >"$tmp/bconcmds"
@$out ${NULL_DEV}
messages
@$out $tmp/status-subscriptions-accounting-incremental.log
run job=$JobName level=Incremental yes
wait
messages
END_OF_DATA

record_file_stats "immediately before Incremental"
run_bconsole
check_for_zombie_jobs storage=File
record_file_stats "immediately after Incremental"
wait_for_accounting_result "$tmp/status-subscriptions-accounting-incremental.log" \
  "AcctTest@bareos-fd[[:space:]]+[1-9][0-9,]*.*[[:space:]]2$"
verify_catalog incremental Incremental
if ! grep -Eq 'SELECTED_ROW .*file=file-1000bytes\.bin .*size=65536 ' \
  "$tmp/accounting-catalog-incremental.log"; then
  echo "catalog did not select the resized file from the Incremental!" >&2
  estat=1
fi
verify_report "$tmp/status-subscriptions-accounting-incremental.log" 2 \
  "after the Incremental (chain dedup broken)"
run_bconsole - <<END_OF_DATA
@$out ${NULL_DEV}
.api 2
@$out $tmp/status-subscriptions-incremental-duration.json
status subscriptions client=bareos-fd
END_OF_DATA
run_python_check "$tmp/status-subscriptions-incremental-duration.json" <<'PY'
metadata = load_result(sys.argv[1])["accounting_snapshot"]
assert metadata["last_full_duration_ms"] == 123
assert metadata["last_full_completed_at"] == "2026-01-01 00:00:00"
PY

# Zero length plus one block is valid outside MSSQL's virtual namespace.
run_bconsole - <<END_OF_DATA
.sql query="CREATE TABLE acct_original_files_backup AS SELECT f.* FROM File f JOIN Job j USING (JobId) JOIN FileSet fs USING (FileSetId) WHERE j.Name='$JobName' AND fs.FileSet='AcctTest'"
END_OF_DATA
update_resized_file "LStat='A A IHA B A A A A QAA B A A A A A A'"
query_accounting empty-regular
run_python_check "$tmp/status-subscriptions-empty-regular.json" <<'PY'
row = find_acct_test_row(load_result(sys.argv[1]))
metadata = load_result(sys.argv[1])["accounting_snapshot"]
assert metadata["last_full_duration_ms"] >= 0
assert metadata["last_full_completed_at"]
assert not row["excluded"]
assert row["logical_bytes"] == 513
PY

# The same placeholder under MSSQL's namespace must fall back for the whole
# combination, even when other measurable files remain in the chain.
run_bconsole - <<END_OF_DATA
.sql query="INSERT INTO Path (Path) VALUES ('/@MSSQL/accounting-test/')"
END_OF_DATA
update_resized_file "PathId=(SELECT PathId FROM Path WHERE Path='/@MSSQL/accounting-test/')"
query_accounting mssql all
run_python_check "$tmp/status-subscriptions-mssql" <<'PY'
base = sys.argv[1]
row = find_acct_test_row(load_result(base + ".json"))
assert not row["excluded"] and row["estimated"] and row["rule"] == "job_bytes", row
assert row["bytes"] > 0 and row["bytes"] == row["logical_bytes"], row
everything = load_result(base + "-all.json")
legacy = load_result(base + "-legacy.json")
for key in ("unit-summary", "unit-detail", "unit-clients", "unit-plugins"):
    assert everything[key] == legacy[key], key
PY

# A genuinely empty file without the MSSQL block-size marker stays measurable.
update_resized_file "LStat='A A IHA B A A A A BAA B A A A A A A'"
query_accounting mssql-empty
run_python_check "$tmp/status-subscriptions-mssql-empty.json" <<'PY'
assert not find_acct_test_row(load_result(sys.argv[1]))["excluded"]
PY

# Replace one real catalog file with the opaque image attributes used by
# barri/NDMP, leaving other measurable files (such as plugin logs) in the chain.
update_resized_file "LStat='A A IHA B A A A -B BAA B A A A A A A'"
query_accounting opaque all
run_python_check "$tmp/status-subscriptions-opaque" <<'PY'
base = sys.argv[1]
row = find_acct_test_row(load_result(base + ".json"))
assert not row["excluded"] and row["estimated"] and row["rule"] == "job_bytes", row
assert row["bytes"] > 0 and row["bytes"] == row["logical_bytes"], row
everything = load_result(base + "-all.json")
legacy = load_result(base + "-legacy.json")
assert everything["subscription_accounting"]["estimated_combinations"] > 0
for key in ("unit-summary", "unit-detail", "unit-clients", "unit-plugins"):
    assert everything[key] == legacy[key], key
PY

# An opaque image whose baseline Full job recorded no size cannot be
# estimated: the combination is excluded and the report flagged incomplete.
run_bconsole - <<END_OF_DATA
.sql query="CREATE TABLE acct_jobsize_backup AS SELECT j.JobId, j.JobBytes, j.ReadBytes FROM Job j JOIN FileSet fs USING (FileSetId) WHERE j.Name='$JobName' AND fs.FileSet='AcctTest'"
.sql query="UPDATE Job SET JobBytes=0, ReadBytes=0 WHERE JobId IN (SELECT JobId FROM acct_jobsize_backup)"
END_OF_DATA
query_accounting opaque-nosize
run_bconsole - <<END_OF_DATA
.sql query="UPDATE Job SET JobBytes=b.JobBytes, ReadBytes=b.ReadBytes FROM acct_jobsize_backup b WHERE Job.JobId=b.JobId"
.sql query="DROP TABLE acct_jobsize_backup"
END_OF_DATA
run_python_check "$tmp/status-subscriptions-opaque-nosize.json" <<'PY'
result = load_result(sys.argv[1])
row = find_acct_test_row(result)
assert row["excluded"] and row["exclusion_reason"] == "opaque_backup_image", row
assert row["baseline_full_jobid"] > 0, row
assert row["accounting_issue"] == (
    "Size unavailable: per-file sizes and a usable baseline Full-job "
    f"(id {row['baseline_full_jobid']}) byte count are both missing."
), row
assert row["accounting_issue"] == (
    row["accounting_issue_before_jobid"] + str(row["baseline_full_jobid"])
    + row["accounting_issue_after_jobid"]
), row
assert "J = job size" in result["accounting_rule_legend"], result
assert "Totals exclude these entries" in result["summary"]["warning"], result["summary"]
assert result["summary"]["incomplete"], result["summary"]
PY

# An NDMP stream container has the same stat, but actual file-history rows
# describe its contents. Preserve their measured accounting.
run_bconsole - <<END_OF_DATA
.sql query="INSERT INTO Path (Path) VALUES ('/@NDMP/accounting-test/')"
END_OF_DATA
update_resized_file "PathId=(SELECT PathId FROM Path WHERE Path='/@NDMP/accounting-test/')"
query_accounting ndmp-history
run_python_check "$tmp/status-subscriptions-ndmp-history.json" <<'PY'
row = find_acct_test_row(load_result(sys.argv[1]))
assert not row["excluded"]
assert row["files"] == 3
assert row["logical_bytes"] == 1513
assert row["bytes"] > 0
PY

# With file history absent, the NDMP container alone must still fall back.
run_bconsole - <<END_OF_DATA
.sql query="DELETE FROM File WHERE JobId IN (SELECT j.JobId FROM Job j JOIN FileSet fs USING (FileSetId) WHERE j.Name='$JobName' AND fs.FileSet='AcctTest') AND PathId NOT IN (SELECT PathId FROM Path WHERE Path='/@NDMP/accounting-test/')"
END_OF_DATA
query_accounting ndmp-no-history
run_python_check "$tmp/status-subscriptions-ndmp-no-history.json" <<'PY'
row = find_acct_test_row(load_result(sys.argv[1]))
assert not row["excluded"] and row["estimated"] and row["rule"] == "job_bytes", row
assert row["bytes"] > 0 and row["bytes"] == row["logical_bytes"], row
assert row["rule_code"] == "J", row
assert "larger of bytes read and bytes stored" in row["rule_explanation"], row
PY

# Native NDMP has no synthetic archive File row; saved restore environment
# still identifies the backup when file history was not requested.
run_bconsole - <<END_OF_DATA
.sql query="CREATE TABLE acct_ndmp_files_backup AS SELECT f.* FROM File f JOIN Job j USING (JobId) JOIN FileSet fs USING (FileSetId) WHERE j.Name='$JobName' AND fs.FileSet='AcctTest'"
.sql query="DELETE FROM File WHERE FileId IN (SELECT FileId FROM acct_ndmp_files_backup)"
.sql query="INSERT INTO NDMPJobEnvironment (JobId,FileIndex,EnvName,EnvValue) SELECT j.JobId,1,'FILESYSTEM','/accounting-test' FROM Job j JOIN FileSet fs USING (FileSetId) WHERE j.Name='$JobName' AND fs.FileSet='AcctTest'"
END_OF_DATA
query_accounting ndmp-native-no-history
run_python_check "$tmp/status-subscriptions-ndmp-native-no-history.json" <<'PY'
row = find_acct_test_row(load_result(sys.argv[1]))
assert not row["excluded"] and row["estimated"] and row["rule"] == "job_bytes", row
assert row["bytes"] > 0 and row["bytes"] == row["logical_bytes"], row
PY
run_bconsole - <<END_OF_DATA
.sql query="CREATE TABLE acct_native_jobsize_backup AS SELECT j.JobId, j.JobBytes, j.ReadBytes FROM Job j JOIN FileSet fs USING (FileSetId) WHERE j.Name='$JobName' AND fs.FileSet='AcctTest'"
.sql query="UPDATE Job SET JobBytes=0, ReadBytes=0 WHERE JobId IN (SELECT JobId FROM acct_native_jobsize_backup)"
END_OF_DATA
query_accounting ndmp-native-no-size
run_python_check "$tmp/status-subscriptions-ndmp-native-no-size.json" <<'PY'
row = find_acct_test_row(load_result(sys.argv[1]))
assert row["excluded"] and row["exclusion_reason"] == "ndmp_no_file_history", row
assert row["baseline_full_jobid"] > 0, row
PY
run_bconsole - <<END_OF_DATA
.sql query="UPDATE Job SET JobBytes=b.JobBytes, ReadBytes=b.ReadBytes FROM acct_native_jobsize_backup b WHERE Job.JobId=b.JobId"
.sql query="DROP TABLE acct_native_jobsize_backup"
.sql query="DELETE FROM NDMPJobEnvironment WHERE JobId IN (SELECT j.JobId FROM Job j JOIN FileSet fs USING (FileSetId) WHERE j.Name='$JobName' AND fs.FileSet='AcctTest') AND EnvName='FILESYSTEM' AND EnvValue='/accounting-test'"
.sql query="INSERT INTO File SELECT * FROM acct_ndmp_files_backup"
.sql query="DROP TABLE acct_ndmp_files_backup"
END_OF_DATA

# Without either file history or a baseline Full size, NDMP remains excluded.
run_bconsole - <<END_OF_DATA
.sql query="CREATE TABLE acct_ndmp_jobsize_backup AS SELECT j.JobId, j.JobBytes, j.ReadBytes FROM Job j JOIN FileSet fs USING (FileSetId) WHERE j.Name='$JobName' AND fs.FileSet='AcctTest'"
.sql query="UPDATE Job SET JobBytes=0, ReadBytes=0 WHERE JobId IN (SELECT JobId FROM acct_ndmp_jobsize_backup)"
END_OF_DATA
query_accounting ndmp-no-size
run_bconsole - <<END_OF_DATA
.sql query="UPDATE Job SET JobBytes=b.JobBytes, ReadBytes=b.ReadBytes FROM acct_ndmp_jobsize_backup b WHERE Job.JobId=b.JobId"
.sql query="DROP TABLE acct_ndmp_jobsize_backup"
END_OF_DATA
run_python_check "$tmp/status-subscriptions-ndmp-no-size.json" <<'PY'
row = find_acct_test_row(load_result(sys.argv[1]))
assert row["excluded"] and row["exclusion_reason"] == "ndmp_no_file_history", row
PY

# Real NDMP_BAREOS and NDMP_NATIVE catalog exports from PR #2855.
# Import each Full first, then its Incremental, to verify both baselines
# and latest-file de-duplication without needing an NDMP server here.
for phase in full incremental; do
  "${PYTHON_EXECUTABLE}" ./ndmp_accounting_fixture.py import "$phase" \
    >"$tmp/ndmp-accounting-import-${phase}.bcon"
  run_bconsole "$tmp/ndmp-accounting-import-${phase}.bcon"
  query_accounting "real-ndmp-${phase}"
  "${PYTHON_EXECUTABLE}" ./ndmp_accounting_fixture.py check "$phase" \
    "$tmp/status-subscriptions-real-ndmp-${phase}.json"
done

# Import privacy-reduced samples of actual Barri, Windows, and MSSQL
# catalog attributes from supplied dumps. The encoded LStat values are
# preserved; names and paths are anonymized in the fixture.
for phase in full incremental; do
  "${PYTHON_EXECUTABLE}" ./plugin_accounting_fixture.py import "$phase" \
    >"$tmp/plugin-accounting-import-${phase}.bcon"
  run_bconsole "$tmp/plugin-accounting-import-${phase}.bcon"
  query_accounting "plugin-fixtures-${phase}"
  "${PYTHON_EXECUTABLE}" ./plugin_accounting_fixture.py check \
    "$phase" "$tmp/status-subscriptions-plugin-fixtures-${phase}.json"
done

"${PYTHON_EXECUTABLE}" ./accounting_edge_fixture.py import \
  >"$tmp/accounting-edges-import.bcon"
run_bconsole "$tmp/accounting-edges-import.bcon"
query_accounting chain-edges
"${PYTHON_EXECUTABLE}" ./accounting_edge_fixture.py check \
  "$tmp/status-subscriptions-chain-edges.json"
{
  echo ".api 2"
  echo "@$out $tmp/status-subscriptions-chain-edges-client.json"
  echo "status subscriptions client=acct-chain-edges"
} >"$tmp/bconcmds"
run_bconsole
"${PYTHON_EXECUTABLE}" ./accounting_edge_fixture.py check-versioned-queue \
  "$tmp/status-subscriptions-chain-edges-client.json"

run_bconsole - <<END_OF_DATA
.sql query="DELETE FROM File WHERE JobId IN (SELECT JobId FROM acct_original_files_backup)"
.sql query="INSERT INTO File SELECT * FROM acct_original_files_backup"
.sql query="DROP TABLE acct_original_files_backup"
END_OF_DATA

# Exercise real sparse and hard-linked files plus Accurate deletion, including
# the per-job delta update before comparing it with a full recalculation.
if ! on_windows_host; then
  ln "$tmp/acct-data/file-512bytes.bin" "$tmp/acct-data/edge/hardlink"
  "${PYTHON_EXECUTABLE}" - "$tmp/acct-data/edge/sparse" <<'PY'
import sys

with open(sys.argv[1], "wb") as stream:
    stream.truncate(8 * 1024 * 1024)
PY
  run_bconsole - <<END_OF_DATA
@$out $tmp/accounting-edges-full.log
run job=$JobName level=Full accurate=yes yes
wait
END_OF_DATA
  wait_for_accounting_result "$tmp/accounting-edges-full.log" \
    "AcctTest@bareos-fd[[:space:]]+5[[:space:]]+.*[[:space:]]1$"
  record_catalog_stats sparse-full
  read -r catalog_files catalog_bytes catalog_logical_bytes \
    < <(catalog_accounted_totals sparse-full)
  run_python_check "$tmp/accounting-edges-full.log.json" "$tmp/acct-data" \
    "$catalog_files" "$catalog_bytes" "$catalog_logical_bytes" <<'PY'
import errno
import os
import stat

row = find_acct_test_row(load_result(sys.argv[1]))
paths = ["file-1byte.bin", "file-512bytes.bin", "file-1000bytes.bin",
         "edge/hardlink", "edge/sparse"]
# Change allocation after the backup without changing the saved file contents.
if hasattr(os, "posix_fallocate"):
    with open(os.path.join(sys.argv[2], "edge/sparse"), "r+b") as stream:
        file_stat = os.fstat(stream.fileno())
        assert stat.S_ISREG(file_stat.st_mode) and file_stat.st_size > 0, file_stat
        unsupported_errors = (errno.ENOSYS, errno.EOPNOTSUPP)
        if sys.platform.startswith("freebsd"):
            # FreeBSD also uses EINVAL for unsupported filesystems.
            unsupported_errors += (errno.EINVAL,)
        try:
            os.posix_fallocate(stream.fileno(), 0, file_stat.st_size)
        except OSError as error:
            if error.errno not in unsupported_errors:
                raise
            print(f"Post-backup allocation probe unsupported: {error}")
        else:
            os.fsync(stream.fileno())
stats = [os.stat(os.path.join(sys.argv[2], path)) for path in paths]
assert row["files"] == len(stats), row
assert row["files"] == int(sys.argv[3]), row
assert row["bytes"] == int(sys.argv[4]), row
assert row["logical_bytes"] == int(sys.argv[5]), row
assert row["logical_bytes"] == sum(stat.st_size for stat in stats), row
assert row["bytes"] < row["logical_bytes"], row
assert stats[1].st_ino == stats[3].st_ino
PY
  rm "$tmp/acct-data/file-1byte.bin"
  sleep 2
  run_bconsole - <<END_OF_DATA
@$out $tmp/accounting-edges-delete.log
run job=$JobName level=Incremental accurate=yes yes
wait
.sql query="SELECT 'DELETED_ACCOUNTING_FILE=' || COUNT(*) FROM File f JOIN Job j USING (JobId) WHERE j.Name='$JobName' AND j.JobId=(SELECT MAX(JobId) FROM Job WHERE Name='$JobName') AND f.Name='file-1byte.bin' AND f.FileIndex=0"
END_OF_DATA
  grep -q 'DELETED_ACCOUNTING_FILE=1' "$tmp/accounting-edges-delete.log"
  wait_for_accounting_result "$tmp/accounting-edges-delete.log" \
    "AcctTest@bareos-fd[[:space:]]+4[[:space:]]+.*[[:space:]]2$"
  record_catalog_stats sparse-incremental
  read -r catalog_files catalog_bytes catalog_logical_bytes \
    < <(catalog_accounted_totals sparse-incremental)
  run_python_check "$tmp/accounting-edges-full.log.json" \
    "$tmp/accounting-edges-delete.log.json" "$tmp/acct-data" \
    "$catalog_files" "$catalog_bytes" "$catalog_logical_bytes" <<'PY'
import os

before = find_acct_test_row(load_result(sys.argv[1]))
after = find_acct_test_row(load_result(sys.argv[2]))
assert after["files"] == before["files"] - 1, (before, after)
assert after["logical_bytes"] == before["logical_bytes"] - 1, (before, after)
paths = ["file-512bytes.bin", "file-1000bytes.bin", "edge/hardlink", "edge/sparse"]
stats = [os.stat(os.path.join(sys.argv[3], path)) for path in paths]
assert after["files"] == len(stats) == int(sys.argv[4]), after
assert after["bytes"] == int(sys.argv[5]), after
assert after["logical_bytes"] == int(sys.argv[6]), after
assert after["logical_bytes"] == sum(stat.st_size for stat in stats), after
PY
  query_accounting accurate-recalculated
  run_python_check "$tmp/accounting-edges-delete.log.json" \
    "$tmp/status-subscriptions-accurate-recalculated.json" <<'PY'
assert find_acct_test_row(load_result(sys.argv[1])) == find_acct_test_row(
    load_result(sys.argv[2]))
PY
fi

# Failed scans must leave the entire last good snapshot and timing intact.
query_accounting before-invalid
run_bconsole - <<END_OF_DATA
.sql query="CREATE TABLE acct_invalid_backup AS SELECT f.FileId, f.LStat FROM File f JOIN Job j USING (JobId) JOIN FileSet fs USING (FileSetId) WHERE j.Name='$JobName' AND fs.FileSet='AcctTest'"
END_OF_DATA
invalid_cases="empty size logical-total-overflow"
if ! on_windows_host; then
  invalid_cases="$invalid_cases blocks block-overflow byte-total-overflow"
fi
for invalid in $invalid_cases; do
  case "$invalid" in
    empty) stat="" ;;
    size) stat="A A IGk B A A A -C BAA B A A A A A A" ;;
    blocks) stat="A A IGk B A A A B BAA -C A A A A A A" ;;
    block-overflow) stat="A A IGk B A A A B BAA CAAAAAAAAA A A A A A A" ;;
    byte-total-overflow) stat="A A IGk B A A A B BAA BAAAAAAAAA A A A A A A" ;;
    logical-total-overflow) stat="A A IGk B A A A H////////// BAA A A A A A A A" ;;
  esac
  run_bconsole - <<END_OF_DATA
.sql query="UPDATE File SET LStat='$stat' WHERE FileId IN (SELECT FileId FROM acct_invalid_backup)"
END_OF_DATA
  query_accounting "invalid-$invalid"
  run_python_check "$tmp/status-subscriptions-before-invalid.json" \
    "$tmp/status-subscriptions-invalid-$invalid.json" "$invalid" <<'PY'
import json

before = load_result(sys.argv[1])
after = load_result(sys.argv[2])
assert after["accounting"] == before["accounting"], (before, after)
assert after["summary"] == before["summary"], (before, after)
old = before["accounting_snapshot"]
new = after["accounting_snapshot"]
if sys.argv[3] in ("empty", "size", "blocks", "block-overflow"):
    assert new["last_refresh_error"].startswith("Invalid file attributes: "), new
    diagnostic = new["last_refresh_diagnostic"]
    assert diagnostic["version"] == 1, diagnostic
    assert diagnostic["type"] == "invalid_lstat", diagnostic
    assert diagnostic["jobid"] > 0 and diagnostic["fileindex"] > 0, diagnostic
    assert diagnostic["path"] and diagnostic["client"] == "bareos-fd", diagnostic
    assert diagnostic["fileset"] == "AcctTest", diagnostic
    assert diagnostic["reason"] in new["last_refresh_error"], diagnostic
    assert diagnostic["path"] in new["last_refresh_error"], diagnostic
else:
    assert new["last_refresh_error"] == "failed to scan file attributes", new
    assert "last_refresh_diagnostic" not in new, new
for key in ("calculated_at", "last_full_completed_at", "last_full_duration_ms"):
    assert new[key] == old[key], (key, old, new)
PY
  run_bconsole - <<END_OF_DATA
@$out $tmp/status-subscriptions-invalid-$invalid-storage.out
.sql query="SELECT 'STRUCTURED_ERROR=' || CASE WHEN LastError LIKE 'bareos-subscription-accounting-error:v1:%' THEN 'true' ELSE 'false' END FROM SubscriptionAccountingSnapshot WHERE SnapshotId=1"
END_OF_DATA
  grep -q 'STRUCTURED_ERROR=true' \
    "$tmp/status-subscriptions-invalid-$invalid-storage.out" || {
    if [[ "$invalid" == "empty" || "$invalid" == "size" ||
      "$invalid" == "blocks" || "$invalid" == "block-overflow" ]]; then
      cat "$tmp/status-subscriptions-invalid-$invalid-storage.out" >&2
      exit 1
    fi
  }
  run_bconsole - <<END_OF_DATA
.sql query="UPDATE File SET LStat=b.LStat FROM acct_invalid_backup b WHERE File.FileId=b.FileId"
END_OF_DATA
done

{
  echo ".api 2"
  echo "@$out $tmp/status-subscriptions-invalid-anonymized.json"
  echo "status subscriptions all anonymize"
} >"$tmp/bconcmds"
run_bconsole
run_python_check "$tmp/status-subscriptions-invalid-empty.json" \
  "$tmp/status-subscriptions-invalid-anonymized.json" <<'PY'
import json

normal = load_result(sys.argv[1])["accounting_snapshot"]
anonymized = load_result(sys.argv[2])["accounting_snapshot"]
assert anonymized["last_refresh_error"] == "Last recalculation failed."
assert "last_refresh_diagnostic" not in anonymized
serialized = json.dumps(anonymized)
for key in ("path", "client", "fileset", "jobid", "fileindex", "reason"):
    assert key not in serialized, serialized
for value in (normal["last_refresh_diagnostic"]["path"],
              normal["last_refresh_diagnostic"]["client"],
              normal["last_refresh_diagnostic"]["fileset"]):
    assert value not in serialized, value
PY

run_bconsole - <<END_OF_DATA
.sql query="UPDATE SubscriptionAccountingSnapshot SET LastError='legacy refresh failed' WHERE SnapshotId=1"
END_OF_DATA
{
  echo ".api 2"
  echo "@$out $tmp/status-subscriptions-legacy-error.json"
  echo "status subscriptions"
} >"$tmp/bconcmds"
run_bconsole
run_python_check "$tmp/status-subscriptions-legacy-error.json" <<'PY'
snapshot = load_result(sys.argv[1])["accounting_snapshot"]
assert snapshot["last_refresh_error"] == "legacy refresh failed", snapshot
assert "last_refresh_diagnostic" not in snapshot
PY

run_bconsole - <<END_OF_DATA
.sql query="UPDATE SubscriptionAccountingSnapshot SET LastError='bareos-subscription-accounting-error:v1:{bad' WHERE SnapshotId=1"
END_OF_DATA
{
  echo ".api 2"
  echo "@$out $tmp/status-subscriptions-malformed-error.json"
  echo "status subscriptions"
} >"$tmp/bconcmds"
run_bconsole
run_python_check "$tmp/status-subscriptions-malformed-error.json" <<'PY'
snapshot = load_result(sys.argv[1])["accounting_snapshot"]
assert "Malformed structured" in snapshot["last_refresh_error"], snapshot
assert "last_refresh_diagnostic" not in snapshot
PY

run_bconsole - <<END_OF_DATA
.sql query="DROP TABLE acct_invalid_backup"
.sql query="UPDATE SubscriptionAccountingSnapshot SET LastError=NULL WHERE SnapshotId=1"
END_OF_DATA
query_accounting after-invalid
run_python_check "$tmp/status-subscriptions-after-invalid.json" <<'PY'
assert not load_result(sys.argv[1])["accounting_snapshot"].get("last_refresh_error")
PY

# Human-readable error text must neutralize terminal controls and backslashes,
# while structured diagnostics retain the original catalog path.
query_accounting before-control-invalid
run_bconsole - <<END_OF_DATA
.sql query="CREATE TABLE acct_control_invalid_backup AS SELECT f.FileId, f.Name, f.LStat FROM File f JOIN Job j USING (JobId) JOIN FileSet fs USING (FileSetId) WHERE j.Name='$JobName' AND fs.FileSet='AcctTest' AND f.Name='file-1000bytes.bin'"
.sql query="UPDATE File SET Name='control' || chr(92) || chr(9) || chr(10) || chr(13) || chr(27) || chr(127) || 'end.bin', LStat='' WHERE FileId IN (SELECT FileId FROM acct_control_invalid_backup)"
END_OF_DATA
query_accounting invalid-control
run_python_check "$tmp/status-subscriptions-before-control-invalid.json" \
  "$tmp/status-subscriptions-invalid-control.json" <<'PY'
before = load_result(sys.argv[1])
after = load_result(sys.argv[2])
old = before["accounting_snapshot"]
new = after["accounting_snapshot"]
assert after["accounting"] == before["accounting"], (before, after)
assert after["summary"] == before["summary"], (before, after)
for key in ("calculated_at", "last_full_completed_at", "last_full_duration_ms"):
    assert new[key] == old[key], (key, old, new)
diagnostic = new["last_refresh_diagnostic"]
raw_name = "control" + chr(92) + chr(9) + chr(10) + chr(13) + chr(27) + chr(127) + "end.bin"
assert diagnostic["path"].endswith(raw_name), diagnostic
message = new["last_refresh_error"]
assert all(token in message for token in ("\\\\", "\\x09", "\\x0A",
                                           "\\x0D", "\\x1B", "\\x7F")), message
assert not any(ord(character) < 0x20 or ord(character) == 0x7f
               for character in message), repr(message)
PY

# A versioned payload's message may predate safe formatting or be tampered
# with; status output must be reconstructed from and escape the raw fields.
run_bconsole - <<END_OF_DATA
.sql query="UPDATE SubscriptionAccountingSnapshot SET LastError=substring(LastError from 1 for length('bareos-subscription-accounting-error:v1:')) || jsonb_set(substring(LastError from length('bareos-subscription-accounting-error:v1:') + 1)::jsonb, '{message}', to_jsonb('UNSAFE' || chr(10) || chr(27) || 'MESSAGE'))::text WHERE SnapshotId=1"
END_OF_DATA
{
  echo ".api 2"
  echo "@$out $tmp/status-subscriptions-malicious-message.json"
  echo "status subscriptions"
} >"$tmp/bconcmds"
run_bconsole
run_python_check "$tmp/status-subscriptions-malicious-message.json" <<'PY'
snapshot = load_result(sys.argv[1])["accounting_snapshot"]
message = snapshot["last_refresh_error"]
assert "UNSAFE" not in message and "MESSAGE" not in message, message
assert not any(ord(character) < 0x20 or ord(character) == 0x7f
               for character in message), repr(message)
assert "\\x0A" in message and "\\x1B" in message, message
assert snapshot["last_refresh_diagnostic"]["path"].endswith(
    "control" + chr(92) + chr(9) + chr(10) + chr(13) + chr(27) + chr(127) + "end.bin"
)
PY
run_bconsole - <<END_OF_DATA
.sql query="UPDATE File SET Name=b.Name, LStat=b.LStat FROM acct_control_invalid_backup b WHERE File.FileId=b.FileId"
.sql query="DROP TABLE acct_control_invalid_backup"
.sql query="UPDATE SubscriptionAccountingSnapshot SET LastError=NULL WHERE SnapshotId=1"
END_OF_DATA
query_accounting after-control-invalid

# A delta must identify the prior effective row, rather than blaming the
# Incremental that exposed its invalid attributes.
query_accounting delta-before
run_bconsole - <<END_OF_DATA
.sql query="CREATE TABLE acct_delta_invalid_backup AS SELECT recent.FileId, recent.JobId, recent.LStat FROM (SELECT DISTINCT ON (f.PathId,f.Name) f.FileId, f.JobId, f.LStat, f.Name FROM File f JOIN Job j USING (JobId) JOIN FileSet fs USING (FileSetId) WHERE j.Name='$JobName' AND fs.FileSet='AcctTest' AND f.Name='file-1000bytes.bin' ORDER BY f.PathId,f.Name,j.JobTDate DESC,j.JobId DESC,f.FileId DESC) recent"
@$out $tmp/accounting-delta-previous-jobid.out
.sql query="SELECT 'DELTA_PREVIOUS_JOBID=' || JobId FROM acct_delta_invalid_backup"
.sql query="UPDATE File SET LStat='' WHERE FileId IN (SELECT FileId FROM acct_delta_invalid_backup)"
END_OF_DATA
delta_previous_jobid="$(
  sed -nE 's/.*DELTA_PREVIOUS_JOBID=([0-9]+).*/\1/p' \
    "$tmp/accounting-delta-previous-jobid.out"
)"
if [[ -z "$delta_previous_jobid" ]]; then
  echo "could not find the previous effective accounting JobId" >&2
  cat "$tmp/accounting-delta-previous-jobid.out" >&2
  exit 1
fi
printf 'delta\n' >>"$tmp/acct-data/file-1000bytes.bin"
run_bconsole - <<END_OF_DATA
@$out $tmp/accounting-invalid-delta-job.log
run job=$JobName level=Incremental accurate=yes yes
wait
END_OF_DATA
delta_before_generation="$(
  "${PYTHON_EXECUTABLE}" - \
    "$tmp/status-subscriptions-delta-before-revision.json" <<'PY'
import json
import sys

text = open(sys.argv[1], encoding="utf-8").read()
result, _ = json.JSONDecoder().raw_decode(text[text.index("{"):])
print(result["result"]["accounting_status"]["generation"])
PY
)"
for _ in {1..60}; do
  {
    echo ".api 2"
    echo "@$out $tmp/accounting-invalid-delta-state.json"
    echo "status subscriptions state"
  } >"$tmp/bconcmds"
  run_bconsole
  if "${PYTHON_EXECUTABLE}" - "$tmp/accounting-invalid-delta-state.json" \
    "$delta_before_generation" <<'PY'; then
import json
import sys

text = open(sys.argv[1], encoding="utf-8").read()
result, _ = json.JSONDecoder().raw_decode(text[text.index("{"):])
status = result["result"]["accounting_status"]
sys.exit(not (
    status["refresh_thread_state"] == "idle"
    and     status["generation"] > int(sys.argv[2])
))
PY
    break
  fi
  sleep 1
done
{
  echo ".api 2"
  echo "@$out $tmp/status-subscriptions-invalid-delta.json"
  echo "status subscriptions"
} >"$tmp/bconcmds"
run_bconsole
run_python_check "$tmp/status-subscriptions-delta-before.json" \
  "$tmp/status-subscriptions-invalid-delta.json" \
  "$delta_previous_jobid" <<'PY'
before = load_result(sys.argv[1])
after = load_result(sys.argv[2])
diagnostic = after["accounting_snapshot"]["last_refresh_diagnostic"]
assert diagnostic["jobid"] == int(sys.argv[3]), diagnostic
assert diagnostic["path"].endswith("file-1000bytes.bin"), diagnostic
assert diagnostic["fileindex"] > 0, diagnostic
assert diagnostic["client"] == "bareos-fd", diagnostic
assert diagnostic["fileset"] == "AcctTest", diagnostic
assert diagnostic["reason"] == "LStat is empty", diagnostic
message = after["accounting_snapshot"]["last_refresh_error"]
assert all(value in message for value in ("JobId=", "FileIndex=", "Client=",
                                           "FileSet=", "Path=", "LStat is empty")), message
assert not any(ord(character) < 0x20 or ord(character) == 0x7f
               for character in message), repr(message)
assert after["accounting"] == before["accounting"], (before, after)
assert after["summary"] == before["summary"], (before, after)
for key in ("calculated_at", "last_full_completed_at", "last_full_duration_ms"):
    assert after["accounting_snapshot"][key] == before["accounting_snapshot"][key]
PY
run_bconsole - <<END_OF_DATA
.sql query="UPDATE File SET LStat=b.LStat FROM acct_delta_invalid_backup b WHERE File.FileId=b.FileId"
.sql query="DROP TABLE acct_delta_invalid_backup"
END_OF_DATA
query_accounting after-invalid-delta

if [[ "${estat:-0}" -ne 0 ]]; then
  echo "=== accounting source/filesystem diagnostics ===" >&2
  if [[ -f "$tmp/accounting-source-stats.log" ]]; then
    cat "$tmp/accounting-source-stats.log" >&2
  fi
  echo "=== catalog after Full ===" >&2
  cat "$tmp/accounting-catalog-full.log" >&2
  echo "=== catalog after Incremental ===" >&2
  cat "$tmp/accounting-catalog-incremental.log" >&2
fi

end_test
