File size: 2,060 Bytes
1ac9f32
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
import pytest
from app import app, GLOBAL_DATA_STORE
import pandas as pd
from unittest.mock import patch

@pytest.fixture
def client():
    app.config['TESTING'] = True
    with app.test_client() as client:
        yield client

def test_clear_session_route(client):
    # Setup session and global data store
    with client.session_transaction() as sess:
        sess['data_id'] = 'test_session_id'

    GLOBAL_DATA_STORE['test_session_id'] = {'dummy': 'data'}

    # Call clear route
    response = client.get('/clear', follow_redirects=True)

    # Verify data is removed and session is cleared
    assert 'test_session_id' not in GLOBAL_DATA_STORE
    with client.session_transaction() as sess:
        assert 'data_id' not in sess

    # Verify redirect
    assert response.status_code == 200

def test_message_limit_enforcement(client):
    # Mock dependencies to reach the limit check
    with patch('app.process_file') as mock_parse:
        # Create a DataFrame with > 50,000 messages
        large_df = pd.DataFrame({
            'timestamp': [pd.Timestamp.now()] * 50001,
            'sender': ['ME'] * 50001,
            'text': ['hello'] * 50001
        })
        mock_parse.return_value = large_df

        # Create dummy file for the request
        import io
        data = {
            'my_name': 'Rahul',
            'partner_name': 'Priya',
            'chat_files': (io.BytesIO(b"dummy content"), 'test.txt')
        }

        response = client.post('/process', data=data)

        assert response.status_code == 400
        assert b"Too many messages. Maximum 50,000 allowed for analysis." in response.data

def test_csp_hardening(client):
    response = client.get('/')
    csp = response.headers.get('Content-Security-Policy', '')
    assert "object-src 'none'" in csp

def test_thread_limit_config():
    # We can't easily test the execution count of ThreadPoolExecutor without complex mocking,
    # but we can verify the code by reading the file (already done).
    # This is a placeholder for architectural verification.
    pass