simbiot-harness/tests/test_member_remove.py

# -*- coding: utf-8 -*-
"""t-261018141319: member_remove — безопасное удаление члена (ядро domains/member_ops.py).

Запуск: ./venv/bin/python tests/test_member_remove.py

Сеть/портал НЕ поднимаются: ядро flask-free (sqlite3 + stdlib), тестируется
на временной базе. Проверяет:
- happy-path: users/login_tokens/project_members зачищены, снимок в
  member_archive + json-файл бэкапа (снимок содержит user/shares/votes);
- паи (shares) и голоса (member_votes) НЕ удаляются — финансовая история;
- guardrail: Председателя удалить нельзя (chairman_protected);
- неизвестный id → not_found, без исключения;
- колбэк on_event (MemberRemoved) вызывается с reason/actor;
- файловый бэкап недоступен (dead dir) — удаление всё равно ок (best effort).
"""
import json
import os
import shutil
import sqlite3
import sys
import tempfile

sys.path.insert(0, "/root/simbiot-harness")
sys.path.insert(0, "/root/coop")

from domains import member_ops  # noqa: E402

fails = []


def check(name, cond, extra=""):
    print(("PASS" if cond else "FAIL"), name, ("" if cond else " | " + str(extra)[:300]))
    if not cond:
        fails.append(name)


_tmp = tempfile.mkdtemp(prefix="simbiot_memberremove_")
DB = os.path.join(_tmp, "coop.db")
ARCH = os.path.join(_tmp, "archive")

db = sqlite3.connect(DB)
db.executescript("""
CREATE TABLE users (id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT UNIQUE NOT NULL,
    name TEXT NOT NULL, password TEXT NOT NULL, email TEXT DEFAULT '', role TEXT DEFAULT 'Член',
    avatar TEXT DEFAULT '', created_at TEXT DEFAULT '', is_board INTEGER DEFAULT 0, bio TEXT DEFAULT '');
CREATE TABLE login_tokens (token TEXT PRIMARY KEY, email TEXT NOT NULL, username TEXT DEFAULT '',
    created_at TEXT DEFAULT '', used INTEGER DEFAULT 0);
CREATE TABLE project_members (id INTEGER PRIMARY KEY AUTOINCREMENT, project_id TEXT NOT NULL,
    member_type TEXT NOT NULL DEFAULT 'user', username TEXT NOT NULL, role TEXT NOT NULL DEFAULT 'member');
CREATE TABLE shares (id INTEGER PRIMARY KEY AUTOINCREMENT, username TEXT NOT NULL,
    usd_amount REAL NOT NULL, note TEXT DEFAULT '');
CREATE TABLE member_votes (application_id INTEGER NOT NULL, username TEXT NOT NULL, vote TEXT NOT NULL);
INSERT INTO users (username, name, password, email, role) VALUES
    ('alex', 'Alex', 'x', 'alex@x.io', 'Председатель'),
    ('serg_test', 'Serg Test', 'x', 'serg.test@x.io', 'Член');
INSERT INTO login_tokens (token, email, username) VALUES ('t1', 'serg.test@x.io', 'serg_test');
INSERT INTO project_members (project_id, member_type, username) VALUES ('vybirai', 'user', 'serg_test');
INSERT INTO shares (username, usd_amount, note) VALUES ('serg_test', 500.0, 'паевой');
INSERT INTO member_votes (application_id, username, vote) VALUES (1, 'serg_test', 'yes');
""")
db.commit()
db.close()


def q(sql, *p):
    db = sqlite3.connect(DB)
    r = db.execute(sql, p).fetchall()
    db.close()
    return r


events = []

# ── guardrail: Председателя не трогаем ───────────────────────────────────────
res = member_ops.remove_member(DB, 1, reason="тест", actor="t", archive_dir=ARCH)
check("chairman protected", res.get("error") == "chairman_protected", res)
check("chairman still in users", q("SELECT COUNT(*) FROM users WHERE username='alex'")[0][0] == 1)

