StyxX65andClaude Opus 5.5
bbe9c62ff9
Starting an M365 scan wrote the sidebar state back into the active profile via _save_settings(), replacing its user_ids with the sidebar selection (minus Google-only users and anything hidden by the role filter). With no profile active it overwrote profiles[0]. Profiles are loaders, so scan start no longer writes them; profile_id is audit-only. The profile editor built its account list from S._allUsers only, so manually added accounts, users not yet loaded, and users gone from the directory were not rendered and got dropped on the next save. They now render as checked "Manual" / "Not loaded" rows, and user_ids "all" is kept when saved before users load. Applying a profile cleared the pending user IDs before the un-awaited Google merge had added google:<email> users, leaving them unticked. Pending IDs are now kept until all are matched or the user list is final, and _mergeGoogleUsers() re-applies them. A saved google:<email> also matches an M365 user merged to platform 'both'. Adds regression tests for scan start leaving profiles untouched. Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
738 lines
29 KiB
Python
738 lines
29 KiB
Python
"""
|
|
Route integration tests — security-sensitive paths and data-correctness contracts.
|
|
|
|
Covers:
|
|
- Viewer token CRUD and scope validation
|
|
- GET /api/db/flagged role and user scope enforcement
|
|
- POST /api/db/disposition/bulk — only updates selected items
|
|
- Viewer PIN set / verify / rate-limit / clear
|
|
- Interface PIN set / gate / clear
|
|
- Scan lock always released (even when run_scan raises)
|
|
- GET /api/db/sessions basic shape
|
|
- Profile routes CRUD and rename
|
|
"""
|
|
from __future__ import annotations
|
|
import time
|
|
from unittest.mock import MagicMock
|
|
|
|
import pytest
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Module-level app fixture (shared with test_routes.py via flask_app)
|
|
# ---------------------------------------------------------------------------
|
|
|
|
@pytest.fixture(scope="module")
|
|
def flask_app():
|
|
import gdpr_scanner
|
|
gdpr_scanner.app.config["TESTING"] = True
|
|
gdpr_scanner.app.config["WTF_CSRF_ENABLED"] = False
|
|
return gdpr_scanner.app
|
|
|
|
|
|
@pytest.fixture()
|
|
def client(flask_app):
|
|
with flask_app.test_client() as c:
|
|
yield c
|
|
|
|
|
|
@pytest.fixture()
|
|
def db_patch(tmp_path, monkeypatch):
|
|
from gdpr_db import ScanDB
|
|
import routes.database, routes.export
|
|
db = ScanDB(str(tmp_path / "test.db"))
|
|
monkeypatch.setattr(routes.database, "_get_db", lambda: db)
|
|
monkeypatch.setattr(routes.database, "DB_OK", True)
|
|
monkeypatch.setattr(routes.export, "_get_db", lambda: db)
|
|
monkeypatch.setattr(routes.export, "DB_OK", True)
|
|
return db
|
|
|
|
|
|
@pytest.fixture()
|
|
def mock_connector(monkeypatch):
|
|
from routes import state
|
|
conn = MagicMock()
|
|
monkeypatch.setattr(state, "connector", conn)
|
|
return conn
|
|
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def clean_state():
|
|
from routes import state
|
|
yield
|
|
state.flagged_items.clear()
|
|
if not state._scan_lock.acquire(blocking=False):
|
|
pass
|
|
else:
|
|
state._scan_lock.release()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Helpers
|
|
# ---------------------------------------------------------------------------
|
|
|
|
def _seed_scan(db, items: list[dict]) -> int:
|
|
"""Create a completed scan and persist items. Returns the scan_id."""
|
|
scan_id = db.begin_scan({"sources": ["email"], "user_ids": [], "options": {}})
|
|
for item in items:
|
|
db.save_item(scan_id, item)
|
|
db.finish_scan(scan_id, total_scanned=len(items))
|
|
return scan_id
|
|
|
|
|
|
def _item(item_id: str, role: str = "staff", account_id: str = "") -> dict:
|
|
return {
|
|
"id": item_id,
|
|
"name": f"{item_id}.docx",
|
|
"source": "email",
|
|
"source_type": "email",
|
|
"account_id": account_id or f"{item_id}@school.dk",
|
|
"user_role": role,
|
|
"cpr_count": 1,
|
|
"face_count": 0,
|
|
"size_kb": 10,
|
|
"modified": "2025-01-01T00:00:00",
|
|
}
|
|
|
|
|
|
def _clear_viewer_pins():
|
|
"""Remove both viewer and interface PINs between tests."""
|
|
from app_config import clear_viewer_pin, clear_interface_pin
|
|
clear_viewer_pin()
|
|
clear_interface_pin()
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Viewer token CRUD
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestViewerTokenCRUD:
|
|
def test_create_and_list(self, client):
|
|
r = client.post("/api/viewer/tokens",
|
|
json={"label": "Test token", "expires_days": 7})
|
|
assert r.status_code == 201
|
|
data = r.get_json()
|
|
assert "token" in data
|
|
tok = data["token"]
|
|
|
|
r2 = client.get("/api/viewer/tokens")
|
|
assert r2.status_code == 200
|
|
tokens = r2.get_json()
|
|
assert any(t["token"] == tok for t in tokens)
|
|
|
|
def test_delete_existing_token(self, client):
|
|
r = client.post("/api/viewer/tokens", json={"label": "to-delete"})
|
|
tok = r.get_json()["token"]
|
|
|
|
r2 = client.delete(f"/api/viewer/tokens/{tok}")
|
|
assert r2.status_code == 200
|
|
assert r2.get_json()["ok"] is True
|
|
|
|
r3 = client.get("/api/viewer/tokens")
|
|
tokens = r3.get_json()
|
|
assert not any(t["token"] == tok for t in tokens)
|
|
|
|
def test_delete_nonexistent_token_returns_404(self, client):
|
|
r = client.delete("/api/viewer/tokens/doesnotexist123")
|
|
assert r.status_code == 404
|
|
|
|
def test_validate_valid_token(self, client):
|
|
tok = client.post("/api/viewer/tokens", json={}).get_json()["token"]
|
|
r = client.post("/api/viewer/tokens/validate", json={"token": tok})
|
|
assert r.status_code == 200
|
|
assert r.get_json()["valid"] is True
|
|
|
|
def test_validate_invalid_token(self, client):
|
|
r = client.post("/api/viewer/tokens/validate",
|
|
json={"token": "notarealtoken00000000"})
|
|
assert r.status_code == 401
|
|
assert r.get_json()["valid"] is False
|
|
|
|
|
|
class TestViewerTokenScopeValidation:
|
|
def test_role_and_user_mutually_exclusive(self, client):
|
|
r = client.post("/api/viewer/tokens", json={
|
|
"scope": {"role": "student", "user": "alice@school.dk"}
|
|
})
|
|
assert r.status_code == 400
|
|
assert "mutually exclusive" in r.get_json()["error"]
|
|
|
|
def test_invalid_role_value(self, client):
|
|
r = client.post("/api/viewer/tokens", json={
|
|
"scope": {"role": "teacher"}
|
|
})
|
|
assert r.status_code == 400
|
|
assert "role" in r.get_json()["error"]
|
|
|
|
def test_user_email_must_contain_at(self, client):
|
|
r = client.post("/api/viewer/tokens", json={
|
|
"scope": {"user": "notanemail"}
|
|
})
|
|
assert r.status_code == 400
|
|
assert "email" in r.get_json()["error"].lower()
|
|
|
|
def test_valid_role_scope_stored(self, client):
|
|
r = client.post("/api/viewer/tokens",
|
|
json={"scope": {"role": "student"}})
|
|
assert r.status_code == 201
|
|
assert r.get_json()["scope"] == {"role": "student"}
|
|
|
|
def test_valid_user_scope_stored(self, client):
|
|
r = client.post("/api/viewer/tokens", json={
|
|
"scope": {
|
|
"user": ["alice@m365.dk", "alice@gws.dk"],
|
|
"display_name": "Alice Smith",
|
|
}
|
|
})
|
|
assert r.status_code == 201
|
|
scope = r.get_json()["scope"]
|
|
assert scope["user"] == ["alice@m365.dk", "alice@gws.dk"]
|
|
assert scope["display_name"] == "Alice Smith"
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/db/flagged — scope enforcement
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestFlaggedScopeEnforcement:
|
|
def test_no_scope_returns_all_items(self, client, db_patch):
|
|
_seed_scan(db_patch, [
|
|
_item("s1", role="student"),
|
|
_item("s2", role="staff"),
|
|
])
|
|
r = client.get("/api/db/flagged")
|
|
assert r.status_code == 200
|
|
ids = {row["id"] for row in r.get_json()}
|
|
assert "s1" in ids
|
|
assert "s2" in ids
|
|
|
|
def test_role_scope_student_excludes_staff(self, client, db_patch):
|
|
_seed_scan(db_patch, [
|
|
_item("r1", role="student"),
|
|
_item("r2", role="staff"),
|
|
])
|
|
with client.session_transaction() as sess:
|
|
sess["viewer_ok"] = True
|
|
sess["viewer_scope"] = {"role": "student"}
|
|
r = client.get("/api/db/flagged")
|
|
ids = {row["id"] for row in r.get_json()}
|
|
assert "r1" in ids
|
|
assert "r2" not in ids
|
|
|
|
def test_role_scope_staff_excludes_students(self, client, db_patch):
|
|
_seed_scan(db_patch, [
|
|
_item("t1", role="student"),
|
|
_item("t2", role="staff"),
|
|
])
|
|
with client.session_transaction() as sess:
|
|
sess["viewer_ok"] = True
|
|
sess["viewer_scope"] = {"role": "staff"}
|
|
r = client.get("/api/db/flagged")
|
|
ids = {row["id"] for row in r.get_json()}
|
|
assert "t2" in ids
|
|
assert "t1" not in ids
|
|
|
|
def test_user_scope_returns_only_matching_account_id(self, client, db_patch):
|
|
_seed_scan(db_patch, [
|
|
_item("u1", account_id="alice@m365.dk"),
|
|
_item("u2", account_id="bob@m365.dk"),
|
|
])
|
|
with client.session_transaction() as sess:
|
|
sess["viewer_ok"] = True
|
|
sess["viewer_scope"] = {"user": ["alice@m365.dk"]}
|
|
r = client.get("/api/db/flagged")
|
|
ids = {row["id"] for row in r.get_json()}
|
|
assert "u1" in ids
|
|
assert "u2" not in ids
|
|
|
|
def test_user_scope_matches_both_platform_emails(self, client, db_patch):
|
|
# Same person — M365 UPN and GWS email both in scope
|
|
_seed_scan(db_patch, [
|
|
_item("p1", account_id="alice@m365.dk"),
|
|
_item("p2", account_id="alice@gws.dk"),
|
|
_item("p3", account_id="bob@m365.dk"),
|
|
])
|
|
with client.session_transaction() as sess:
|
|
sess["viewer_ok"] = True
|
|
sess["viewer_scope"] = {"user": ["alice@m365.dk", "alice@gws.dk"]}
|
|
r = client.get("/api/db/flagged")
|
|
ids = {row["id"] for row in r.get_json()}
|
|
assert "p1" in ids
|
|
assert "p2" in ids
|
|
assert "p3" not in ids
|
|
|
|
def test_user_scope_case_insensitive(self, client, db_patch):
|
|
_seed_scan(db_patch, [_item("ci1", account_id="Alice@M365.dk")])
|
|
with client.session_transaction() as sess:
|
|
sess["viewer_ok"] = True
|
|
sess["viewer_scope"] = {"user": ["alice@m365.dk"]}
|
|
r = client.get("/api/db/flagged")
|
|
ids = {row["id"] for row in r.get_json()}
|
|
assert "ci1" in ids
|
|
|
|
def test_no_ref_returns_open_items_across_all_sessions(self, client, db_patch):
|
|
# Two scans in separate session windows. The default (no-ref) view must
|
|
# surface unactioned items from BOTH, not just the latest session.
|
|
old_id = _seed_scan(db_patch, [_item("o1")])
|
|
db_patch._connect().execute(
|
|
"UPDATE scans SET started_at = started_at - 400 WHERE id = ?", (old_id,)
|
|
)
|
|
db_patch._connect().commit()
|
|
_seed_scan(db_patch, [_item("o2")])
|
|
|
|
r = client.get("/api/db/flagged")
|
|
ids = {row["id"] for row in r.get_json()}
|
|
assert ids == {"o1", "o2"}
|
|
|
|
def test_no_ref_excludes_items_with_a_disposition(self, client, db_patch):
|
|
_seed_scan(db_patch, [_item("d1"), _item("d2")])
|
|
db_patch.set_disposition("d1", "kept")
|
|
|
|
r = client.get("/api/db/flagged")
|
|
ids = {row["id"] for row in r.get_json()}
|
|
assert "d2" in ids # untouched → still open
|
|
assert "d1" not in ids # action taken → hidden
|
|
|
|
def test_no_ref_unreviewed_disposition_stays_open(self, client, db_patch):
|
|
_seed_scan(db_patch, [_item("u1")])
|
|
db_patch.set_disposition("u1", "unreviewed")
|
|
|
|
r = client.get("/api/db/flagged")
|
|
ids = {row["id"] for row in r.get_json()}
|
|
assert "u1" in ids # 'unreviewed' status is not an action
|
|
|
|
def test_no_ref_dedupes_rescanned_item_to_latest(self, client, db_patch):
|
|
# Same item flagged by two scans → appears once.
|
|
old_id = _seed_scan(db_patch, [_item("k1")])
|
|
db_patch._connect().execute(
|
|
"UPDATE scans SET started_at = started_at - 400 WHERE id = ?", (old_id,)
|
|
)
|
|
db_patch._connect().commit()
|
|
_seed_scan(db_patch, [_item("k1")])
|
|
|
|
rows = [row for row in client.get("/api/db/flagged").get_json() if row["id"] == "k1"]
|
|
assert len(rows) == 1
|
|
|
|
def test_ref_param_loads_historical_session(self, client, db_patch):
|
|
# Push first scan >300 s into the past so it occupies its own session window.
|
|
old_id = _seed_scan(db_patch, [_item("h1")])
|
|
db_patch._connect().execute(
|
|
"UPDATE scans SET started_at = started_at - 400 WHERE id = ?", (old_id,)
|
|
)
|
|
db_patch._connect().commit()
|
|
_seed_scan(db_patch, [_item("h2")])
|
|
|
|
r = client.get(f"/api/db/flagged?ref={old_id}")
|
|
ids = {row["id"] for row in r.get_json()}
|
|
assert "h1" in ids
|
|
# h2 belongs to a different (newer) session window — must not appear
|
|
assert "h2" not in ids
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# POST /api/db/disposition/bulk
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestBulkDisposition:
|
|
def test_updates_selected_items(self, client, db_patch):
|
|
_seed_scan(db_patch, [_item("b1"), _item("b2"), _item("b3")])
|
|
r = client.post("/api/db/disposition/bulk", json={
|
|
"item_ids": ["b1", "b2"],
|
|
"status": "retain-legal",
|
|
})
|
|
assert r.status_code == 200
|
|
assert r.get_json()["saved"] == 2
|
|
|
|
assert db_patch.get_disposition("b1")["status"] == "retain-legal"
|
|
assert db_patch.get_disposition("b2")["status"] == "retain-legal"
|
|
|
|
def test_unselected_item_unchanged(self, client, db_patch):
|
|
_seed_scan(db_patch, [_item("c1"), _item("c2")])
|
|
client.post("/api/db/disposition/bulk", json={
|
|
"item_ids": ["c1"],
|
|
"status": "delete-scheduled",
|
|
})
|
|
d = db_patch.get_disposition("c2")
|
|
# c2 was not in the bulk request — must remain unreviewed
|
|
assert d is None or d.get("status", "unreviewed") == "unreviewed"
|
|
|
|
def test_missing_item_ids_returns_400(self, client, db_patch):
|
|
r = client.post("/api/db/disposition/bulk",
|
|
json={"status": "retain-legal"})
|
|
assert r.status_code == 400
|
|
|
|
def test_missing_status_returns_400(self, client, db_patch):
|
|
r = client.post("/api/db/disposition/bulk",
|
|
json={"item_ids": ["x"]})
|
|
assert r.status_code == 400
|
|
|
|
def test_without_db_returns_503(self, client, monkeypatch):
|
|
import routes.database
|
|
monkeypatch.setattr(routes.database, "DB_OK", False)
|
|
r = client.post("/api/db/disposition/bulk",
|
|
json={"item_ids": ["x"], "status": "retain-legal"})
|
|
assert r.status_code == 503
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Viewer PIN
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestViewerPin:
|
|
def setup_method(self):
|
|
_clear_viewer_pins()
|
|
|
|
def teardown_method(self):
|
|
_clear_viewer_pins()
|
|
|
|
def test_status_no_pin(self, client):
|
|
r = client.get("/api/viewer/pin")
|
|
assert r.status_code == 200
|
|
assert r.get_json()["pin_set"] is False
|
|
|
|
def test_set_and_status_reflects_set(self, client):
|
|
client.post("/api/viewer/pin", json={"pin": "1234"})
|
|
r = client.get("/api/viewer/pin")
|
|
assert r.get_json()["pin_set"] is True
|
|
|
|
def test_set_too_short_rejected(self, client):
|
|
r = client.post("/api/viewer/pin", json={"pin": "123"})
|
|
assert r.status_code == 400
|
|
|
|
def test_set_too_long_rejected(self, client):
|
|
r = client.post("/api/viewer/pin", json={"pin": "123456789"})
|
|
assert r.status_code == 400
|
|
|
|
def test_set_non_digits_rejected(self, client):
|
|
r = client.post("/api/viewer/pin", json={"pin": "abcd"})
|
|
assert r.status_code == 400
|
|
|
|
def test_verify_correct_pin_sets_session(self, client):
|
|
client.post("/api/viewer/pin", json={"pin": "4321"})
|
|
r = client.post("/api/viewer/pin/verify", json={"pin": "4321"})
|
|
assert r.status_code == 200
|
|
assert r.get_json()["ok"] is True
|
|
|
|
def test_verify_wrong_pin_returns_401(self, client):
|
|
client.post("/api/viewer/pin", json={"pin": "4321"})
|
|
r = client.post("/api/viewer/pin/verify", json={"pin": "9999"})
|
|
assert r.status_code == 401
|
|
|
|
def test_verify_rate_limit_after_5_failures(self, client):
|
|
client.post("/api/viewer/pin", json={"pin": "5678"})
|
|
from routes.viewer import _pin_attempts
|
|
_pin_attempts.clear()
|
|
for _ in range(5):
|
|
client.post("/api/viewer/pin/verify", json={"pin": "0000"})
|
|
r = client.post("/api/viewer/pin/verify", json={"pin": "0000"})
|
|
assert r.status_code == 429
|
|
_pin_attempts.clear()
|
|
|
|
def test_change_pin_requires_current(self, client):
|
|
client.post("/api/viewer/pin", json={"pin": "1111"})
|
|
r = client.post("/api/viewer/pin",
|
|
json={"pin": "2222", "current_pin": "9999"})
|
|
assert r.status_code == 403
|
|
|
|
def test_change_pin_with_correct_current(self, client):
|
|
client.post("/api/viewer/pin", json={"pin": "1111"})
|
|
r = client.post("/api/viewer/pin",
|
|
json={"pin": "2222", "current_pin": "1111"})
|
|
assert r.status_code == 200
|
|
# Old PIN no longer valid
|
|
r2 = client.post("/api/viewer/pin/verify", json={"pin": "1111"})
|
|
assert r2.status_code == 401
|
|
|
|
def test_clear_pin_requires_current(self, client):
|
|
client.post("/api/viewer/pin", json={"pin": "3333"})
|
|
r = client.delete("/api/viewer/pin", json={"current_pin": "0000"})
|
|
assert r.status_code == 403
|
|
|
|
def test_clear_pin_with_correct_current(self, client):
|
|
client.post("/api/viewer/pin", json={"pin": "3333"})
|
|
r = client.delete("/api/viewer/pin", json={"current_pin": "3333"})
|
|
assert r.status_code == 200
|
|
assert client.get("/api/viewer/pin").get_json()["pin_set"] is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Interface PIN
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestInterfacePin:
|
|
def setup_method(self):
|
|
_clear_viewer_pins()
|
|
|
|
def teardown_method(self):
|
|
_clear_viewer_pins()
|
|
|
|
def test_status_no_pin(self, client):
|
|
r = client.get("/api/interface/pin")
|
|
assert r.get_json()["pin_set"] is False
|
|
|
|
def test_set_and_verify(self, client):
|
|
r = client.post("/api/interface/pin", json={"pin": "7777"})
|
|
assert r.status_code == 200
|
|
# Gate is now active — authenticate before the status check
|
|
with client.session_transaction() as sess:
|
|
sess["interface_ok"] = True
|
|
assert client.get("/api/interface/pin").get_json()["pin_set"] is True
|
|
|
|
def test_non_digit_rejected(self, client):
|
|
r = client.post("/api/interface/pin", json={"pin": "abcd"})
|
|
assert r.status_code == 400
|
|
|
|
def test_set_requires_current_when_set(self, client):
|
|
client.post("/api/interface/pin", json={"pin": "7777"})
|
|
with client.session_transaction() as sess:
|
|
sess["interface_ok"] = True
|
|
r = client.post("/api/interface/pin",
|
|
json={"pin": "8888", "current_pin": "0000"})
|
|
assert r.status_code == 403
|
|
|
|
def test_clear_requires_current(self, client):
|
|
client.post("/api/interface/pin", json={"pin": "7777"})
|
|
with client.session_transaction() as sess:
|
|
sess["interface_ok"] = True
|
|
r = client.delete("/api/interface/pin", json={"current_pin": "0000"})
|
|
assert r.status_code == 403
|
|
|
|
def test_clear_with_correct_current(self, client):
|
|
client.post("/api/interface/pin", json={"pin": "7777"})
|
|
with client.session_transaction() as sess:
|
|
sess["interface_ok"] = True
|
|
r = client.delete("/api/interface/pin", json={"current_pin": "7777"})
|
|
assert r.status_code == 200
|
|
assert client.get("/api/interface/pin").get_json()["pin_set"] is False
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scan lock released on run_scan() exception
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestScanLockReleasedOnError:
|
|
def test_lock_released_when_run_scan_raises(self, client, mock_connector,
|
|
monkeypatch):
|
|
import scan_engine
|
|
from routes import state
|
|
|
|
def _boom(opts):
|
|
raise RuntimeError("simulated scan failure")
|
|
|
|
monkeypatch.setattr(scan_engine, "run_scan", _boom)
|
|
r = client.post("/api/scan/start", json={"sources": ["email"]})
|
|
assert r.status_code == 200
|
|
|
|
# Wait for the background thread to finish and release the lock
|
|
deadline = time.time() + 2.0
|
|
while True:
|
|
acquired = state._scan_lock.acquire(blocking=False)
|
|
if acquired:
|
|
state._scan_lock.release()
|
|
break
|
|
assert time.time() < deadline, "scan lock was never released after exception"
|
|
time.sleep(0.05)
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# GET /api/db/sessions
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestDbSessions:
|
|
def test_returns_list(self, client, db_patch):
|
|
r = client.get("/api/db/sessions")
|
|
assert r.status_code == 200
|
|
assert isinstance(r.get_json(), list)
|
|
|
|
def test_completed_scan_appears_in_sessions(self, client, db_patch):
|
|
_seed_scan(db_patch, [_item("sess1")])
|
|
r = client.get("/api/db/sessions")
|
|
sessions = r.get_json()
|
|
assert len(sessions) >= 1
|
|
s = sessions[0]
|
|
assert "ref_scan_id" in s
|
|
assert "flagged_count" in s
|
|
assert s["flagged_count"] == 1
|
|
|
|
def test_sessions_ordered_newest_first(self, client, db_patch):
|
|
# Create two scans >300 s apart so each forms its own session window.
|
|
old_id = _seed_scan(db_patch, [_item("old1")])
|
|
db_patch._connect().execute(
|
|
"UPDATE scans SET started_at = started_at - 400 WHERE id = ?", (old_id,)
|
|
)
|
|
db_patch._connect().commit()
|
|
_seed_scan(db_patch, [_item("new1")])
|
|
sessions = client.get("/api/db/sessions").get_json()
|
|
assert len(sessions) == 2
|
|
# Newest session (highest ref_scan_id) must be first
|
|
assert sessions[0]["ref_scan_id"] > sessions[1]["ref_scan_id"]
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Profile routes
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestProfileRoutes:
|
|
"""
|
|
Tests for GET /api/profiles, POST /api/profiles/save,
|
|
GET /api/profiles/get, and POST /api/profiles/delete.
|
|
|
|
Each test monkeypatches the profile storage path to a tmp directory so
|
|
tests are fully isolated from the real ~/.gdprscanner/settings.json.
|
|
"""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolate(self, tmp_path, monkeypatch):
|
|
import app_config
|
|
monkeypatch.setattr(app_config, "_SETTINGS_PATH", tmp_path / "settings.json")
|
|
|
|
def test_list_returns_empty_list_initially(self, client):
|
|
r = client.get("/api/profiles")
|
|
assert r.status_code == 200
|
|
assert r.get_json()["profiles"] == []
|
|
|
|
def test_save_missing_name_returns_400(self, client):
|
|
r = client.post("/api/profiles/save", json={"sources": ["email"]})
|
|
assert r.status_code == 400
|
|
assert "error" in r.get_json()
|
|
|
|
def test_save_creates_profile_and_returns_it(self, client):
|
|
r = client.post("/api/profiles/save", json={
|
|
"id": "", "name": "Alpha", "sources": ["email"], "options": {}
|
|
})
|
|
assert r.status_code == 200
|
|
data = r.get_json()
|
|
assert data["status"] == "saved"
|
|
assert data["profile"]["name"] == "Alpha"
|
|
assert data["profile"]["id"] # server assigned a non-empty id
|
|
|
|
def test_saved_profile_appears_in_list(self, client):
|
|
client.post("/api/profiles/save", json={"name": "Beta", "sources": [], "options": {}})
|
|
profiles = client.get("/api/profiles").get_json()["profiles"]
|
|
assert any(p["name"] == "Beta" for p in profiles)
|
|
|
|
def test_rename_updates_name_in_list(self, client):
|
|
"""Regression: _pmgmtSaveFullEdit renames the copy — the API must
|
|
persist the new name so loadProfiles() returns fresh data for the
|
|
left-column re-render."""
|
|
r = client.post("/api/profiles/save", json={
|
|
"id": "", "name": "LOCAL-TEST (copy)", "sources": [], "options": {}
|
|
})
|
|
profile_id = r.get_json()["profile"]["id"]
|
|
|
|
# Simulate the user renaming the copy in the editor and clicking Save
|
|
r2 = client.post("/api/profiles/save", json={
|
|
"id": profile_id, "name": "LOCAL-TEST-2", "sources": [], "options": {}
|
|
})
|
|
assert r2.status_code == 200
|
|
assert r2.get_json()["profile"]["name"] == "LOCAL-TEST-2"
|
|
|
|
profiles = client.get("/api/profiles").get_json()["profiles"]
|
|
names = [p["name"] for p in profiles]
|
|
assert "LOCAL-TEST-2" in names
|
|
assert "LOCAL-TEST (copy)" not in names
|
|
|
|
def test_get_by_id(self, client):
|
|
r = client.post("/api/profiles/save", json={
|
|
"id": "fixed-id-1", "name": "Gamma", "sources": [], "options": {}
|
|
})
|
|
profile_id = r.get_json()["profile"]["id"]
|
|
r2 = client.get(f"/api/profiles/get?id={profile_id}")
|
|
assert r2.status_code == 200
|
|
assert r2.get_json()["profile"]["name"] == "Gamma"
|
|
|
|
def test_get_nonexistent_returns_404(self, client):
|
|
r = client.get("/api/profiles/get?id=does-not-exist")
|
|
assert r.status_code == 404
|
|
|
|
def test_delete_removes_profile(self, client):
|
|
client.post("/api/profiles/save", json={"name": "ToDelete", "sources": [], "options": {}})
|
|
r = client.post("/api/profiles/delete", json={"name": "ToDelete"})
|
|
assert r.status_code == 200
|
|
assert r.get_json()["status"] == "deleted"
|
|
profiles = client.get("/api/profiles").get_json()["profiles"]
|
|
assert not any(p["name"] == "ToDelete" for p in profiles)
|
|
|
|
def test_delete_nonexistent_returns_not_found(self, client):
|
|
r = client.post("/api/profiles/delete", json={"name": "Ghost"})
|
|
assert r.status_code == 200
|
|
assert r.get_json()["status"] == "not_found"
|
|
|
|
def test_delete_missing_key_returns_400(self, client):
|
|
r = client.post("/api/profiles/delete", json={})
|
|
assert r.status_code == 400
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Scan start must not write profiles
|
|
# ---------------------------------------------------------------------------
|
|
|
|
class TestScanStartLeavesProfilesAlone:
|
|
"""
|
|
Regression: POST /api/scan/start used to call _save_settings() with the
|
|
sidebar state, overwriting the active profile's user_ids (dropping
|
|
Google-only users) — or profiles[0] when no profile was active.
|
|
"""
|
|
|
|
@pytest.fixture(autouse=True)
|
|
def _isolate(self, tmp_path, monkeypatch):
|
|
import app_config, scan_engine
|
|
monkeypatch.setattr(app_config, "_SETTINGS_PATH", tmp_path / "settings.json")
|
|
monkeypatch.setattr(scan_engine, "run_scan", lambda opts: None)
|
|
|
|
@staticmethod
|
|
def _wait_for_lock():
|
|
from routes import state
|
|
deadline = time.time() + 2.0
|
|
while not state._scan_lock.acquire(blocking=False):
|
|
assert time.time() < deadline, "scan lock was never released"
|
|
time.sleep(0.05)
|
|
state._scan_lock.release()
|
|
|
|
def _save(self, client, name, user_ids):
|
|
r = client.post("/api/profiles/save", json={
|
|
"name": name, "sources": ["email", "gmail"],
|
|
"google_sources": ["gmail"], "user_ids": user_ids, "options": {},
|
|
})
|
|
return r.get_json()["profile"]
|
|
|
|
def _get(self, client, pid):
|
|
return client.get(f"/api/profiles/get?id={pid}").get_json()["profile"]
|
|
|
|
def test_active_profile_users_unchanged(self, client, mock_connector):
|
|
users = ["m365-a", "google:b@school.dk", "manual:c@school.dk"]
|
|
p = self._save(client, "Staff", users)
|
|
r = client.post("/api/scan/start", json={
|
|
"sources": ["email"],
|
|
"user_ids": [{"id": "m365-z", "displayName": "Z"}],
|
|
"options": {"older_than_days": 0},
|
|
"profile_id": p["id"],
|
|
})
|
|
assert r.status_code == 200
|
|
self._wait_for_lock()
|
|
after = self._get(client, p["id"])
|
|
assert after["user_ids"] == users
|
|
assert after["sources"] == ["email", "gmail"]
|
|
|
|
def test_no_active_profile_leaves_first_profile_unchanged(self, client, mock_connector):
|
|
first = self._save(client, "First", ["m365-a", "m365-b"])
|
|
self._save(client, "Second", ["m365-c"])
|
|
r = client.post("/api/scan/start", json={
|
|
"sources": ["email"],
|
|
"user_ids": [{"id": "m365-z", "displayName": "Z"}],
|
|
"options": {},
|
|
"profile_id": None,
|
|
})
|
|
assert r.status_code == 200
|
|
self._wait_for_lock()
|
|
assert self._get(client, first["id"])["user_ids"] == ["m365-a", "m365-b"]
|
|
assert len(client.get("/api/profiles").get_json()["profiles"]) == 2
|
|
|
|
def test_scan_start_creates_no_profile(self, client, mock_connector):
|
|
r = client.post("/api/scan/start", json={
|
|
"sources": ["email"], "user_ids": [{"id": "m365-z"}], "options": {},
|
|
})
|
|
assert r.status_code == 200
|
|
self._wait_for_lock()
|
|
assert client.get("/api/profiles").get_json()["profiles"] == []
|