Keith Smith
126 lines
4.0 KiB
Python
126 lines
4.0 KiB
Python
from fastapi.testclient import TestClient
|
|
from sqlalchemy.orm import Session
|
|
|
|
from app.core.secrets import decrypt_secret
|
|
from app.models import Credential
|
|
|
|
|
|
def test_create_snmp_profile_encrypts_community_and_masks_secret(client: TestClient, db_session: Session) -> None:
|
|
response = client.post(
|
|
"/credentials/snmp",
|
|
json={
|
|
"name": "Core Switch Read Only",
|
|
"version": "2c",
|
|
"community": "public-read",
|
|
"port": 161,
|
|
"timeout_seconds": 7,
|
|
"retries": 2,
|
|
},
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body["name"] == "Core Switch Read Only"
|
|
assert body["credential_type"] == "snmp"
|
|
assert body["version"] == "2c"
|
|
assert body["port"] == 161
|
|
assert body["timeout_seconds"] == 7
|
|
assert body["retries"] == 2
|
|
assert body["has_secret"] is True
|
|
assert "community" not in body
|
|
assert "encrypted_secret" not in body
|
|
|
|
profile = db_session.get(Credential, body["id"])
|
|
assert profile is not None
|
|
assert profile.credential_type == "snmp"
|
|
assert profile.encrypted_secret != "public-read"
|
|
assert decrypt_secret(profile.encrypted_secret) == "public-read"
|
|
assert profile.extra == {"version": "2c", "port": 161, "timeout_seconds": 7, "retries": 2}
|
|
|
|
|
|
def test_update_snmp_profile_without_community_preserves_saved_secret(client: TestClient, db_session: Session) -> None:
|
|
create_response = client.post(
|
|
"/credentials/snmp",
|
|
json={
|
|
"name": "Access Switch",
|
|
"community": "access-read",
|
|
},
|
|
)
|
|
profile_id = create_response.json()["id"]
|
|
original_secret = db_session.get(Credential, profile_id).encrypted_secret
|
|
|
|
update_response = client.patch(
|
|
f"/credentials/snmp/{profile_id}",
|
|
json={
|
|
"name": "Access Switches",
|
|
"timeout_seconds": 9,
|
|
"retries": 3,
|
|
},
|
|
)
|
|
|
|
assert update_response.status_code == 200
|
|
body = update_response.json()
|
|
assert body["name"] == "Access Switches"
|
|
assert body["timeout_seconds"] == 9
|
|
assert body["retries"] == 3
|
|
assert body["has_secret"] is True
|
|
assert "community" not in body
|
|
assert "encrypted_secret" not in body
|
|
|
|
profile = db_session.get(Credential, profile_id)
|
|
assert profile is not None
|
|
assert profile.encrypted_secret == original_secret
|
|
assert decrypt_secret(profile.encrypted_secret) == "access-read"
|
|
|
|
|
|
def test_update_snmp_profile_can_rotate_community(client: TestClient, db_session: Session) -> None:
|
|
create_response = client.post(
|
|
"/credentials/snmp",
|
|
json={
|
|
"name": "Router",
|
|
"community": "old-community",
|
|
},
|
|
)
|
|
profile_id = create_response.json()["id"]
|
|
|
|
update_response = client.patch(
|
|
f"/credentials/snmp/{profile_id}",
|
|
json={
|
|
"community": "new-community",
|
|
},
|
|
)
|
|
|
|
assert update_response.status_code == 200
|
|
assert update_response.json()["has_secret"] is True
|
|
|
|
profile = db_session.get(Credential, profile_id)
|
|
assert profile is not None
|
|
assert decrypt_secret(profile.encrypted_secret) == "new-community"
|
|
|
|
|
|
def test_list_and_delete_snmp_profiles(client: TestClient) -> None:
|
|
create_response = client.post(
|
|
"/credentials/snmp",
|
|
json={
|
|
"name": "Distribution Switch",
|
|
"community": "dist-read",
|
|
},
|
|
)
|
|
profile_id = create_response.json()["id"]
|
|
|
|
list_response = client.get("/credentials/snmp")
|
|
assert list_response.status_code == 200
|
|
profiles = list_response.json()
|
|
assert len(profiles) == 1
|
|
assert profiles[0]["id"] == profile_id
|
|
assert profiles[0]["has_secret"] is True
|
|
assert "community" not in profiles[0]
|
|
assert "encrypted_secret" not in profiles[0]
|
|
|
|
delete_response = client.delete(f"/credentials/snmp/{profile_id}")
|
|
assert delete_response.status_code == 204
|
|
|
|
list_after_delete_response = client.get("/credentials/snmp")
|
|
assert list_after_delete_response.status_code == 200
|
|
assert list_after_delete_response.json() == []
|