ObsiGate/tests/test_user_home.py

# tests/test_user_home.py — #194 dossier personnel par utilisateur
import asyncio
import json
import os

import pytest


def _run(coro):
    loop = asyncio.new_event_loop()
    try:
        return loop.run_until_complete(coro)
    finally:
        loop.close()


@pytest.fixture
def home_env(tmp_path, monkeypatch):
    """Racine Home + registre vaults + users.json isolés, globals sauvegardés."""
    root = tmp_path / "Home"
    root.mkdir()
    monkeypatch.setenv("OBSIGATE_HOME_ROOT", str(root))

    from backend import indexer

    monkeypatch.setattr(indexer, "DYNAMIC_VAULTS_FILE", tmp_path / "vaults.json")

    from backend.auth import user_store

    monkeypatch.setattr(user_store, "USERS_FILE", tmp_path / "users.json")

    from backend import watcher_state

    monkeypatch.setattr(watcher_state, "get_watcher", lambda: None)

    saved_index = dict(indexer.index)
    saved_config = dict(indexer.vault_config)
    yield root
    indexer.index.clear()
    indexer.index.update(saved_index)
    indexer.vault_config.clear()
    indexer.vault_config.update(saved_config)


def _create_user(name: str) -> dict:
    from backend.auth.user_store import create_user

    return create_user(name, "Motdepasse1!", role="user")


def _registry() -> dict:
    """Contenu du registre dynamique (chemin monkeypatché par home_env)."""
    from backend import indexer

    return json.loads(indexer.DYNAMIC_VAULTS_FILE.read_text(encoding="utf-8"))


class TestEnsureUserHome:
    def test_creates_folder_vault_and_grant(self, home_env):
        from backend.auth.user_store import get_user
        from backend.indexer import index
        from backend.user_home import ensure_user_home

        _create_user("alice")
        name = _run(ensure_user_home("alice"))

        assert name == "home-alice"
        assert (home_env / "alice").is_dir()
        assert "home-alice" in index
        assert "home-alice" in get_user("alice")["vaults"]
        # Registre persistant rempli → survit au prochain rebuild/redémarrage
        assert "home-alice" in _registry()["vaults"]

    def test_is_idempotent(self, home_env):
        from backend.auth.user_store import get_user
        from backend.indexer import index
        from backend.user_home import ensure_user_home

        _create_user("bob")
        _run(ensure_user_home("bob"))
        index_snapshot = dict(index)
        _run(ensure_user_home("bob"))

        assert get_user("bob")["vaults"].count("home-bob") == 1
        assert index == index_snapshot

    def test_invalid_username_never_touches_disk(self, home_env):
        from backend.indexer import index
        from backend.user_home import ensure_user_home

        assert _run(ensure_user_home("../evil")) is None
        assert _run(ensure_user_home("a/b")) is None
        assert not (home_env.parent / "evil").exists()
        assert not any(name.startswith("home-") for name in index)

    def test_disabled_without_env(self, home_env, monkeypatch):
        from backend.user_home import ensure_user_home

        monkeypatch.delenv("OBSIGATE_HOME_ROOT")
        assert _run(ensure_user_home("alice")) is None
        assert not (home_env / "alice").exists()


class TestPersistence:
    def test_registry_reloaded_by_load_vault_config(self, home_env):
        """data/vaults.json doit être relu par load_vault_config (rebuild/restart)."""
        from backend import indexer

        _create_user("carol")
        from backend.user_home import ensure_user_home

        _run(ensure_user_home("carol"))

        # Simule un redémarrage : vault_config vidé, rechargé depuis la source
        indexer.vault_config.clear()
        config = indexer.load_vault_config()
        assert config["home-carol"]["path"] == str(home_env / "carol")

    def test_unpersist_drops_registry_entry(self, home_env):
        from backend import indexer
        from backend.user_home import ensure_user_home

        _create_user("dan")
        _run(ensure_user_home("dan"))
        indexer.unpersist_vault("home-dan")
        assert "home-dan" not in _registry()["vaults"]


class TestRelease:
    def test_release_closes_vault_but_keeps_folder(self, home_env):
        from backend.indexer import index
        from backend.user_home import ensure_user_home, release_user_home

        _create_user("erin")
        _run(ensure_user_home("erin"))
        assert "home-erin" in index

        _run(release_user_home("erin"))
        assert "home-erin" not in index
        assert (home_env / "erin").is_dir()  # décision : dossier conservé
        assert "home-erin" not in _registry()["vaults"]


class TestOrphanSweep:
    def test_orphan_home_released_when_user_deleted_outside_route(self, home_env):
        """create_admin.py delete ne passe pas par la route → réparé au boot."""
        from backend.auth.user_store import delete_user
        from backend.indexer import index
        from backend.user_home import ensure_all_user_homes, ensure_user_home

        _create_user("hank")
        _run(ensure_user_home("hank"))
        delete_user("hank")
        assert "home-hank" in index

        _run(ensure_all_user_homes())
        assert "home-hank" not in index
        assert (home_env / "hank").is_dir()

    def test_vault_named_home_outside_root_is_left_alone(self, home_env):
        """Un vault admin nommé « home-… » ailleurs dans le FS n'est pas touché."""
        from backend.indexer import index
        from backend.user_home import ensure_all_user_homes

        other = home_env.parent / "Elsewhere"
        other.mkdir()
        _run(add_vault("home-keep", str(other)))

        _run(ensure_all_user_homes())
        assert "home-keep" in index


def add_vault(name: str, path: str):
    from backend.indexer import add_vault_to_index

    return add_vault_to_index(name, path)


class TestIsolation:
    def test_user_cannot_access_another_home_vault(self, home_env):
        from backend.auth.middleware import check_vault_access
        from backend.auth.user_store import get_user
        from backend.user_home import ensure_user_home

        _create_user("alice")
        _create_user("bob")
        _run(ensure_user_home("alice"))
        _run(ensure_user_home("bob"))

        assert check_vault_access("home-alice", get_user("alice")) is True
        assert check_vault_access("home-alice", get_user("bob")) is False
        assert check_vault_access("home-bob", get_user("alice")) is False


class TestEnsureAll:
    def test_repairs_every_existing_user(self, home_env):
        from backend.auth.user_store import get_user
        from backend.user_home import ensure_all_user_homes

        _create_user("fab")
        _create_user("gab")
        # Dossier supprimé à la main + user créé hors API (pas de vault)
        (home_env / "fab").mkdir()
        (home_env / "fab").rmdir()

        created = _run(ensure_all_user_homes())

        assert created == 2
        assert (home_env / "fab").is_dir()
        assert (home_env / "gab").is_dir()
        assert "home-fab" in get_user("fab")["vaults"]
        assert "home-gab" in get_user("gab")["vaults"]


def test_home_root_env_contract():
    """OBSIGATE_HOME_ROOT absent → fonctionnalité inactive (dev/tests/desktop)."""
    from backend.user_home import home_root

    saved = os.environ.pop("OBSIGATE_HOME_ROOT", None)
    try:
        assert home_root() is None
    finally:
        if saved is not None:
            os.environ["OBSIGATE_HOME_ROOT"] = saved