|
5 | 5 | from werkzeug.datastructures import FileStorage |
6 | 6 | from flask.testing import FlaskClient |
7 | 7 | from werkzeug.test import TestResponse |
| 8 | +from pathlib import Path |
| 9 | +import json |
| 10 | +import zipfile |
8 | 11 |
|
9 | 12 | # Local application imports |
10 | 13 | from opengeodeweb_microservice.database.data import Data |
@@ -457,3 +460,139 @@ def test_model_components(client: FlaskClient) -> None: |
457 | 460 | assert isinstance(collection_component["items"], list) |
458 | 461 | for item_uuid in collection_component["items"]: |
459 | 462 | assert isinstance(item_uuid, str) |
| 463 | + |
| 464 | + |
| 465 | +def test_export_project_route(client: FlaskClient, tmp_path: Path) -> None: |
| 466 | + route = "/opengeodeweb_back/export_project" |
| 467 | + snapshot = { |
| 468 | + "styles": {"1": {"visibility": True, "opacity": 1.0, "color": [0.2, 0.6, 0.9]}} |
| 469 | + } |
| 470 | + filename = "export_project_test.vease" |
| 471 | + project_folder = client.application.config["DATA_FOLDER_PATH"] |
| 472 | + os.makedirs(project_folder, exist_ok=True) |
| 473 | + database_root_path = os.path.join(project_folder, "project.db") |
| 474 | + with open(database_root_path, "wb") as f: |
| 475 | + f.write(b"test_project_db") |
| 476 | + |
| 477 | + with get_session() as session: |
| 478 | + session.query(Data).delete() |
| 479 | + session.commit() |
| 480 | + |
| 481 | + data1 = Data( |
| 482 | + id="test_data_1", |
| 483 | + geode_object="BRep", |
| 484 | + viewer_object="BRep", |
| 485 | + viewer_elements_type="default", |
| 486 | + native_file="native.txt", |
| 487 | + ) |
| 488 | + data2 = Data( |
| 489 | + id="test_data_2", |
| 490 | + geode_object="Section", |
| 491 | + viewer_object="Section", |
| 492 | + viewer_elements_type="default", |
| 493 | + native_file="native.txt", |
| 494 | + ) |
| 495 | + session.add(data1) |
| 496 | + session.add(data2) |
| 497 | + session.commit() |
| 498 | + |
| 499 | + data1_dir = os.path.join(project_folder, "test_data_1") |
| 500 | + os.makedirs(data1_dir, exist_ok=True) |
| 501 | + with open(os.path.join(data1_dir, "native.txt"), "w") as f: |
| 502 | + f.write("native file content") |
| 503 | + |
| 504 | + data2_dir = os.path.join(project_folder, "test_data_2") |
| 505 | + os.makedirs(data2_dir, exist_ok=True) |
| 506 | + with open(os.path.join(data2_dir, "native.txt"), "w") as f: |
| 507 | + f.write("native file content") |
| 508 | + |
| 509 | + response = client.post(route, json={"snapshot": snapshot, "filename": filename}) |
| 510 | + assert response.status_code == 200 |
| 511 | + assert response.headers.get("new-file-name") == filename |
| 512 | + assert response.mimetype == "application/octet-binary" |
| 513 | + response.direct_passthrough = False |
| 514 | + zip_bytes = response.get_data() |
| 515 | + tmp_zip_path = tmp_path / filename |
| 516 | + tmp_zip_path.write_bytes(zip_bytes) |
| 517 | + |
| 518 | + with zipfile.ZipFile(tmp_zip_path, "r") as zip_file: |
| 519 | + names = zip_file.namelist() |
| 520 | + assert "snapshot.json" in names |
| 521 | + parsed = json.loads(zip_file.read("snapshot.json").decode("utf-8")) |
| 522 | + assert parsed == snapshot |
| 523 | + assert "project.db" in names |
| 524 | + assert "test_data_1/native.txt" in names |
| 525 | + assert "test_data_2/native.txt" in names |
| 526 | + |
| 527 | + response.close() |
| 528 | + |
| 529 | + export_path = os.path.join(project_folder, filename) |
| 530 | + if os.path.exists(export_path): |
| 531 | + os.remove(export_path) |
| 532 | + |
| 533 | + |
| 534 | +def test_import_project_route(client: FlaskClient, tmp_path: Path) -> None: |
| 535 | + route = "/opengeodeweb_back/import_project" |
| 536 | + snapshot = { |
| 537 | + "styles": {"1": {"visibility": True, "opacity": 1.0, "color": [0.2, 0.6, 0.9]}} |
| 538 | + } |
| 539 | + |
| 540 | + original_data_folder = client.application.config["DATA_FOLDER_PATH"] |
| 541 | + client.application.config["DATA_FOLDER_PATH"] = os.path.join( |
| 542 | + str(tmp_path), "project_data" |
| 543 | + ) |
| 544 | + db_path = os.path.join(client.application.config["DATA_FOLDER_PATH"], "project.db") |
| 545 | + |
| 546 | + import sqlite3, zipfile, json |
| 547 | + |
| 548 | + temp_db = tmp_path / "temp_project.db" |
| 549 | + conn = sqlite3.connect(str(temp_db)) |
| 550 | + conn.execute( |
| 551 | + "CREATE TABLE datas (id TEXT PRIMARY KEY, geode_object TEXT, viewer_object TEXT, viewer_elements_type TEXT, native_file TEXT, " |
| 552 | + "viewable_file TEXT, light_viewable_file TEXT)" |
| 553 | + ) |
| 554 | + conn.commit() |
| 555 | + conn.close() |
| 556 | + |
| 557 | + z = tmp_path / "import_project_test.vease" |
| 558 | + with zipfile.ZipFile(z, "w", compression=zipfile.ZIP_DEFLATED) as zipf: |
| 559 | + zipf.writestr("snapshot.json", json.dumps(snapshot)) |
| 560 | + zipf.write(str(temp_db), "project.db") |
| 561 | + |
| 562 | + with open(z, "rb") as f: |
| 563 | + resp = client.post( |
| 564 | + route, |
| 565 | + data={"file": (f, "import_project_test.vease")}, |
| 566 | + content_type="multipart/form-data", |
| 567 | + ) |
| 568 | + |
| 569 | + assert resp.status_code == 200 |
| 570 | + assert resp.get_json().get("snapshot") == snapshot |
| 571 | + assert os.path.exists(db_path) |
| 572 | + |
| 573 | + from opengeodeweb_microservice.database import connection |
| 574 | + |
| 575 | + client.application.config["DATA_FOLDER_PATH"] = original_data_folder |
| 576 | + test_db_path = os.environ.get("TEST_DB_PATH") |
| 577 | + if test_db_path: |
| 578 | + connection.init_database(test_db_path, create_tables=True) |
| 579 | + |
| 580 | + client.application.config["DATA_FOLDER_PATH"] = original_data_folder |
| 581 | + |
| 582 | + |
| 583 | +def test_save_viewable_workflow_from_object(client: FlaskClient) -> None: |
| 584 | + route = "/opengeodeweb_back/create/point" |
| 585 | + point_data = { |
| 586 | + "name": "workflow_point_3d", |
| 587 | + "x": 0.0, |
| 588 | + "y": 0.0, |
| 589 | + "z": 0.0, |
| 590 | + } |
| 591 | + |
| 592 | + response = client.post(route, json=point_data) |
| 593 | + assert response.status_code == 200 |
| 594 | + |
| 595 | + data_id = response.get_json()["id"] |
| 596 | + assert isinstance(data_id, str) and len(data_id) > 0 |
| 597 | + assert response.get_json()["geode_object_type"] == "PointSet3D" |
| 598 | + assert response.get_json()["viewable_file"].endswith(".vtp") |
0 commit comments