diff --git a/tests/test_utils.py b/tests/test_utils.py index 3a19bbf38..f9f2005de 100644 --- a/tests/test_utils.py +++ b/tests/test_utils.py @@ -1593,7 +1593,7 @@ def test_add_uploaded_files_to_db_other_failed_op(client: flask.testing.FlaskCli assert "Incorrect 'failed_op'." in errors["file1.txt"]["error"] -def test_add_uploaded_files_to_db_file_not_found(client: flask.testing.FlaskClient, capfd): +def test_add_uploaded_files_to_db_file_not_found(client: flask.testing.FlaskClient): """Test the return values of the function when file is not found on S3.""" from botocore.exceptions import ClientError @@ -1632,3 +1632,83 @@ def test_add_uploaded_files_to_db_file_not_found(client: flask.testing.FlaskClie assert file not in files_added assert files_added == [] assert "File not found in S3" in errors["file1.txt"]["error"] + + +def test_add_uploaded_files_to_db_file_already_in_db(client: flask.testing.FlaskClient): + """Function should return error if file is already in the database.""" + + # get a project and an existing file from this project + proj_in_db = models.Project.query.first() + file_in_db = proj_in_db.files[0] + assert file_in_db + + log = { + file_in_db.name: { + "status": {"failed_op": "add_file_db"}, + "path_remote": "path/to/file1.txt", + "subpath": "subpath", + "size_raw": 100, + "size_processed": 200, + "compressed": False, + "public_key": "public_key", + "salt": "salt", + "checksum": "checksum", + } + } + + # Mock the S3 connector and head_object method + mock_s3conn = MagicMock() + mock_s3conn.resource.meta.client.head_object.return_value = None + + # Call the function + with patch("dds_web.api.api_s3_connector.ApiS3Connector", return_value=mock_s3conn): + files_added, errors = utils.add_uploaded_files_to_db(proj_in_db, log) + + assert files_added == [] + assert file_in_db.name in errors + assert "File already in database." in errors[file_in_db.name]["error"] + + +def test_add_uploaded_files_to_db_sql_error(client: flask.testing.FlaskClient): + """Test the return values of the function when sqlalchemy error occurs.""" + import sqlalchemy.exc + from dds_web import db + + # Prepare input data + proj_in_db = models.Project.query.first() + log = { + "file1.txt": { + "status": {"failed_op": "add_file_db"}, + "path_remote": "path/to/file1.txt", + "subpath": "subpath", + "size_raw": 100, + "size_processed": 200, + "compressed": False, + "public_key": "public_key", + "salt": "salt", + "checksum": "checksum", + } + } + + # mock ApiS3Connector + mock_s3conn = MagicMock() + mock_s3conn.resource.meta.client.head_object.return_value = None + + # mock db.session.commit + db_session_commit_mock = MagicMock( + side_effect=sqlalchemy.exc.OperationalError("OperationalError", "test", "sqlalchemy") + ) + + # call add_uploaded_files_to_db + with patch("dds_web.api.api_s3_connector.ApiS3Connector", return_value=mock_s3conn): + with patch("dds_web.db.session.commit", db_session_commit_mock): + files_added, errors = utils.add_uploaded_files_to_db(proj_in_db, log) + + # check that the file is not added to the database + file = models.File.query.filter_by(name="file1.txt").first() + assert not file + + # check that the error is returned and files_added is empty + assert file not in files_added + assert files_added == [] + assert "OperationalError" in errors["file1.txt"]["error"]