Initial commit - Mold Cost Calculator Application (Security Fixed)

2025-06-24 10:28:55 +08:00
commit 9b84cee5be
108 changed files with 15886 additions and 0 deletions
@@ -0,0 +1,82 @@
+import pytest
+from app import create_app, db
+from app.models import User
+from app.config import TestingConfig
+
+@pytest.fixture(scope='session')
+def app():
+    """Create and configure a Flask app for testing."""
+    app = create_app(TestingConfig)
+    
+    # Create the database and load test data
+    with app.app_context():
+        db.create_all()
+        yield app
+        db.session.remove()
+        db.drop_all()
+
+@pytest.fixture(scope='session')
+def client(app):
+    """A test client for the app."""
+    return app.test_client()
+
+@pytest.fixture(scope='session')
+def runner(app):
+    """A test CLI runner for the app."""
+    return app.test_cli_runner()
+
+@pytest.fixture(scope='function')
+def test_db(app):
+    """Create a fresh database for each test."""
+    with app.app_context():
+        db.create_all()
+        yield db
+        db.session.remove()
+        db.drop_all()
+
+@pytest.fixture
+def test_user(app):
+    """Create a test user."""
+    with app.app_context():
+        user = User.create(
+            username='testuser',
+            email='test@example.com',
+            company='Test Company',
+            password='Test123!'
+        )
+        db.session.commit()
+        return user
+
+@pytest.fixture
+def test_admin(app):
+    """Create a test admin user."""
+    with app.app_context():
+        admin = User.create(
+            username='admin',
+            email='admin@example.com',
+            company='Admin Company',
+            password='Admin123!'
+        )
+        admin.is_admin = True
+        db.session.commit()
+        return admin
+
+@pytest.fixture
+def auth_client(client, test_user):
+    """A test client that is authenticated."""
+    client.post('/login', data={
+        'email': 'test@example.com',
+        'password': 'Test123!',
+        'remember_me': False
+    })
+    return client
+
+@pytest.fixture
+def admin_client(client, test_admin):
+    """A test client that is authenticated as admin."""
+    client.post('/login', data={
+        'email': 'admin@example.com',
+        'password': 'Admin123!',
+        'remember_me': False
+    })
+    return client 
@@ -0,0 +1,184 @@
+import requests
+import json
+from datetime import datetime
+import time
+import re
+
+def get_csrf_token(html_content):
+    """从HTML内容中提取CSRF token"""
+    match = re.search(r'name="csrf_token" type="hidden" value="([^"]+)"', html_content)
+    if match:
+        return match.group(1)
+    return None
+
+def test_calculation(session, test_data, scenario_name):
+    """执行单个测试场景"""
+    print(f"\n测试场景: {scenario_name}")
+    print("输入数据:")
+    print(json.dumps(test_data, indent=2, ensure_ascii=False))
+    
+    try:
+        # 发送计算请求
+        calc_response = session.post('https://moldcost.jimmygan.com/calculate', data=test_data)
+        
+        # 打印响应状态和头信息
+        print(f"\n响应状态码: {calc_response.status_code}")
+        print("响应头:")
+        print(json.dumps(dict(calc_response.headers), indent=2, ensure_ascii=False))
+        
+        # 打印响应内容
+        print("\n响应内容:")
+        print(calc_response.text)
+        
+        # 尝试解析JSON
+        try:
+            result = calc_response.json()
+            print("\n解析后的JSON:")
+            print(json.dumps(result, indent=2, ensure_ascii=False))
+            return result
+        except json.JSONDecodeError as e:
+            print(f"\nJSON解析错误: {str(e)}")
+            return {'status': 'error', 'message': 'Invalid JSON response'}
+            
+    except requests.exceptions.RequestException as e:
+        print(f"\n请求错误: {str(e)}")
+        return {'status': 'error', 'message': str(e)}
+
+def run_tests():
+    # 基础URL
+    base_url = 'https://moldcost.jimmygan.com'
+    
+    # 创建会话以保持cookie
+    session = requests.Session()
+    session.verify = False  # 忽略SSL验证
+    
+    try:
+        # 1. 首先获取登录页面以获取CSRF token
+        print("\n获取登录页面...")
+        login_page = session.get(f'{base_url}/auth/login')
+        print(f"登录页面状态码: {login_page.status_code}")
+        
+        # 从页面内容获取CSRF token
+        csrf_token = get_csrf_token(login_page.text)
+        print(f"CSRF Token: {csrf_token}")
+        
+        if not csrf_token:
+            print("无法获取CSRF token")
+            return
+        
+        # 2. 登录
+        print("\n尝试登录...")
+        login_data = {
+            'email': 'admin@moldcost.com',
+            'password': 'XViQLV6XE92HeYb1r3Xk3Q',
+            'csrf_token': csrf_token,
+            'remember_me': False
+        }
+        
+        login_response = session.post(f'{base_url}/auth/login', data=login_data)
+        print(f"登录响应状态码: {login_response.status_code}")
+        print("登录响应头:")
+        print(json.dumps(dict(login_response.headers), indent=2, ensure_ascii=False))
+        
+        # 检查登录是否成功
+        if login_response.status_code != 200:
+            print("登录失败")
+            print("响应内容:")
+            print(login_response.text)
+            return
+        
+        # 等待一下确保登录完成
+        time.sleep(1)
+        
+        # 获取计算页面以获取新的CSRF token
+        calc_page = session.get(f'{base_url}/calculator')
+        csrf_token = get_csrf_token(calc_page.text)
+        
+        if not csrf_token:
+            print("无法获取计算页面的CSRF token")
+            return
+        
+        # 基础数据
+        base_data = {
+            'mold_shop_name': 'ACMI',
+            't1_factory_name': 'Test Factory 1',
+            't2_component_factory': 'Test Factory 2',
+            'model_name': 'Test Model',
+            'tooling_id': '12345',
+            'season': 'SS26',
+            'stage': 'CR0',
+            'country': 'China',
+            'issue_date': datetime.now().strftime('%Y-%m-%d'),
+            'csrf_token': csrf_token
+        }
+        
+        # 测试场景1：基础场景
+        test_data_1 = base_data.copy()
+        test_data_1.update({
+            'mold_main_type': 'Flat Rubber',
+            'mold_sub_type': 'Flat & Thin',
+            'complexity_high_sidewall': 'false',
+            'sliders_count': '0',
+            'cavity_count': '1',
+            'digital_texture_type': 'adidas Digital Texture Library'
+        })
+        
+        # 测试场景2：复杂场景
+        test_data_2 = base_data.copy()
+        test_data_2.update({
+            'mold_main_type': 'Cupsole',
+            'mold_sub_type': '3 Plates Mold',
+            'complexity_high_sidewall': 'true',
+            'sliders_count': '2',
+            'cavity_count': '2',
+            'digital_texture_type': 'Customized'
+        })
+        
+        # 测试场景3：边界场景
+        test_data_3 = base_data.copy()
+        test_data_3.update({
+            'mold_main_type': 'Rubber Sole',
+            'mold_sub_type': 'Visible PL',
+            'complexity_high_sidewall': 'true',
+            'sliders_count': '4',
+            'cavity_count': '4',
+            'digital_texture_type': 'Lasering'
+        })
+        
+        # 执行测试
+        results = []
+        results.append(test_calculation(session, test_data_1, "基础场景"))
+        time.sleep(1)  # 添加延迟
+        results.append(test_calculation(session, test_data_2, "复杂场景"))
+        time.sleep(1)  # 添加延迟
+        results.append(test_calculation(session, test_data_3, "边界场景"))
+        
+        # 分析结果
+        print("\n测试结果分析:")
+        for i, result in enumerate(results):
+            print(f"\n场景 {i+1} 分析:")
+            if result.get('status') == 'success':
+                calc = result.get('calculation', {})
+                print(f"基础价格: ${calc.get('net_base', 0)}")
+                print(f"模具加成: ${calc.get('mold_factor', 0)}")
+                print(f"复杂度加成: ${calc.get('complexity', 0)}")
+                print(f"总价: ${calc.get('total', 0)}")
+                
+                # 验证计算
+                expected_total = (calc.get('net_base', 0) + 
+                                calc.get('mold_factor', 0) + 
+                                calc.get('complexity', 0))
+                if abs(expected_total - calc.get('total', 0)) < 0.01:
+                    print("✓ 计算正确")
+                else:
+                    print("✗ 计算错误")
+                    print(f"预期总价: ${expected_total}")
+                    print(f"实际总价: ${calc.get('total', 0)}")
+            else:
+                print(f"测试失败: {result.get('message', 'Unknown error')}")
+                
+    except Exception as e:
+        print(f"\n测试过程中发生错误: {str(e)}")
+
+if __name__ == '__main__':
+    run_tests() 
@@ -0,0 +1,57 @@
+import os
+import pytest
+from app import create_app, db
+from app.config import Config
+
+class TestConfig(Config):
+    """Test configuration that overrides the base configuration."""
+    TESTING = True
+    DEBUG = True
+    SQLALCHEMY_DATABASE_URI = 'sqlite:///:memory:'
+    WTF_CSRF_ENABLED = False
+    LOG_LEVEL = 'DEBUG'
+    
+    # Override required settings for testing
+    SECRET_KEY = 'testing-secret-key-not-used-in-production'
+    MAIL_SUPPRESS_SEND = True
+    RATELIMIT_ENABLED = False
+    
+    # Testing-specific settings
+    RATELIMIT_DEFAULT = "1000 per day;100 per hour;20 per minute"
+    CACHE_TYPE = "SimpleCache"
+
+def get_test_config():
+    """Get test configuration."""
+    return TestConfig()
+
+@pytest.fixture(scope='session')
+def app():
+    """Create and configure a Flask app for testing."""
+    app = create_app('testing')
+    app.config.update(get_test_config())
+    
+    # Create the database and load test data
+    with app.app_context():
+        db.create_all()
+        yield app
+        db.session.remove()
+        db.drop_all()
+
+@pytest.fixture(scope='session')
+def client(app):
+    """A test client for the app."""
+    return app.test_client()
+
+@pytest.fixture(scope='session')
+def runner(app):
+    """A test CLI runner for the app."""
+    return app.test_cli_runner()
+
+@pytest.fixture(scope='function')
+def test_db(app):
+    """Create a fresh database for each test."""
+    with app.app_context():
+        db.create_all()
+        yield db
+        db.session.remove()
+        db.drop_all() 
@@ -0,0 +1,64 @@
+import pytest
+import os
+from app import create_app, db
+
+def test_production_config():
+    """Test production configuration settings."""
+    app = create_app('production')
+    
+    # Test essential production settings
+    assert not app.config['DEBUG']
+    assert app.config['TESTING'] is False
+    assert 'SECRET_KEY' in app.config
+    assert app.config['SQLALCHEMY_DATABASE_URI'] is not None
+    assert app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] is False
+
+def test_database_connection(app):
+    """Test database connection in production environment."""
+    with app.app_context():
+        try:
+            # Test database connection
+            db.engine.connect()
+            assert True
+        except Exception as e:
+            pytest.fail(f"Database connection failed: {str(e)}")
+
+def test_environment_variables():
+    """Test required environment variables are set."""
+    required_vars = [
+        'FLASK_APP',
+        'FLASK_ENV',
+        'SECRET_KEY',
+        'DATABASE_URL'
+    ]
+    
+    for var in required_vars:
+        assert os.getenv(var) is not None, f"Required environment variable {var} is not set"
+
+def test_gunicorn_config():
+    """Test Gunicorn configuration settings."""
+    import gunicorn_config
+    
+    assert hasattr(gunicorn_config, 'bind')
+    assert hasattr(gunicorn_config, 'workers')
+    assert hasattr(gunicorn_config, 'timeout')
+    assert gunicorn_config.forwarded_allow_ips != '*'
+
+def test_container_health_check(client):
+    """Test container health check endpoint."""
+    response = client.get('/health')
+    assert response.status_code == 200
+    assert response.json['status'] == 'healthy'
+
+def test_error_handling(client):
+    """Test error handling for common deployment issues."""
+    # Test database connection error
+    with pytest.raises(Exception):
+        client.get('/calculator')
+    
+    # Test rate limiting
+    for _ in range(100):
+        client.get('/')
+    
+    response = client.get('/')
+    assert response.status_code == 429  # Too Many Requests 
@@ -0,0 +1,65 @@
+import os
+os.environ["SECRET_KEY"] = os.getenv("TEST_SECRET_KEY", "test-secret-key-for-testing-only")
+
+from app import create_app, db
+from app.models import User, Quotation
+from datetime import datetime, date
+
+def test_user_quotation_relationship():
+    app = create_app()
+    with app.app_context():
+        # Create a test user
+        test_user = User(
+            username='testuser',
+            email='test@example.com',
+            password_hash='test_hash'
+        )
+        db.session.add(test_user)
+        db.session.commit()
+        
+        print(f"Created test user: {test_user}")
+        
+        # Create a test quotation
+        test_quotation = Quotation(
+            user_id=test_user.id,
+            mold_shop_name='Test Shop',
+            model_name='Test Model',
+            tooling_id='12345',
+            season='Spring',
+            stage='Development',
+            country='USA',
+            issue_date=date.today(),
+            approver='Test Approver',
+            approval_date=date.today(),
+            mold_main_type='Injection',
+            mold_sub_type='Standard',
+            sliders_count=2,
+            cavity_count=1,
+            digital_texture_type='Basic',
+            base_price=1000.0,
+            total_cost=1100.0
+        )
+        db.session.add(test_quotation)
+        db.session.commit()
+        
+        print(f"Created test quotation: {test_quotation}")
+        
+        # Test the relationship from User to Quotation
+        user_quotations = test_user.quotations
+        print(f"\nUser's quotations: {user_quotations}")
+        print(f"Number of quotations: {len(user_quotations)}")
+        
+        # Test the relationship from Quotation to User
+        quotation_user = test_quotation.user
+        print(f"\nQuotation's user: {quotation_user}")
+        print(f"User ID matches: {quotation_user.id == test_user.id}")
+        
+        # Clean up
+        db.session.delete(test_quotation)
+        db.session.delete(test_user)
+        db.session.commit()
+        
+        print("\nTest completed successfully!")
+
+if __name__ == '__main__':
+    test_user_quotation_relationship() 
@@ -0,0 +1,116 @@
+import pytest
+from flask import url_for
+from app.models import User, Quotation
+from datetime import datetime, timezone
+import os
+
+def test_home_page(client):
+    response = client.get('/')
+    assert response.status_code == 200
+
+def test_login_page(client):
+    response = client.get('/login')
+    assert response.status_code == 200
+    assert b'Login' in response.data
+
+def test_register_page(client):
+    response = client.get('/register')
+    assert response.status_code == 200
+    assert b'Register' in response.data
+
+def test_calculator_page_redirect(client):
+    response = client.get('/calculator')
+    assert response.status_code == 302
+    assert '/login' in response.headers['Location']
+
+def test_login_success(client, app):
+    with app.app_context():
+        test_password = os.getenv('TEST_USER_PASSWORD', 'Test123!')
+        user = User.create(
+            username='testuser',
+            email='test@example.com',
+            company='Test Company',
+            password=test_password
+        )
+        db.session.commit()
+        
+        response = client.post('/login', data={
+            'email': 'test@example.com',
+            'password': test_password,
+            'remember_me': False
+        }, follow_redirects=True)
+        
+        assert response.status_code == 200
+        assert b'Calculator' in response.data
+
+def test_login_failure(client):
+    response = client.post('/login', data={
+        'email': 'wrong@example.com',
+        'password': 'WrongPass123!',
+        'remember_me': False
+    }, follow_redirects=True)
+    
+    assert response.status_code == 200
+    assert b'Invalid email or password' in response.data
+
+def test_register_success(client, app):
+    response = client.post('/register', data={
+        'username': 'newuser',
+        'email': 'jenny.tay@adidas.com',
+        'company': 'New Company',
+        'password': 'NewPass123!',
+        'password_confirm': 'NewPass123!',
+        'terms': True
+    }, follow_redirects=True)
+    
+    assert response.status_code == 200
+    assert b'Registration successful' in response.data
+
+def test_register_duplicate_email(client, app):
+    with app.app_context():
+        user = User.create(
+            username='testuser',
+            email='jenny.tay@adidas.com',
+            company='Test Company',
+            password='Test123!'
+        )
+        db.session.commit()
+        
+        response = client.post('/register', data={
+            'username': 'newuser',
+            'email': 'jenny.tay@adidas.com',
+            'company': 'New Company',
+            'password': 'NewPass123!',
+            'password_confirm': 'NewPass123!',
+            'terms': True
+        }, follow_redirects=True)
+        
+        assert response.status_code == 200
+        assert b'Email already registered' in response.data
+
+def test_admin_management_unauthorized(client):
+    response = client.get('/admin/management')
+    assert response.status_code == 302
+    assert '/login' in response.headers['Location']
+
+def test_admin_management_authorized(client, app):
+    with app.app_context():
+        admin = User.create(
+            username='admin',
+            email='admin@example.com',
+            company='Admin Company',
+            password='Admin123!'
+        )
+        admin.is_admin = True
+        db.session.commit()
+        
+        client.post('/login', data={
+            'email': 'admin@example.com',
+            'password': 'Admin123!',
+            'remember_me': False
+        })
+        
+        response = client.get('/admin/management')
+        assert response.status_code == 200
+        assert b'User Management' in response.data
+