backend.tests.unit.components.workers.test_worker_service

  1import pytest
  2import numpy as np
  3from datetime import datetime, timedelta
  4from unittest.mock import patch, MagicMock
  5import backend.components.workers.workerService
  6from backend.components.workers.workerService import (
  7    create_worker,
  8    get_worker_by_id,
  9    update_worker_name,
 10    extend_worker_expiration,
 11    get_worker_from_qr_code,
 12    get_worker_embedding,
 13    create_worker_embedding,
 14    generate_worker_secret,
 15    decrypt_worker_secret
 16)
 17from backend.components.camera_verification.qrcode.qrcodeService import (
 18    ExpiredCodeError,
 19    InvalidCodeError,
 20    NoCodeFoundError
 21)
 22
 23
 24# ============================================================================
 25# Test Worker Management (CRUD & Utils)
 26# ============================================================================
 27
 28@patch('backend.components.workers.workerService.face_recognition.face_encodings')
 29def test_create_worker_success(mock_face_encodings, db_session):
 30    """
 31    Test successful creation of a new worker.
 32
 33    Verifies that:
 34    1. The worker is assigned an ID.
 35    2. The face embedding is generated and stored.
 36    3. The secret key is generated (replacing TEMP_SECRET).
 37    4. The expiration date is set correctly.
 38    """
 39    fake_embedding = [np.random.rand(128)]
 40    mock_face_encodings.return_value = fake_embedding
 41
 42    # Input data
 43    name = "Testowy Janusz"
 44    fake_image = np.zeros((100, 100, 3), dtype=np.uint8)  # Empty dummy image
 45    exp_date = datetime.now() + timedelta(days=30)
 46
 47    # Action
 48    worker = create_worker(name, fake_image, exp_date)
 49
 50    # Assertions
 51    assert worker.id is not None
 52    assert worker.name == name
 53    assert worker.expiration_date == exp_date
 54    assert worker.secret != "TEMP_SECRET"  # Secret should be updated
 55    assert len(worker.face_embedding) > 0  # Blob should contain data
 56
 57
 58def test_get_worker_by_id_success(db_session, created_worker):
 59    """
 60    Verifies retrieval of an existing worker by their database ID.
 61    """
 62    fetched = get_worker_by_id(created_worker.id)
 63    assert fetched.id == created_worker.id
 64    assert fetched.name == created_worker.name
 65
 66
 67def test_get_worker_by_id_not_found(db_session):
 68    """
 69    Verifies that a ValueError is raised when requesting a non-existent worker ID.
 70    """
 71    with pytest.raises(ValueError, match="not found"):
 72        get_worker_by_id(999999)
 73
 74
 75def test_update_worker_name(db_session, created_worker):
 76    """
 77    Verifies the functionality to update a worker's name.
 78    """
 79    new_name = "Zmieniony Imię"
 80    update_worker_name(created_worker, new_name)
 81
 82    # Refresh object from database to ensure persistence
 83    db_session.session.refresh(created_worker)
 84    assert created_worker.name == new_name
 85
 86
 87def test_extend_worker_expiration(db_session, created_worker):
 88    """
 89    Verifies extending the worker's access expiration date.
 90    """
 91    new_date = datetime.now() + timedelta(days=365)
 92    extend_worker_expiration(created_worker, new_date)
 93
 94    db_session.session.refresh(created_worker)
 95    # Compare timestamps (allowing for microsecond differences during DB write)
 96    assert abs((created_worker.expiration_date - new_date).total_seconds()) < 1
 97
 98
 99# ============================================================================
