aitutor_demo/test_event_processor_unit.py at main · codeprimate123/aitutor_demo · GitHub

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
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
#!/usr/bin/env python
"""
Test event processor without database dependency.
This file can be run directly: python test_event_processor_unit.py
"""

import sys
from unittest.mock import MagicMock

sys.path.insert(0, '/home/aram/dev/tutor_demo')

from app.services.event_processor import EventProcessor


class TestSchemaCaching:
    """Test schema caching functionality"""

    def test_schema_is_cached_after_first_load(self):
        """Test that schema is cached and only loaded once per path"""
        processor = EventProcessor()

        schema_path = "data/procedures/pipette_verification.json"

        mock_load = MagicMock(return_value={"id": "test", "steps": []})
        processor.load_schema_func = mock_load

        processor._get_schema(schema_path)
        processor._get_schema(schema_path)
        processor._get_schema(schema_path)

        assert mock_load.call_count == 1, f"Expected 1 call, got {mock_load.call_count}"
        print("✓ test_schema_is_cached_after_first_load passed")

    def test_different_schema_paths_load_independently(self):
        """Test that different schema paths load independently"""
        processor = EventProcessor()

        mock_load = MagicMock(side_effect=[
            {"id": "schema1", "steps": []},
            {"id": "schema2", "steps": []},
        ])
        processor.load_schema_func = mock_load

        processor._get_schema("path1.json")
        processor._get_schema("path2.json")
        processor._get_schema("path1.json")
        processor._get_schema("path2.json")

        assert mock_load.call_count == 2, f"Expected 2 calls, got {mock_load.call_count}"
        print("✓ test_different_schema_paths_load_independently passed")

    def test_cache_persists_for_processor_lifetime(self):
        """Test that cache persists across multiple operations"""
        processor = EventProcessor()

        mock_load = MagicMock(return_value={"id": "test", "steps": []})
        processor.load_schema_func = mock_load

        processor._get_schema("test.json")

        assert "test.json" in processor._schema_cache
        assert processor._schema_cache["test.json"] == {"id": "test", "steps": []}
        print("✓ test_cache_persists_for_processor_lifetime passed")


class TestEventProcessing:
    """Test event processing through evaluation pipeline"""

    def test_process_unknown_event_type_returns_none(self):
        """Test that unknown event types return None"""
        processor = EventProcessor()

        mock_db_session = MagicMock()
        mock_session = MagicMock()

        event_data = {"type": "unknown_type"}
        result = processor.process_event(mock_db_session, mock_session, event_data, "schema.json")

        assert result is None
        print("✓ test_process_unknown_event_type_returns_none passed")


if __name__ == "__main__":
    test_suite = TestSchemaCaching()
    test_suite.test_schema_is_cached_after_first_load()
    test_suite.test_different_schema_paths_load_independently()
    test_suite.test_cache_persists_for_processor_lifetime()

    test_suite2 = TestEventProcessing()
    test_suite2.test_process_unknown_event_type_returns_none()

    print("\n✅ All event processor tests passed!")