Unit test improvement

Author: NeptuneHub

tests/unit/test_app_alchemy_anchor.py

@@ -0,0 +1,50 @@
+import pytest
+from unittest.mock import patch
+from flask import Flask
+
+from app_alchemy import alchemy_bp
+
+
+@pytest.fixture
+def app():
+    app = Flask(__name__)
+    app.register_blueprint(alchemy_bp)
+    app.config['TESTING'] = True
+    return app
+
+
+@pytest.fixture
+def client(app):
+    return app.test_client()
+
+
+class TestCreateAnchorValidation:
+    @patch('app_helper.save_alchemy_anchor')
+    def test_whitespace_only_name_returns_400(self, mock_save, client):
+        response = client.post('/api/anchors', json={'name': '   ', 'centroid': [0.1, 0.2]})
+        assert response.status_code == 400
+        assert response.get_json() == {'error': 'Anchor name is required'}
+        mock_save.assert_not_called()
+
+    @patch('app_helper.save_alchemy_anchor')
+    def test_non_list_centroid_returns_400(self, mock_save, client):
+        response = client.post('/api/anchors', json={'name': 'My Anchor', 'centroid': 'not-a-list'})
+        assert response.status_code == 400
+        assert response.get_json() == {'error': 'Anchor centroid is required and must be a list'}
+        mock_save.assert_not_called()
+
+    @patch('app_helper.save_alchemy_anchor')
+    def test_empty_list_centroid_returns_400(self, mock_save, client):
+        response = client.post('/api/anchors', json={'name': 'My Anchor', 'centroid': []})
+        assert response.status_code == 400
+        assert response.get_json() == {'error': 'Anchor centroid is required and must be a list'}
+        mock_save.assert_not_called()
+
+
+class TestRenameAnchorValidation:
+    @patch('app_helper.update_alchemy_anchor_name')
+    def test_whitespace_only_name_returns_400(self, mock_update, client):
+        response = client.put('/api/anchors/7', json={'name': '   '})
+        assert response.status_code == 400
+        assert response.get_json() == {'error': 'Anchor name is required'}
+        mock_update.assert_not_called()

tests/unit/test_app_alchemy_payload.py

@@ -0,0 +1,38 @@
+import pytest
+from unittest.mock import patch
+from flask import Flask
+
+from app_alchemy import alchemy_bp
+
+
+@pytest.fixture
+def app():
+    app = Flask(__name__)
+    app.register_blueprint(alchemy_bp)
+    app.config['TESTING'] = True
+    return app
+
+
+@pytest.fixture
+def client(app):
+    return app.test_client()
+
+
+class TestAlchemyApiPayloadValidation:
+    def test_items_without_any_add_op_returns_400(self, client):
+        response = client.post('/api/alchemy', json={'items': [{'id': 'song-1', 'op': 'SUBTRACT'}]})
+        assert response.status_code == 400
+        assert response.get_json() == {'error': 'Invalid request'}
+
+    def test_add_item_missing_id_returns_400(self, client):
+        response = client.post('/api/alchemy', json={'items': [{'op': 'ADD', 'type': 'song'}]})
+        assert response.status_code == 400
+        assert response.get_json() == {'error': 'Invalid request'}
+
+    @patch('app_alchemy.song_alchemy')
+    def test_add_items_without_id_are_filtered_before_dispatch(self, mock_alchemy, client):
+        mock_alchemy.side_effect = ValueError('At least one item must be in the ADD set')
+        response = client.post('/api/alchemy', json={'items': [{'op': 'ADD'}, {'op': 'ADD', 'id': ''}]})
+        assert response.status_code == 400
+        assert mock_alchemy.call_args.kwargs['add_items'] == []
+        assert mock_alchemy.call_args.kwargs['subtract_items'] == []

tests/unit/test_app_backup_restore.py