100# Test QR Code Logic (Scan -> Worker)
101# ============================================================================
102
103@patch('backend.components.workers.workerService.decode_qr_image')
104def test_get_worker_from_qr_success(mock_decode, db_session, created_worker):
105    """
106    Tests the scenario where the QR code is valid and active.
107
108    We mock the decoder to return the secret assigned to the 'created_worker' fixture.
109    """
110    mock_decode.return_value = created_worker.secret
111
112    # The image array doesn't matter here as the decoder is mocked
113    found_worker = get_worker_from_qr_code(np.array([]))
114
115    assert found_worker.id == created_worker.id
116
117
118@patch('backend.components.workers.workerService.decode_qr_image')
119def test_get_worker_from_qr_expired(mock_decode, db_session, created_worker):
120    """
121    Tests the scenario where the QR code is valid, but the expiration date has passed.
122    """
123    # Set worker expiration to the past
124    created_worker.expiration_date = datetime.now() - timedelta(days=1)
125    db_session.session.commit()
126
127    mock_decode.return_value = created_worker.secret
128
129    with pytest.raises(ExpiredCodeError, match="wygasła"):
130        get_worker_from_qr_code(np.array([]))
131
132
133@patch('backend.components.workers.workerService.decode_qr_image')
134def test_get_worker_from_qr_invalid_secret(mock_decode, db_session):
135    """
136    Tests the scenario where the QR code contains a secret that does not exist in the DB.
137    """
138    mock_decode.return_value = "non_existent_secret_123"
139
140    with pytest.raises(InvalidCodeError, match="niepoprawny kod QR"):
141        get_worker_from_qr_code(np.array([]))
142
143
144@patch('backend.components.workers.workerService.decode_qr_image')
145def test_get_worker_from_qr_no_code(mock_decode, db_session):
146    """
147    Tests the situation where the decoding library cannot find any QR code in the image.
148    """
149    mock_decode.side_effect = NoCodeFoundError("Nie znaleziono kodu")
150
151    with pytest.raises(NoCodeFoundError):
152        get_worker_from_qr_code(np.array([]))
153
154
155# ============================================================================
156# Test Technical Implementation (Embeddings & Secrets)
157# ============================================================================
158
159def test_embedding_serialization_roundtrip():
160    """
161    Verifies the NumPy Array -> BLOB -> NumPy Array serialization cycle.
162
163    This is crucial because face embeddings are stored as binary BLOBs in SQLite.
164    The test simulates the process used inside `create_worker_embedding` and `get_worker_embedding`.
165    """
166    # 1. Create a synthetic face vector (128 floats)
167    original_embedding = np.random.rand(128)
168
169    # Note: create_worker_embedding uses face_recognition.face_encodings internally.
170    # To test strictly the IO/NumPy logic without the heavy ML library, we mock it.
171    with patch('backend.components.workers.workerService.face_recognition.face_encodings') as mock_enc:
172        # Mock returning a list containing our synthetic embedding
173        mock_enc.return_value = [original_embedding]
174
175        # 2. Create BLOB (simulate passing an image)
176        # The function converts the array to bytes via io.BytesIO
177        blob = create_worker_embedding(np.zeros((10, 10)))
178
179        # 3. Create a fake worker object holding this blob
180        fake_worker = MagicMock()
181        fake_worker.face_embedding = blob
182
183        # 4. Read BLOB back into a NumPy array
184        restored_arr = get_worker_embedding(fake_worker)
185
186        # 5. Compare the original vs restored array
187        # create_worker_embedding grabs the first face [0], and np.save saves that array.
188        # The restored array should be identical (within float precision).
189        np.testing.assert_array_almost_equal(restored_arr[0], original_embedding)
190
191
192def test_generate_and_decrypt_secret_functional():
193    """
194    Functional test for secret generation and decryption.
195
196    Verifies that a secret generated for a specific worker can be successfully
197    decrypted to retrieve the original worker ID and name.
198    """
199    worker_id = 67
200    worker_name = "Six Seven"
201
202    # Create a mock worker
203    mock_worker = MagicMock()
204    mock_worker.id = worker_id
205    mock_worker.name = worker_name
206
207    # Call the actual generation function
208    secret = generate_worker_secret(mock_worker)
209
210    # Decrypt the generated secret
211    data = decrypt_worker_secret(secret)
212
213    # Assertion 1: Ensure decrypted data is a dictionary
214    assert isinstance(data, dict)
215
216    # Assertion 2: Check if critical, deterministic data matches
217    assert data['worker_id'] == worker_id
218    assert data['name'] == worker_name
219
220    # Assertion 3: Check if the random value exists (salt/entropy)
221    assert 'rand_value' in data
222    assert len(data['rand_value']) == 6
223
224
225def test_decrypt_secret_static_check():
226    """
227    Sanity check with a static Fernet token.
228
229    WARNING: This test might fail if the Fernet key in the configuration changes.
230    It serves as a regression test for a specific token structure.
231    """
232    # Example token (valid if the key hasn't changed and token hasn't expired relative to TTL)
233    # Note: In a real environment, mocking the time or key is recommended for stability.
234    # We proceed assuming the dev environment uses a fixed key or this token was just generated.
235    secret = 'gAAAAABpPdLUcBJbhCLwEX5HKf8mzB-sUIzAYQaQencHd--KaC4wbHRHlmdIfHSioWUMoZ_woRxjTsBVr30YQBRYv5xoicHjaERw2aGLvQ5Wgud1gaFNR7_zgTpNqzu96fsY-dQt3NvdRUXFmMKWiWV-9VgE99_HBg=='
236
237    # We wrap this in a try-except or rely on the fact that if the key differs, it raises generic error
238    try:
239        data = decrypt_worker_secret(secret)
240        assert data['worker_id'] == 67
241        assert data['name'] == "Six Seven"
242    except Exception:
243        pytest.skip("Skipping static token test - encryption key might have changed.")
@patch('backend.components.workers.workerService.face_recognition.face_encodings')
def test_create_worker_success(mock_face_encodings, db_session):
29@patch('backend.components.workers.workerService.face_recognition.face_encodings')
30def test_create_worker_success(mock_face_encodings, db_session):
31    """
32    Test successful creation of a new worker.
33
34    Verifies that:
35    1. The worker is assigned an ID.
36    2. The face embedding is generated and stored.
37    3. The secret key is generated (replacing TEMP_SECRET).
38    4. The expiration date is set correctly.
39    """
40    fake_embedding = [np.random.rand(128)]
41    mock_face_encodings.return_value = fake_embedding
42
43    # Input data
44    name = "Testowy Janusz"
45    fake_image = np.zeros((100, 100, 3), dtype=np.uint8)  # Empty dummy image
46    exp_date = datetime.now() + timedelta(days=30)
47
48    # Action
49    worker = create_worker(name, fake_image, exp_date)
50
51    # Assertions
52    assert worker.id is not None
53    assert worker.name == name
54    assert worker.expiration_date == exp_date
55    assert worker.secret != "TEMP_SECRET"  # Secret should be updated
56    assert len(worker.face_embedding) > 0  # Blob should contain data

