264 lines
9.7 KiB
Python
264 lines
9.7 KiB
Python
"""
|
|
Tests for ZynkTime application.
|
|
"""
|
|
import io
|
|
import os
|
|
from datetime import datetime
|
|
from unittest.mock import Mock, patch
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
from openpyxl import Workbook
|
|
|
|
from config import Config
|
|
from main import app
|
|
from services import (
|
|
create_time_event,
|
|
extract_work_time,
|
|
get_user_by_name,
|
|
parse_csv,
|
|
parse_xlsx,
|
|
validate_date,
|
|
)
|
|
|
|
|
|
# Test client for FastAPI
|
|
client = TestClient(app)
|
|
|
|
|
|
class TestServices:
|
|
"""Tests for service functions."""
|
|
|
|
def test_validate_date_valid(self):
|
|
"""Test date validation with valid date."""
|
|
assert validate_date("2024-01-15") is True
|
|
|
|
def test_validate_date_invalid(self):
|
|
"""Test date validation with invalid date."""
|
|
assert validate_date("invalid-date") is False
|
|
assert validate_date("2024-13-01") is False
|
|
|
|
def test_parse_csv_semicolon(self):
|
|
"""Test CSV parsing with semicolon delimiter."""
|
|
csv_content = "col1;col2;col3\nval1;val2;val3"
|
|
result = parse_csv(csv_content)
|
|
assert len(result) == 2
|
|
assert result[0] == ["col1", "col2", "col3"]
|
|
assert result[1] == ["val1", "val2", "val3"]
|
|
|
|
def test_parse_csv_comma(self):
|
|
"""Test CSV parsing with comma delimiter."""
|
|
csv_content = "col1,col2,col3\nval1,val2,val3"
|
|
result = parse_csv(csv_content)
|
|
assert len(result) == 2
|
|
assert result[0] == ["col1", "col2", "col3"]
|
|
assert result[1] == ["val1", "val2", "val3"]
|
|
|
|
def test_extract_work_time(self):
|
|
"""Test work time extraction from CSV rows."""
|
|
rows = [
|
|
["h1", "h2", "h3", "h4", "h5", "01/15/2024", "h7", "h8", "h9", "h10", "8.5"],
|
|
["v1", "v2", "v3", "v4", "v5", "01/16/2024", "v7", "v8", "v9", "v10", "7,5"],
|
|
]
|
|
result = extract_work_time(rows)
|
|
assert "2024-01-15" in result
|
|
assert "2024-01-16" in result
|
|
assert result["2024-01-15"] == 8.5
|
|
assert result["2024-01-16"] == 7.5
|
|
|
|
def test_extract_work_time_with_header(self):
|
|
"""Test work time extraction with header row."""
|
|
rows = [
|
|
["reg_period", "xreg_period", "T", "Time code", "Time code (T)", "Vouch.date",
|
|
"Project Activity", "Project Activity (T)", "Work package", "Work package (T)",
|
|
"Hours", "Workflow status", "Text"],
|
|
["202544", "", "A", "0", "Standard Time", "10/27/2025", "", "", "", "", "2.00 ", "N", ""],
|
|
["202544", "", "A", "0", "Standard Time", "10/27/2025", "", "", "", "", "4.00 ", "N", ""],
|
|
]
|
|
result = extract_work_time(rows)
|
|
assert "2025-10-27" in result
|
|
assert result["2025-10-27"] == 6.0
|
|
|
|
def test_extract_work_time_below_threshold(self):
|
|
"""Test that hours below threshold are filtered out."""
|
|
rows = [
|
|
["h1", "h2", "h3", "h4", "h5", "01/15/2024", "h7", "h8", "h9", "h10", "0.1"],
|
|
]
|
|
result = extract_work_time(rows)
|
|
assert len(result) == 0
|
|
|
|
def test_parse_xlsx_agresso_sheet(self):
|
|
"""Test XLSX parsing using the AGRESSO worksheet."""
|
|
workbook = Workbook()
|
|
workbook.active.title = "Other"
|
|
sheet = workbook.create_sheet("AGRESSO")
|
|
sheet.append(["Date", "Hours"])
|
|
sheet.append(["01/15/2024", "8.5"])
|
|
buffer = io.BytesIO()
|
|
workbook.save(buffer)
|
|
|
|
rows = parse_xlsx(buffer.getvalue())
|
|
assert rows[0] == ["Date", "Hours"]
|
|
assert rows[1][0] == "01/15/2024"
|
|
assert rows[1][1] == "8.5"
|
|
|
|
def test_parse_xlsx_missing_agresso(self):
|
|
"""Test XLSX parsing failure when AGRESSO worksheet is missing."""
|
|
workbook = Workbook()
|
|
workbook.active.title = "Other"
|
|
buffer = io.BytesIO()
|
|
workbook.save(buffer)
|
|
|
|
with pytest.raises(ValueError, match="Worksheet 'AGRESSO' not found"):
|
|
parse_xlsx(buffer.getvalue())
|
|
|
|
def test_create_time_event(self):
|
|
"""Test time event creation."""
|
|
event = create_time_event(
|
|
user_id=123,
|
|
project_id=456,
|
|
activity={"id": 789},
|
|
date="2024-01-15",
|
|
hours=8.5
|
|
)
|
|
assert event["user"]["id"] == 123
|
|
assert event["client-project"] == 456
|
|
assert event["activity"]["id"] == 789
|
|
assert event["date"] == "2024-01-15"
|
|
assert event["hours"] == 8.5
|
|
|
|
def test_get_user_by_name_success(self):
|
|
"""Test getting user by name successfully."""
|
|
users_data = {
|
|
"users": [
|
|
{"id": 1, "name": "Alice"},
|
|
{"id": 2, "name": "Bob"}
|
|
]
|
|
}
|
|
user_id = get_user_by_name(users_data, "Bob")
|
|
assert user_id == 2
|
|
|
|
def test_get_user_by_name_not_found(self):
|
|
"""Test getting user by name when user not found."""
|
|
users_data = {
|
|
"users": [
|
|
{"id": 1, "name": "Alice"}
|
|
]
|
|
}
|
|
with pytest.raises(ValueError, match="User 'Bob' not found"):
|
|
get_user_by_name(users_data, "Bob")
|
|
|
|
def test_get_user_by_name_invalid_data(self):
|
|
"""Test getting user by name with invalid data."""
|
|
with pytest.raises(ValueError, match="Invalid user data"):
|
|
get_user_by_name(None, "Bob")
|
|
|
|
|
|
class TestAPI:
|
|
"""Tests for FastAPI endpoints."""
|
|
|
|
def test_root_endpoint(self):
|
|
"""Test root endpoint."""
|
|
response = client.get("/")
|
|
assert response.status_code == 200
|
|
assert "name" in response.json()
|
|
assert response.json()["name"] == "ZynkTime API"
|
|
|
|
def test_health_check_without_api_key(self):
|
|
"""Test health check without API key configured."""
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
response = client.get("/health")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "unhealthy"
|
|
assert data["api_key_configured"] is False
|
|
|
|
def test_health_check_with_api_key(self):
|
|
"""Test health check with API key configured."""
|
|
with patch.dict(os.environ, {"KLEER_API_KEY": "test-key"}):
|
|
response = client.get("/health")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["status"] == "healthy"
|
|
assert data["api_key_configured"] is True
|
|
|
|
@patch('main.KleerAPI')
|
|
def test_list_projects_success(self, mock_kleer_api):
|
|
"""Test listing projects successfully."""
|
|
# Set up mock
|
|
mock_instance = Mock()
|
|
mock_instance.get_user_info.return_value = {
|
|
"users": [{"id": 1, "name": "Test User"}]
|
|
}
|
|
mock_instance.get_projects.return_value = {
|
|
"Project A": {"id": 1, "activity": {"id": 10}}
|
|
}
|
|
mock_kleer_api.return_value = mock_instance
|
|
|
|
with patch.dict(os.environ, {"KLEER_API_KEY": "test-key", "KLEER_USERNAME": "Test User"}):
|
|
response = client.get("/projects")
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert "Project A" in data
|
|
|
|
@patch('main.KleerAPI')
|
|
def test_list_events_success(self, mock_kleer_api):
|
|
"""Test listing events within date range."""
|
|
mock_instance = Mock()
|
|
mock_instance.get_user_info.return_value = {"users": [{"id": 1, "name": "Test User"}]}
|
|
mock_instance.get_events.return_value = {"event-readables": [{"id": {"id": 1}, "date": "2020-07-20"}]}
|
|
mock_kleer_api.return_value = mock_instance
|
|
|
|
with patch.dict(os.environ, {"KLEER_API_KEY": "test-key", "KLEER_USERNAME": "Test User"}):
|
|
response = client.get("/events", params={"start_date": "2020-07-20", "end_date": "2020-07-21"})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["user_id"] == 1
|
|
assert len(data["events"]) == 1
|
|
|
|
@patch('main.KleerAPI')
|
|
def test_approve_events_success(self, mock_kleer_api):
|
|
"""Test approving events."""
|
|
mock_instance = Mock()
|
|
mock_instance.get_user_info.return_value = {"users": [{"id": 1, "name": "Test User"}]}
|
|
mock_instance.approve_events.return_value = {"id": 3118}
|
|
mock_kleer_api.return_value = mock_instance
|
|
|
|
with patch.dict(os.environ, {"KLEER_API_KEY": "test-key", "KLEER_USERNAME": "Test User"}):
|
|
response = client.post("/approve-events", params={"start_date": "2020-07-20", "end_date": "2020-07-21"})
|
|
assert response.status_code == 200
|
|
data = response.json()
|
|
assert data["approval"] == {"id": 3118}
|
|
assert data["user_id"] == 1
|
|
|
|
|
|
class TestConfig:
|
|
"""Tests for configuration."""
|
|
|
|
def test_get_username_default(self):
|
|
"""Test getting default username."""
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
username = Config.get_username()
|
|
assert username == "Christopher Juhlin"
|
|
|
|
def test_get_username_from_env(self):
|
|
"""Test getting username from environment."""
|
|
with patch.dict(os.environ, {"KLEER_USERNAME": "Custom User"}):
|
|
username = Config.get_username()
|
|
assert username == "Custom User"
|
|
|
|
def test_validate_config_no_api_key(self):
|
|
"""Test config validation without API key."""
|
|
with patch.dict(os.environ, {}, clear=True):
|
|
with pytest.raises(ValueError, match="KLEER_API_KEY"):
|
|
Config.validate_config()
|
|
|
|
def test_validate_config_with_api_key(self):
|
|
"""Test config validation with API key."""
|
|
with patch.dict(os.environ, {"KLEER_API_KEY": "test-key"}):
|
|
Config.validate_config() # Should not raise
|
|
|
|
|
|
if __name__ == "__main__":
|
|
pytest.main([__file__, "-v"])
|