Spaces:
Sleeping
Sleeping
File size: 6,304 Bytes
da9494d |
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 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 |
"""
Comprehensive Tests for Blink Router
Tests cover:
1. POST /blink - Data submission
2. Client-user linking
3. Encryption/decryption flow
4. Rate limiting
5. Authentication requirements
Uses mocked database and encryption services.
"""
import pytest
from unittest.mock import MagicMock, AsyncMock, patch
from fastapi.testclient import TestClient
from fastapi import FastAPI
# ============================================================================
# 1. Blink Data Submission Tests
# ============================================================================
class TestBlinkDataSubmission:
"""Test blink data collection endpoint."""
def test_blink_endpoint_exists(self):
"""Blink endpoint is accessible."""
from routers.blink import router
app = FastAPI()
app.include_router(router)
client = TestClient(app)
# Should accept POST requests
response = client.post("/blink")
# May return error without proper data, but endpoint exists
assert response.status_code in [200, 204, 400, 401, 422, 500]
def test_blink_without_auth(self):
"""Blink endpoint works without authentication."""
from routers.blink import router
from core.database import get_db
app = FastAPI()
# Mock database
async def mock_get_db():
mock_db = AsyncMock()
yield mock_db
app.dependency_overrides[get_db] = mock_get_db
app.include_router(router)
client = TestClient(app)
with patch('routers.blink.check_rate_limit', return_value=True):
response = client.post(
"/blink",
json={"client_user_id": "temp_123", "data": {}}
)
# Should work (may be 204 No Content or 200)
assert response.status_code in [200, 204]
def test_blink_rate_limited(self):
"""Blink endpoint respects rate limiting."""
from routers.blink import router
from core.database import get_db
app = FastAPI()
async def mock_get_db():
mock_db = AsyncMock()
yield mock_db
app.dependency_overrides[get_db] = mock_get_db
app.include_router(router)
client = TestClient(app)
# Mock rate limit exceeded
with patch('routers.blink.check_rate_limit', return_value=False):
response = client.post(
"/blink",
json={"client_user_id": "temp_123", "data": {}}
)
assert response.status_code == 429 # Too Many Requests
# ============================================================================
# 2. Client-User Linking Tests
# ============================================================================
class TestClientUserLinking:
"""Test client-user linking functionality."""
@pytest.mark.asyncio
async def test_creates_client_user_entry(self, db_session):
"""Blink creates ClientUser entry if not exists."""
from core.models import ClientUser
from sqlalchemy import select
# Simulate blink creating client user
client_user = ClientUser(
client_user_id="blink_test_123",
ip_address="192.168.1.1"
)
db_session.add(client_user)
await db_session.commit()
# Verify created
result = await db_session.execute(
select(ClientUser).where(ClientUser.client_user_id == "blink_test_123")
)
found = result.scalar_one_or_none()
assert found is not None
assert found.ip_address == "192.168.1.1"
@pytest.mark.asyncio
async def test_links_to_authenticated_user(self, db_session):
"""Authenticated blink links to user."""
from core.models import User, ClientUser
# Create user
user = User(user_id="usr_blink", email="blink@example.com")
db_session.add(user)
await db_session.commit()
# Create linked client user
client_user = ClientUser(
user_id=user.id,
client_user_id="auth_blink_123",
ip_address="10.0.0.1"
)
db_session.add(client_user)
await db_session.commit()
assert client_user.user_id == user.id
# ============================================================================
# 3. Data Validation Tests
# ============================================================================
class TestBlinkDataValidation:
"""Test blink data validation."""
def test_accepts_valid_json(self):
"""Accepts valid JSON data."""
from routers.blink import router
from core.database import get_db
app = FastAPI()
async def mock_get_db():
mock_db = AsyncMock()
yield mock_db
app.dependency_overrides[get_db] = mock_get_db
app.include_router(router)
client = TestClient(app)
with patch('routers.blink.check_rate_limit', return_value=True):
response = client.post(
"/blink",
json={
"client_user_id": "test_456",
"data": {"event": "page_view", "page": "/home"}
}
)
assert response.status_code in [200, 204]
def test_handles_missing_fields(self):
"""Handles requests with missing fields gracefully."""
from routers.blink import router
from core.database import get_db
app = FastAPI()
async def mock_get_db():
mock_db = AsyncMock()
yield mock_db
app.dependency_overrides[get_db] = mock_get_db
app.include_router(router)
client = TestClient(app)
with patch('routers.blink.check_rate_limit', return_value=True):
response = client.post("/blink", json={})
# Should handle gracefully (may return error or success)
assert response.status_code in [200, 204, 400, 422]
if __name__ == "__main__":
pytest.main([__file__, "-v"])
|