feat: Initial commit - Task Reporter incident response system
Complete implementation of the production line incident response system (生產線異常即時反應系統) including: Backend (FastAPI): - User authentication with AD integration and session management - Chat room management (create, list, update, members, roles) - Real-time messaging via WebSocket (typing indicators, reactions) - File storage with MinIO (upload, download, image preview) Frontend (React + Vite): - Authentication flow with token management - Room list with filtering, search, and pagination - Real-time chat interface with WebSocket - File upload with drag-and-drop and image preview - Member management and room settings - Breadcrumb navigation - 53 unit tests (Vitest) Specifications: - authentication: AD auth, sessions, JWT tokens - chat-room: rooms, members, templates - realtime-messaging: WebSocket, messages, reactions - file-storage: MinIO integration, file management - frontend-core: React SPA structure 🤖 Generated with [Claude Code](https://claude.com/claude-code) Co-Authored-By: Claude <noreply@anthropic.com>
This commit is contained in:
210
tests/test_authentication.py
Normal file
210
tests/test_authentication.py
Normal file
@@ -0,0 +1,210 @@
|
||||
"""Unit and integration tests for authentication module
|
||||
|
||||
測試範圍:
|
||||
- EncryptionService (加密/解密)
|
||||
- ADAuthService (AD API 整合)
|
||||
- SessionService (資料庫操作)
|
||||
- Login/Logout endpoints (整合測試)
|
||||
"""
|
||||
import pytest
|
||||
from app.modules.auth.services.encryption import encryption_service
|
||||
from app.modules.auth.services.session_service import session_service
|
||||
from datetime import datetime, timedelta
|
||||
|
||||
|
||||
class TestEncryptionService:
|
||||
"""測試 EncryptionService"""
|
||||
|
||||
def test_encrypt_decrypt_roundtrip(self):
|
||||
"""測試加密/解密往返"""
|
||||
plaintext = "4RFV5tgb6yhn"
|
||||
encrypted = encryption_service.encrypt_password(plaintext)
|
||||
decrypted = encryption_service.decrypt_password(encrypted)
|
||||
|
||||
assert decrypted == plaintext
|
||||
assert encrypted != plaintext # 密文不應等於明文
|
||||
|
||||
def test_encrypted_output_differs(self):
|
||||
"""測試加密輸出與明文不同"""
|
||||
plaintext = "test_password_123"
|
||||
encrypted = encryption_service.encrypt_password(plaintext)
|
||||
|
||||
assert encrypted != plaintext
|
||||
assert len(encrypted) > len(plaintext) # 密文通常更長
|
||||
|
||||
|
||||
class TestSessionService:
|
||||
"""測試 SessionService"""
|
||||
|
||||
def test_create_session(self, db_session):
|
||||
"""測試建立 session"""
|
||||
expires_at = datetime.utcnow() + timedelta(hours=1)
|
||||
|
||||
session = session_service.create_session(
|
||||
db=db_session,
|
||||
username="test@example.com",
|
||||
display_name="Test User",
|
||||
ad_token="test_ad_token",
|
||||
encrypted_password="encrypted_pwd",
|
||||
ad_token_expires_at=expires_at,
|
||||
)
|
||||
|
||||
assert session.id is not None
|
||||
assert session.username == "test@example.com"
|
||||
assert session.display_name == "Test User"
|
||||
assert session.internal_token is not None # UUID4 generated
|
||||
assert session.refresh_attempt_count == 0
|
||||
|
||||
def test_get_session_by_token(self, db_session):
|
||||
"""測試根據 token 查詢 session"""
|
||||
expires_at = datetime.utcnow() + timedelta(hours=1)
|
||||
|
||||
# Create session
|
||||
created_session = session_service.create_session(
|
||||
db=db_session,
|
||||
username="test@example.com",
|
||||
display_name="Test User",
|
||||
ad_token="test_ad_token",
|
||||
encrypted_password="encrypted_pwd",
|
||||
ad_token_expires_at=expires_at,
|
||||
)
|
||||
|
||||
# Retrieve by token
|
||||
retrieved_session = session_service.get_session_by_token(
|
||||
db=db_session, internal_token=created_session.internal_token
|
||||
)
|
||||
|
||||
assert retrieved_session is not None
|
||||
assert retrieved_session.id == created_session.id
|
||||
assert retrieved_session.username == created_session.username
|
||||
|
||||
def test_update_activity(self, db_session):
|
||||
"""測試更新活動時間"""
|
||||
expires_at = datetime.utcnow() + timedelta(hours=1)
|
||||
|
||||
session = session_service.create_session(
|
||||
db=db_session,
|
||||
username="test@example.com",
|
||||
display_name="Test User",
|
||||
ad_token="test_ad_token",
|
||||
encrypted_password="encrypted_pwd",
|
||||
ad_token_expires_at=expires_at,
|
||||
)
|
||||
|
||||
original_activity = session.last_activity
|
||||
|
||||
# Wait a moment and update
|
||||
import time
|
||||
|
||||
time.sleep(0.1)
|
||||
session_service.update_activity(db=db_session, session_id=session.id)
|
||||
|
||||
# Retrieve updated session
|
||||
updated_session = session_service.get_session_by_token(
|
||||
db=db_session, internal_token=session.internal_token
|
||||
)
|
||||
|
||||
assert updated_session.last_activity > original_activity
|
||||
|
||||
def test_increment_refresh_attempts(self, db_session):
|
||||
"""測試增加重試計數器"""
|
||||
expires_at = datetime.utcnow() + timedelta(hours=1)
|
||||
|
||||
session = session_service.create_session(
|
||||
db=db_session,
|
||||
username="test@example.com",
|
||||
display_name="Test User",
|
||||
ad_token="test_ad_token",
|
||||
encrypted_password="encrypted_pwd",
|
||||
ad_token_expires_at=expires_at,
|
||||
)
|
||||
|
||||
assert session.refresh_attempt_count == 0
|
||||
|
||||
# Increment
|
||||
new_count = session_service.increment_refresh_attempts(db=db_session, session_id=session.id)
|
||||
assert new_count == 1
|
||||
|
||||
new_count = session_service.increment_refresh_attempts(db=db_session, session_id=session.id)
|
||||
assert new_count == 2
|
||||
|
||||
def test_delete_session(self, db_session):
|
||||
"""測試刪除 session"""
|
||||
expires_at = datetime.utcnow() + timedelta(hours=1)
|
||||
|
||||
session = session_service.create_session(
|
||||
db=db_session,
|
||||
username="test@example.com",
|
||||
display_name="Test User",
|
||||
ad_token="test_ad_token",
|
||||
encrypted_password="encrypted_pwd",
|
||||
ad_token_expires_at=expires_at,
|
||||
)
|
||||
|
||||
session_id = session.id
|
||||
internal_token = session.internal_token
|
||||
|
||||
# Delete
|
||||
session_service.delete_session(db=db_session, session_id=session_id)
|
||||
|
||||
# Verify deleted
|
||||
retrieved = session_service.get_session_by_token(db=db_session, internal_token=internal_token)
|
||||
assert retrieved is None
|
||||
|
||||
|
||||
class TestAuthenticationEndpoints:
|
||||
"""測試認證 API 端點 (整合測試)
|
||||
|
||||
注意:這些測試會實際呼叫 AD API,需要網路連線
|
||||
如果要在 CI/CD 中執行,應該 mock AD API
|
||||
"""
|
||||
|
||||
@pytest.mark.skip(reason="Requires actual AD API credentials")
|
||||
def test_login_success(self, client):
|
||||
"""測試成功登入"""
|
||||
response = client.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "ymirliu@panjit.com.tw", "password": "4RFV5tgb6yhn"},
|
||||
)
|
||||
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
assert "token" in data
|
||||
assert "display_name" in data
|
||||
assert data["display_name"] == "ymirliu 劉念萱"
|
||||
|
||||
def test_login_invalid_credentials(self, client):
|
||||
"""測試錯誤憑證登入"""
|
||||
response = client.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "wrong@example.com", "password": "wrongpassword"},
|
||||
)
|
||||
|
||||
assert response.status_code == 401
|
||||
data = response.json()
|
||||
assert "detail" in data
|
||||
|
||||
def test_logout_without_token(self, client):
|
||||
"""測試無 token 登出"""
|
||||
response = client.post("/api/auth/logout")
|
||||
|
||||
assert response.status_code == 401
|
||||
|
||||
@pytest.mark.skip(reason="Requires actual login first")
|
||||
def test_logout_with_valid_token(self, client):
|
||||
"""測試有效 token 登出"""
|
||||
# First login
|
||||
login_response = client.post(
|
||||
"/api/auth/login",
|
||||
json={"username": "ymirliu@panjit.com.tw", "password": "4RFV5tgb6yhn"},
|
||||
)
|
||||
token = login_response.json()["token"]
|
||||
|
||||
# Then logout
|
||||
logout_response = client.post(
|
||||
"/api/auth/logout", headers={"Authorization": f"Bearer {token}"}
|
||||
)
|
||||
|
||||
assert logout_response.status_code == 200
|
||||
data = logout_response.json()
|
||||
assert data["message"] == "Logout successful"
|
||||
Reference in New Issue
Block a user