feat(tests,users): blocking procedure-suggestion review on test detail, multi-role switcher, Power Automate webhook payload
Aegis CI / lint-and-test (push) Has been cancelled
Snyk Security Scan / Python vulnerabilities (backend) (push) Has been cancelled
Snyk Security Scan / npm vulnerabilities (frontend) (push) Has been cancelled
Snyk Security Scan / Docker image vulnerabilities (backend) (push) Has been cancelled
Aegis CI / lint-and-test (push) Has been cancelled
Snyk Security Scan / Python vulnerabilities (backend) (push) Has been cancelled
Snyk Security Scan / npm vulnerabilities (frontend) (push) Has been cancelled
Snyk Security Scan / Docker image vulnerabilities (backend) (push) Has been cancelled
- A lead opening a test with a pending procedure suggestion awaiting
their review now gets a blocking popup (approve/reject only) instead
of discovering it later in a separate queue.
- Users can be granted more than one role via extra_roles; only one is
ever active at a time (no permission mixing) and a top-bar switcher
lets the user swap which one, taking effect immediately since role
is read fresh from the DB on every request.
- The password-setup/reset webhook now posts the agreed Power Automate
contract ({to, subject, body} with the platform's standard greeting
and signature template) and supports an admin-configured API key
sent as an x-api-key header.
This commit is contained in:
@@ -0,0 +1,96 @@
|
||||
"""A user can be granted more than one role but only ever acts under a
|
||||
single active one at a time — switching swaps which role is active
|
||||
instead of ever mixing permissions from more than one.
|
||||
"""
|
||||
|
||||
|
||||
def test_admin_can_grant_extra_roles(api, auth_headers):
|
||||
created = api(
|
||||
"post", "/api/v1/users", auth_headers,
|
||||
json={"full_name": "Multi Role User", "email": "multirole@test.com", "role": "red_tech"},
|
||||
)
|
||||
assert created.status_code == 201, created.text
|
||||
user_id = created.json()["id"]
|
||||
|
||||
resp = api(
|
||||
"patch", f"/api/v1/users/{user_id}", auth_headers,
|
||||
json={"extra_roles": ["blue_tech", "red_lead"]},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["role"] == "red_tech"
|
||||
assert set(resp.json()["extra_roles"]) == {"blue_tech", "red_lead"}
|
||||
|
||||
|
||||
def test_admin_cannot_grant_invalid_extra_role(api, auth_headers):
|
||||
created = api(
|
||||
"post", "/api/v1/users", auth_headers,
|
||||
json={"full_name": "Bad Extra Role", "email": "badextrarole@test.com", "role": "viewer"},
|
||||
)
|
||||
user_id = created.json()["id"]
|
||||
|
||||
resp = api(
|
||||
"patch", f"/api/v1/users/{user_id}", auth_headers,
|
||||
json={"extra_roles": ["superuser"]},
|
||||
)
|
||||
assert resp.status_code == 400
|
||||
|
||||
|
||||
def test_user_can_switch_to_a_granted_extra_role(api, db, auth_headers):
|
||||
from app.auth import hash_password
|
||||
from app.models.user import User
|
||||
|
||||
user = User(
|
||||
username="switcher@test.com",
|
||||
email="switcher@test.com",
|
||||
full_name="Switcher",
|
||||
hashed_password=hash_password("SwitcherPass123!@#"),
|
||||
role="red_tech",
|
||||
extra_roles=["blue_tech"],
|
||||
must_change_password=False,
|
||||
)
|
||||
db.add(user)
|
||||
db.commit()
|
||||
|
||||
login = api(
|
||||
"post", "/api/v1/auth/login", {},
|
||||
data={"username": "switcher@test.com", "password": "SwitcherPass123!@#"},
|
||||
)
|
||||
assert login.status_code == 200, login.text
|
||||
token = login.json()["access_token"]
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
|
||||
resp = api("post", "/api/v1/users/me/switch-role", headers, json={"role": "blue_tech"})
|
||||
assert resp.status_code == 200, resp.text
|
||||
body = resp.json()
|
||||
assert body["role"] == "blue_tech"
|
||||
assert body["extra_roles"] == ["red_tech"]
|
||||
|
||||
me = api("get", "/api/v1/auth/me", headers)
|
||||
assert me.json()["role"] == "blue_tech"
|
||||
|
||||
|
||||
def test_user_cannot_switch_to_an_ungranted_role(api, db, auth_headers):
|
||||
from app.auth import hash_password
|
||||
from app.models.user import User
|
||||
|
||||
user = User(
|
||||
username="switcher2@test.com",
|
||||
email="switcher2@test.com",
|
||||
full_name="Switcher Two",
|
||||
hashed_password=hash_password("SwitcherPass123!@#"),
|
||||
role="red_tech",
|
||||
extra_roles=["blue_tech"],
|
||||
must_change_password=False,
|
||||
)
|
||||
db.add(user)
|
||||
db.commit()
|
||||
|
||||
login = api(
|
||||
"post", "/api/v1/auth/login", {},
|
||||
data={"username": "switcher2@test.com", "password": "SwitcherPass123!@#"},
|
||||
)
|
||||
token = login.json()["access_token"]
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
|
||||
resp = api("post", "/api/v1/users/me/switch-role", headers, json={"role": "admin"})
|
||||
assert resp.status_code == 400
|
||||
@@ -36,6 +36,29 @@ def test_admin_can_configure_password_webhook(api, auth_headers):
|
||||
assert get_resp.json()["url"] == "https://example.com/power-automate-hook"
|
||||
|
||||
|
||||
def test_admin_can_configure_webhook_api_key(api, auth_headers):
|
||||
resp = api(
|
||||
"patch", "/api/v1/system/password-webhook-config", auth_headers,
|
||||
json={"url": "https://example.com/power-automate-hook", "api_key": "super-secret-key"},
|
||||
)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json()["api_key_set"] is True
|
||||
# The key itself is never echoed back.
|
||||
assert "api_key" not in resp.json()
|
||||
assert "super-secret-key" not in resp.text
|
||||
|
||||
|
||||
def test_send_password_email_sends_api_key_header(api, db, auth_headers, new_user_id):
|
||||
api(
|
||||
"patch", "/api/v1/system/password-webhook-config", auth_headers,
|
||||
json={"url": "https://example.com/power-automate-hook", "api_key": "super-secret-key"},
|
||||
)
|
||||
with patch("app.services.password_setup_service.requests.post") as mock_post:
|
||||
api("post", f"/api/v1/users/{new_user_id}/send-password-email", auth_headers)
|
||||
call_kwargs = mock_post.call_args
|
||||
assert call_kwargs.kwargs["headers"] == {"x-api-key": "super-secret-key"}
|
||||
|
||||
|
||||
def test_send_password_email_posts_to_webhook_and_issues_token(api, db, auth_headers, new_user_id):
|
||||
api(
|
||||
"patch", "/api/v1/system/password-webhook-config", auth_headers,
|
||||
@@ -48,8 +71,12 @@ def test_send_password_email_posts_to_webhook_and_issues_token(api, db, auth_hea
|
||||
mock_post.assert_called_once()
|
||||
call_kwargs = mock_post.call_args
|
||||
assert call_kwargs.args[0] == "https://example.com/power-automate-hook"
|
||||
assert call_kwargs.kwargs["json"]["email"] == "setpw@test.com"
|
||||
assert "token=" in call_kwargs.kwargs["json"]["set_password_url"]
|
||||
payload = call_kwargs.kwargs["json"]
|
||||
assert payload["to"] == "setpw@test.com"
|
||||
assert payload["subject"] == "Set Your Password"
|
||||
assert "HI Set Password User" in payload["body"]
|
||||
assert "token=" in payload["body"]
|
||||
assert "Purple Team Engineering" in payload["body"]
|
||||
|
||||
from app.models.password_setup_token import PasswordSetupToken
|
||||
token_row = db.query(PasswordSetupToken).filter(PasswordSetupToken.user_id == uuid.UUID(new_user_id)).first()
|
||||
|
||||
@@ -306,3 +306,67 @@ def test_detect_procedure_defaults_to_template_expected_detection(
|
||||
)
|
||||
assert resp.status_code == 201, resp.text
|
||||
assert resp.json()["detect_procedure"] == "Check process creation logs."
|
||||
|
||||
|
||||
class TestSuggestionsForTest:
|
||||
"""GET /procedure-suggestions/for-test/{test_id} — powers the blocking
|
||||
review popup a lead sees when opening a test with a pending suggestion."""
|
||||
|
||||
def test_returns_pending_suggestion_for_the_right_test(
|
||||
self, client, db, api, test_from_template, red_tech_headers, red_lead_headers,
|
||||
):
|
||||
test_id = test_from_template
|
||||
api("post", f"/api/v1/tests/{test_id}/start-execution", red_tech_headers)
|
||||
api(
|
||||
"patch", f"/api/v1/tests/{test_id}/red", red_tech_headers,
|
||||
json={"procedure_text": "Ran discovery.\nwhoami /priv\nDone."},
|
||||
)
|
||||
_add_evidence(db, test_id, TeamSide.red)
|
||||
api("post", f"/api/v1/tests/{test_id}/submit-red", red_tech_headers)
|
||||
|
||||
resp = api("get", f"/api/v1/procedure-suggestions/for-test/{test_id}", red_lead_headers)
|
||||
assert resp.status_code == 200, resp.text
|
||||
suggestions = resp.json()
|
||||
assert len(suggestions) == 1
|
||||
assert suggestions[0]["suggested_text"] == "whoami /priv"
|
||||
|
||||
def test_returns_empty_for_a_test_with_no_suggestion(
|
||||
self, client, db, api, test_from_template, red_lead_headers,
|
||||
):
|
||||
resp = api("get", f"/api/v1/procedure-suggestions/for-test/{test_from_template}", red_lead_headers)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json() == []
|
||||
|
||||
def test_blue_lead_does_not_see_a_red_suggestion_for_the_test(
|
||||
self, client, db, api, test_from_template, red_tech_headers, blue_lead_headers,
|
||||
):
|
||||
test_id = test_from_template
|
||||
api("post", f"/api/v1/tests/{test_id}/start-execution", red_tech_headers)
|
||||
api(
|
||||
"patch", f"/api/v1/tests/{test_id}/red", red_tech_headers,
|
||||
json={"procedure_text": "Ran discovery.\nwhoami /priv\nDone."},
|
||||
)
|
||||
_add_evidence(db, test_id, TeamSide.red)
|
||||
api("post", f"/api/v1/tests/{test_id}/submit-red", red_tech_headers)
|
||||
|
||||
resp = api("get", f"/api/v1/procedure-suggestions/for-test/{test_id}", blue_lead_headers)
|
||||
assert resp.status_code == 200, resp.text
|
||||
assert resp.json() == []
|
||||
|
||||
def test_suggestion_disappears_from_for_test_once_approved(
|
||||
self, client, db, api, test_from_template, red_tech_headers, red_lead_headers,
|
||||
):
|
||||
test_id = test_from_template
|
||||
api("post", f"/api/v1/tests/{test_id}/start-execution", red_tech_headers)
|
||||
api(
|
||||
"patch", f"/api/v1/tests/{test_id}/red", red_tech_headers,
|
||||
json={"procedure_text": "Ran discovery.\nwhoami /priv\nDone."},
|
||||
)
|
||||
_add_evidence(db, test_id, TeamSide.red)
|
||||
api("post", f"/api/v1/tests/{test_id}/submit-red", red_tech_headers)
|
||||
|
||||
suggestion = api("get", f"/api/v1/procedure-suggestions/for-test/{test_id}", red_lead_headers).json()[0]
|
||||
api("post", f"/api/v1/procedure-suggestions/{suggestion['id']}/approve", red_lead_headers)
|
||||
|
||||
resp = api("get", f"/api/v1/procedure-suggestions/for-test/{test_id}", red_lead_headers)
|
||||
assert resp.json() == []
|
||||
|
||||
Reference in New Issue
Block a user