# ── not_found ────────────────────────────────────────────────────────────────
res = member_ops.remove_member(DB, 999, reason="нет такого", actor="t", archive_dir=ARCH)
check("not_found without exception", res.get("error") == "not_found", res)

# ── happy-path: удаление члена с бэкапом и аудитом ───────────────────────────
def on_event(username, payload, actor):
    events.append((username, payload, actor))


res = member_ops.remove_member(DB, 2, reason="вышел из Labs, m-516175137",
                               actor="mcp:simbiot", on_event=on_event,
                               archive_dir=ARCH)
check("ok removed", bool(res.get("ok")), res)
check("username in result", res.get("username") == "serg_test", res)
check("users row gone", q("SELECT COUNT(*) FROM users WHERE username='serg_test'")[0][0] == 0)
check("login_tokens cleaned", q("SELECT COUNT(*) FROM login_tokens WHERE username='serg_test' OR email='serg.test@x.io'")[0][0] == 0)
check("project_members cleaned", q("SELECT COUNT(*) FROM project_members WHERE username='serg_test'")[0][0] == 0)
check("shares kept (финансовая история)", q("SELECT COUNT(*) FROM shares WHERE username='serg_test'")[0][0] == 1)
check("votes kept", q("SELECT COUNT(*) FROM member_votes WHERE username='serg_test'")[0][0] == 1)

arch_rows = q("SELECT user_id, username, snapshot, reason, actor FROM member_archive")
check("member_archive row", len(arch_rows) == 1, arch_rows)
if arch_rows:
    snap = json.loads(arch_rows[0][2])
    check("snapshot user email", snap["user"]["email"] == "serg.test@x.io", snap.get("user"))
    check("snapshot shares inside", len(snap.get("shares", [])) == 1, snap.get("shares"))
    check("snapshot votes inside", len(snap.get("member_votes", [])) == 1, snap.get("member_votes"))
    check("archive reason/actor", arch_rows[0][3] == "вышел из Labs, m-516175137" and arch_rows[0][4] == "mcp:simbiot")

check("json backup file written", isinstance(res.get("archive_file"), str)
      and os.path.exists(res["archive_file"]), res.get("archive_file"))
if res.get("archive_file"):
    disk = json.load(open(res["archive_file"], encoding="utf-8"))
    check("json backup has user+shares", disk["user"]["username"] == "serg_test"
          and len(disk.get("shares", [])) == 1)

check("on_event fired once with payload", len(events) == 1
      and events[0][0] == "serg_test"
      and events[0][1].get("reason") == "вышел из Labs, m-516175137"
      and events[0][1].get("user_id") == 2, events)

# ── файловый бэкап недоступен — удаление всё равно проходит ─────────────────
db = sqlite3.connect(DB)
db.execute("INSERT INTO users (username, name, password, email, role) VALUES ('tmp2', 'T2', 'x', 't2@x.io', 'Член')")
db.commit()
# путь-ловушка: archive_dir указывает на обычный файл — makedirs упадёт с OSError
# (chmod для root не барьер, поэтому симулируем отказ так)
not_a_dir = os.path.join(_tmp, "blocked_file.json")
open(not_a_dir, "w").write("i am a file")
res2 = member_ops.remove_member(DB, q("SELECT id FROM users WHERE username='tmp2'")[0][0],
                                reason="x", actor="t", archive_dir=not_a_dir)
check("removed even if file backup fails", bool(res2.get("ok")) and res2.get("archive_file") is None, res2)
check("second archive row", q("SELECT COUNT(*) FROM member_archive")[0][0] == 2)

# ── идемпотентность повторного вызова: уже удалён → not_found ───────────────
res3 = member_ops.remove_member(DB, 2, reason="повтор", actor="t", archive_dir=ARCH)
check("re-remove → not_found", res3.get("error") == "not_found", res3)

shutil.rmtree(_tmp, ignore_errors=True)
print(("ALL PASS" if not fails else "FAILURES: " + ", ".join(fails)))
sys.exit(1 if fails else 0)