|
2 | 2 | from urllib.error import HTTPError |
3 | 3 |
|
4 | 4 | import pytest |
| 5 | +from flask import Response |
5 | 6 | from werkzeug.datastructures import MultiDict |
6 | 7 |
|
7 | 8 | from app import create_app |
| 9 | +from app.modules.dataset import routes as dataset_routes |
8 | 10 | from app.modules.dataset.services import ( |
9 | 11 | DatasetMetadataUpdateError, |
10 | 12 | DatasetMetadataValidationError, |
@@ -389,3 +391,116 @@ def test_update_metadata_from_request_unsynced_dataset_to_zenodo_publishes_and_s |
389 | 391 | assert dataset.ds_meta_data.deposition_id == 101 |
390 | 392 | assert dataset.ds_meta_data.dataset_doi == "10.5072/zenodo.101" |
391 | 393 | service.repository.session.commit.assert_called_once() |
| 394 | + |
| 395 | + |
| 396 | +def test_resolve_download_formats_defaults_to_all_available(): |
| 397 | + service = DataSetService() |
| 398 | + result = service.resolve_download_formats(None) |
| 399 | + assert result == list(service.AVAILABLE_DOWNLOAD_FORMATS) |
| 400 | + |
| 401 | + |
| 402 | +def test_resolve_download_formats_rejects_invalid_formats(): |
| 403 | + service = DataSetService() |
| 404 | + with pytest.raises(ValueError, match="Invalid format"): |
| 405 | + service.resolve_download_formats(["uvl", "invalid_format"]) |
| 406 | + |
| 407 | + |
| 408 | +def test_resolve_download_formats_rejects_empty_selection(): |
| 409 | + service = DataSetService() |
| 410 | + with pytest.raises(ValueError, match="No download formats selected"): |
| 411 | + service.resolve_download_formats([" "]) |
| 412 | + |
| 413 | + |
| 414 | +def test_zip_from_storage_filters_files_by_selected_formats(tmp_path, monkeypatch): |
| 415 | + service = DataSetService() |
| 416 | + dataset = MagicMock() |
| 417 | + dataset.user_id = 42 |
| 418 | + dataset.id = 99 |
| 419 | + |
| 420 | + base = tmp_path / "uploads" / "user_42" / "dataset_99" |
| 421 | + (base / "uvl").mkdir(parents=True) |
| 422 | + (base / "glencoe").mkdir(parents=True) |
| 423 | + (base / "dimacs").mkdir(parents=True) |
| 424 | + |
| 425 | + (base / "uvl" / "model.uvl").write_text("uvl", encoding="utf-8") |
| 426 | + (base / "glencoe" / "model.json").write_text("json", encoding="utf-8") |
| 427 | + (base / "dimacs" / "model.cnf").write_text("cnf", encoding="utf-8") |
| 428 | + |
| 429 | + monkeypatch.setenv("WORKING_DIR", str(tmp_path)) |
| 430 | + |
| 431 | + zip_path = service.zip_from_storage(dataset, formats=["uvl", "dimacs"]) |
| 432 | + assert zip_path is not None |
| 433 | + |
| 434 | + import zipfile |
| 435 | + |
| 436 | + with zipfile.ZipFile(zip_path, "r") as zipf: |
| 437 | + names = set(zipf.namelist()) |
| 438 | + assert "uvl/model.uvl" in names |
| 439 | + assert "dimacs/model.cnf" in names |
| 440 | + assert "glencoe/model.json" not in names |
| 441 | + |
| 442 | + |
| 443 | +def test_zip_all_datasets_by_formats_only_includes_selected_format(tmp_path, monkeypatch): |
| 444 | + service = DataSetService() |
| 445 | + monkeypatch.chdir(tmp_path) |
| 446 | + |
| 447 | + ds10 = tmp_path / "uploads" / "user_1" / "dataset_10" |
| 448 | + ds10.joinpath("uvl").mkdir(parents=True) |
| 449 | + ds10.joinpath("glencoe").mkdir(parents=True) |
| 450 | + ds10.joinpath("uvl", "a.uvl").write_text("a", encoding="utf-8") |
| 451 | + ds10.joinpath("glencoe", "a.json").write_text("a", encoding="utf-8") |
| 452 | + |
| 453 | + ds11 = tmp_path / "uploads" / "user_1" / "dataset_11" |
| 454 | + ds11.joinpath("glencoe").mkdir(parents=True) |
| 455 | + ds11.joinpath("glencoe", "b.json").write_text("b", encoding="utf-8") |
| 456 | + |
| 457 | + service.is_synchronized = MagicMock(side_effect=lambda dataset_id: dataset_id == 10) |
| 458 | + |
| 459 | + zip_path = tmp_path / "bulk.zip" |
| 460 | + service.zip_all_datasets_by_formats(str(zip_path), formats=["glencoe"]) |
| 461 | + |
| 462 | + import zipfile |
| 463 | + |
| 464 | + with zipfile.ZipFile(zip_path, "r") as zipf: |
| 465 | + names = set(zipf.namelist()) |
| 466 | + assert "dataset_10/glencoe/a.json" in names |
| 467 | + assert "dataset_10/uvl/a.uvl" not in names |
| 468 | + assert not any(name.startswith("dataset_11/") for name in names) |
| 469 | + |
| 470 | + |
| 471 | +def test_download_dataset_route_passes_selected_formats_to_service(): |
| 472 | + app = create_app("testing") |
| 473 | + dataset = MagicMock() |
| 474 | + dataset.id = 5 |
| 475 | + |
| 476 | + with app.test_request_context("/datasets/download/5?formats=uvl&formats=dimacs"): |
| 477 | + with ( |
| 478 | + patch.object(dataset_routes.dataset_service, "get_or_404", return_value=dataset), |
| 479 | + patch.object( |
| 480 | + dataset_routes.dataset_service, "zip_from_storage", return_value="C:\\tmp\\dataset_5.zip" |
| 481 | + ) as p_zip, |
| 482 | + patch("app.modules.dataset.routes.os.path.exists", return_value=True), |
| 483 | + patch.object(dataset_routes.ds_download_record_service, "create_cookie", return_value="cookie-1"), |
| 484 | + patch("app.modules.dataset.routes.send_file", return_value=Response("ok", status=200)), |
| 485 | + ): |
| 486 | + response = dataset_routes.download_dataset(5) |
| 487 | + |
| 488 | + assert response.status_code == 200 |
| 489 | + p_zip.assert_called_once_with(dataset, formats=["uvl", "dimacs"]) |
| 490 | + |
| 491 | + |
| 492 | +def test_download_all_dataset_route_passes_selected_formats_to_service(): |
| 493 | + app = create_app("testing") |
| 494 | + |
| 495 | + with app.test_request_context("/datasets/download/all?formats=uvl&formats=splot"): |
| 496 | + with ( |
| 497 | + patch.object(dataset_routes.dataset_service, "zip_all_datasets_by_formats") as p_zip_all, |
| 498 | + patch("app.modules.dataset.routes.send_file", return_value=Response("ok", status=200)), |
| 499 | + patch("app.modules.dataset.routes.os.path.exists", return_value=False), |
| 500 | + patch("app.modules.dataset.routes.shutil.rmtree"), |
| 501 | + ): |
| 502 | + response = dataset_routes.download_all_dataset() |
| 503 | + |
| 504 | + assert response.status_code == 200 |
| 505 | + _, kwargs = p_zip_all.call_args |
| 506 | + assert kwargs["formats"] == ["uvl", "splot"] |
0 commit comments