GDPRScanner/tests/test_manual_users.py

"""
Manual accounts in scan profiles.

The profile editor stores manually added accounts either with their directory
id (looked up via GET /api/users/lookup) or, when M365 wasn't reachable / for
profiles saved by older versions, as "manual:<email>". The scan engine must
resolve the latter to a Graph user id or they are never scanned.
"""
from __future__ import annotations
from unittest.mock import MagicMock

import pytest

import scan_engine


class _Conn:
    def __init__(self, directory: dict[str, dict]):
        self.directory = directory
        self.lookups: list[str] = []

    def lookup_user(self, upn: str) -> dict:
        self.lookups.append(upn)
        if upn not in self.directory:
            raise RuntimeError("Request_ResourceNotFound")
        return self.directory[upn]


@pytest.fixture()
def broadcasts(monkeypatch):
    sent = []
    monkeypatch.setattr(scan_engine, "broadcast", lambda ev, data: sent.append((ev, data)))
    return sent


class TestResolveManualUsers:
    def test_manual_email_resolved_to_directory_id(self, broadcasts):
        conn = _Conn({"ann@school.dk": {"id": "guid-ann", "displayName": "Ann Andersen",
                                        "mail": "ann@school.dk"}})
        out = scan_engine._resolve_scan_users(conn, [
            {"id": "manual:ann@school.dk", "displayName": "manual:ann@school.dk", "userRole": "other"},
        ])
        assert out == [{"id": "guid-ann", "displayName": "Ann Andersen", "userRole": "other"}]
        assert conn.lookups == ["ann@school.dk"]
        assert broadcasts == []

    def test_regular_and_resolved_manual_ids_pass_through_without_lookup(self, broadcasts):
        conn = _Conn({})
        users = [{"id": "guid-1", "displayName": "One"},
                 {"id": "guid-2", "displayName": "Two", "manual": True, "email": "two@school.dk"}]
        assert scan_engine._resolve_scan_users(conn, users) == users
        assert conn.lookups == []

    def test_unknown_manual_account_skipped_and_reported(self, broadcasts):
        conn = _Conn({})
        out = scan_engine._resolve_scan_users(conn, [
            {"id": "guid-1", "displayName": "One"},
            {"id": "manual:ghost@school.dk", "displayName": "ghost"},
        ])
        assert out == [{"id": "guid-1", "displayName": "One"}]
        assert broadcasts and broadcasts[0][0] == "scan_error"
        assert "ghost@school.dk" in broadcasts[0][1]["file"]

    def test_manual_account_already_selected_is_deduplicated(self, broadcasts):
        conn = _Conn({"ann@school.dk": {"id": "guid-ann", "displayName": "Ann"}})
        out = scan_engine._resolve_scan_users(conn, [
            {"id": "guid-ann", "displayName": "Ann"},
            {"id": "manual:ann@school.dk", "displayName": "ann"},
        ])
        assert [u["id"] for u in out] == ["guid-ann"]

    def test_all_manual_unresolvable_gives_empty_list_not_all_users(self, broadcasts):
        # run_scan resolves inside the non-empty branch, so an empty result must
        # stay empty — it must never widen to "scan every tenant user".
        conn = _Conn({})
        assert scan_engine._resolve_scan_users(
            conn, [{"id": "manual:x@school.dk", "displayName": "x"}]) == []


class TestGoogleOnlyAccountsDropped:
    """Regression: google:<email> ids reached the M365 scan and Graph 404'd
    ("/users/google:anne@school.dk/mailFolders")."""

    def test_google_only_accounts_are_dropped_and_reported(self, broadcasts):
        conn = _Conn({})
        out = scan_engine._resolve_scan_users(conn, [
            {"id": "guid-1", "displayName": "One"},
            {"id": "google:anne@school.dk", "displayName": "Anne"},
            {"id": "google:appleadmin@adm.school.dk", "displayName": "Apple Admin"},
        ])
        assert out == [{"id": "guid-1", "displayName": "One"}]
        assert conn.lookups == []
        phases = [d["phase"] for e, d in broadcasts if e == "scan_phase"]
        assert phases and "2" in phases[0]

    def test_only_google_accounts_gives_empty_list_not_all_users(self, broadcasts):
        assert scan_engine._resolve_scan_users(_Conn({}), [
            {"id": "google:anne@school.dk", "displayName": "Anne"}]) == []

    def test_run_scan_never_calls_graph_for_google_ids(self, broadcasts, monkeypatch):
        from routes import state
        monkeypatch.setattr(scan_engine, "DB_OK", False)
        monkeypatch.setattr(scan_engine, "_load_checkpoint", lambda key, **kw: None)
        monkeypatch.setattr(scan_engine, "_save_checkpoint", lambda *a, **kw: None)
        monkeypatch.setattr(scan_engine, "_clear_checkpoint", lambda **kw: None)
        monkeypatch.setattr(state, "flagged_items", [])
        conn = MagicMock()
        conn.is_app_mode = True
        conn.list_all_mail_folders_for.return_value = []
        monkeypatch.setattr(state, "connector", conn)
        import sse
        saved = sse._current_scan_id
        try:
            scan_engine.run_scan({"sources": ["email"], "options": {}, "user_ids": [
                {"id": "google:anne@school.dk", "displayName": "Anne"},
                {"id": "guid-1", "displayName": "One"},
            ]})
        finally:
            sse._current_scan_id = saved
        called_ids = {c.args[0] for c in conn.list_all_mail_folders_for.call_args_list}
        assert "guid-1" in called_ids
        assert not any(str(i).startswith("google:") for i in called_ids)
        conn.list_users.assert_not_called()  # never widened to the whole tenant


class TestLookupRoute:
    @pytest.fixture()
    def client(self):
        import gdpr_scanner
        gdpr_scanner.app.config["TESTING"] = True
        with gdpr_scanner.app.test_client() as c:
            yield c

    def test_lookup_uses_connector(self, client, monkeypatch):
        from routes import state
        conn = MagicMock()
        conn.lookup_user.return_value = {"id": "guid-ann", "displayName": "Ann",
                                         "mail": "ann@school.dk"}
        monkeypatch.setattr(state, "connector", conn)
        r = client.get("/api/users/lookup?upn=ann@school.dk")
        assert r.status_code == 200
        assert r.get_json()["id"] == "guid-ann"
        conn.lookup_user.assert_called_once_with("ann@school.dk")

    def test_lookup_not_found_is_404(self, client, monkeypatch):
        from routes import state
        conn = MagicMock()
        conn.lookup_user.side_effect = RuntimeError("Request_ResourceNotFound")
        monkeypatch.setattr(state, "connector", conn)
        assert client.get("/api/users/lookup?upn=ghost@school.dk").status_code == 404