@@ -0,0 +1,154 @@
+"""Unit tests for app_backup.restore_backup() chunk and lock validation.
+
+Exercises the /api/backup/restore route through a Flask test client with
+multipart/form-data uploads. The module-level Redis lock helpers are patched
+(the functions, not redis), BACKUP_DIR is redirected to a pytest tmp dir, and
+no test ever completes a chunk set, so the detached restore subprocess is
+never spawned.
+"""
+import io
+import os
+from unittest.mock import MagicMock
+
+import pytest
+from flask import Flask
+
+import app_backup
+
+CONFIRMATION = "I want to restore the database from the backup. This action is not reversible"
+
+
+@pytest.fixture
+def client():
+    app = Flask(__name__)
+    app.config['TESTING'] = True
+    app.register_blueprint(app_backup.backup_bp)
+    return app.test_client()
+
+
+def _form(confirmation=CONFIRMATION, chunk_num=None, total_chunks=None, with_file=True):
+    data = {'confirmation': confirmation}
+    if chunk_num is not None:
+        data['chunk_num'] = str(chunk_num)
+    if total_chunks is not None:
+        data['total_chunks'] = str(total_chunks)
+    if with_file:
+        data['file'] = (io.BytesIO(b'SELECT 1;\n'), 'backup.sql')
+    return data
+
+
+def _post(client, **kwargs):
+    return client.post(
+        '/api/backup/restore',
+        data=_form(**kwargs),
+        content_type='multipart/form-data',
+    )
+
+
+class TestRestoreValidation:
+    def test_wrong_confirmation_is_400(self, client):
+        resp = _post(client, confirmation='nope')
+        assert resp.status_code == 400
+        assert 'Confirmation' in resp.get_json()['error']
+
+    def test_missing_confirmation_is_400(self, client):
+        resp = _post(client, confirmation='')
+        assert resp.status_code == 400
+
+    def test_missing_file_is_400(self, client):
+        resp = _post(client, with_file=False)
+        assert resp.status_code == 400
+        assert resp.get_json()['error'] == 'No file uploaded.'
+
+    def test_non_integer_chunk_fields_are_400(self, client):
+        resp = _post(client, chunk_num='abc', total_chunks='3')
+        assert resp.status_code == 400
+        assert 'must be integers' in resp.get_json()['error']
+
+    @pytest.mark.parametrize('chunk_num,total_chunks', [
+        (0, 3),
+        (4, 3),
+        (2, 1),
+        (-1, 3),
+        (0, 0),
+    ])
+    def test_chunk_num_out_of_range_is_400(self, client, chunk_num, total_chunks):
+        resp = _post(client, chunk_num=chunk_num, total_chunks=total_chunks)
+        assert resp.status_code == 400
+        assert 'Invalid chunk numbers' in resp.get_json()['error']
+
+
+class TestRestoreLock:
+    def test_first_chunk_lock_already_held_is_409(self, client, monkeypatch, tmp_path):
+        monkeypatch.setattr(app_backup, 'BACKUP_DIR', str(tmp_path))
+        monkeypatch.setattr(app_backup, '_acquire_restore_lock', lambda: False)
+        resp = _post(client, chunk_num=1, total_chunks=3)
+        assert resp.status_code == 409
+        assert 'already in progress' in resp.get_json()['error']
+
+    def test_later_chunk_lock_not_held_is_409(self, client, monkeypatch, tmp_path):
+        monkeypatch.setattr(app_backup, 'BACKUP_DIR', str(tmp_path))
+        monkeypatch.setattr(app_backup, '_restore_lock_held', lambda: False)
+        resp = _post(client, chunk_num=2, total_chunks=3)
+        assert resp.status_code == 409
+        assert 'Restart the upload from chunk 1' in resp.get_json()['error']
+
+    def test_later_chunk_never_tries_to_acquire(self, client, monkeypatch, tmp_path):
+        monkeypatch.setattr(app_backup, 'BACKUP_DIR', str(tmp_path))
+        acquire = MagicMock(return_value=True)
+        monkeypatch.setattr(app_backup, '_acquire_restore_lock', acquire)
+        monkeypatch.setattr(app_backup, '_restore_lock_held', lambda: False)
+        resp = _post(client, chunk_num=2, total_chunks=3)
+        assert resp.status_code == 409
+        acquire.assert_not_called()
+
+    def test_single_file_upload_lock_held_is_409(self, client, monkeypatch):
+        monkeypatch.setattr(app_backup, '_acquire_restore_lock', lambda: False)
+        resp = _post(client)
+        assert resp.status_code == 409
+        assert 'already in progress' in resp.get_json()['error']
+
+
+class TestRestoreChunkProgress:
+    def test_intermediate_chunk_is_acknowledged(self, client, monkeypatch, tmp_path):
+        monkeypatch.setattr(app_backup, 'BACKUP_DIR', str(tmp_path))
+        monkeypatch.setattr(app_backup, '_acquire_restore_lock', lambda: True)
+        resp = _post(client, chunk_num=1, total_chunks=3)
+        assert resp.status_code == 200
+        body = resp.get_json()
+        assert body['success'] is True
+        assert body['all_chunks_received'] is False
+        assert body['chunk_num'] == 1
+        assert body['total_chunks'] == 3
+        assert body['received_chunks'] == [1]
+        assert body['missing_chunks'] == [2, 3]
+        assert os.path.exists(os.path.join(str(tmp_path), 'chunks', 'backup_1_of_3.sql'))
+
+    def test_first_chunk_wipes_leftover_chunks(self, client, monkeypatch, tmp_path):
+        monkeypatch.setattr(app_backup, 'BACKUP_DIR', str(tmp_path))
+        monkeypatch.setattr(app_backup, '_acquire_restore_lock', lambda: True)
+        chunks_dir = tmp_path / 'chunks'
+        chunks_dir.mkdir()
+        leftover = chunks_dir / 'backup_2_of_3.sql'
+        leftover.write_bytes(b'stale data')
+        resp = _post(client, chunk_num=1, total_chunks=3)
+        assert resp.status_code == 200
+        body = resp.get_json()
+        assert body['received_chunks'] == [1]
+        assert body['missing_chunks'] == [2, 3]
+        assert not leftover.exists()
+
+    def test_second_chunk_keeps_existing_chunks(self, client, monkeypatch, tmp_path):
+        monkeypatch.setattr(app_backup, 'BACKUP_DIR', str(tmp_path))
+        monkeypatch.setattr(app_backup, '_restore_lock_held', lambda: True)
+        chunks_dir = tmp_path / 'chunks'
+        chunks_dir.mkdir()
+        (chunks_dir / 'backup_1_of_3.sql').write_bytes(b'first chunk')
+        resp = _post(client, chunk_num=2, total_chunks=3)
+        assert resp.status_code == 200
+        body = resp.get_json()
+        assert body['all_chunks_received'] is False
+        assert body['received_chunks'] == [1, 2]
+        assert body['missing_chunks'] == [3]
+        assert (chunks_dir / 'backup_1_of_3.sql').exists()
+        assert (chunks_dir / 'backup_2_of_3.sql').exists()

