231 lines
7.1 KiB
Python
231 lines
7.1 KiB
Python
from urllib.parse import parse_qs, urlparse
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from services.auth_service.app import app
|
|
from services.shared.security import decode_app_token
|
|
|
|
|
|
def test_login_success():
|
|
client = TestClient(app)
|
|
response = client.post('/auth/login', json={'username': 'admin', 'password': 'admin123'})
|
|
assert response.status_code == 200
|
|
body = response.json()
|
|
assert body['access_token']
|
|
assert body['role'] == 'admin'
|
|
assert body['auth_source'] == 'local'
|
|
claims = decode_app_token(body['access_token'])
|
|
assert claims['username'] == 'admin'
|
|
assert claims['role'] == 'admin'
|
|
assert claims['auth_source'] == 'local'
|
|
|
|
|
|
def test_create_user_as_admin():
|
|
client = TestClient(app)
|
|
response = client.post(
|
|
'/users',
|
|
json={
|
|
'username': 'newoperator',
|
|
'password': 'password1',
|
|
'full_name': 'New Operator',
|
|
'role': 'operator',
|
|
},
|
|
headers={'X-User': 'admin', 'X-Role': 'admin'},
|
|
)
|
|
assert response.status_code == 200
|
|
assert response.json()['username'] == 'newoperator'
|
|
|
|
|
|
def test_operator_cannot_create_user():
|
|
client = TestClient(app)
|
|
response = client.post(
|
|
'/users',
|
|
json={
|
|
'username': 'blockeduser',
|
|
'password': 'password1',
|
|
'full_name': 'Blocked User',
|
|
'role': 'operator',
|
|
},
|
|
headers={'X-User': 'operator', 'X-Role': 'operator'},
|
|
)
|
|
assert response.status_code == 403
|
|
|
|
|
|
def test_admin_can_update_user():
|
|
client = TestClient(app)
|
|
created = client.post(
|
|
'/users',
|
|
json={
|
|
'username': 'updatableuser',
|
|
'password': 'password1',
|
|
'full_name': 'Updatable User',
|
|
'role': 'operator',
|
|
},
|
|
headers={'X-User': 'admin', 'X-Role': 'admin'},
|
|
)
|
|
assert created.status_code == 200
|
|
|
|
updated = client.patch(
|
|
f"/users/{created.json()['user_id']}",
|
|
json={
|
|
'full_name': 'Updated User',
|
|
'role': 'analyst',
|
|
'password': 'password2',
|
|
},
|
|
headers={'X-User': 'admin', 'X-Role': 'admin'},
|
|
)
|
|
assert updated.status_code == 200
|
|
assert updated.json()['full_name'] == 'Updated User'
|
|
assert updated.json()['role'] == 'analyst'
|
|
|
|
|
|
def test_supervisor_cannot_update_user():
|
|
client = TestClient(app)
|
|
created = client.post(
|
|
'/users',
|
|
json={
|
|
'username': 'guardeduser',
|
|
'password': 'password1',
|
|
'full_name': 'Guarded User',
|
|
'role': 'operator',
|
|
},
|
|
headers={'X-User': 'admin', 'X-Role': 'admin'},
|
|
)
|
|
assert created.status_code == 200
|
|
|
|
denied = client.patch(
|
|
f"/users/{created.json()['user_id']}",
|
|
json={'full_name': 'Should Not Update'},
|
|
headers={'X-User': 'supervisor', 'X-Role': 'supervisor'},
|
|
)
|
|
assert denied.status_code == 403
|
|
|
|
|
|
def test_admin_can_delete_user():
|
|
client = TestClient(app)
|
|
created = client.post(
|
|
'/users',
|
|
json={
|
|
'username': 'deletecandidate',
|
|
'password': 'password1',
|
|
'full_name': 'Delete Candidate',
|
|
'role': 'operator',
|
|
},
|
|
headers={'X-User': 'admin', 'X-Role': 'admin'},
|
|
)
|
|
assert created.status_code == 200
|
|
|
|
deleted = client.delete(
|
|
f"/users/{created.json()['user_id']}",
|
|
headers={'X-User': 'admin', 'X-Role': 'admin'},
|
|
)
|
|
assert deleted.status_code == 200
|
|
assert deleted.json()['deleted'] is True
|
|
|
|
lookup = client.get(
|
|
f"/users/{created.json()['user_id']}",
|
|
headers={'X-User': 'admin', 'X-Role': 'admin'},
|
|
)
|
|
assert lookup.status_code == 404
|
|
|
|
|
|
def test_admin_cannot_delete_current_user():
|
|
client = TestClient(app)
|
|
|
|
users = client.get('/users', headers={'X-User': 'admin', 'X-Role': 'admin'})
|
|
assert users.status_code == 200
|
|
admin_user = next(item for item in users.json() if item['username'] == 'admin')
|
|
|
|
deleted = client.delete(
|
|
f"/users/{admin_user['user_id']}",
|
|
headers={'X-User': 'admin', 'X-Role': 'admin'},
|
|
)
|
|
assert deleted.status_code == 409
|
|
|
|
|
|
def test_supervisor_cannot_delete_user():
|
|
client = TestClient(app)
|
|
created = client.post(
|
|
'/users',
|
|
json={
|
|
'username': 'nodeletebysupervisor',
|
|
'password': 'password1',
|
|
'full_name': 'No Delete',
|
|
'role': 'operator',
|
|
},
|
|
headers={'X-User': 'admin', 'X-Role': 'admin'},
|
|
)
|
|
assert created.status_code == 200
|
|
|
|
denied = client.delete(
|
|
f"/users/{created.json()['user_id']}",
|
|
headers={'X-User': 'supervisor', 'X-Role': 'supervisor'},
|
|
)
|
|
assert denied.status_code == 403
|
|
|
|
|
|
def test_oidc_config_reflects_disabled_state(monkeypatch):
|
|
monkeypatch.setenv('OIDC_ENABLED', '0')
|
|
client = TestClient(app)
|
|
|
|
response = client.get('/auth/oidc/config')
|
|
|
|
assert response.status_code == 200
|
|
assert response.json()['enabled'] is False
|
|
|
|
|
|
def test_oidc_start_redirects_with_override_urls(monkeypatch):
|
|
monkeypatch.setenv('OIDC_ENABLED', '1')
|
|
monkeypatch.setenv('OIDC_AUTHORIZE_URL', 'http://keycloak.local/auth')
|
|
monkeypatch.setenv('OIDC_TOKEN_URL', 'http://keycloak.local/token')
|
|
client = TestClient(app)
|
|
|
|
response = client.get('/auth/oidc/start', follow_redirects=False)
|
|
|
|
assert response.status_code == 302
|
|
location = response.headers['location']
|
|
assert location.startswith('http://keycloak.local/auth?')
|
|
params = parse_qs(urlparse(location).query)
|
|
assert params['client_id'][0] == 'mvp-contact-center'
|
|
assert params['response_type'][0] == 'code'
|
|
assert 'state' in params
|
|
|
|
|
|
def test_oidc_callback_accepts_mapped_mock_user(monkeypatch):
|
|
monkeypatch.setenv('OIDC_ENABLED', '1')
|
|
monkeypatch.setenv('OIDC_TEST_MODE', '1')
|
|
monkeypatch.setenv('OIDC_AUTHORIZE_URL', 'http://keycloak.local/auth')
|
|
monkeypatch.setenv('OIDC_TOKEN_URL', 'http://keycloak.local/token')
|
|
client = TestClient(app)
|
|
|
|
start = client.get('/auth/oidc/start', follow_redirects=False)
|
|
params = parse_qs(urlparse(start.headers['location']).query)
|
|
state = params['state'][0]
|
|
|
|
response = client.get(
|
|
f'/auth/oidc/callback?state={state}&mock_sub=sub-1&mock_username=kc.user&mock_groups=kc_operator'
|
|
)
|
|
|
|
assert response.status_code == 200
|
|
assert 'oidc-login' in response.text
|
|
assert 'Corporate login completed' in response.text
|
|
|
|
|
|
def test_oidc_callback_rejects_unmapped_mock_user(monkeypatch):
|
|
monkeypatch.setenv('OIDC_ENABLED', '1')
|
|
monkeypatch.setenv('OIDC_TEST_MODE', '1')
|
|
monkeypatch.setenv('OIDC_AUTHORIZE_URL', 'http://keycloak.local/auth')
|
|
monkeypatch.setenv('OIDC_TOKEN_URL', 'http://keycloak.local/token')
|
|
client = TestClient(app)
|
|
|
|
start = client.get('/auth/oidc/start', follow_redirects=False)
|
|
params = parse_qs(urlparse(start.headers['location']).query)
|
|
state = params['state'][0]
|
|
|
|
response = client.get(
|
|
f'/auth/oidc/callback?state={state}&mock_sub=sub-2&mock_username=kc.bad&mock_groups=unknown_group'
|
|
)
|
|
|
|
assert response.status_code == 403
|
|
assert 'OIDC role is not mapped' in response.text
|