backend.tests.unit.components.workers.test_worker_service_manual

  1import pytest
  2import os
  3import io
  4import numpy as np
  5import face_recognition
  6from unittest.mock import patch, MagicMock
  7from datetime import datetime
  8from backend.app import create_app, db
  9from backend.database.models import Worker
 10from backend.components.workers import workerService
 11
 12
 13# ============================================================================
 14# Test Setup & Fixtures
 15# ============================================================================
 16
 17@pytest.fixture
 18def app_context():
 19    """
 20    Creates an isolated application context with an in-memory SQLite database.
 21    Ensures a clean state for every test function using this fixture.
 22    """
 23    app = create_app()
 24    app.config.update({
 25        "TESTING": True,
 26        "SQLALCHEMY_DATABASE_URI": "sqlite:///:memory:"
 27    })
 28
 29    with app.app_context():
 30        db.create_all()
 31        yield app
 32        db.session.remove()
 33        db.drop_all()
 34
 35
 36# ============================================================================
 37# Test: Embedding Generation (Integration)
 38# ============================================================================
 39
 40def test_create_worker_embedding_from_file():
 41    """
 42    Integration test for face embedding generation using a real image file.
 43
 44    Verifies that:
 45    1. The test image file exists.
 46    2. The service converts the image to a binary BLOB (bytes).
 47    3. The BLOB can be deserialized back into a valid NumPy array of shape (128,).
 48    """
 49    current_dir = os.path.dirname(os.path.abspath(__file__))
 50    # Calculate path: backend/tests/unit/components/workers -> backend/tests/assets
 51    image_path = os.path.normpath(os.path.join(current_dir, "../../../assets/testimg.jpg"))
 52
 53    print(f"Looking for test image at: {image_path}")
 54    if not os.path.exists(image_path):
 55        pytest.skip(f"Skipped: Test image not found at {image_path}")
 56
 57    image_input = face_recognition.load_image_file(image_path)
 58
 59    # Execute service function
 60    result_blob = workerService.create_worker_embedding(image_input)
 61
 62    # Basic assertions
 63    assert isinstance(result_blob, bytes)
 64    assert len(result_blob) > 0
 65
 66    # Verify that the blob is a valid serialized NumPy array
 67    buffer = io.BytesIO(result_blob)
 68    embeddings = np.load(buffer, allow_pickle=True)
 69
 70    assert len(embeddings) > 0, "No faces detected in the test image."
 71    # face_recognition library generates 128-dimensional vectors
 72    assert len(embeddings[0]) == 128
 73
 74
 75# ============================================================================
 76# Test: Worker CRUD Operations
 77# ============================================================================
 78
 79@patch('backend.components.workers.workerService.create_worker_embedding')
 80def test_create_worker_adds_to_db(mock_create_embedding, app_context):
 81    """
 82    Tests if the `create_worker` function correctly persists a new worker to the database.
 83
 84    We mock the embedding generation to avoid the overhead of the ML library
 85    and focus on database interactions.
 86    """
 87    mock_create_embedding.return_value = b'test_blob_data'
 88    name = "John Doe"
 89    fake_image = MagicMock()  # Mock the image object
 90    expiration = datetime(2025, 12, 31)
 91
 92    with app_context.app_context():
 93        # Action: Create worker via service
 94        worker = workerService.create_worker(name, fake_image, expiration)
 95
 96        assert worker.id is not None
 97        assert worker.name == name
 98
 99        # Verification: Fetch from DB to ensure persistence
100        fetched_worker = db.session.get(Worker, worker.id)
101        assert fetched_worker is not None
102        assert fetched_worker.name == "John Doe"
103        assert fetched_worker.face_embedding == b'test_blob_data'
104
105
106def test_extend_worker_expiration_success(app_context):
107    """
108    Tests the `extend_worker_expiration` functionality.
109
110    Verifies that the expiration date is updated in the database object.
111    """
112    initial_date = datetime(2024, 1, 1)
113    new_date = datetime(2030, 1, 1)
114
115    with app_context.app_context():
116        # 1. Setup: Create a worker with an initial date
117        worker = Worker(
118            name="Mark ToChange",
119            face_embedding=b'dummy_data',
120            expiration_date=initial_date,
121            secret="dummy_secret"
122        )
123        db.session.add(worker)
124        db.session.commit()
125
126        # Retrieve the object (service requires an object, not ID)
127        worker_to_update = db.session.get(Worker, worker.id)
128
129        # 2. Action: Extend expiration
130        updated_worker = workerService.extend_worker_expiration(worker_to_update, new_date)
131
132        # 3. Verification
133        assert updated_worker.expiration_date == new_date
134
135        # Force session refresh to check DB state
136        db.session.expire_all()
137        fetched_worker = db.session.get(Worker, worker.id)
138        assert fetched_worker.expiration_date == new_date
139
140
141def test_update_worker_name_success(app_context):
142    """
143    Tests the `update_worker_name` functionality.
144
145    Verifies that the worker's name is successfully updated in the database.
146    """
147    initial_name = "Jane Doe"
148    new_name = "Jane Smith"
149
150    with app_context.app_context():
151        # Setup
152        worker = Worker(
153            name=initial_name,
154            face_embedding=b'dummy_data',
155            expiration_date=datetime(2025, 12, 12),
156            secret="dummy_secret"
157        )
158        db.session.add(worker)
159        db.session.commit()
160
161        # Action
162        worker_to_update = db.session.get(Worker, worker.id)
163        workerService.update_worker_name(worker_to_update, new_name)
164
165        # Verification
166        db.session.expire_all()
167        fetched_worker = db.session.get(Worker, worker.id)
168        assert fetched_worker.name == new_name
169
170
171def test_update_worker_face_embedding_success(app_context):
172    """
173    Tests the `update_worker_face_image` functionality.
174
175    Verifies that providing a new image triggers the embedding recalculation
176    and updates the database field.
177    """
178    initial_name = "Test Face Update"
179    initial_face_data = b'old_dummy_embedding'
180    new_raw_image = b'new_raw_image_data_jpg'
181    mocked_embedding_result = b'new_calculated_embedding_123'
182
183    with app_context.app_context():
184        # Setup
185        worker = Worker(
186            name=initial_name,
187            face_embedding=initial_face_data,
188            expiration_date=datetime(2025, 12, 12),
189            secret="dummy_secret"
190        )
191        db.session.add(worker)
192        db.session.commit()
193
194        # Mock the internal embedding function to isolate logic
195        with patch('backend.components.workers.workerService.create_worker_embedding') as mock_embedding:
196            mock_embedding.return_value = mocked_embedding_result
197
198            # Action
199            worker_to_update = db.session.get(Worker, worker.id)
200            workerService.update_worker_face_image(worker_to_update, new_raw_image)
201
202            # Assert internal call
203            mock_embedding.assert_called_once_with(new_raw_image)
204
205        # Verification
206        db.session.expire_all()
207        fetched_worker = db.session.get(Worker, worker.id)
208        assert fetched_worker.face_embedding == mocked_embedding_result