mirror of
https://github.com/bec-project/bec_atlas.git
synced 2025-07-14 07:01:48 +02:00
tests: added more backend tests
This commit is contained in:
96
backend/tests/test_deployment_access_router.py
Normal file
96
backend/tests/test_deployment_access_router.py
Normal file
@ -0,0 +1,96 @@
|
||||
import pytest
|
||||
|
||||
from bec_atlas.model.model import DeploymentAccess
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def logged_in_client(backend):
|
||||
client, _ = backend
|
||||
response = client.post(
|
||||
"/api/v1/user/login", json={"username": "admin@bec_atlas.ch", "password": "admin"}
|
||||
)
|
||||
assert response.status_code == 200
|
||||
token = response.json()
|
||||
assert isinstance(token, str)
|
||||
assert len(token) > 20
|
||||
client.headers.update({"Authorization": f"Bearer {token}"})
|
||||
return client
|
||||
|
||||
|
||||
def test_deployment_access_router_invalid_deployment_id(logged_in_client):
|
||||
"""
|
||||
Test that the deployment access endpoint returns a 400 when the deployment id is invalid.
|
||||
"""
|
||||
response = logged_in_client.get("/api/v1/deployment_access", params={"deployment_id": "test"})
|
||||
assert response.status_code == 400
|
||||
assert response.json() == {"detail": "Invalid deployment ID"}
|
||||
|
||||
|
||||
def test_deployment_access_router(logged_in_client):
|
||||
"""
|
||||
Test that the deployment access endpoint returns a 200 when the deployment id is valid.
|
||||
"""
|
||||
deployments = logged_in_client.get(
|
||||
"/api/v1/deployments/realm", params={"realm": "demo_beamline_1"}
|
||||
).json()
|
||||
deployment_id = deployments[0]["_id"]
|
||||
|
||||
response = logged_in_client.get(
|
||||
"/api/v1/deployment_access", params={"deployment_id": deployment_id}
|
||||
)
|
||||
assert response.status_code == 200
|
||||
out = response.json()
|
||||
out = DeploymentAccess(**out)
|
||||
|
||||
|
||||
def test_patch_deployment_access(logged_in_client):
|
||||
"""
|
||||
Test that the deployment access endpoint returns a 200 when the deployment id is valid.
|
||||
"""
|
||||
deployments = logged_in_client.get(
|
||||
"/api/v1/deployments/realm", params={"realm": "demo_beamline_1"}
|
||||
).json()
|
||||
deployment_id = deployments[0]["_id"]
|
||||
|
||||
response = logged_in_client.get(
|
||||
"/api/v1/deployment_access", params={"deployment_id": deployment_id}
|
||||
)
|
||||
assert response.status_code == 200
|
||||
out = response.json()
|
||||
out = DeploymentAccess(**out)
|
||||
|
||||
response = logged_in_client.patch(
|
||||
"/api/v1/deployment_access",
|
||||
params={"deployment_id": deployment_id},
|
||||
json={
|
||||
"user_read_access": ["test1"],
|
||||
"user_write_access": ["test2"],
|
||||
"su_read_access": ["test3"],
|
||||
"su_write_access": ["test4"],
|
||||
"remote_read_access": ["test5"],
|
||||
"remote_write_access": ["test6"],
|
||||
},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
out = response.json()
|
||||
out = DeploymentAccess(**out)
|
||||
assert out.user_read_access == ["test1"]
|
||||
assert out.user_write_access == ["test2"]
|
||||
assert out.su_read_access == ["test3"]
|
||||
assert out.su_write_access == ["test4"]
|
||||
assert out.remote_read_access == ["test5"]
|
||||
assert out.remote_write_access == ["test6"]
|
||||
|
||||
for user in ["test1", "test2", "test3", "test4"]:
|
||||
out = logged_in_client.get(
|
||||
"/api/v1/bec_access", params={"deployment_id": deployment_id, "user": user}
|
||||
)
|
||||
assert out.status_code == 200
|
||||
out = out.json()
|
||||
assert "token" in out
|
||||
|
||||
for user in ["test5", "test6"]:
|
||||
out = logged_in_client.get(
|
||||
"/api/v1/bec_access", params={"deployment_id": deployment_id, "user": user}
|
||||
)
|
||||
assert out.status_code == 404
|
@ -18,7 +18,7 @@ def logged_in_client(backend):
|
||||
@pytest.mark.timeout(60)
|
||||
def test_get_deployment_credentials(logged_in_client):
|
||||
"""
|
||||
Test that the login endpoint returns a token.
|
||||
Test that the deployment credentials endpoint returns a token.
|
||||
"""
|
||||
client = logged_in_client
|
||||
|
||||
@ -34,7 +34,7 @@ def test_get_deployment_credentials(logged_in_client):
|
||||
@pytest.mark.timeout(60)
|
||||
def test_refresh_deployment_credentials(logged_in_client):
|
||||
"""
|
||||
Test that the login endpoint returns a token.
|
||||
Test that the refresh deployment credentials endpoint returns a new token.
|
||||
"""
|
||||
client = logged_in_client
|
||||
|
||||
@ -54,3 +54,87 @@ def test_refresh_deployment_credentials(logged_in_client):
|
||||
out = response.json()
|
||||
assert out == {"_id": deployment_id, "credential": out["credential"]}
|
||||
assert out["credential"] != old_token
|
||||
|
||||
|
||||
@pytest.mark.timeout(60)
|
||||
def test_deployment_credential_rejects_unauthorized_user(backend):
|
||||
"""
|
||||
Test that the deployment credentials endpoint returns a 403
|
||||
when the user is not authorized.
|
||||
"""
|
||||
client, _ = backend
|
||||
response = client.post(
|
||||
"/api/v1/user/login", json={"username": "jane.doe@bec_atlas.ch", "password": "atlas"}
|
||||
)
|
||||
assert response.status_code == 200
|
||||
token = response.json()
|
||||
assert isinstance(token, str)
|
||||
assert len(token) > 20
|
||||
client.headers.update({"Authorization": f"Bearer {token}"})
|
||||
|
||||
deployments = client.get(
|
||||
"/api/v1/deployments/realm", params={"realm": "demo_beamline_1"}
|
||||
).json()
|
||||
deployment_id = deployments[0]["_id"]
|
||||
|
||||
response = client.get("/api/v1/deploymentCredentials", params={"deployment_id": deployment_id})
|
||||
assert response.status_code == 403
|
||||
assert response.json() == {"detail": "User does not have permission to access this resource."}
|
||||
|
||||
|
||||
@pytest.mark.timeout(60)
|
||||
def test_refresh_deployment_credentials_rejects_unauthorized_user(backend):
|
||||
"""
|
||||
Test that the refresh deployment credentials endpoint returns a 403
|
||||
when the user is not authorized.
|
||||
"""
|
||||
client, _ = backend
|
||||
response = client.post(
|
||||
"/api/v1/user/login", json={"username": "jane.doe@bec_atlas.ch", "password": "atlas"}
|
||||
)
|
||||
assert response.status_code == 200
|
||||
token = response.json()
|
||||
assert isinstance(token, str)
|
||||
assert len(token) > 20
|
||||
client.headers.update({"Authorization": f"Bearer {token}"})
|
||||
|
||||
deployments = client.get(
|
||||
"/api/v1/deployments/realm", params={"realm": "demo_beamline_1"}
|
||||
).json()
|
||||
deployment_id = deployments[0]["_id"]
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/deploymentCredentials/refresh", params={"deployment_id": deployment_id}
|
||||
)
|
||||
assert response.status_code == 403
|
||||
assert response.json() == {"detail": "User does not have permission to access this resource."}
|
||||
|
||||
|
||||
@pytest.mark.timeout(60)
|
||||
def test_get_deployment_credentials_wrong_id(logged_in_client):
|
||||
"""
|
||||
Test that the deployment credentials endpoint returns a 400
|
||||
when the deployment ID is invalid.
|
||||
"""
|
||||
client = logged_in_client
|
||||
|
||||
response = client.get("/api/v1/deploymentCredentials", params={"deployment_id": "wrong_id"})
|
||||
assert response.status_code == 400
|
||||
assert response.json() == {"detail": "Invalid deployment ID"}
|
||||
|
||||
|
||||
@pytest.mark.timeout(60)
|
||||
def test_deployment_credentials_refresh_not_found(logged_in_client):
|
||||
"""
|
||||
Test that the deployment credentials refresh endpoint returns a 404
|
||||
when the deployment is not found.
|
||||
"""
|
||||
client = logged_in_client
|
||||
|
||||
response = client.post(
|
||||
"/api/v1/deploymentCredentials/refresh",
|
||||
params={"deployment_id": "678aa8d4875568640bd92000"},
|
||||
)
|
||||
assert response.status_code == 404
|
||||
out = response.json()
|
||||
assert out == {"detail": "Deployment not found"}
|
||||
|
@ -7,7 +7,7 @@ def backend_client(backend):
|
||||
return client
|
||||
|
||||
|
||||
@pytest.mark.timeout(10)
|
||||
@pytest.mark.timeout(20)
|
||||
def test_login(backend_client):
|
||||
"""
|
||||
Test that the login endpoint returns a token.
|
||||
@ -33,7 +33,7 @@ def test_login_wrong_password(backend_client):
|
||||
assert response.json() == {"detail": "User not found or password is incorrect"}
|
||||
|
||||
|
||||
@pytest.mark.timeout(10)
|
||||
@pytest.mark.timeout(20)
|
||||
def test_login_unknown_user(backend_client):
|
||||
"""
|
||||
Test that the login returns a 401 when the user is unknown.
|
||||
|
Reference in New Issue
Block a user