Merge pull request #327 from christianlouis/copilot/increase-test-coverage
test: increase coverage for wizard.py and upload_to_nextcloud.py to 98%+
This commit is contained in:
@@ -266,3 +266,130 @@ class TestUploadToNextcloud:
|
|||||||
result = upload_to_nextcloud.apply(args=[str(test_file)], kwargs={"file_id": 1}).get()
|
result = upload_to_nextcloud.apply(args=[str(test_file)], kwargs={"file_id": 1}).get()
|
||||||
|
|
||||||
assert result["status"] == "Completed"
|
assert result["status"] == "Completed"
|
||||||
|
|
||||||
|
@patch("app.tasks.upload_to_nextcloud.get_unique_filename")
|
||||||
|
@patch("app.tasks.upload_to_nextcloud.extract_remote_path")
|
||||||
|
@patch("app.tasks.upload_to_nextcloud.requests")
|
||||||
|
@patch("app.tasks.upload_to_nextcloud.log_task_progress")
|
||||||
|
@patch("app.tasks.upload_to_nextcloud.settings")
|
||||||
|
def test_file_exists_check_returns_true(
|
||||||
|
self, mock_settings, mock_log, mock_requests, mock_extract, mock_unique, tmp_path
|
||||||
|
):
|
||||||
|
"""Test that check_exists_in_nextcloud correctly identifies existing files."""
|
||||||
|
from app.tasks.upload_to_nextcloud import upload_to_nextcloud
|
||||||
|
|
||||||
|
mock_settings.nextcloud_upload_url = "https://nextcloud.example.com/remote.php/dav/"
|
||||||
|
mock_settings.nextcloud_username = "user"
|
||||||
|
mock_settings.nextcloud_password = "pass" # noqa: S105
|
||||||
|
mock_settings.nextcloud_folder = ""
|
||||||
|
mock_settings.workdir = str(tmp_path)
|
||||||
|
mock_settings.http_request_timeout = 30
|
||||||
|
|
||||||
|
test_file = tmp_path / "test.pdf"
|
||||||
|
test_file.write_bytes(b"test content")
|
||||||
|
|
||||||
|
mock_extract.return_value = "test.pdf"
|
||||||
|
|
||||||
|
# Mock PROPFIND to return file exists (path in response text)
|
||||||
|
mock_propfind_response = Mock()
|
||||||
|
mock_propfind_response.text = "test.pdf"
|
||||||
|
mock_requests.request.return_value = mock_propfind_response
|
||||||
|
|
||||||
|
# get_unique_filename should be called and will use check_exists_in_nextcloud
|
||||||
|
def mock_get_unique(path, check_fn):
|
||||||
|
# Call check_fn to exercise the inner function
|
||||||
|
exists = check_fn(path)
|
||||||
|
return "test_1.pdf" if exists else path
|
||||||
|
|
||||||
|
mock_unique.side_effect = mock_get_unique
|
||||||
|
|
||||||
|
mock_put_response = Mock()
|
||||||
|
mock_put_response.status_code = 201
|
||||||
|
mock_requests.put.return_value = mock_put_response
|
||||||
|
|
||||||
|
result = upload_to_nextcloud.apply(args=[str(test_file)], kwargs={"file_id": 1}).get()
|
||||||
|
|
||||||
|
assert result["status"] == "Completed"
|
||||||
|
|
||||||
|
@patch("app.tasks.upload_to_nextcloud.get_unique_filename")
|
||||||
|
@patch("app.tasks.upload_to_nextcloud.extract_remote_path")
|
||||||
|
@patch("app.tasks.upload_to_nextcloud.requests")
|
||||||
|
@patch("app.tasks.upload_to_nextcloud.log_task_progress")
|
||||||
|
@patch("app.tasks.upload_to_nextcloud.settings")
|
||||||
|
def test_file_exists_check_exception_handling(
|
||||||
|
self, mock_settings, mock_log, mock_requests, mock_extract, mock_unique, tmp_path
|
||||||
|
):
|
||||||
|
"""Test that check_exists_in_nextcloud handles exceptions gracefully."""
|
||||||
|
from app.tasks.upload_to_nextcloud import upload_to_nextcloud
|
||||||
|
|
||||||
|
mock_settings.nextcloud_upload_url = "https://nextcloud.example.com/remote.php/dav/"
|
||||||
|
mock_settings.nextcloud_username = "user"
|
||||||
|
mock_settings.nextcloud_password = "pass" # noqa: S105
|
||||||
|
mock_settings.nextcloud_folder = ""
|
||||||
|
mock_settings.workdir = str(tmp_path)
|
||||||
|
mock_settings.http_request_timeout = 30
|
||||||
|
|
||||||
|
test_file = tmp_path / "test.pdf"
|
||||||
|
test_file.write_bytes(b"test content")
|
||||||
|
|
||||||
|
mock_extract.return_value = "test.pdf"
|
||||||
|
|
||||||
|
# Mock get_unique_filename to call check function with exception
|
||||||
|
def mock_get_unique(path, check_fn):
|
||||||
|
# Mock PROPFIND to raise exception
|
||||||
|
mock_requests.request.side_effect = Exception("Network error")
|
||||||
|
# Call check_fn to exercise exception handling
|
||||||
|
exists = check_fn(path)
|
||||||
|
# Should return False when exception occurs
|
||||||
|
assert exists is False
|
||||||
|
return path
|
||||||
|
|
||||||
|
mock_unique.side_effect = mock_get_unique
|
||||||
|
|
||||||
|
mock_put_response = Mock()
|
||||||
|
mock_put_response.status_code = 201
|
||||||
|
mock_requests.put.return_value = mock_put_response
|
||||||
|
|
||||||
|
result = upload_to_nextcloud.apply(args=[str(test_file)], kwargs={"file_id": 1}).get()
|
||||||
|
|
||||||
|
assert result["status"] == "Completed"
|
||||||
|
|
||||||
|
@patch("app.tasks.upload_to_nextcloud.get_unique_filename")
|
||||||
|
@patch("app.tasks.upload_to_nextcloud.extract_remote_path")
|
||||||
|
@patch("app.tasks.upload_to_nextcloud.requests")
|
||||||
|
@patch("app.tasks.upload_to_nextcloud.log_task_progress")
|
||||||
|
@patch("app.tasks.upload_to_nextcloud.settings")
|
||||||
|
def test_empty_parent_dirs_handling(
|
||||||
|
self, mock_settings, mock_log, mock_requests, mock_extract, mock_unique, tmp_path
|
||||||
|
):
|
||||||
|
"""Test handling of empty parent directory paths."""
|
||||||
|
from app.tasks.upload_to_nextcloud import upload_to_nextcloud
|
||||||
|
|
||||||
|
mock_settings.nextcloud_upload_url = "https://nextcloud.example.com/remote.php/dav/"
|
||||||
|
mock_settings.nextcloud_username = "user"
|
||||||
|
mock_settings.nextcloud_password = "pass" # noqa: S105
|
||||||
|
mock_settings.nextcloud_folder = ""
|
||||||
|
mock_settings.workdir = str(tmp_path)
|
||||||
|
mock_settings.http_request_timeout = 30
|
||||||
|
|
||||||
|
test_file = tmp_path / "test.pdf"
|
||||||
|
test_file.write_bytes(b"test content")
|
||||||
|
|
||||||
|
# Return a path with no parent directory (file in root)
|
||||||
|
mock_extract.return_value = "test.pdf"
|
||||||
|
mock_unique.return_value = "test.pdf"
|
||||||
|
|
||||||
|
mock_put_response = Mock()
|
||||||
|
mock_put_response.status_code = 201
|
||||||
|
mock_requests.put.return_value = mock_put_response
|
||||||
|
|
||||||
|
mock_propfind_response = Mock()
|
||||||
|
mock_propfind_response.text = ""
|
||||||
|
mock_requests.request.return_value = mock_propfind_response
|
||||||
|
|
||||||
|
result = upload_to_nextcloud.apply(args=[str(test_file)], kwargs={"file_id": 1}).get()
|
||||||
|
|
||||||
|
assert result["status"] == "Completed"
|
||||||
|
# No MKCOL calls should be made for root-level files
|
||||||
|
mkcol_calls = [c for c in mock_requests.request.call_args_list if c[0][0] == "MKCOL"]
|
||||||
|
assert len(mkcol_calls) == 0
|
||||||
|
|||||||
@@ -1,5 +1,7 @@
|
|||||||
"""Tests for app/views/wizard.py module."""
|
"""Tests for app/views/wizard.py module."""
|
||||||
|
|
||||||
|
from unittest.mock import patch
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
|
||||||
|
|
||||||
@@ -36,3 +38,154 @@ class TestWizardViews:
|
|||||||
"""Test skipping the setup wizard."""
|
"""Test skipping the setup wizard."""
|
||||||
response = client.get("/setup/skip", follow_redirects=False)
|
response = client.get("/setup/skip", follow_redirects=False)
|
||||||
assert response.status_code in (200, 303)
|
assert response.status_code in (200, 303)
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.integration
|
||||||
|
class TestWizardViewsPost:
|
||||||
|
"""Tests for wizard view POST routes."""
|
||||||
|
|
||||||
|
@patch("app.views.wizard.save_setting_to_db")
|
||||||
|
def test_setup_wizard_save_valid_data(self, mock_save, client):
|
||||||
|
"""Test saving valid wizard settings."""
|
||||||
|
mock_save.return_value = True
|
||||||
|
|
||||||
|
response = client.post(
|
||||||
|
"/setup",
|
||||||
|
data={
|
||||||
|
"step": "1",
|
||||||
|
"database_url": "sqlite:///test.db",
|
||||||
|
"redis_url": "redis://localhost:6379/0",
|
||||||
|
},
|
||||||
|
follow_redirects=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 303
|
||||||
|
assert "/setup?step=2" in response.headers["location"]
|
||||||
|
# At least one save should have been called
|
||||||
|
assert mock_save.call_count >= 1
|
||||||
|
|
||||||
|
@patch("app.views.wizard.save_setting_to_db")
|
||||||
|
def test_setup_wizard_save_empty_values_skipped(self, mock_save, client):
|
||||||
|
"""Test that empty values are skipped during save."""
|
||||||
|
mock_save.return_value = True
|
||||||
|
|
||||||
|
response = client.post(
|
||||||
|
"/setup",
|
||||||
|
data={
|
||||||
|
"step": "1",
|
||||||
|
"openai_api_key": "", # Empty value should be skipped
|
||||||
|
"azure_endpoint": " ", # Whitespace only should be skipped
|
||||||
|
},
|
||||||
|
follow_redirects=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 303
|
||||||
|
# Should not have called save for empty values
|
||||||
|
assert mock_save.call_count == 0
|
||||||
|
|
||||||
|
@patch("app.views.wizard.save_setting_to_db")
|
||||||
|
@patch("app.views.wizard.secrets.token_hex")
|
||||||
|
def test_setup_wizard_auto_generate_session_secret(self, mock_token, mock_save, client):
|
||||||
|
"""Test auto-generation of session secret."""
|
||||||
|
mock_token.return_value = "auto_generated_secret_token_12345678"
|
||||||
|
mock_save.return_value = True
|
||||||
|
|
||||||
|
response = client.post(
|
||||||
|
"/setup",
|
||||||
|
data={
|
||||||
|
"step": "2", # session_secret is in step 2
|
||||||
|
"session_secret": "auto-generate",
|
||||||
|
},
|
||||||
|
follow_redirects=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 303
|
||||||
|
mock_token.assert_called_once_with(32)
|
||||||
|
# Verify that the auto-generated token was saved
|
||||||
|
mock_save.assert_called_once()
|
||||||
|
call_args = mock_save.call_args[0]
|
||||||
|
assert call_args[1] == "session_secret"
|
||||||
|
assert call_args[2] == "auto_generated_secret_token_12345678"
|
||||||
|
|
||||||
|
@patch("app.views.wizard.save_setting_to_db")
|
||||||
|
def test_setup_wizard_save_last_step_redirects_home(self, mock_save, client):
|
||||||
|
"""Test that last step redirects to home."""
|
||||||
|
mock_save.return_value = True
|
||||||
|
|
||||||
|
# Step 3 is typically the last step
|
||||||
|
response = client.post(
|
||||||
|
"/setup",
|
||||||
|
data={
|
||||||
|
"step": "3",
|
||||||
|
"some_setting": "value",
|
||||||
|
},
|
||||||
|
follow_redirects=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 303
|
||||||
|
assert "/?setup=complete" in response.headers["location"]
|
||||||
|
|
||||||
|
@patch("app.views.wizard.save_setting_to_db")
|
||||||
|
def test_setup_wizard_save_failed_setting(self, mock_save, client):
|
||||||
|
"""Test handling when save_setting_to_db returns False."""
|
||||||
|
mock_save.return_value = False
|
||||||
|
|
||||||
|
response = client.post(
|
||||||
|
"/setup",
|
||||||
|
data={
|
||||||
|
"step": "1",
|
||||||
|
"some_key": "some_value",
|
||||||
|
},
|
||||||
|
follow_redirects=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
# Should still continue even if save fails
|
||||||
|
assert response.status_code == 303
|
||||||
|
|
||||||
|
@patch("app.views.wizard.save_setting_to_db")
|
||||||
|
def test_setup_wizard_save_exception_handling(self, mock_save, client):
|
||||||
|
"""Test exception handling in setup_wizard_save."""
|
||||||
|
mock_save.side_effect = Exception("Database error")
|
||||||
|
|
||||||
|
response = client.post(
|
||||||
|
"/setup",
|
||||||
|
data={
|
||||||
|
"step": "1",
|
||||||
|
"database_url": "sqlite:///test.db",
|
||||||
|
},
|
||||||
|
follow_redirects=False,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert response.status_code == 303
|
||||||
|
assert "error=save_failed" in response.headers["location"]
|
||||||
|
assert "step=1" in response.headers["location"]
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.integration
|
||||||
|
class TestWizardSkip:
|
||||||
|
"""Tests for wizard skip functionality."""
|
||||||
|
|
||||||
|
@patch("app.views.wizard.save_setting_to_db")
|
||||||
|
def test_setup_wizard_skip_success(self, mock_save, client):
|
||||||
|
"""Test successful skipping of setup wizard."""
|
||||||
|
mock_save.return_value = True
|
||||||
|
|
||||||
|
response = client.get("/setup/skip", follow_redirects=False)
|
||||||
|
|
||||||
|
assert response.status_code == 303
|
||||||
|
assert response.headers["location"] == "/"
|
||||||
|
mock_save.assert_called_once()
|
||||||
|
call_args = mock_save.call_args[0]
|
||||||
|
assert call_args[1] == "_setup_wizard_skipped"
|
||||||
|
assert call_args[2] == "true"
|
||||||
|
|
||||||
|
@patch("app.views.wizard.save_setting_to_db")
|
||||||
|
def test_setup_wizard_skip_exception_handling(self, mock_save, client):
|
||||||
|
"""Test exception handling when skipping wizard."""
|
||||||
|
mock_save.side_effect = Exception("Database error")
|
||||||
|
|
||||||
|
response = client.get("/setup/skip", follow_redirects=False)
|
||||||
|
|
||||||
|
# Should still redirect to home even on error
|
||||||
|
assert response.status_code == 303
|
||||||
|
assert response.headers["location"] == "/"
|
||||||
|
|||||||
Reference in New Issue
Block a user