tests/unit/test_app_cron_parsing.py

@@ -0,0 +1,92 @@
+import os
+import time
+
+import pytest
+
+from app_cron import _field_matches, cron_matches_now
+
+
+FIXED_TS = 1700000000
+
+requires_tzset = pytest.mark.skipif(
+    not hasattr(time, 'tzset'), reason='time.tzset not available on this platform'
+)
+
+
+@pytest.fixture
+def utc_tz():
+    old = os.environ.get('TZ')
+    os.environ['TZ'] = 'UTC'
+    time.tzset()
+    yield
+    if old is None:
+        os.environ.pop('TZ', None)
+    else:
+        os.environ['TZ'] = old
+    time.tzset()
+
+
+def test_field_matches_star_matches_any_value():
+    assert _field_matches('*', 0) is True
+    assert _field_matches('*', 59) is True
+    assert _field_matches(' * ', 7) is True
+
+
+def test_field_matches_exact_value():
+    assert _field_matches('5', 5) is True
+    assert _field_matches('5', 6) is False
+
+
+def test_field_matches_range_inclusive():
+    assert _field_matches('1-5', 1) is True
+    assert _field_matches('1-5', 3) is True
+    assert _field_matches('1-5', 5) is True
+
+
+def test_field_matches_range_outside():
+    assert _field_matches('1-5', 0) is False
+    assert _field_matches('1-5', 6) is False
+
+
+def test_field_matches_comma_list():
+    assert _field_matches('1,3,5', 1) is True
+    assert _field_matches('1,3,5', 3) is True
+    assert _field_matches('1,3,5', 5) is True
+    assert _field_matches('1,3,5', 2) is False
+    assert _field_matches('1,3,5', 4) is False
+
+
+def test_field_matches_malformed_range_returns_false():
+    assert _field_matches('1-', 1) is False
+    assert _field_matches('1-', 0) is False
+    assert _field_matches('-5', 3) is False
+
+
+def test_field_matches_non_numeric_returns_false():
+    assert _field_matches('abc', 1) is False
+
+
+def test_cron_matches_now_short_expression_returns_false():
+    assert cron_matches_now('* * * *', FIXED_TS) is False
+
+
+@requires_tzset
+def test_cron_matches_now_matching_expression(utc_tz):
+    assert cron_matches_now('13 22 14 11 *', FIXED_TS) is True
+
+
+@requires_tzset
+def test_cron_matches_now_matching_dow(utc_tz):
+    assert cron_matches_now('13 22 * * 2', FIXED_TS) is True
+
+
+@requires_tzset
+def test_cron_matches_now_non_matching_minute(utc_tz):
+    assert cron_matches_now('14 22 14 11 *', FIXED_TS) is False
+
+
+@requires_tzset
+def test_cron_matches_now_dom_dow_either_matches(utc_tz):
+    assert cron_matches_now('13 22 1 11 2', FIXED_TS) is True
+    assert cron_matches_now('13 22 14 11 5', FIXED_TS) is True
+    assert cron_matches_now('13 22 1 11 5', FIXED_TS) is False

