mcp/tests/test_request_context.py

import unittest

from services.request_context import RequestContextParser


class TestRequestContextParser(unittest.TestCase):
    def setUp(self):
        self.parser = RequestContextParser()

    def test_parse_from_header(self):
        headers = {'X-Gateway-Session': 'GWS_abc123'}
        context = self.parser.parse(headers)

        self.assertTrue(context.has_session())
        self.assertEqual(context.gateway_session_id, 'GWS_abc123')
        self.assertEqual(context.source, 'header')

    def test_parse_from_authorization_bearer(self):
        headers = {'Authorization': 'Bearer GWS_xyz789'}
        context = self.parser.parse(headers)

        self.assertTrue(context.has_session())
        self.assertEqual(context.gateway_session_id, 'GWS_xyz789')
        self.assertEqual(context.source, 'authorization')

    def test_parse_from_cookie(self):
        headers = {'Cookie': 'gateway_session_id=GWS_cookie123; other=value'}
        context = self.parser.parse(headers)

        self.assertTrue(context.has_session())
        self.assertEqual(context.gateway_session_id, 'GWS_cookie123')
        self.assertEqual(context.source, 'cookie')

    def test_parse_case_insensitive_headers(self):
        # Test with different case
        headers = {'x-gateway-session': 'GWS_test', 'AUTHORIZATION': 'Bearer GWS_test2'}
        context = self.parser.parse(headers)

        # Should prioritize header over authorization
        self.assertEqual(context.gateway_session_id, 'GWS_test')
        self.assertEqual(context.source, 'header')

    def test_parse_priority_header_over_authorization(self):
        headers = {
            'X-Gateway-Session': 'GWS_from_header',
            'Authorization': 'Bearer GWS_from_auth'
        }
        context = self.parser.parse(headers)

        self.assertEqual(context.gateway_session_id, 'GWS_from_header')
        self.assertEqual(context.source, 'header')

    def test_parse_priority_authorization_over_cookie(self):
        headers = {
            'Authorization': 'Bearer GWS_from_auth',
            'Cookie': 'gateway_session_id=GWS_from_cookie'
        }
        context = self.parser.parse(headers)

        self.assertEqual(context.gateway_session_id, 'GWS_from_auth')
        self.assertEqual(context.source, 'authorization')

    def test_parse_no_session(self):
        headers = {'Content-Type': 'application/json'}
        context = self.parser.parse(headers)

        self.assertFalse(context.has_session())
        self.assertEqual(context.gateway_session_id, '')
        self.assertEqual(context.source, '')

    def test_parse_invalid_session_id_format(self):
        # Session ID doesn't start with GWS_
        headers = {'X-Gateway-Session': 'INVALID_123'}
        context = self.parser.parse(headers)

        self.assertFalse(context.has_session())

    def test_parse_authorization_without_bearer_prefix(self):
        headers = {'Authorization': 'Basic xyz123'}
        context = self.parser.parse(headers)

        self.assertFalse(context.has_session())

    def test_parse_authorization_with_non_gws_token(self):
        headers = {'Authorization': 'Bearer MT_regular_token'}
        context = self.parser.parse(headers)

        self.assertFalse(context.has_session())

    def test_parse_empty_headers(self):
        context = self.parser.parse({})

        self.assertFalse(context.has_session())

    def test_parse_none_headers(self):
        context = self.parser.parse(None)

        self.assertFalse(context.has_session())

    def test_parse_cookie_with_multiple_values(self):
        headers = {'Cookie': 'session=abc; gateway_session_id=GWS_multi; lang=en'}
        context = self.parser.parse(headers)

        self.assertTrue(context.has_session())
        self.assertEqual(context.gateway_session_id, 'GWS_multi')

    def test_parse_whitespace_handling(self):
        headers = {'Authorization': '  Bearer   GWS_spaces  '}
        context = self.parser.parse(headers)

        self.assertTrue(context.has_session())
        self.assertEqual(context.gateway_session_id, 'GWS_spaces')

    def test_parse_cookie_with_invalid_session_id_format(self):
        """Test cookie with non-GWS_ prefixed session ID"""
        headers = {'Cookie': 'gateway_session_id=INVALID_123; other=value'}
        context = self.parser.parse(headers)

        self.assertFalse(context.has_session())
        self.assertEqual(context.gateway_session_id, '')


if __name__ == '__main__':
    unittest.main()