Test successful creation of a new worker.

Verifies that:

  1. The worker is assigned an ID.
  2. The face embedding is generated and stored.
  3. The secret key is generated (replacing TEMP_SECRET).
  4. The expiration date is set correctly.
def test_get_worker_by_id_success(db_session, created_worker):
59def test_get_worker_by_id_success(db_session, created_worker):
60    """
61    Verifies retrieval of an existing worker by their database ID.
62    """
63    fetched = get_worker_by_id(created_worker.id)
64    assert fetched.id == created_worker.id
65    assert fetched.name == created_worker.name

Verifies retrieval of an existing worker by their database ID.

def test_get_worker_by_id_not_found(db_session):
68def test_get_worker_by_id_not_found(db_session):
69    """
70    Verifies that a ValueError is raised when requesting a non-existent worker ID.
71    """
72    with pytest.raises(ValueError, match="not found"):
73        get_worker_by_id(999999)

Verifies that a ValueError is raised when requesting a non-existent worker ID.

def test_update_worker_name(db_session, created_worker):
76def test_update_worker_name(db_session, created_worker):
77    """
78    Verifies the functionality to update a worker's name.
79    """
80    new_name = "Zmieniony Imię"
81    update_worker_name(created_worker, new_name)
82
83    # Refresh object from database to ensure persistence
84    db_session.session.refresh(created_worker)
85    assert created_worker.name == new_name