tests/unit/test_app_dashboard_parsing.py

@@ -0,0 +1,35 @@
+from app_dashboard import _parse_keyval
+
+
+class TestParseKeyval:
+    def test_empty_string_returns_empty_dict(self):
+        assert _parse_keyval('') == {}
+
+    def test_none_returns_empty_dict(self):
+        assert _parse_keyval(None) == {}
+
+    def test_invalid_value_pair_is_skipped(self):
+        result = _parse_keyval('key1:invalid,key2:5.5')
+        assert 'key1' not in result
+        assert result == {'key2': 5.5}
+
+    def test_pair_without_separator_is_skipped(self):
+        result = _parse_keyval('key1,key2:5.5')
+        assert 'key1' not in result
+        assert result == {'key2': 5.5}
+
+    def test_valid_multi_pair_string_parses_fully(self):
+        result = _parse_keyval('rock:0.9,jazz:0.05,pop:0.12')
+        assert result == {'rock': 0.9, 'jazz': 0.05, 'pop': 0.12}
+
+    def test_key_whitespace_is_stripped(self):
+        result = _parse_keyval(' rock :0.9, jazz:0.1')
+        assert result == {'rock': 0.9, 'jazz': 0.1}
+
+    def test_empty_key_is_skipped(self):
+        result = _parse_keyval(':0.5,pop:0.2')
+        assert result == {'pop': 0.2}
+
+    def test_value_with_surrounding_whitespace_parses(self):
+        result = _parse_keyval('rock: 0.9 ')
+        assert result == {'rock': 0.9}