work-statistics/backend/tests/test_export.py

"""Tests for Export API endpoints."""
import pytest
from datetime import date
from io import BytesIO
from openpyxl import load_workbook
from flask_bcrypt import Bcrypt
from app import create_app, db
from app.models.person import Person
from app.models.item import Item
from app.models.work_record import WorkRecord
from app.models.admin import Admin
from app.services.auth_service import AuthService

bcrypt = Bcrypt()


@pytest.fixture
def app():
    """Create application for testing."""
    app = create_app('testing')
    with app.app_context():
        db.create_all()
        yield app
        db.session.remove()
        db.drop_all()


@pytest.fixture
def client(app):
    """Create test client."""
    return app.test_client()


@pytest.fixture
def auth_headers(app):
    """Create auth headers for testing."""
    with app.app_context():
        password_hash = bcrypt.generate_password_hash('testpassword').decode('utf-8')
        admin = Admin(username='exporttestadmin', password_hash=password_hash)
        db.session.add(admin)
        db.session.commit()
        token = AuthService.generate_token(admin)
        return {'Authorization': f'Bearer {token}'}


@pytest.fixture
def sample_data(app, auth_headers):
    """Create sample data for export tests."""
    with app.app_context():
        # Create persons
        person1 = Person(name='张三')
        person2 = Person(name='李四')
        db.session.add_all([person1, person2])
        db.session.commit()
        
        # Create items
        item1 = Item(name='物品A', unit_price=10.50)
        item2 = Item(name='物品B', unit_price=20.75)
        db.session.add_all([item1, item2])
        db.session.commit()
        
        # Create work records for January 2024
        records = [
            WorkRecord(person_id=person1.id, item_id=item1.id, work_date=date(2024, 1, 5), quantity=5),
            WorkRecord(person_id=person1.id, item_id=item2.id, work_date=date(2024, 1, 10), quantity=3),
            WorkRecord(person_id=person2.id, item_id=item1.id, work_date=date(2024, 1, 15), quantity=8),
            # February 2024
            WorkRecord(person_id=person1.id, item_id=item1.id, work_date=date(2024, 2, 5), quantity=4),
            WorkRecord(person_id=person2.id, item_id=item2.id, work_date=date(2024, 2, 10), quantity=6),
        ]
        db.session.add_all(records)
        db.session.commit()
        
        return {
            'person1_id': person1.id,
            'person2_id': person2.id,
            'item1_id': item1.id,
            'item2_id': item2.id
        }


