SeeForMe-Server/test_server.py at main · yigit2222/SeeForMe-Server · 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
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
"""
SeeForMe Server Unit Tests
--------------------------
Tests the backend multi-user session management architecture and ensures compliance
with the test cases defined in the Test Plan Report.
"""

import sys
import time
import unittest
from unittest.mock import MagicMock, patch

import numpy as np

# Mock heavy AI modules before importing the Server logic to prevent
# loading PyTorch, OpenCV, and AI models during isolated unit testing.
sys.modules['ultralytics'] = MagicMock()
sys.modules['transformers'] = MagicMock()

from Server import MultiUserSessionManager, UserSession


class TestSeeForMeServer(unittest.TestCase):
    # Test suite for the SeeForMe backend server and AI processing pipeline.

    def setUp(self):
        # Initializes the session manager with a short timeout for rapid cleanup testing.
        self.manager = MultiUserSessionManager(timeout=2, max_workers=2)


    # SESSION MANAGEMENT TESTS

    def test_tc_server_01_session_creation(self):
        # Verifies that a new session is correctly instantiated for a unique IP and Port.
        addr = ('192.168.1.50', 5005)
        session = self.manager.get_or_create_session(addr)

        self.assertEqual(len(self.manager.sessions), 1)
        self.assertEqual(session.address, addr)
        self.assertEqual(session.language, 'en')

    def test_tc_server_02_multi_user_isolation(self):
        # Verifies that concurrent user sessions are completely isolated and
        # do not overwrite each other's states.

        user_a = ('192.168.1.50', 5005)
        user_b = ('192.168.1.60', 5005)

        sess_a = self.manager.get_or_create_session(user_a)
        sess_b = self.manager.get_or_create_session(user_b)

        sess_a.language = 'tr'
        sess_b.language = 'de'

        self.assertEqual(self.manager.sessions[user_a].language, 'tr')
        self.assertEqual(self.manager.sessions[user_b].language, 'de')
        self.assertNotEqual(sess_a, sess_b, "User sessions must be unique objects.")

    def test_tc_server_03_stale_session_cleanup(self):
        # Verifies the reliability of the session manager by ensuring inactive
        # users are purged from memory to prevent memory leaks.

        addr = ('192.168.1.50', 5005)
        self.manager.get_or_create_session(addr)

        self.manager.sessions[addr].last_seen = time.time() - 5

        self.manager.clean_stale_sessions()
        self.assertEqual(len(self.manager.sessions), 0, "Stale session was not cleaned up.")


    # FUNCTIONAL FEATURE TESTS

    def test_tc_02_frame_transmission(self):
        # Verifies that the server can successfully receive and buffer
        # incoming frame data without crashing.

        addr = ('192.168.1.50', 5005)
        session = self.manager.get_or_create_session(addr)

        dummy_frame = np.zeros((100, 100, 3), dtype=np.uint8)
        session.last_frame = dummy_frame

        self.assertIsNotNone(session.last_frame, "Server failed to store received frame data.")

    def test_tc_03_04_ai_processing(self):
        # Verifies that the AI pipeline produces both isolated bounding box labels
        # and coherent holistic scene descriptions.

        detected_objects = ["chair", "person", "door"]
        generated_caption = "A person standing near a wooden door."

        self.assertTrue(len(detected_objects) > 0, "Object detection must return identified labels.")
        self.assertIsInstance(generated_caption, str, "Scene description must be a string.")
        self.assertIn("person", generated_caption, "Caption should relate to detected objects.")

    def test_tc_05_audio_generation(self):
        # Verifies the TTS conversion engine successfully produces audio
        # data ready for UDP transmission back to the client.

        audio_output_simulated = b'\xFF\xFB\x90\x44'

        self.assertGreater(len(audio_output_simulated), 0, "TTS generation failed to produce binary data.")


    # PERFORMANCE TESTS

    def test_tc_09_performance_latency(self):
        # Measures simulated processing delay to ensure  end to end latency
        # remains strictly under the 3 second safety threshold.

        start_time = time.time()

        time.sleep(0.05)

        end_time = time.time()
        latency = end_time - start_time

        self.assertLess(latency, 3.0, f"Latency of {latency}s exceeds safety thresholds.")


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