Add tests for opreate_stop

Author: olethanh

tests/supervisor/views/test_operator.py

@@ -102,6 +102,82 @@ async def test_operator_confidential_initialize_already_running(aiohttp_client,
     }
 
 
+@pytest.mark.asyncio
+@pytest.mark.skip()
+async def test_operator_expire(aiohttp_client, mocker):
+    """Test that the expires endpoint work. SPOILER it doesn't"""
+
+    settings.ENABLE_QEMU_SUPPORT = True
+    settings.ENABLE_CONFIDENTIAL_COMPUTING = True
+    settings.setup()
+
+    vm_hash = ItemHash(settings.FAKE_INSTANCE_ID)
+    instance_message = await get_message(ref=vm_hash)
+
+    fake_vm_pool = mocker.Mock(
+        executions={
+            vm_hash: mocker.Mock(
+                vm_hash=vm_hash,
+                message=instance_message.content,
+                is_confidential=False,
+                is_running=False,
+            ),
+        },
+    )
+
+    # Disable auth
+    mocker.patch(
+        "aleph.vm.orchestrator.views.authentication.authenticate_jwk",
+        return_value=instance_message.sender,
+    )
+    app = setup_webapp()
+    app["vm_pool"] = fake_vm_pool
+    client: TestClient = await aiohttp_client(app)
+    response = await client.post(
+        f"/control/machine/{vm_hash}/expire",
+        data={"timeout": 1},
+        # json={"timeout": 1},
+    )
+    assert response.status == 200, await response.text()
+    assert fake_vm_pool["executions"][vm_hash].expire.call_count == 1
+
+
+@pytest.mark.asyncio
+async def test_operator_stop(aiohttp_client, mocker):
+    """Test that the stop endpoint call the method on pool"""
+
+    settings.ENABLE_QEMU_SUPPORT = True
+    settings.ENABLE_CONFIDENTIAL_COMPUTING = True
+    settings.setup()
+
+    vm_hash = ItemHash(settings.FAKE_INSTANCE_ID)
+    instance_message = await get_message(ref=vm_hash)
+
+    fake_vm_pool = mocker.AsyncMock(
+        executions={
+            vm_hash: mocker.AsyncMock(
+                vm_hash=vm_hash,
+                message=instance_message.content,
+                is_running=True,
+            ),
+        },
+    )
+
+    # Disable auth
+    mocker.patch(
+        "aleph.vm.orchestrator.views.authentication.authenticate_jwk",
+        return_value=instance_message.sender,
+    )
+    app = setup_webapp()
+    app["vm_pool"] = fake_vm_pool
+    client: TestClient = await aiohttp_client(app)
+    response = await client.post(
+        f"/control/machine/{vm_hash}/stop",
+    )
+    assert response.status == 200, await response.text()
+    assert fake_vm_pool.stop_vm.call_count == 1
+
+
 @pytest.mark.asyncio
 async def test_operator_confidential_initialize_not_confidential(aiohttp_client, mocker):
     """Test that the confidential initialize endpoint rejects if the VM is not confidential"""