Replace snapshots by import / export

Fix #1042

tests/controller/test_controller.py

@@ -338,11 +338,6 @@ def test_load_project(controller, async_run, tmpdir):
     node1 = project.get_node("50d66d7b-0dd7-4e9f-b720-6eb621ae6543")
     assert node1.name == "PC1"
 
-    # Reload the same project should do nothing
-    with asyncio_patch("gns3server.controller.Controller.add_project") as mock_add_project:
-        project = async_run(controller.load_project(str(tmpdir / "test.gns3")))
-    assert not mock_add_project.called
-
 
 def test_get_free_project_name(controller, async_run):
 

tests/controller/test_export_project.py

@@ -54,6 +54,7 @@ def test_filter_files():
     assert _filter_files("project-files/tmp")
     assert _filter_files("project-files/test_log.txt")
     assert _filter_files("project-files/test.log")
+    assert _filter_files("test/snapshots")
     assert _filter_files("test/project-files/snapshots")
     assert _filter_files("test/project-files/snapshots/test.gns3p")
 

tests/controller/test_import_project.py

@@ -56,7 +56,7 @@ def test_import_project(async_run, tmpdir, controller):
         project = async_run(import_project(controller, project_id, f))
 
     assert project.name == "test"
-    assert project.id == project_id  # The project should changed
+    assert project.id == project_id
 
     assert os.path.exists(os.path.join(project.path, "b.png"))
     assert not os.path.exists(os.path.join(project.path, "project.gns3"))
@@ -70,6 +70,41 @@ def test_import_project(async_run, tmpdir, controller):
     assert project.name != "test"
 
 