class TestExportAPI:
    """Test cases for Export API endpoints."""
    
    def test_monthly_export_success(self, client, sample_data, auth_headers):
        """Test successful monthly export."""
        response = client.get('/api/export/monthly?year=2024&month=1', headers=auth_headers)
        
        assert response.status_code == 200
        assert response.content_type == 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
        
        # Verify Excel content
        wb = load_workbook(BytesIO(response.data))
        assert len(wb.sheetnames) == 2
        assert '2024年1月明细' in wb.sheetnames
        assert '月度汇总' in wb.sheetnames

    def test_monthly_export_detail_sheet(self, client, sample_data, auth_headers):
        """Test monthly export detail sheet content."""
        response = client.get('/api/export/monthly?year=2024&month=1', headers=auth_headers)
        
        wb = load_workbook(BytesIO(response.data))
        detail_sheet = wb['2024年1月明细']
        
        # Check headers (now includes supplier and settlement status columns)
        headers = [cell.value for cell in detail_sheet[1]]
        assert headers == ['人员', '日期', '供应商', '物品', '单价', '数量', '总价', '结算状态']
        
        # Check data rows (3 records in January)
        data_rows = list(detail_sheet.iter_rows(min_row=2, values_only=True))
        assert len(data_rows) == 3
        
        # Check settlement status column values
        for row in data_rows:
            assert row[7] in ['已结算', '未结算']
    
    def test_monthly_export_summary_sheet(self, client, sample_data, auth_headers):
        """Test monthly export summary sheet content."""
        response = client.get('/api/export/monthly?year=2024&month=1', headers=auth_headers)
        
        wb = load_workbook(BytesIO(response.data))
        summary_sheet = wb['月度汇总']
        
        # Check headers (now includes supplier and settlement status columns)
        headers = [cell.value for cell in summary_sheet[1]]
        assert headers == ['人员', '供应商', '总金额', '结算状态']
        
        # Check that summary has person+supplier+settlement rows plus total row
        data_rows = list(summary_sheet.iter_rows(min_row=2, values_only=True))
        assert len(data_rows) >= 2  # At least 2 persons
        
        # Last row should be total
        last_row = data_rows[-1]
        assert last_row[0] == '合计'
        
        # Check settlement status column values (except total row)
        for row in data_rows[:-1]:
            assert row[3] in ['已结算', '未结算']
    
    def test_monthly_export_missing_year(self, client, sample_data, auth_headers):
        """Test monthly export with missing year parameter."""
        response = client.get('/api/export/monthly?month=1', headers=auth_headers)
        
        assert response.status_code == 400
        data = response.get_json()
        assert data['success'] is False
        assert 'Year' in data['error']
    
    def test_monthly_export_missing_month(self, client, sample_data, auth_headers):
        """Test monthly export with missing month parameter."""
        response = client.get('/api/export/monthly?year=2024', headers=auth_headers)
        
        assert response.status_code == 400
        data = response.get_json()
        assert data['success'] is False
        assert 'Month' in data['error']
    
    def test_monthly_export_invalid_month(self, client, sample_data, auth_headers):
        """Test monthly export with invalid month."""
        response = client.get('/api/export/monthly?year=2024&month=13', headers=auth_headers)
        
        assert response.status_code == 400
        data = response.get_json()
        assert data['success'] is False
    
    def test_monthly_export_empty_month(self, client, sample_data, auth_headers):
        """Test monthly export for month with no records."""
        response = client.get('/api/export/monthly?year=2024&month=12', headers=auth_headers)
        
        assert response.status_code == 200
        
        wb = load_workbook(BytesIO(response.data))
        detail_sheet = wb.active
        
        # Should have headers but no data rows
        data_rows = list(detail_sheet.iter_rows(min_row=2, values_only=True))
        assert len(data_rows) == 0


    def test_yearly_export_success(self, client, sample_data, auth_headers):
        """Test successful yearly export."""
        response = client.get('/api/export/yearly?year=2024', headers=auth_headers)
        
        assert response.status_code == 200
        assert response.content_type == 'application/vnd.openxmlformats-officedocument.spreadsheetml.sheet'
        
        # Verify Excel content
        wb = load_workbook(BytesIO(response.data))
        assert len(wb.sheetnames) == 2
        assert '2024年明细' in wb.sheetnames
        assert '年度汇总' in wb.sheetnames
    
    def test_yearly_export_detail_sheet(self, client, sample_data, auth_headers):
        """Test yearly export detail sheet content."""
        response = client.get('/api/export/yearly?year=2024', headers=auth_headers)
        
        wb = load_workbook(BytesIO(response.data))
        detail_sheet = wb['2024年明细']
        
        # Check headers (now includes supplier and settlement status columns)
        headers = [cell.value for cell in detail_sheet[1]]
        assert headers == ['人员', '日期', '供应商', '物品', '单价', '数量', '总价', '结算状态']
        
        # Check data rows (5 records total in 2024)
        data_rows = list(detail_sheet.iter_rows(min_row=2, values_only=True))
        assert len(data_rows) == 5
        
        # Check settlement status column values
        for row in data_rows:
            assert row[7] in ['已结算', '未结算']
    
    def test_yearly_export_summary_sheet(self, client, sample_data, auth_headers):
        """Test yearly export summary sheet with monthly breakdown."""
        response = client.get('/api/export/yearly?year=2024', headers=auth_headers)
        
        wb = load_workbook(BytesIO(response.data))
        summary_sheet = wb['年度汇总']
        
        # Check headers: 人员, 1月-12月, 年度合计, 已结算, 未结算
        headers = [cell.value for cell in summary_sheet[1]]
        assert headers[0] == '人员'
        assert headers[1] == '1月'
        assert headers[12] == '12月'
        assert headers[13] == '年度合计'
        assert headers[14] == '已结算'
        assert headers[15] == '未结算'
        
        # Check that summary has person rows plus total row
        data_rows = list(summary_sheet.iter_rows(min_row=2, values_only=True))
        assert len(data_rows) >= 2  # At least 2 persons
        
        # Last row should be total
        last_row = data_rows[-1]
        assert last_row[0] == '合计'
        
        # Check that settled + unsettled = yearly total for each person row
        for row in data_rows:
            yearly_total = row[13]
            settled = row[14]
            unsettled = row[15]
            if yearly_total is not None and settled is not None and unsettled is not None:
                assert abs(yearly_total - (settled + unsettled)) < 0.01
    
    def test_yearly_export_missing_year(self, client, sample_data, auth_headers):
        """Test yearly export with missing year parameter."""
        response = client.get('/api/export/yearly', headers=auth_headers)
        
        assert response.status_code == 400
        data = response.get_json()
        assert data['success'] is False
        assert 'Year' in data['error']
    
    def test_yearly_export_empty_year(self, client, sample_data, auth_headers):
        """Test yearly export for year with no records."""
        response = client.get('/api/export/yearly?year=2020', headers=auth_headers)
        
        assert response.status_code == 200
        
        wb = load_workbook(BytesIO(response.data))
        detail_sheet = wb.active
        
        # Should have headers but no data rows
        data_rows = list(detail_sheet.iter_rows(min_row=2, values_only=True))
        assert len(data_rows) == 0
    
    def test_export_total_price_calculation(self, client, sample_data, auth_headers):
        """Test that total_price is correctly calculated in export."""
        response = client.get('/api/export/monthly?year=2024&month=1', headers=auth_headers)
        
        wb = load_workbook(BytesIO(response.data))
        detail_sheet = wb['2024年1月明细']
        
        # Check each row's total_price = unit_price * quantity
        # Column indices: 0=人员, 1=日期, 2=供应商, 3=物品, 4=单价, 5=数量, 6=总价
        for row in detail_sheet.iter_rows(min_row=2, values_only=True):
            if row[0] is not None:  # Skip empty rows
                unit_price = row[4]
                quantity = row[5]
                total_price = row[6]
                assert abs(total_price - (unit_price * quantity)) < 0.01
    
    def test_unauthorized_access(self, client, app):
        """Test that export endpoints require authentication."""
        with app.app_context():
            db.create_all()
        response = client.get('/api/export/monthly?year=2024&month=1')
        assert response.status_code == 401
        data = response.get_json()
        assert data['success'] is False
        assert data['code'] == 'UNAUTHORIZED'