File size: 4,972 Bytes
cf898d5
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
cede2ad
cf898d5
 
 
cede2ad
cf898d5
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
cede2ad
cf898d5
 
 
 
 
cede2ad
cf898d5
 
 
 
 
 
 
 
 
 
 
 
 
cede2ad
cf898d5
 
 
 
 
cede2ad
cf898d5
 
 
 
 
 
 
 
 
 
 
 
 
 
cede2ad
 
cf898d5
 
 
 
 
 
 
 
 
 
cede2ad
cf898d5
 
 
 
cede2ad
cf898d5
 
 
 
 
cede2ad
cf898d5
 
 
 
 
 
 
 
 
cede2ad
cf898d5
 
 
 
 
cede2ad
cf898d5
 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
"""Tests for the auth router endpoints."""

from unittest.mock import MagicMock, patch
import pytest
from fastapi.testclient import TestClient

from synthra.execution.exceptions import (
    ExecutionAuthenticationError,
    VerificationRequiredError,
)


@pytest.fixture
def mock_service_state():
    """Fixture providing a mocked ServiceState."""
    state = MagicMock()
    state.auth_state = "Logged Out"
    state.auth_username = None
    state.auth_verification_url = None
    state.execution_client = MagicMock()

    async def async_noop() -> None:
        pass

    state.initialize = async_noop
    state.shutdown = async_noop
    return state


@pytest.fixture
def client(mock_service_state):
    """Provide a TestClient with mocked ServiceState initialization."""
    with patch(
        "synthra.api.app.ServiceState.get_instance", return_value=mock_service_state
    ):
        from synthra.api.app import create_app

        application = create_app()
        # Ensure service state is attached to application
        application.state.service = mock_service_state
        with TestClient(application, raise_server_exceptions=True) as tc:
            yield tc


def test_login_success(client, mock_service_state):
    """POST /api/auth/login returns success on valid credentials."""
    mock_service_state.execution_client.authenticate.return_value = None

    response = client.post(
        "/api/auth/login",
        json={"username": "test_user", "password": "test_password", "remember": True},
    )
    assert response.status_code == 200
    assert response.json() == {
        "status": "success",
        "message": None,
        "verification_url": None,
    }

    # Verify session cookie was set
    assert "session_id" in response.cookies
    assert mock_service_state.auth_state == "Authenticated"
    assert mock_service_state.auth_username == "test_user"


def test_login_invalid_credentials(client, mock_service_state):
    """POST /api/auth/login returns error on invalid credentials."""
    mock_service_state.execution_client.authenticate.side_effect = (
        ExecutionAuthenticationError("Auth failed")
    )

    response = client.post(
        "/api/auth/login",
        json={"username": "bad_user", "password": "bad_password", "remember": False},
    )
    assert response.status_code == 200
    assert response.json() == {
        "status": "error",
        "message": "Invalid username or password",
        "verification_url": None,
    }
    assert "session_id" not in response.cookies
    assert mock_service_state.auth_state == "Logged Out"


def test_login_verification_required(client, mock_service_state):
    """POST /api/auth/login returns verification_required status on MFA."""
    mock_service_state.execution_client.authenticate.side_effect = (
        VerificationRequiredError("https://mfa.example.test")
    )

    response = client.post(
        "/api/auth/login",
        json={"username": "mfa_user", "password": "mfa_password", "remember": False},
    )
    assert response.status_code == 200
    assert response.json() == {
        "status": "verification_required",
        "verification_url": "https://mfa.example.test",
        "message": None,
    }
    assert "session_id" not in response.cookies
    assert mock_service_state.auth_state == "Waiting for Biometric Verification"
    assert mock_service_state.auth_verification_url == "https://mfa.example.test"


def test_status_unauthenticated(client):
    """GET /api/auth/status returns unauthenticated when no session exists."""
    response = client.get("/api/auth/status")
    assert response.status_code == 200
    assert response.json() == {
        "authenticated": False,
        "username": None,
        "state": "Logged Out",
        "verification_url": None,
    }


def test_status_authenticated_and_logout(client, mock_service_state):
    """GET /api/auth/status returns authenticated and logout clears session."""
    mock_service_state.execution_client.authenticate.return_value = None

    # 1. Login to establish session
    login_resp = client.post(
        "/api/auth/login",
        json={"username": "user1", "password": "pass1", "remember": False},
    )
    assert login_resp.status_code == 200

    # 2. Get status with cookie
    status_resp = client.get("/api/auth/status")
    assert status_resp.status_code == 200
    assert status_resp.json() == {
        "authenticated": True,
        "username": "user1",
        "state": "Authenticated",
        "verification_url": None,
    }

    # 3. Invalidate via logout
    logout_resp = client.post("/api/auth/logout")
    assert logout_resp.status_code == 200
    assert logout_resp.json() == {"status": "success", "message": None}

    # Cookie should be deleted
    # 4. Status should now be unauthenticated
    status_resp_after = client.get("/api/auth/status")
    assert status_resp_after.status_code == 200
    assert status_resp_after.json()["authenticated"] is False