+def test_import_project_override(async_run, tmpdir, controller):
+    """
+    In the case of snapshot we will import a project for
+    override the previous keeping the same project id & location
+    """
+    project_id = str(uuid.uuid4())
+
+    topology = {
+        "project_id": project_id,
+        "name": "test",
+        "topology": {
+        },
+        "version": "2.0.0"
+    }
+
+    with open(str(tmpdir / "project.gns3"), 'w+') as f:
+        json.dump(topology, f)
+
+    zip_path = str(tmpdir / "project.zip")
+    with zipfile.ZipFile(zip_path, 'w') as myzip:
+        myzip.write(str(tmpdir / "project.gns3"), "project.gns3")
+
+    with open(zip_path, "rb") as f:
+        project = async_run(import_project(controller, project_id, f, location=str(tmpdir)))
+
+    assert project.name == "test"
+    assert project.id == project_id
+
+    # Overide the project with same project
+    with open(zip_path, "rb") as f:
+        project = async_run(import_project(controller, project_id, f, location=str(tmpdir)))
+    assert project.id == project_id
+    assert project.name == "test"
+
+
 def test_import_upgrade(async_run, tmpdir, controller):
     """
     Topology made for previous GNS3 version are upgraded during the process

tests/controller/test_project.py

@@ -373,3 +373,60 @@ def test_duplicate(project, async_run, controller):
 
     assert new_project.get_node(remote_vpcs.id).compute.id == "remote"
     assert new_project.get_node(remote_virtualbox.id).compute.id == "remote"
+
+
+def test_snapshots(project):
+    """
+    List the snapshots
+    """
+    os.makedirs(os.path.join(project.path, "snapshots"))
+    open(os.path.join(project.path, "snapshots", "test1_260716_103713.gns3project"), "w+").close()
+    project.reset()
+
+    assert len(project.snapshots) == 1
+    assert list(project.snapshots.values())[0].name == "test1"
+
+
+def test_get_snapshot(project):
+    os.makedirs(os.path.join(project.path, "snapshots"))
+    open(os.path.join(project.path, "snapshots", "test1.gns3project"), "w+").close()
+    project.reset()
+
+    snapshot = list(project.snapshots.values())[0]
+    assert project.get_snapshot(snapshot.id) == snapshot
+
+    with pytest.raises(aiohttp.web_exceptions.HTTPNotFound):
+        project.get_snapshot("BLU")
+
+
+def test_delete_snapshot(project, async_run):
+    os.makedirs(os.path.join(project.path, "snapshots"))
+    open(os.path.join(project.path, "snapshots", "test1_260716_103713.gns3project"), "w+").close()
+    project.reset()
+
+    snapshot = list(project.snapshots.values())[0]
+    assert project.get_snapshot(snapshot.id) == snapshot
+
+    async_run(project.delete_snapshot(snapshot.id))
+
+    with pytest.raises(aiohttp.web_exceptions.HTTPNotFound):
+        project.get_snapshot(snapshot.id)
+
+    assert not os.path.exists(os.path.join(project.path, "snapshots", "test1.gns3project"))
+
+
+def test_snapshot(project, async_run):
+    """
+    Create a snapshot
+    """
+    assert len(project.snapshots) == 0
+
+    snapshot = async_run(project.snapshot("test1"))
+    assert snapshot.name == "test1"
+
+    assert len(project.snapshots) == 1
+    assert list(project.snapshots.values())[0].name == "test1"
+
+    # Raise a conflict if name is already use
+    with pytest.raises(aiohttp.web_exceptions.HTTPConflict):
+        snapshot = async_run(project.snapshot("test1"))

tests/controller/test_snapshot.py

@@ -0,0 +1,98 @@
+#!/usr/bin/env python
+#
+# Copyright (C) 2016 GNS3 Technologies Inc.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import os
+import pytest
+from unittest.mock import patch
+
+from gns3server.controller.project import Project
+from gns3server.controller.snapshot import Snapshot
+
+from tests.utils import AsyncioMagicMock
+
+
+@pytest.fixture
+def project(controller):
+    project = Project(controller=controller, name="Test")
+    controller._projects[project.id] = project
+    return project
+
+
+def test_snapshot_name(project):
+    """
+    Test create a snapshot object with a name
+    """
+    snapshot = Snapshot(project, name="test1")
+    assert snapshot.name == "test1"
+    assert snapshot._created_at > 0
+    assert snapshot.path.startswith(os.path.join(project.path, "snapshots", "test1_"))
+    assert snapshot.path.endswith(".gns3project")
+
+    # Check if UTC conversion doesn't corrupt the path
+    snap2 = Snapshot(project, filename=os.path.basename(snapshot.path))
+    assert snap2.path == snapshot.path
+
+
+def test_snapshot_filename(project):
+    """
+    Test create a snapshot object with a filename
+    """
+    snapshot = Snapshot(project, filename="test1_260716_100439.gns3project")
+    assert snapshot.name == "test1"
+    assert snapshot._created_at == 1469527479.0
+    assert snapshot.path == os.path.join(project.path, "snapshots", "test1_260716_100439.gns3project")
+
+
+def test_json(project):
+    snapshot = Snapshot(project, filename="test1_260716_100439.gns3project")
+    assert snapshot.__json__() == {
+        "snapshot_id": snapshot._id,
+        "name": "test1",
+        "project_id": project.id,
+        "created_at": 1469527479
+    }
+
+
+def test_restore(project, controller, async_run):
+    compute = AsyncioMagicMock()
+    compute.id = "local"
+    controller._computes["local"] = compute
+    response = AsyncioMagicMock()
+    response.json = {"console": 2048}
+    compute.post = AsyncioMagicMock(return_value=response)
+
+    async_run(project.add_node(compute, "test1", None, node_type="vpcs", properties={"startup_config": "test.cfg"}))
+
+    snapshot = async_run(project.snapshot(name="test"))
+
+    # We add a node after the snapshots
+    async_run(project.add_node(compute, "test2", None, node_type="vpcs", properties={"startup_config": "test.cfg"}))
+
+    # project-files should be reset when reimporting
+    test_file = os.path.join(project.path, "project-files", "test.txt")
+    os.makedirs(os.path.join(project.path, "project-files"))
+    open(test_file, "a+").close()
+
+    assert os.path.exists(test_file)
+    assert len(project.nodes) == 2
+
+    with patch("gns3server.config.Config.get_section_config", return_value={"local": True}):
+        async_run(snapshot.restore())
+
+    project = controller.get_project(project.id)
+    assert not os.path.exists(test_file)
+    assert len(project.nodes) == 1

tests/handlers/api/controller/test_snapshot.py

@@ -0,0 +1,60 @@
+#!/usr/bin/env python
+#
+# Copyright (C) 2016 GNS3 Technologies Inc.
+#
+# This program is free software: you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation, either version 3 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+
+import os
+import uuid
+import pytest
+
+
+@pytest.fixture
+def project(http_controller, controller):
+    u = str(uuid.uuid4())
+    query = {"name": "test", "project_id": u}
+    response = http_controller.post("/projects", query)
+    project = controller.get_project(u)
+
+    return project
+
+
+@pytest.fixture
+def snapshot(project, async_run):
+    snapshot = async_run(project.snapshot("test"))
+    return snapshot
+
+
+def test_list_snapshots(http_controller, project, snapshot):
+    response = http_controller.get("/projects/{}/snapshots".format(project.id), example=True)
+    assert response.status == 200
+    assert len(response.json) == 1
+
+
+def test_delete_snapshot(http_controller, project, snapshot):
+    response = http_controller.delete("/projects/{}/snapshots/{}".format(project.id, snapshot.id), example=True)
+    assert response.status == 204
+    assert not os.path.exists(snapshot.path)
+
+
+def test_restore_snapshot(http_controller, project, snapshot):
+    response = http_controller.post("/projects/{}/snapshots/{}/restore".format(project.id, snapshot.id), example=True)
+    assert response.status == 201
+    assert response.json["name"] == project.name
+
+
+def test_create_snapshot(http_controller, project):
+    response = http_controller.post("/projects/{}/snapshots".format(project.id), {"name": "snap1"}, example=True)
+    assert response.status == 201
+    assert len(os.listdir(os.path.join(project.path, "snapshots"))) == 1