Verifies the functionality to update a worker's name.

def test_extend_worker_expiration(db_session, created_worker):
88def test_extend_worker_expiration(db_session, created_worker):
89    """
90    Verifies extending the worker's access expiration date.
91    """
92    new_date = datetime.now() + timedelta(days=365)
93    extend_worker_expiration(created_worker, new_date)
94
95    db_session.session.refresh(created_worker)
96    # Compare timestamps (allowing for microsecond differences during DB write)
97    assert abs((created_worker.expiration_date - new_date).total_seconds()) < 1

Verifies extending the worker's access expiration date.

@patch('backend.components.workers.workerService.decode_qr_image')
def test_get_worker_from_qr_success(mock_decode, db_session, created_worker):
104@patch('backend.components.workers.workerService.decode_qr_image')
105def test_get_worker_from_qr_success(mock_decode, db_session, created_worker):
106    """
107    Tests the scenario where the QR code is valid and active.
108
109    We mock the decoder to return the secret assigned to the 'created_worker' fixture.
110    """
111    mock_decode.return_value = created_worker.secret
112
113    # The image array doesn't matter here as the decoder is mocked
114    found_worker = get_worker_from_qr_code(np.array([]))
115
116    assert found_worker.id == created_worker.id

Tests the scenario where the QR code is valid and active.

We mock the decoder to return the secret assigned to the 'created_worker' fixture.

@patch('backend.components.workers.workerService.decode_qr_image')
def test_get_worker_from_qr_expired(mock_decode, db_session, created_worker):
119@patch('backend.components.workers.workerService.decode_qr_image')
120def test_get_worker_from_qr_expired(mock_decode, db_session, created_worker):
121    """
122    Tests the scenario where the QR code is valid, but the expiration date has passed.
123    """
124    # Set worker expiration to the past
125    created_worker.expiration_date = datetime.now() - timedelta(days=1)
126    db_session.session.commit()
127
128    mock_decode.return_value = created_worker.secret
129
130    with pytest.raises(ExpiredCodeError, match="wygasła"):
131        get_worker_from_qr_code(np.array([]))

Tests the scenario where the QR code is valid, but the expiration date has passed.

@patch('backend.components.workers.workerService.decode_qr_image')
def test_get_worker_from_qr_invalid_secret(mock_decode, db_session):
134@patch('backend.components.workers.workerService.decode_qr_image')
135def test_get_worker_from_qr_invalid_secret(mock_decode, db_session):
136    """
137    Tests the scenario where the QR code contains a secret that does not exist in the DB.
138    """
139    mock_decode.return_value = "non_existent_secret_123"
140
141    with pytest.raises(InvalidCodeError, match="niepoprawny kod QR"):
142        get_worker_from_qr_code(np.array([]))

Tests the scenario where the QR code contains a secret that does not exist in the DB.

@patch('backend.components.workers.workerService.decode_qr_image')
def test_get_worker_from_qr_no_code(mock_decode, db_session):
145@patch('backend.components.workers.workerService.decode_qr_image')
146def test_get_worker_from_qr_no_code(mock_decode, db_session):
147    """
148    Tests the situation where the decoding library cannot find any QR code in the image.
149    """
150    mock_decode.side_effect = NoCodeFoundError("Nie znaleziono kodu")
151
152    with pytest.raises(NoCodeFoundError):
153        get_worker_from_qr_code(np.array([]))

Tests the situation where the decoding library cannot find any QR code in the image.

