API: Improved vm_status and added some tests. Skip vm tests in VM
This commit is contained in:
parent
8d38ba93ff
commit
f224deace9
1
.gitignore
vendored
1
.gitignore
vendored
|
@ -3,6 +3,7 @@ result*
|
||||||
/pkgs/clan-cli/clan_cli/nixpkgs
|
/pkgs/clan-cli/clan_cli/nixpkgs
|
||||||
/pkgs/clan-cli/clan_cli/webui/assets
|
/pkgs/clan-cli/clan_cli/webui/assets
|
||||||
/machines
|
/machines
|
||||||
|
nixos.qcow2
|
||||||
|
|
||||||
# python
|
# python
|
||||||
__pycache__
|
__pycache__
|
||||||
|
|
|
@ -115,13 +115,15 @@ command output:
|
||||||
|
|
||||||
|
|
||||||
@router.get("/api/vms/{uuid}/status")
|
@router.get("/api/vms/{uuid}/status")
|
||||||
async def get_status(uuid: UUID) -> VmStatusResponse:
|
async def vm_status(uuid: UUID) -> VmStatusResponse:
|
||||||
task = get_task(uuid)
|
task = get_task(uuid)
|
||||||
return VmStatusResponse(running=not task.finished, status=0)
|
status: list[int | None] = list(map(lambda x: x.returncode, task.procs))
|
||||||
|
log.debug(msg=f"returncodes: {status}. task.finished: {task.finished}")
|
||||||
|
return VmStatusResponse(running=not task.finished, returncode=status)
|
||||||
|
|
||||||
|
|
||||||
@router.get("/api/vms/{uuid}/logs")
|
@router.get("/api/vms/{uuid}/logs")
|
||||||
async def get_logs(uuid: UUID) -> StreamingResponse:
|
async def get_vm_logs(uuid: UUID) -> StreamingResponse:
|
||||||
# Generator function that yields log lines as they are available
|
# Generator function that yields log lines as they are available
|
||||||
def stream_logs() -> Iterator[str]:
|
def stream_logs() -> Iterator[str]:
|
||||||
task = get_task(uuid)
|
task = get_task(uuid)
|
||||||
|
|
|
@ -45,7 +45,7 @@ class VmConfig(BaseModel):
|
||||||
|
|
||||||
|
|
||||||
class VmStatusResponse(BaseModel):
|
class VmStatusResponse(BaseModel):
|
||||||
status: int
|
returncode: list[int | None]
|
||||||
running: bool
|
running: bool
|
||||||
|
|
||||||
|
|
||||||
|
|
|
@ -1,25 +1,58 @@
|
||||||
|
import os
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
from api import TestClient
|
from api import TestClient
|
||||||
from httpx import SyncByteStream
|
from httpx import SyncByteStream
|
||||||
|
|
||||||
# @pytest.mark.impure
|
|
||||||
# def test_inspect(api: TestClient, test_flake_with_core: Path) -> None:
|
def is_running_in_ci() -> bool:
|
||||||
# response = api.post(
|
# Check if pytest is running in GitHub Actions
|
||||||
# "/api/vms/inspect",
|
if os.getenv("GITHUB_ACTIONS") == "true":
|
||||||
# json=dict(flake_url=str(test_flake_with_core), flake_attr="vm1"),
|
print("Running on GitHub Actions")
|
||||||
# )
|
return True
|
||||||
# assert response.status_code == 200, "Failed to inspect vm"
|
|
||||||
# config = response.json()["config"]
|
# Check if pytest is running in Travis CI
|
||||||
# assert config.get("flake_attr") == "vm1"
|
if os.getenv("TRAVIS") == "true":
|
||||||
# assert config.get("cores") == 1
|
print("Running on Travis CI")
|
||||||
# assert config.get("memory_size") == 1024
|
return True
|
||||||
# assert config.get("graphics") is True
|
|
||||||
|
# Check if pytest is running in Circle CI
|
||||||
|
if os.getenv("CIRCLECI") == "true":
|
||||||
|
print("Running on Circle CI")
|
||||||
|
return True
|
||||||
|
return False
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.impure
|
||||||
|
def test_inspect(api: TestClient, test_flake_with_core: Path) -> None:
|
||||||
|
response = api.post(
|
||||||
|
"/api/vms/inspect",
|
||||||
|
json=dict(flake_url=str(test_flake_with_core), flake_attr="vm1"),
|
||||||
|
)
|
||||||
|
assert response.status_code == 200, "Failed to inspect vm"
|
||||||
|
config = response.json()["config"]
|
||||||
|
assert config.get("flake_attr") == "vm1"
|
||||||
|
assert config.get("cores") == 1
|
||||||
|
assert config.get("memory_size") == 1024
|
||||||
|
assert config.get("graphics") is True
|
||||||
|
|
||||||
|
|
||||||
|
def test_incorrect_uuid(api: TestClient) -> None:
|
||||||
|
uuid_endpoints = [
|
||||||
|
"/api/vms/{}/status",
|
||||||
|
"/api/vms/{}/logs",
|
||||||
|
]
|
||||||
|
|
||||||
|
for endpoint in uuid_endpoints:
|
||||||
|
response = api.get(endpoint.format("1234"))
|
||||||
|
assert response.status_code == 422, "Failed to get vm status"
|
||||||
|
|
||||||
|
|
||||||
@pytest.mark.impure
|
@pytest.mark.impure
|
||||||
def test_create(api: TestClient, test_flake_with_core: Path) -> None:
|
def test_create(api: TestClient, test_flake_with_core: Path) -> None:
|
||||||
|
if is_running_in_ci():
|
||||||
|
pytest.skip("Skipping test in CI. As it requires KVM")
|
||||||
print(f"flake_url: {test_flake_with_core} ")
|
print(f"flake_url: {test_flake_with_core} ")
|
||||||
response = api.post(
|
response = api.post(
|
||||||
"/api/vms/create",
|
"/api/vms/create",
|
||||||
|
@ -57,3 +90,10 @@ def test_create(api: TestClient, test_flake_with_core: Path) -> None:
|
||||||
print(line.decode("utf-8"), end="")
|
print(line.decode("utf-8"), end="")
|
||||||
print("=========END LOGS==========")
|
print("=========END LOGS==========")
|
||||||
assert response.status_code == 200, "Failed to get vm logs"
|
assert response.status_code == 200, "Failed to get vm logs"
|
||||||
|
|
||||||
|
response = api.get(f"/api/vms/{uuid}/status")
|
||||||
|
assert response.status_code == 200, "Failed to get vm status"
|
||||||
|
returncodes = response.json()["returncode"]
|
||||||
|
assert response.json()["running"] is False, "VM is still running. Should be stopped"
|
||||||
|
for exit_code in returncodes:
|
||||||
|
assert exit_code == 0, "One VM failed with exit code != 0"
|
||||||
|
|
Loading…
Reference in New Issue
Block a user