def test_embedding_serialization_roundtrip():
160def test_embedding_serialization_roundtrip():
161    """
162    Verifies the NumPy Array -> BLOB -> NumPy Array serialization cycle.
163
164    This is crucial because face embeddings are stored as binary BLOBs in SQLite.
165    The test simulates the process used inside `create_worker_embedding` and `get_worker_embedding`.
166    """
167    # 1. Create a synthetic face vector (128 floats)
168    original_embedding = np.random.rand(128)
169
170    # Note: create_worker_embedding uses face_recognition.face_encodings internally.
171    # To test strictly the IO/NumPy logic without the heavy ML library, we mock it.
172    with patch('backend.components.workers.workerService.face_recognition.face_encodings') as mock_enc:
173        # Mock returning a list containing our synthetic embedding
174        mock_enc.return_value = [original_embedding]
175
176        # 2. Create BLOB (simulate passing an image)
177        # The function converts the array to bytes via io.BytesIO
178        blob = create_worker_embedding(np.zeros((10, 10)))
179
180        # 3. Create a fake worker object holding this blob
181        fake_worker = MagicMock()
182        fake_worker.face_embedding = blob
183
184        # 4. Read BLOB back into a NumPy array
185        restored_arr = get_worker_embedding(fake_worker)
186
187        # 5. Compare the original vs restored array
188        # create_worker_embedding grabs the first face [0], and np.save saves that array.
189        # The restored array should be identical (within float precision).
190        np.testing.assert_array_almost_equal(restored_arr[0], original_embedding)

Verifies the NumPy Array -> BLOB -> NumPy Array serialization cycle.

This is crucial because face embeddings are stored as binary BLOBs in SQLite. The test simulates the process used inside create_worker_embedding and get_worker_embedding.

def test_generate_and_decrypt_secret_functional():
193def test_generate_and_decrypt_secret_functional():
194    """
195    Functional test for secret generation and decryption.
196
197    Verifies that a secret generated for a specific worker can be successfully
198    decrypted to retrieve the original worker ID and name.
199    """
200    worker_id = 67
201    worker_name = "Six Seven"
202
203    # Create a mock worker
204    mock_worker = MagicMock()
205    mock_worker.id = worker_id
206    mock_worker.name = worker_name
207
208    # Call the actual generation function
209    secret = generate_worker_secret(mock_worker)
210
211    # Decrypt the generated secret
212    data = decrypt_worker_secret(secret)
213
214    # Assertion 1: Ensure decrypted data is a dictionary
215    assert isinstance(data, dict)
216
217    # Assertion 2: Check if critical, deterministic data matches
218    assert data['worker_id'] == worker_id
219    assert data['name'] == worker_name
220
221    # Assertion 3: Check if the random value exists (salt/entropy)
222    assert 'rand_value' in data
223    assert len(data['rand_value']) == 6

Functional test for secret generation and decryption.

Verifies that a secret generated for a specific worker can be successfully decrypted to retrieve the original worker ID and name.

def test_decrypt_secret_static_check():
226def test_decrypt_secret_static_check():
227    """
228    Sanity check with a static Fernet token.
229
230    WARNING: This test might fail if the Fernet key in the configuration changes.
231    It serves as a regression test for a specific token structure.
232    """
233    # Example token (valid if the key hasn't changed and token hasn't expired relative to TTL)
234    # Note: In a real environment, mocking the time or key is recommended for stability.
235    # We proceed assuming the dev environment uses a fixed key or this token was just generated.
236    secret = 'gAAAAABpPdLUcBJbhCLwEX5HKf8mzB-sUIzAYQaQencHd--KaC4wbHRHlmdIfHSioWUMoZ_woRxjTsBVr30YQBRYv5xoicHjaERw2aGLvQ5Wgud1gaFNR7_zgTpNqzu96fsY-dQt3NvdRUXFmMKWiWV-9VgE99_HBg=='
237
238    # We wrap this in a try-except or rely on the fact that if the key differs, it raises generic error
239    try:
240        data = decrypt_worker_secret(secret)
241        assert data['worker_id'] == 67
242        assert data['name'] == "Six Seven"
243    except Exception:
244        pytest.skip("Skipping static token test - encryption key might have changed.")

Sanity check with a static Fernet token.

WARNING: This test might fail if the Fernet key in the configuration changes. It serves as a regression test for a specific token structure.